From 53d1571d3a69e985e36e49a7d15d5437550191ac Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 4 Jul 2026 10:56:08 +0100 Subject: [PATCH 01/15] refactor(server): remove ownership dance from handleReconnect (P3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit handleReconnect no longer does reattach/rekey/zombie cleanup — that's all handled by handleSendV2 on the first user message. Reconnect now only does: watch + cursor replay + suspend resume + boot context. Co-Authored-By: Claude Opus 4.6 --- server/__tests__/ws-handler-v2.test.ts | 184 ++----------------------- server/ws-handler-v2.ts | 53 ++----- 2 files changed, 23 insertions(+), 214 deletions(-) diff --git a/server/__tests__/ws-handler-v2.test.ts b/server/__tests__/ws-handler-v2.test.ts index bad44d60..fdbaac41 100644 --- a/server/__tests__/ws-handler-v2.test.ts +++ b/server/__tests__/ws-handler-v2.test.ts @@ -242,36 +242,14 @@ describe('handleReconnect', () => { ]); }); - it('reattaches detached session on reconnect', () => { - (reattachChat as ReturnType).mockClear(); - - const sessionReg = mockSessionRegistry(); - sessionReg.findBySessionId.mockReturnValue({ clientId: 'c1:sess-1' }); - sessionReg.isActive.mockReturnValue(true); - sessionReg.isAttached.mockReturnValue(false); - - const ctx = createContext({ - sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], - }); - const transport = mockTransport(); - ctx.connRegistry.register('c1', transport); - - handleReconnect( - 'c1', - { type: 'reconnect', sessions: [{ sessionId: 'sess-1', lastSeq: 0 }] }, - ctx, - ); - - expect(reattachChat).toHaveBeenCalledWith('c1:sess-1', transport); - }); - - it('does not reattach if session is already attached', () => { + it('does not reattach or rekey on reconnect (deferred to handleSendV2)', () => { (reattachChat as ReturnType).mockClear(); + (rekeyChat as ReturnType).mockClear(); const sessionReg = mockSessionRegistry(); sessionReg.findBySessionId.mockReturnValue({ clientId: 'c1:sess-1' }); sessionReg.isActive.mockReturnValue(true); - sessionReg.isAttached.mockReturnValue(true); // already attached + sessionReg.isAttached.mockReturnValue(false); // detached — but reconnect should NOT reattach const ctx = createContext({ sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], @@ -286,6 +264,7 @@ describe('handleReconnect', () => { ); expect(reattachChat).not.toHaveBeenCalled(); + expect(rekeyChat).not.toHaveBeenCalled(); }); it('resets cursor to client lastSeq immediately after watch (before replay)', () => { @@ -1086,17 +1065,14 @@ describe('handleReconnect reconnected summary (P1)', () => { expect(summary.sessions[0]).toHaveProperty('replayed'); }); - it('removes stale session from registry when store state is ENDED (zombie)', () => { + it('does not remove stale sessions on reconnect (deferred to handleSendV2)', () => { const sessionReg = mockSessionRegistry(); sessionReg.findBySessionId.mockReturnValue({ clientId: 'driver-1' }); sessionReg.isActive.mockReturnValue(true); - sessionReg.isAttached.mockReturnValue(false); const eventStore = mockEventStore(); eventStore.getSessionState.mockReturnValue('ENDED'); - (reattachChat as ReturnType).mockClear(); - const ctx = createContext({ sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], eventStore: eventStore as unknown as V2HandlerContext['eventStore'], @@ -1110,8 +1086,8 @@ describe('handleReconnect reconnected summary (P1)', () => { ctx, ); - expect(sessionReg.remove).toHaveBeenCalledWith('driver-1'); - expect(reattachChat).not.toHaveBeenCalled(); + // Reconnect no longer does zombie cleanup — handleSendV2 handles it + expect(sessionReg.remove).not.toHaveBeenCalled(); }); it('replays multiple events in sequence order', () => { @@ -2150,81 +2126,9 @@ describe('handleInterruptV2 state-based routing', () => { }); }); -// ─── handleReconnect — ownership guard ────────────────────────────────────── - -describe('handleReconnect ownership guard', () => { - it('does not reattach session when original owner connection is still active', () => { - (reattachChat as ReturnType).mockClear(); - - const sessionReg = mockSessionRegistry(); - sessionReg.findBySessionId.mockReturnValue({ clientId: 'other-conn:sess-1' }); - sessionReg.isActive.mockReturnValue(true); - sessionReg.isAttached.mockReturnValue(false); // detached - - const ctx = createContext({ - sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], - }); - const transport = mockTransport(); - ctx.connRegistry.register('c1', transport); - // Register the original owner so it's still "alive" - ctx.connRegistry.register('other-conn', mockTransport()); - - handleReconnect( - 'c1', - { type: 'reconnect', sessions: [{ sessionId: 'sess-1', lastSeq: 0 }] }, - ctx, - ); - - expect(reattachChat).not.toHaveBeenCalled(); - }); - - it('reattaches detached session when original owner connection is gone', () => { - (reattachChat as ReturnType).mockClear(); - - const sessionReg = mockSessionRegistry(); - sessionReg.findBySessionId.mockReturnValue({ clientId: 'other-conn:sess-1' }); - sessionReg.isActive.mockReturnValue(true); - sessionReg.isAttached.mockReturnValue(false); // detached - - const ctx = createContext({ - sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], - }); - const transport = mockTransport(); - ctx.connRegistry.register('c1', transport); - // other-conn is NOT registered — it disconnected - - handleReconnect( - 'c1', - { type: 'reconnect', sessions: [{ sessionId: 'sess-1', lastSeq: 0 }] }, - ctx, - ); - - expect(reattachChat).toHaveBeenCalledWith('other-conn:sess-1', transport); - }); - - it('reattaches session driven by the same connection', () => { - (reattachChat as ReturnType).mockClear(); - - const sessionReg = mockSessionRegistry(); - sessionReg.findBySessionId.mockReturnValue({ clientId: 'c1:sess-1' }); - sessionReg.isActive.mockReturnValue(true); - sessionReg.isAttached.mockReturnValue(false); - - const ctx = createContext({ - sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], - }); - const transport = mockTransport(); - ctx.connRegistry.register('c1', transport); - - handleReconnect( - 'c1', - { type: 'reconnect', sessions: [{ sessionId: 'sess-1', lastSeq: 0 }] }, - ctx, - ); - - expect(reattachChat).toHaveBeenCalledWith('c1:sess-1', transport); - }); -}); +// ─── handleReconnect — no ownership dance (P3) ────────────────────────────── +// Ownership (reattach/rekey/zombie) is handled by handleSendV2 on first message. +// These tests verify reconnect does NOT attempt ownership operations. // ─── handleInterruptV2 — images and contextBlocks forwarding ─────────────── @@ -2630,60 +2534,8 @@ describe('handleInterruptV2 connection ownership', () => { }); }); -// ─── rekey after reattach — ownership transfer ──────────────────────────────── - -describe('handleReconnect rekey after reattach', () => { - it('rekeys session to new connection after reattach so subsequent sends pass ownership', () => { - (reattachChat as ReturnType).mockClear(); - (rekeyChat as ReturnType).mockClear(); - - const sessionReg = mockSessionRegistry(); - sessionReg.findBySessionId.mockReturnValue({ clientId: 'old-conn:sess-1' }); - sessionReg.isActive.mockReturnValue(true); - sessionReg.isAttached.mockReturnValue(false); // detached - - const ctx = createContext({ - sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], - }); - const transport = mockTransport(); - ctx.connRegistry.register('new-conn', transport); - // old-conn is NOT registered — it disconnected - - handleReconnect( - 'new-conn', - { type: 'reconnect', sessions: [{ sessionId: 'sess-1', lastSeq: 0 }] }, - ctx, - ); - - expect(reattachChat).toHaveBeenCalledWith('old-conn:sess-1', transport); - expect(rekeyChat).toHaveBeenCalledWith('old-conn:sess-1', 'new-conn:sess-1'); - }); - - it('skips rekey when connectionId already matches (same connection reconnects)', () => { - (reattachChat as ReturnType).mockClear(); - (rekeyChat as ReturnType).mockClear(); - - const sessionReg = mockSessionRegistry(); - sessionReg.findBySessionId.mockReturnValue({ clientId: 'c1:sess-1' }); - sessionReg.isActive.mockReturnValue(true); - sessionReg.isAttached.mockReturnValue(false); - - const ctx = createContext({ - sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], - }); - const transport = mockTransport(); - ctx.connRegistry.register('c1', transport); - - handleReconnect( - 'c1', - { type: 'reconnect', sessions: [{ sessionId: 'sess-1', lastSeq: 0 }] }, - ctx, - ); - - expect(reattachChat).toHaveBeenCalled(); - expect(rekeyChat).not.toHaveBeenCalled(); - }); -}); +// rekey after reattach tests removed — reconnect no longer does ownership transfer (P3). +// handleSendV2 rekey tests (below) still cover the rekey-on-send path. describe('handleSendV2 rekey after detached reattach', () => { it('rekeys and uses new clientId for sendToChat when taking over detached session', () => { @@ -2951,31 +2803,24 @@ describe('handleSessionSuspend', () => { // ─── handleReconnect — suspend resume ─────────────────────────────────────── describe('handleReconnect suspend resume', () => { - it('replays buffered events for suspended sessions', () => { + it('clears suspend state and sends session_resumed', () => { const sessionReg = mockSessionRegistry(); sessionReg.findBySessionId.mockReturnValue({ clientId: 'conn-1:sess-1', session: { sessionId: 'sess-1' }, }); sessionReg.isActive.mockReturnValue(true); - sessionReg.isAttached.mockReturnValue(false); sessionReg.isSuspended.mockReturnValue(true); sessionReg.resume.mockReturnValue([ { v: 2, type: 'block_delta', delta: 'buffered-text', sessionId: 'sess-1' }, ]); - const eventStore = mockEventStore(); - eventStore.getSession.mockReturnValue({ isActive: true }); - const ctx = createContext({ sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], - eventStore: eventStore as unknown as V2HandlerContext['eventStore'], }); const transport = mockTransport(); ctx.connRegistry.register('conn-1', transport); - (reattachChat as ReturnType).mockReturnValue(true); - handleReconnect( 'conn-1', { type: 'reconnect', sessions: [{ sessionId: 'sess-1', lastSeq: 0 }] }, @@ -2984,12 +2829,13 @@ describe('handleReconnect suspend resume', () => { expect(sessionReg.resume).toHaveBeenCalledWith('conn-1:sess-1'); // Buffered events should NOT be replayed — EventStore replay covers them. - // resume() is called only to clear suspend state. expect( transport.sent.some((m) => m.type === 'block_delta' && m.delta === 'buffered-text'), ).toBe(false); // Should have sent session_resumed with total replayed count expect(transport.sent.some((m) => m.type === 'session_resumed' && m.replayed === 1)).toBe(true); + // No reattach — ownership deferred to handleSendV2 + expect(reattachChat).not.toHaveBeenCalled(); }); }); diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index 32d9d5a1..e59f1685 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -247,58 +247,21 @@ export function handleReconnect( } as Record); } - // Reset cursor to last replayed seq — prevents duplicate delivery from - // periodic sync. If no events replayed, cursor stays at client's lastSeq. + // Reset cursor to last replayed seq so broadcast() doesn't re-deliver. + // If no events replayed, cursor stays at client's lastSeq. const newCursor = events.length > 0 ? events[events.length - 1].seq : entry.lastSeq; ctx.connRegistry.resetCursor(connectionId, entry.sessionId, newCursor); - // Cross-reference with the durable EventStore: state=ENDED in the store - // is ground truth that the query loop has finished (P1: use state, not is_active). - const found = ctx.sessionRegistry.findBySessionId(entry.sessionId); - const storeState = ctx.eventStore.getSessionState(entry.sessionId); - let running = found ? ctx.sessionRegistry.isActive(found.clientId) : false; - if (running && (storeState === 'ENDED' || storeState === 'CLOSING')) { - running = false; - log.info('removing stale session from registry (state-based)', { - connectionId, - sessionId: entry.sessionId, - clientId: found!.clientId, - storeState, - }); - ctx.sessionRegistry.remove(found!.clientId); - } - if (found && running && !ctx.sessionRegistry.isAttached(found.clientId)) { - const ownerConnection = getOwnerConnection(found.clientId); - const ownerGone = !ctx.connRegistry.get(ownerConnection); - const isOwner = ownerConnection === connectionId; - if (isOwner || ownerGone) { - const conn = ctx.connRegistry.get(connectionId); - if (conn) { - reattachChat(found.clientId, conn.transport); - const newClientId = `${connectionId}:${entry.sessionId}`; - if (found.clientId !== newClientId) { - rekeyChat(found.clientId, newClientId); - log.info('rekeyed session to new connection', { - connectionId, - sessionId: entry.sessionId, - oldClientId: found.clientId, - newClientId, - }); - } - log.info('reattached detached session on reconnect', { - connectionId, - sessionId: entry.sessionId, - clientId: newClientId, - ownerGone, - }); - } - } - } + // Ownership dance (reattach/rekey/zombie cleanup) is NOT done here. + // handleSendV2 handles all of that on the first user message — reconnect + // only needs to restore the event stream and boot context. // If the session was suspended, clear suspend state. Don't replay // buffered events — they were already replayed from EventStore above // (sendOrBuffer appends to both stores, so EventStore covers the // suspend period). resume() just clears the suspend flag + buffer. + const found = ctx.sessionRegistry.findBySessionId(entry.sessionId); + const running = found ? ctx.sessionRegistry.isActive(found.clientId) : false; let suspendReplayed = 0; if (found && running && ctx.sessionRegistry.isSuspended(found.clientId)) { const buffered = ctx.sessionRegistry.resume(found.clientId); @@ -950,7 +913,7 @@ export async function dispatchV2Message( // Already handled at routing layer, ignore duplicate break; case 'reconnect': - handleReconnect(connectionId, msg, ctx); + // Reconnect is handled via REST POST, not WS. Ignore if received over WS. break; case 'watch': handleWatch(connectionId, msg, ctx); From 9c1cbb6982bd109a1e71a34c655c05b74ae326d5 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 4 Jul 2026 12:04:39 +0100 Subject: [PATCH 02/15] refactor(harness): remove periodic sync from ConnectionRegistry (P3) Periodic sync (5s timer retrying missed events) is redundant now that reconnect replays via EventStore cursor on welcome. Removes setEventStore, startPeriodicSync, stopPeriodicSync, EventStoreAdapter interface, and all associated tests and wiring. Co-Authored-By: Claude Opus 4.6 --- .../__tests__/connection-registry.test.ts | 236 +----------------- packages/harness/src/connection-registry.ts | 128 +--------- packages/harness/src/index.ts | 2 +- server/index.ts | 14 -- server/ws-handler-v2.ts | 4 +- 5 files changed, 11 insertions(+), 373 deletions(-) diff --git a/packages/harness/__tests__/connection-registry.test.ts b/packages/harness/__tests__/connection-registry.test.ts index 6c4cd851..a5bc2f2f 100644 --- a/packages/harness/__tests__/connection-registry.test.ts +++ b/packages/harness/__tests__/connection-registry.test.ts @@ -1,5 +1,5 @@ -import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest'; -import { ConnectionRegistry, type EventStoreAdapter } from '../src/connection-registry.js'; +import { describe, it, expect, beforeEach, vi } from 'vitest'; +import { ConnectionRegistry } from '../src/connection-registry.js'; import type { SessionTransport } from '../src/session-transport.js'; function mockTransport(open = true): SessionTransport { @@ -9,17 +9,6 @@ function mockTransport(open = true): SessionTransport { }; } -function mockEventStore( - events: Array<{ seq: number; payload: Record }> = [], -): EventStoreAdapter { - return { - getEventsAfter: vi.fn((sessionId: string, afterSeq: number, limit?: number) => { - const filtered = events.filter((e) => e.seq > afterSeq); - return limit ? filtered.slice(0, limit) : filtered; - }), - }; -} - describe('ConnectionRegistry', () => { let registry: ConnectionRegistry; @@ -297,230 +286,15 @@ describe('ConnectionRegistry', () => { }); }); - describe('periodic sync', () => { - afterEach(() => { - vi.useRealTimers(); - }); - - it('retries missed events from EventStore', async () => { - vi.useFakeTimers(); - const t = mockTransport(true); - const store = mockEventStore([ - { seq: 5, payload: { type: 'msg1', data: 'a' } }, - { seq: 10, payload: { type: 'msg2', data: 'b' } }, - { seq: 15, payload: { type: 'msg3', data: 'c' } }, - ]); - - registry.setEventStore(store); - registry.register('conn-1', t); - registry.watch('conn-1', 'sess-a'); - - // Simulate cursor at 0 (never delivered anything) - registry.startPeriodicSync(); - - // Advance time to trigger sync - await vi.advanceTimersByTimeAsync(5000); - - // Should fetch events > 0 from store and deliver them - expect(store.getEventsAfter).toHaveBeenCalledWith('sess-a', 0, 50); - expect(t.send).toHaveBeenCalledTimes(3); - expect(t.send).toHaveBeenCalledWith({ type: 'msg1', data: 'a', seq: 5 }); - expect(t.send).toHaveBeenCalledWith({ type: 'msg2', data: 'b', seq: 10 }); - expect(t.send).toHaveBeenCalledWith({ type: 'msg3', data: 'c', seq: 15 }); - - registry.stopPeriodicSync(); - }); - - it('stops retrying on first send failure in a batch', async () => { - vi.useFakeTimers(); - const t = mockTransport(true); - let callCount = 0; - (t.send as ReturnType).mockImplementation(() => { - callCount++; - if (callCount === 2) throw new Error('socket dead'); - }); - - const store = mockEventStore([ - { seq: 5, payload: { type: 'msg1' } }, - { seq: 10, payload: { type: 'msg2' } }, - { seq: 15, payload: { type: 'msg3' } }, - ]); - - registry.setEventStore(store); - registry.register('conn-1', t); - registry.watch('conn-1', 'sess-a'); - registry.startPeriodicSync(); - - await vi.advanceTimersByTimeAsync(5000); - - // Should send msg1 (success), msg2 (fail), then stop - expect(t.send).toHaveBeenCalledTimes(2); - - registry.stopPeriodicSync(); - }); - - it('skips connections with closed transports', async () => { - vi.useFakeTimers(); - const tClosed = mockTransport(false); - const store = mockEventStore([{ seq: 5, payload: { type: 'test' } }]); - - registry.setEventStore(store); - registry.register('conn-closed', tClosed); - registry.watch('conn-closed', 'sess-a'); - registry.startPeriodicSync(); - - await vi.advanceTimersByTimeAsync(5000); - - expect(tClosed.send).not.toHaveBeenCalled(); - - registry.stopPeriodicSync(); - }); - - it('respects SYNC_BATCH_LIMIT to avoid overwhelming slow clients', async () => { - vi.useFakeTimers(); - const t = mockTransport(true); - const manyEvents = Array.from({ length: 100 }, (_, i) => ({ - seq: i + 1, - payload: { type: 'msg', i }, - })); - const store = mockEventStore(manyEvents); - - registry.setEventStore(store); - registry.register('conn-1', t); - registry.watch('conn-1', 'sess-a'); - registry.startPeriodicSync(); - - await vi.advanceTimersByTimeAsync(5000); - - // Should fetch with limit=50 - expect(store.getEventsAfter).toHaveBeenCalledWith('sess-a', 0, 50); - expect(t.send).toHaveBeenCalledTimes(50); - - registry.stopPeriodicSync(); - }); - - it('handles EventStore fetch errors gracefully', async () => { - vi.useFakeTimers(); - const t = mockTransport(true); - const store: EventStoreAdapter = { - getEventsAfter: vi.fn(() => { - throw new Error('database locked'); - }), - }; - - registry.setEventStore(store); - registry.register('conn-1', t); - registry.watch('conn-1', 'sess-a'); - registry.startPeriodicSync(); - - // Sync should not throw even when EventStore fetch fails - await expect(vi.advanceTimersByTimeAsync(5000)).resolves.not.toThrow(); - - expect(t.send).not.toHaveBeenCalled(); - - registry.stopPeriodicSync(); - }); - - it('is a no-op when EventStore not set', () => { - const registry2 = new ConnectionRegistry(); - expect(() => registry2.startPeriodicSync()).not.toThrow(); - // No timer started, so no cleanup needed - }); - - it('warns when starting sync twice', () => { - const registry2 = new ConnectionRegistry(); - const store = mockEventStore(); - registry2.setEventStore(store); - registry2.startPeriodicSync(); - // Second call should warn but not crash - expect(() => registry2.startPeriodicSync()).not.toThrow(); - registry2.stopPeriodicSync(); - }); - - it('stops periodic sync and clears timer', () => { - vi.useFakeTimers(); - const store = mockEventStore(); - registry.setEventStore(store); - registry.startPeriodicSync(); - registry.stopPeriodicSync(); - - // Timer should be cleared — no sync fires - const t = mockTransport(true); - registry.register('conn-1', t); - registry.watch('conn-1', 'sess-a'); - - vi.advanceTimersByTime(5000); - expect(t.send).not.toHaveBeenCalled(); - }); - - it('skips ended sessions when isSessionActive is provided', async () => { - vi.useFakeTimers(); - const t = mockTransport(true); - const store = mockEventStore([ - { seq: 5, payload: { type: 'msg1' } }, - { seq: 10, payload: { type: 'msg2' } }, - ]); - - // Add isSessionActive — sess-ended is inactive, sess-active is active - (store as EventStoreAdapter & { isSessionActive?: (id: string) => boolean }).isSessionActive = - (id: string) => id !== 'sess-ended'; - - registry.setEventStore(store); - registry.register('conn-1', t); - registry.watch('conn-1', 'sess-ended'); - registry.watch('conn-1', 'sess-active'); - registry.startPeriodicSync(); - - await vi.advanceTimersByTimeAsync(5000); - - // Should only fetch events for sess-active, not sess-ended - const calls = (store.getEventsAfter as ReturnType).mock.calls; - const sessionIds = calls.map((c: unknown[]) => c[0]); - expect(sessionIds).toContain('sess-active'); - expect(sessionIds).not.toContain('sess-ended'); - - registry.stopPeriodicSync(); - }); - - it('still syncs all sessions when isSessionActive is not provided', async () => { - vi.useFakeTimers(); - const t = mockTransport(true); - const store = mockEventStore([{ seq: 5, payload: { type: 'msg1' } }]); - - // No isSessionActive — backwards compatible - registry.setEventStore(store); - registry.register('conn-1', t); - registry.watch('conn-1', 'sess-a'); - registry.watch('conn-1', 'sess-b'); - registry.startPeriodicSync(); - - await vi.advanceTimersByTimeAsync(5000); - - // Should fetch events for both sessions (no filtering) - const calls = (store.getEventsAfter as ReturnType).mock.calls; - const sessionIds = calls.map((c: unknown[]) => c[0]); - expect(sessionIds).toContain('sess-a'); - expect(sessionIds).toContain('sess-b'); - - registry.stopPeriodicSync(); - }); - }); - describe('dispose', () => { - it('stops periodic sync and clears all state', () => { - vi.useFakeTimers(); - const store = mockEventStore(); - registry.setEventStore(store); + it('clears all connections and cursors', () => { registry.register('conn-1', mockTransport()); - registry.startPeriodicSync(); + registry.watch('conn-1', 'sess-a'); + registry.resetCursor('conn-1', 'sess-a', 10); registry.dispose(); expect(registry.get('conn-1')).toBeUndefined(); - - // Timer stopped — no sync fires - vi.advanceTimersByTime(5000); - expect(store.getEventsAfter).not.toHaveBeenCalled(); }); }); }); diff --git a/packages/harness/src/connection-registry.ts b/packages/harness/src/connection-registry.ts index 33470ab6..94fd319d 100644 --- a/packages/harness/src/connection-registry.ts +++ b/packages/harness/src/connection-registry.ts @@ -12,7 +12,7 @@ * Delivery Guarantee: * - Tracks per-connection per-session cursors (last delivered seq) * - broadcast() updates cursor on successful send - * - Periodic sync retries events beyond cursor (handles WS races, iOS kills) + * - Reconnect replays missed events via EventStore cursor on welcome * - Reconnect resets cursor to client's lastSeq to prevent duplicate replay */ @@ -28,32 +28,10 @@ export interface Connection { activeSession: string | null; } -/** Event store interface for periodic sync — injected to avoid circular deps */ -export interface EventStoreAdapter { - getEventsAfter( - sessionId: string, - afterSeq: number, - limit?: number, - ): Array<{ - seq: number; - payload: Record; - }>; - /** Optional: check if a session is still active. When provided, periodic sync - * skips ended sessions to avoid unnecessary EventStore queries. */ - isSessionActive?(sessionId: string): boolean; -} - -// Periodic sync fires every 5s to retry missed events -const SYNC_INTERVAL_MS = 5000; -// Limit events per sync round per connection to avoid overwhelming slow clients -const SYNC_BATCH_LIMIT = 50; - export class ConnectionRegistry { private connections = new Map(); // Per-connection per-session cursors: last successfully delivered seq private cursors = new Map>(); - private syncTimer: ReturnType | null = null; - private eventStore: EventStoreAdapter | null = null; register(connectionId: string, transport: SessionTransport): void { this.connections.set(connectionId, { @@ -76,14 +54,6 @@ export class ConnectionRegistry { this.cursors.delete(connectionId); } - /** - * Set the EventStore adapter for periodic sync. - * Must be called before starting periodic sync. - */ - setEventStore(eventStore: EventStoreAdapter): void { - this.eventStore = eventStore; - } - watch(connectionId: string, sessionId: string): void { const conn = this.connections.get(connectionId); if (!conn) return; @@ -160,7 +130,7 @@ export class ConnectionRegistry { } } catch { log.warn('broadcast send failed', { connectionId, sessionId, seq }); - // Cursor not updated → periodic sync will retry + // Cursor not updated — reconnect replay will cover the gap } } } @@ -195,101 +165,9 @@ export class ConnectionRegistry { } /** - * Start periodic sync — retries missed events for all connections. - * Runs every SYNC_INTERVAL_MS, bounded by SYNC_BATCH_LIMIT per connection. - * Call this once during server startup after setEventStore(). - */ - startPeriodicSync(): void { - if (this.syncTimer) { - log.warn('periodic sync already running'); - return; - } - if (!this.eventStore) { - log.error('cannot start periodic sync: EventStore not set'); - return; - } - - log.info('starting periodic sync', { intervalMs: SYNC_INTERVAL_MS }); - - this.syncTimer = setInterval(() => { - if (!this.eventStore) return; - - for (const [connectionId, conn] of this.connections.entries()) { - if (!conn.transport.isOpen()) continue; - - const connCursors = this.cursors.get(connectionId); - if (!connCursors) continue; - - for (const sessionId of conn.watchedSessions) { - // Skip ended sessions to avoid unnecessary EventStore queries - if (this.eventStore.isSessionActive && !this.eventStore.isSessionActive(sessionId)) { - continue; - } - - const cursor = connCursors.get(sessionId) ?? 0; - - // Fetch missed events from EventStore - let missedEvents: Array<{ seq: number; payload: Record }>; - try { - missedEvents = this.eventStore.getEventsAfter(sessionId, cursor, SYNC_BATCH_LIMIT); - } catch (err) { - log.warn('periodic sync: EventStore fetch failed', { - connectionId, - sessionId, - error: err instanceof Error ? err.message : String(err), - }); - continue; - } - - if (missedEvents.length === 0) continue; - - log.info('periodic sync: retrying missed events', { - connectionId, - sessionId, - cursor, - missedCount: missedEvents.length, - }); - - // Retry delivery - for (const evt of missedEvents) { - try { - conn.transport.send({ ...evt.payload, seq: evt.seq }); - // Update cursor on success - const current = connCursors.get(sessionId) ?? 0; - if (evt.seq > current) { - connCursors.set(sessionId, evt.seq); - } - } catch { - // Still failing — stop here, retry next sync round - log.warn('periodic sync: retry failed, stopping batch', { - connectionId, - sessionId, - failedSeq: evt.seq, - }); - break; - } - } - } - } - }, SYNC_INTERVAL_MS); - } - - /** - * Stop periodic sync and clean up timer. Call during graceful shutdown. - */ - stopPeriodicSync(): void { - if (this.syncTimer) { - clearInterval(this.syncTimer); - this.syncTimer = null; - log.info('periodic sync stopped'); - } - } - - /** - * Dispose: stop sync, clear all state. Used for graceful shutdown. + * Dispose: clear all state. Used for graceful shutdown. */ dispose(): void { - this.stopPeriodicSync(); this.connections.clear(); this.cursors.clear(); } diff --git a/packages/harness/src/index.ts b/packages/harness/src/index.ts index ba7eb243..638e35b3 100644 --- a/packages/harness/src/index.ts +++ b/packages/harness/src/index.ts @@ -13,7 +13,7 @@ export type { // Connection registry (v2 single-WS protocol) export { ConnectionRegistry } from './connection-registry.js'; -export type { Connection, EventStoreAdapter } from './connection-registry.js'; +export type { Connection } from './connection-registry.js'; // SSE registry (broadcast events) export { SseRegistry } from './sse-registry.js'; diff --git a/server/index.ts b/server/index.ts index 06b7ae64..fa6ae76b 100644 --- a/server/index.ts +++ b/server/index.ts @@ -100,17 +100,6 @@ const nativeCommands = new NativeCommandRegistry(); const connRegistry = new ConnectionRegistry(); setConnectionRegistry(connRegistry); -// Wire up EventStore for periodic sync (enables delivery guarantee). -// Provide isSessionActive so periodic sync skips ended sessions (P1: use state, not is_active). -connRegistry.setEventStore({ - getEventsAfter: (sessionId, afterSeq, limit) => - eventStore.getEventsAfter(sessionId, afterSeq, limit), - isSessionActive: (sessionId) => { - const state = eventStore.getSessionState(sessionId); - return state !== null && state !== 'ENDED' && state !== 'CLOSING'; - }, -}); - // Resolve cert paths relative to the project root (where package.json lives) const __filename = fileURLToPath(import.meta.url); const PROJECT_ROOT = join(__filename, '..', '..'); @@ -1025,9 +1014,6 @@ checkPort(PORT).then((inUse) => { const protocol = USE_TLS ? 'https' : 'http'; log.info(`Chat Agent running on ${protocol}://localhost:${PORT}${USE_TLS ? ' (TLS)' : ''}`); - // Start periodic sync for connection-level delivery guarantee - connRegistry.startPeriodicSync(); - // Recover sessions left in incomplete states after crash/restart (Transport SSOT P0). // Must run before reconcileSessionsBackground() so reconciliation sees ENDED states. // recoverStaleSessions() logs internally — no need to log here. diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index e59f1685..eb32690e 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -235,8 +235,8 @@ export function handleReconnect( for (const entry of msg.sessions) { ctx.connRegistry.watch(connectionId, entry.sessionId); - // Set cursor to client's lastSeq BEFORE replay, so periodic sync - // sees a reasonable cursor during replay instead of 0. + // Set cursor to client's lastSeq BEFORE replay so broadcast() + // doesn't re-deliver events that are about to be replayed. ctx.connRegistry.resetCursor(connectionId, entry.sessionId, entry.lastSeq); const events = ctx.eventStore.getEventsAfter(entry.sessionId, entry.lastSeq); From 6c91ae94418a725f8db9b0ff0b90a2915b16a25a Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 4 Jul 2026 12:08:07 +0100 Subject: [PATCH 03/15] refactor(client): fire-and-forget reconnect POST in SseConnection (P3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reconnect POST no longer defers _connected — the client marks connected immediately on welcome. handleSendV2 handles ownership on first message, and replayed events arrive via SSE regardless of POST outcome. Removes doReconnectPost, scheduleReconnect, reconnectTimer, and reconnectDelayMs. Replaces 12 deferred/stale/failure tests with 4 fire-and-forget tests. Co-Authored-By: Claude Opus 4.6 --- .../src/__tests__/sse-connection.test.ts | 376 +----------------- packages/client/src/sse-connection.ts | 100 +---- 2 files changed, 26 insertions(+), 450 deletions(-) diff --git a/packages/client/src/__tests__/sse-connection.test.ts b/packages/client/src/__tests__/sse-connection.test.ts index e00e62ae..88111d8a 100644 --- a/packages/client/src/__tests__/sse-connection.test.ts +++ b/packages/client/src/__tests__/sse-connection.test.ts @@ -290,7 +290,7 @@ describe('SseConnection', () => { expect(conn.isConnected()).toBe(false); }); - it('sends reconnect POST on welcome when has tracked sessions', () => { + it('sends reconnect POST fire-and-forget on reconnect welcome', () => { const mockFetch = vi.fn().mockResolvedValue({ ok: true }); const conn = new SseConnection(createConfig({ fetch: mockFetch })); conn.connect(); @@ -318,54 +318,8 @@ describe('SseConnection', () => { ); }); - it('defers _connected until reconnect POST completes', async () => { - let resolveReconnect!: (v: { ok: true }) => void; - const mockFetch = vi.fn().mockImplementation((_url: string) => { - if (_url.includes('/reconnect')) { - return new Promise((resolve) => { - resolveReconnect = resolve; - }); - } - return Promise.resolve({ ok: true }); - }); - const conn = new SseConnection(createConfig({ fetch: mockFetch })); - const listener = vi.fn(); - conn.onMessage(listener); - conn.connect(); - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); - conn.trackSeq('sess-1', 10); - - // Force reconnect - conn.checkAndReconnect(true); - mockFetch.mockClear(); - listener.mockClear(); - - // New welcome — reconnect POST fires but doesn't resolve yet - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - - // _connected should still be false while POST is in-flight - expect(conn.isConnected()).toBe(false); - expect(listener).not.toHaveBeenCalledWith({ type: '_open' }); - - // Resolve the reconnect POST - resolveReconnect({ ok: true }); - await vi.runAllTimersAsync(); - - // Now _connected should be true and _open emitted - expect(conn.isConnected()).toBe(true); - expect(listener).toHaveBeenCalledWith({ type: '_open' }); - }); - - it('bails out if disconnect() called during in-flight reconnect POST', async () => { - let resolveReconnect!: (v: { ok: true }) => void; - const mockFetch = vi.fn().mockImplementation((_url: string) => { - if (_url.includes('/reconnect')) { - return new Promise((resolve) => { - resolveReconnect = resolve; - }); - } - return Promise.resolve({ ok: true }); - }); + it('marks connected immediately on reconnect welcome', () => { + const mockFetch = vi.fn().mockResolvedValue({ ok: true }); const conn = new SseConnection(createConfig({ fetch: mockFetch })); const listener = vi.fn(); conn.onMessage(listener); @@ -375,135 +329,16 @@ describe('SseConnection', () => { // Force reconnect conn.checkAndReconnect(true); - - // New welcome — reconnect POST in-flight - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - - // Disconnect while POST is in-flight - conn.disconnect(); - expect(conn.isConnected()).toBe(false); listener.mockClear(); - // Resolve the reconnect POST — staleness guard should bail out - resolveReconnect({ ok: true }); - await vi.runAllTimersAsync(); - - // Must remain disconnected — .finally() must not overwrite - expect(conn.isConnected()).toBe(false); - expect(listener).not.toHaveBeenCalledWith({ type: '_open' }); - }); - - it('ignores stale reconnect POST when a newer welcome arrives', async () => { - const reconnectCalls: Array<(v: { ok: true }) => void> = []; - const mockFetch = vi.fn().mockImplementation((_url: string) => { - if (_url.includes('/reconnect')) { - return new Promise((resolve) => { - reconnectCalls.push(resolve); - }); - } - return Promise.resolve({ ok: true }); - }); - const conn = new SseConnection(createConfig({ fetch: mockFetch })); - const listener = vi.fn(); - conn.onMessage(listener); - conn.connect(); - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); - conn.trackSeq('sess-1', 10); - - // Force reconnect - conn.checkAndReconnect(true); - - // First welcome — reconnect POST #1 in-flight + // New welcome — should be connected immediately (fire-and-forget) lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - const resolveFirst = reconnectCalls[0]; - - // Second welcome arrives (rapid reconnect race) — reconnect POST #2 in-flight - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-ghi' }); - const resolveSecond = reconnectCalls[1]; - - // Resolve the FIRST (stale) reconnect POST - resolveFirst({ ok: true }); - await vi.runAllTimersAsync(); - - // Must NOT set _connected — connectionId has moved on to conn-ghi - expect(conn.isConnected()).toBe(false); - expect(conn.getConnectionId()).toBe('conn-ghi'); - - // Resolve the SECOND (current) reconnect POST - listener.mockClear(); - resolveSecond({ ok: true }); - await vi.runAllTimersAsync(); - // Now _connected should be true expect(conn.isConnected()).toBe(true); expect(listener).toHaveBeenCalledWith({ type: '_open' }); }); - it('flushes pending sends only after reconnect POST completes', async () => { - let resolveReconnect!: (v: { ok: true }) => void; - const postEndpoints: string[] = []; - const mockFetch = vi.fn().mockImplementation((url: string) => { - const endpoint = url.replace('https://localhost:3100/api/chat/', ''); - postEndpoints.push(endpoint); - if (url.includes('/reconnect')) { - return new Promise((resolve) => { - resolveReconnect = resolve; - }); - } - return Promise.resolve({ ok: true }); - }); - const conn = new SseConnection(createConfig({ fetch: mockFetch })); - conn.connect(); - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); - conn.trackSeq('sess-1', 10); - - // Force reconnect — sends are now queued - conn.checkAndReconnect(true); - conn.send({ type: 'send', prompt: 'queued msg', clientMsgId: 'q-1' }); - postEndpoints.length = 0; - - // Welcome — reconnect POST fires, queued send waits - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - - // Only reconnect POST should have fired, not the queued send - expect(postEndpoints).toEqual(['reconnect']); - - // Resolve reconnect — now the queued send should flush - resolveReconnect({ ok: true }); - await vi.runAllTimersAsync(); - - expect(postEndpoints).toEqual(['reconnect', 'send']); - }); - - it('stays disconnected when reconnect POST fails', async () => { - const mockFetch = vi.fn().mockImplementation((url: string) => { - if (url.includes('/reconnect')) { - return Promise.resolve({ ok: false, status: 500 }); - } - return Promise.resolve({ ok: true }); - }); - const conn = new SseConnection(createConfig({ fetch: mockFetch })); - const listener = vi.fn(); - conn.onMessage(listener); - conn.connect(); - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); - conn.trackSeq('sess-1', 10); - - // Force reconnect - conn.checkAndReconnect(true); - conn.send({ type: 'send', prompt: 'should stay queued', clientMsgId: 'q-1' }); - listener.mockClear(); - - // New welcome — reconnect POST will fail - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - await vi.runAllTimersAsync(); - - // Must stay disconnected — server never ran handleReconnect - expect(conn.isConnected()).toBe(false); - expect(listener).not.toHaveBeenCalledWith({ type: '_open' }); - }); - - it('stays disconnected when reconnect POST throws network error', async () => { + it('POST failure does not affect connection state', async () => { const mockFetch = vi.fn().mockImplementation((url: string) => { if (url.includes('/reconnect')) { return Promise.reject(new Error('network error')); @@ -520,101 +355,18 @@ describe('SseConnection', () => { conn.checkAndReconnect(true); listener.mockClear(); + // Welcome — reconnect POST fires (and will fail), but connection is immediate lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - await vi.runAllTimersAsync(); - - expect(conn.isConnected()).toBe(false); - expect(listener).not.toHaveBeenCalledWith({ type: '_open' }); - }); - it('recovers after failed reconnect when EventSource auto-reconnects', async () => { - let callCount = 0; - const mockFetch = vi.fn().mockImplementation((url: string) => { - if (url.includes('/reconnect')) { - callCount++; - // First reconnect fails, second succeeds - if (callCount === 1) return Promise.resolve({ ok: false, status: 500 }); - return Promise.resolve({ ok: true }); - } - return Promise.resolve({ ok: true }); - }); - const conn = new SseConnection(createConfig({ fetch: mockFetch })); - const listener = vi.fn(); - conn.onMessage(listener); - conn.connect(); - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); - conn.trackSeq('sess-1', 10); - - // Force reconnect - conn.checkAndReconnect(true); - listener.mockClear(); - - // First welcome — reconnect POST fails - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - await vi.runAllTimersAsync(); - expect(conn.isConnected()).toBe(false); - - // EventSource auto-reconnect fires a new welcome - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-ghi' }); - await vi.runAllTimersAsync(); - - // Second attempt succeeds + // Connected immediately regardless of POST outcome expect(conn.isConnected()).toBe(true); expect(listener).toHaveBeenCalledWith({ type: '_open' }); }); - it('dispatches SSE events to listener while reconnect POST is in-flight', async () => { - let resolveReconnect!: (v: { ok: boolean }) => void; - const mockFetch = vi.fn().mockImplementation((url: string) => { - if (url.includes('/reconnect')) { - return new Promise((resolve) => { - resolveReconnect = resolve; - }); - } - return Promise.resolve({ ok: true }); - }); - const conn = new SseConnection(createConfig({ fetch: mockFetch })); - const listener = vi.fn(); - conn.onMessage(listener); - conn.connect(); - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); - conn.trackSeq('sess-1', 10); - - conn.checkAndReconnect(true); - listener.mockClear(); - - // Welcome — reconnect POST in-flight, _connected = false - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - expect(conn.isConnected()).toBe(false); - - // Server replays events via SSE while reconnect POST is processing. - // onmessage is independent of _connected — these must still dispatch. - lastES()._emit('block_delta', { - type: 'block_delta', - sessionId: 'sess-1', - seq: 11, - delta: 'replayed', - }); - - expect(listener).toHaveBeenCalledWith( - expect.objectContaining({ type: 'block_delta', delta: 'replayed' }), - ); - - resolveReconnect({ ok: true }); - await vi.runAllTimersAsync(); - }); - - it('queued sends survive POST failure and flush on successful retry', async () => { - let callCount = 0; + it('flushes pending sends immediately on reconnect welcome', () => { const postEndpoints: string[] = []; const mockFetch = vi.fn().mockImplementation((url: string) => { const endpoint = url.replace('https://localhost:3100/api/chat/', ''); - if (url.includes('/reconnect')) { - callCount++; - if (callCount === 1) return Promise.resolve({ ok: false, status: 500 }); - postEndpoints.push(endpoint); - return Promise.resolve({ ok: true }); - } postEndpoints.push(endpoint); return Promise.resolve({ ok: true }); }); @@ -623,84 +375,17 @@ describe('SseConnection', () => { lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); conn.trackSeq('sess-1', 10); - // Force reconnect and queue a send + // Force reconnect — sends are now queued conn.checkAndReconnect(true); - conn.send({ type: 'send', prompt: 'must survive', clientMsgId: 'q-1' }); + conn.send({ type: 'send', prompt: 'queued msg', clientMsgId: 'q-1' }); postEndpoints.length = 0; - // First welcome — reconnect fails, send stays queued + // Welcome — reconnect POST + queued send both fire immediately lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - await vi.runAllTimersAsync(); - expect(conn.isConnected()).toBe(false); - expect(postEndpoints).toEqual([]); - // Second welcome — reconnect succeeds, queued send flushes - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-ghi' }); - await vi.runAllTimersAsync(); - expect(conn.isConnected()).toBe(true); expect(postEndpoints).toEqual(['reconnect', 'send']); }); - it('schedules delayed reconnect when reconnect POST fails', async () => { - const mockFetch = vi.fn().mockImplementation((url: string) => { - if (url.includes('/reconnect')) { - return Promise.resolve({ ok: false, status: 500 }); - } - return Promise.resolve({ ok: true }); - }); - const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); - const conn = new SseConnection(createConfig({ fetch: mockFetch })); - conn.connect(); - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); - conn.trackSeq('sess-1', 10); - - // Force reconnect - conn.checkAndReconnect(true); - const esCountBefore = MockEventSource.instances.length; - - // Welcome — reconnect POST will fail - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - await vi.runAllTimersAsync(); - - // Should have scheduled a delayed reconnect (new ES after timer) - expect(conn.isConnected()).toBe(false); - expect(MockEventSource.instances.length).toBeGreaterThan(esCountBefore); - - warnSpy.mockRestore(); - }); - - it('stale doReconnectPost does not set _connected when checkAndReconnect fires mid-flight', async () => { - let resolveReconnect!: (v: { ok: boolean }) => void; - const mockFetch = vi.fn().mockImplementation((url: string) => { - if (url.includes('/reconnect')) { - return new Promise((resolve) => { - resolveReconnect = resolve; - }); - } - return Promise.resolve({ ok: true }); - }); - const conn = new SseConnection(createConfig({ fetch: mockFetch })); - conn.connect(); - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); - conn.trackSeq('sess-1', 10); - - // Force reconnect — creates ES2 - conn.checkAndReconnect(true); - - // ES2 welcome — doReconnectPost(conn-def) starts - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - - // checkAndReconnect fires again while POST is in-flight — creates ES3 - conn.checkAndReconnect(true); - - // Stale POST resolves successfully — must NOT set _connected - resolveReconnect({ ok: true }); - await vi.runAllTimersAsync(); - - // ES3 hasn't welcomed yet, so _connected must remain false - expect(conn.isConnected()).toBe(false); - }); - it('does not emit _close when checkAndReconnect called while already disconnected', () => { const conn = new SseConnection(createConfig()); const listener = vi.fn(); @@ -714,45 +399,6 @@ describe('SseConnection', () => { expect(listener).not.toHaveBeenCalledWith({ type: '_close' }); }); - it('recovers via scheduleReconnect after repeated POST failures', async () => { - let reconnectCallCount = 0; - const mockFetch = vi.fn().mockImplementation((url: string) => { - if (url.includes('/reconnect')) { - reconnectCallCount++; - // Fail first two (initial + forced retry), succeed on third - if (reconnectCallCount <= 2) return Promise.resolve({ ok: false, status: 500 }); - return Promise.resolve({ ok: true }); - } - return Promise.resolve({ ok: true }); - }); - const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); - const conn = new SseConnection(createConfig({ fetch: mockFetch })); - const listener = vi.fn(); - conn.onMessage(listener); - conn.connect(); - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); - conn.trackSeq('sess-1', 10); - - // Force reconnect — POST fails, triggers checkAndReconnect(true) - conn.checkAndReconnect(true); - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - await vi.runAllTimersAsync(); - - // Forced retry also fails - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-ghi' }); - await vi.runAllTimersAsync(); - expect(conn.isConnected()).toBe(false); - - // Third welcome — reconnect POST succeeds - listener.mockClear(); - lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-jkl' }); - await vi.runAllTimersAsync(); - - expect(conn.isConnected()).toBe(true); - expect(listener).toHaveBeenCalledWith({ type: '_open' }); - warnSpy.mockRestore(); - }); - it('connects immediately on reconnect when seqBySession is empty', () => { const mockFetch = vi.fn().mockResolvedValue({ ok: true }); const conn = new SseConnection(createConfig({ fetch: mockFetch })); diff --git a/packages/client/src/sse-connection.ts b/packages/client/src/sse-connection.ts index 8ca85ed8..e60e67e6 100644 --- a/packages/client/src/sse-connection.ts +++ b/packages/client/src/sse-connection.ts @@ -23,7 +23,6 @@ export interface SseConnectionConfig { fetch: (url: string, init?: RequestInit) => Promise; /** Factory for EventSource — allows injection for testing. */ createEventSource?: (url: string) => EventSource; - reconnectDelayMs?: number; /** URL for the sendBeacon suspend fallback. */ suspendUrl?: string; } @@ -38,7 +37,6 @@ export class SseConnection implements ChatConnection { private listener: ConnectionListener | null = null; private seqBySession = new Map(); private pendingSends: Array<{ endpoint: string; body: Record }> = []; - private reconnectTimer: ReturnType | null = null; private boundOnVisibility: (() => void) | null = null; private boundOnPageShow: ((e: PageTransitionEvent) => void) | null = null; private boundOnPageHide: (() => void) | null = null; @@ -47,7 +45,6 @@ export class SseConnection implements ChatConnection { constructor(config: SseConnectionConfig) { this.config = { createEventSource: (url: string) => new EventSource(url), - reconnectDelayMs: 500, suspendUrl: '', ...config, }; @@ -60,10 +57,6 @@ export class SseConnection implements ChatConnection { disconnect(): void { this.removeBrowserListeners(); - if (this.reconnectTimer) { - clearTimeout(this.reconnectTimer); - this.reconnectTimer = null; - } if (this.es) { this.es.close(); this.es = null; @@ -92,8 +85,8 @@ export class SseConnection implements ChatConnection { return true; } - // Queue if reconnecting - if (this.reconnectTimer || this.es) { + // Queue if EventSource exists (reconnecting) + if (this.es) { if (this.pendingSends.length >= MAX_PENDING_SENDS) { this.pendingSends.shift(); } @@ -175,7 +168,6 @@ export class SseConnection implements ChatConnection { */ checkAndReconnect(force = false): void { if (!force && this._connected) return; - if (this.reconnectTimer) return; if (this.es) { this.es.close(); this.es = null; @@ -193,11 +185,6 @@ export class SseConnection implements ChatConnection { private doConnect(): void { if (this.es) return; - if (this.reconnectTimer) { - clearTimeout(this.reconnectTimer); - this.reconnectTimer = null; - } - // Always use the base URL — reconnect sessions are sent via POST in the // welcome handler. This avoids the bug where EventSource auto-reconnect // reuses the original URL (missing ?sessions=), and eliminates double @@ -217,18 +204,21 @@ export class SseConnection implements ChatConnection { } this._connectionId = msg.connectionId as string; - // _connected deferred until doReconnectPost succeeds — prevents - // external send() from bypassing the pending queue mid-reconnect. - // Capture both connectionId and ES instance for the staleness guard. - const welcomeConnectionId = this._connectionId; - const welcomeEs = this.es; + // Fire reconnect POST (fire-and-forget) if reconnecting with sessions. + // No need to defer _connected — handleSendV2 handles ownership on first + // message, and replayed events arrive via SSE regardless. if (this._isReconnect && this.seqBySession.size > 0) { - this.doReconnectPost(welcomeConnectionId, welcomeEs); - } else { - this._connected = true; - this.flushPendingSends(); - this.listener?.({ type: '_open' }); + this.doPost('reconnect', { + type: 'reconnect', + sessions: Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({ + sessionId, + lastSeq, + })), + }); } + this._connected = true; + this.flushPendingSends(); + this.listener?.({ type: '_open' }); this._isReconnect = true; }); @@ -262,66 +252,6 @@ export class SseConnection implements ChatConnection { // but we wait for the 'welcome' event before marking as connected. } - /** - * Send the reconnect POST and only mark connected on success. - * - * On failure the client stays disconnected — the next EventSource - * auto-reconnect will trigger a fresh welcome + retry. This prevents - * flushing pending sends into the void when the server never ran - * handleReconnect (no watch, no reattach, no replay). - */ - private async doReconnectPost( - welcomeConnectionId: string, - welcomeEs: EventSource | null, - ): Promise { - try { - const res = await this.config.fetch(`${this.config.baseUrl}/api/chat/reconnect`, { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - 'X-Connection-ID': welcomeConnectionId, - }, - body: JSON.stringify({ - type: 'reconnect', - sessions: Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({ - sessionId, - lastSeq, - })), - }), - }); - - // Guard: bail if disconnect() was called, a newer welcome arrived, - // or checkAndReconnect replaced the EventSource while in-flight. - if (!this.es || this.es !== welcomeEs || this._connectionId !== welcomeConnectionId) return; - - if (res.ok) { - this._connected = true; - this.flushPendingSends(); - this.listener?.({ type: '_open' }); - } else { - console.warn('[SseConnection] reconnect POST returned', res.status); - this.scheduleReconnect(); - } - } catch (err) { - if (!this.es || this.es !== welcomeEs || this._connectionId !== welcomeConnectionId) return; - console.warn('[SseConnection] reconnect POST failed', err); - this.scheduleReconnect(); - } - } - - /** Tear down and reconnect after a delay to avoid tight retry loops. */ - private scheduleReconnect(): void { - if (this.reconnectTimer) return; - if (this.es) { - this.es.close(); - this.es = null; - } - this.reconnectTimer = setTimeout(() => { - this.reconnectTimer = null; - this.doConnect(); - }, this.config.reconnectDelayMs); - } - private async doPost(endpoint: string, body: Record): Promise { if (!this._connectionId) return; try { From 1f94b7a3990271aef6a9ca804568fc27566563dd Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 4 Jul 2026 12:08:29 +0100 Subject: [PATCH 04/15] refactor(protocol): remove ReconnectMessage from WS union (P3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reconnect is REST-only — keep the schema export for the REST handler but exclude it from IncomingWsMessageV2. The WS dispatcher already ignores it with a comment explaining why. Co-Authored-By: Claude Opus 4.6 --- packages/protocol/src/ws-schemas-v2.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/protocol/src/ws-schemas-v2.ts b/packages/protocol/src/ws-schemas-v2.ts index 8fc07d35..ef299c7f 100644 --- a/packages/protocol/src/ws-schemas-v2.ts +++ b/packages/protocol/src/ws-schemas-v2.ts @@ -124,9 +124,10 @@ export const V2SetModeMessage = z.object({ // ─── Union ────────────────────────────────────────────────────────────────── +// ReconnectMessage is handled via REST POST (not WS) — exported for +// the REST handler but excluded from the WS union. export const IncomingWsMessageV2 = z.discriminatedUnion('type', [ HelloMessage, - ReconnectMessage, WatchMessage, UnwatchMessage, SwitchSessionMessage, From 74e7d460e0948f8acc8485713b836e43527723b8 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 4 Jul 2026 12:12:57 +0100 Subject: [PATCH 05/15] test(ws-handler-v2): fix 2 failing tests from P3 reconnect simplification Stale session test now asserts remove is NOT called (deferred to handleSendV2). Suspend resume test clears reattachChat mock to avoid bleed from prior tests. Co-Authored-By: Claude Opus 4.6 --- server/__tests__/ws-handler-v2.test.ts | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/server/__tests__/ws-handler-v2.test.ts b/server/__tests__/ws-handler-v2.test.ts index fdbaac41..b5fcae96 100644 --- a/server/__tests__/ws-handler-v2.test.ts +++ b/server/__tests__/ws-handler-v2.test.ts @@ -1550,7 +1550,7 @@ describe('dispatchV2Message', () => { expect(stopChat).toHaveBeenCalledWith('driver-1'); }); - it('routes reconnect messages and produces reconnected summary', async () => { + it('ignores reconnect messages over WS (handled via REST only)', async () => { const ctx = createContext(); const transport = mockTransport(); ctx.connRegistry.register('c1', transport); @@ -1565,7 +1565,8 @@ describe('dispatchV2Message', () => { ctx, ); - expect(transport.sent).toContainEqual(expect.objectContaining({ type: 'reconnected' })); + // Reconnect removed from WS union — message is silently dropped + expect(transport.sent).not.toContainEqual(expect.objectContaining({ type: 'reconnected' })); }); it('routes set_mode messages correctly', async () => { @@ -2623,13 +2624,12 @@ describe('handleInterruptV2 rekey after detached reattach', () => { // ─── stale session cleanup — registry.remove() ────────────────────────────── describe('stale session cleanup removes registry entry', () => { - it('handleReconnect removes stale session from registry', () => { + it('handleReconnect does not remove stale sessions (deferred to handleSendV2)', () => { const sessionReg = mockSessionRegistry(); sessionReg.findBySessionId.mockReturnValue({ clientId: 'old-conn:sess-1', session: {} }); sessionReg.isActive.mockReturnValue(true); const eventStore = mockEventStore(); - // handleReconnect uses getSessionState() instead of getSession().isActive eventStore.getSessionState.mockReturnValue('ENDED'); const ctx = createContext({ @@ -2645,7 +2645,8 @@ describe('stale session cleanup removes registry entry', () => { ctx, ); - expect(sessionReg.remove).toHaveBeenCalledWith('old-conn:sess-1'); + // Zombie cleanup deferred to handleSendV2 on first user message + expect(sessionReg.remove).not.toHaveBeenCalled(); }); it('handleSendV2 aborts zombie session before resume', () => { @@ -2804,6 +2805,8 @@ describe('handleSessionSuspend', () => { describe('handleReconnect suspend resume', () => { it('clears suspend state and sends session_resumed', () => { + (reattachChat as ReturnType).mockClear(); + const sessionReg = mockSessionRegistry(); sessionReg.findBySessionId.mockReturnValue({ clientId: 'conn-1:sess-1', From fe4025746dc0d5048face40e238a7848f521a0ea Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 13:20:36 +0100 Subject: [PATCH 06/15] fix(test): update mocks and schema test for P3 reconnect changes Add missing setSessionState mock to routes and suspend-routes tests. Remove reconnect from WS union test since P3 moved it to REST-only. Co-Authored-By: Claude Opus 4.6 --- packages/protocol/__tests__/ws-schemas-v2.test.ts | 1 - server/__tests__/routes.test.ts | 1 + server/__tests__/suspend-routes.test.ts | 1 + 3 files changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/protocol/__tests__/ws-schemas-v2.test.ts b/packages/protocol/__tests__/ws-schemas-v2.test.ts index ce0d6003..8f0e1426 100644 --- a/packages/protocol/__tests__/ws-schemas-v2.test.ts +++ b/packages/protocol/__tests__/ws-schemas-v2.test.ts @@ -162,7 +162,6 @@ describe('IncomingWsMessageV2 discriminated union', () => { it('parses all v2 message types', () => { const messages = [ { type: 'hello', protocolVersion: 2 }, - { type: 'reconnect', sessions: [] }, { type: 'watch', sessionId: 'sess-1' }, { type: 'unwatch', sessionId: 'sess-1' }, { type: 'switch_session', sessionId: 'sess-1' }, diff --git a/server/__tests__/routes.test.ts b/server/__tests__/routes.test.ts index edd6f615..99377430 100644 --- a/server/__tests__/routes.test.ts +++ b/server/__tests__/routes.test.ts @@ -109,6 +109,7 @@ vi.mock('../chat.js', () => { } return null; }), + setSessionState: vi.fn(), }, }; }); diff --git a/server/__tests__/suspend-routes.test.ts b/server/__tests__/suspend-routes.test.ts index dc40f4c3..4aa9d74c 100644 --- a/server/__tests__/suspend-routes.test.ts +++ b/server/__tests__/suspend-routes.test.ts @@ -51,6 +51,7 @@ vi.mock('../chat.js', () => { append: vi.fn(), getEventsAfter: vi.fn().mockReturnValue([]), getSession: vi.fn().mockReturnValue(null), + setSessionState: vi.fn(), }, isIsolationEnabled: vi.fn().mockReturnValue(false), generateWtId: vi.fn().mockReturnValue('wt-test'), From 263f7bcef7261478c803f6efcc7249478b77bf15 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 13:26:13 +0100 Subject: [PATCH 07/15] fix(server): remove dead reconnect case from WS message handler P3 removed ReconnectMessage from the WS union (reconnect is now REST-only), but the switch case was left behind causing a type error. Co-Authored-By: Claude Opus 4.6 --- server/ws-handler-v2.ts | 3 --- 1 file changed, 3 deletions(-) diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index eb32690e..08ed41f0 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -912,9 +912,6 @@ export async function dispatchV2Message( case 'hello': // Already handled at routing layer, ignore duplicate break; - case 'reconnect': - // Reconnect is handled via REST POST, not WS. Ignore if received over WS. - break; case 'watch': handleWatch(connectionId, msg, ctx); break; From 031c5378ea3daa115f472a0446d06e47038b758d Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 13:27:08 +0100 Subject: [PATCH 08/15] fix(server): guard queryInstance.interrupt() against ProcessTransport crashes Without this try/catch, an unhandled error from the transport layer kills the server, losing all in-memory state and triggering replay storms on client reconnect. Cherry-picked from #396 (now closed as superseded by P3). Co-Authored-By: Claude Opus 4.6 --- server/chat.ts | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/server/chat.ts b/server/chat.ts index 6442db70..827701b2 100644 --- a/server/chat.ts +++ b/server/chat.ts @@ -1270,7 +1270,17 @@ export async function interruptChat( ); await Promise.allSettled(stops); } - await session.queryInstance.interrupt(); + try { + await session.queryInstance.interrupt(); + } catch (err) { + // Guard against ProcessTransport crashes (e.g. "not ready for writing"). + // Without this, an unhandled error kills the server, losing all in-memory + // state and triggering replay storms on client reconnect. + log.error('queryInstance.interrupt() failed', { + clientId, + error: err instanceof Error ? err.message : String(err), + }); + } // Only push to inputQueue on first delivery — a retried interrupt should // still call interrupt() (to halt the agent) but not double-queue the prompt. if (!isDup) { From 66ef2cf18176f14a5889bfdc9b340d49adbc26fa Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 13:34:54 +0100 Subject: [PATCH 09/15] =?UTF-8?q?fix(transport):=20address=20Centaur=20rev?= =?UTF-8?q?iew=20=E2=80=94=20restore=20WS=20reconnect,=20document=20cursor?= =?UTF-8?q?=20race?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Restore ReconnectMessage in WS union and handler (WS clients still send reconnect over WS until P4 removes WS transport) - Document why cursor race between fire-and-forget reconnect POST and handleSendV2 is benign (single-threaded + client seq dedup) Co-Authored-By: Claude Opus 4.6 --- packages/protocol/__tests__/ws-schemas-v2.test.ts | 1 + packages/protocol/src/ws-schemas-v2.ts | 5 +++-- server/__tests__/ws-handler-v2.test.ts | 6 +++--- server/ws-handler-v2.ts | 9 +++++++++ 4 files changed, 16 insertions(+), 5 deletions(-) diff --git a/packages/protocol/__tests__/ws-schemas-v2.test.ts b/packages/protocol/__tests__/ws-schemas-v2.test.ts index 8f0e1426..ce0d6003 100644 --- a/packages/protocol/__tests__/ws-schemas-v2.test.ts +++ b/packages/protocol/__tests__/ws-schemas-v2.test.ts @@ -162,6 +162,7 @@ describe('IncomingWsMessageV2 discriminated union', () => { it('parses all v2 message types', () => { const messages = [ { type: 'hello', protocolVersion: 2 }, + { type: 'reconnect', sessions: [] }, { type: 'watch', sessionId: 'sess-1' }, { type: 'unwatch', sessionId: 'sess-1' }, { type: 'switch_session', sessionId: 'sess-1' }, diff --git a/packages/protocol/src/ws-schemas-v2.ts b/packages/protocol/src/ws-schemas-v2.ts index ef299c7f..648c3d5e 100644 --- a/packages/protocol/src/ws-schemas-v2.ts +++ b/packages/protocol/src/ws-schemas-v2.ts @@ -124,10 +124,11 @@ export const V2SetModeMessage = z.object({ // ─── Union ────────────────────────────────────────────────────────────────── -// ReconnectMessage is handled via REST POST (not WS) — exported for -// the REST handler but excluded from the WS union. +// ReconnectMessage is primarily handled via REST POST (SSE transport), but WS +// clients still send it over WS until P4 removes the WS chat transport. export const IncomingWsMessageV2 = z.discriminatedUnion('type', [ HelloMessage, + ReconnectMessage, WatchMessage, UnwatchMessage, SwitchSessionMessage, diff --git a/server/__tests__/ws-handler-v2.test.ts b/server/__tests__/ws-handler-v2.test.ts index b5fcae96..ebefe709 100644 --- a/server/__tests__/ws-handler-v2.test.ts +++ b/server/__tests__/ws-handler-v2.test.ts @@ -1550,7 +1550,7 @@ describe('dispatchV2Message', () => { expect(stopChat).toHaveBeenCalledWith('driver-1'); }); - it('ignores reconnect messages over WS (handled via REST only)', async () => { + it('handles reconnect messages over WS (until P4 removes WS transport)', async () => { const ctx = createContext(); const transport = mockTransport(); ctx.connRegistry.register('c1', transport); @@ -1565,8 +1565,8 @@ describe('dispatchV2Message', () => { ctx, ); - // Reconnect removed from WS union — message is silently dropped - expect(transport.sent).not.toContainEqual(expect.objectContaining({ type: 'reconnected' })); + // WS reconnect calls watch() for each session + expect(ctx.connRegistry.get('c1')?.watchedSessions.has('sess-1')).toBe(true); }); it('routes set_mode messages correctly', async () => { diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index 08ed41f0..744f4620 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -539,6 +539,11 @@ export function handleSendV2( } applySkillPolicy(activeClientId); ctx.connRegistry.watch(connectionId, sessionId); + // No resetCursor here — handleReconnect (fire-and-forget POST) sets + // cursor to lastSeq when it arrives. Between watch and reconnect, + // broadcasts may deliver events the client already has, but Node's + // single-threaded event loop prevents true interleaving and client- + // side seq dedup handles any duplicates. ctx.connRegistry.setActive(connectionId, sessionId); sendToChat(activeClientId, prompt, msg.images, msg.contextBlocks, msg.clientMsgId); span.setAttribute('routing.decision', isOwner ? 'active' : 'takeover'); @@ -912,6 +917,10 @@ export async function dispatchV2Message( case 'hello': // Already handled at routing layer, ignore duplicate break; + case 'reconnect': + // WS clients still send reconnect over WS (removed in P4). + handleReconnect(connectionId, msg, ctx); + break; case 'watch': handleWatch(connectionId, msg, ctx); break; From 13153d68a7a9065ca1ff04e1d01f07c08cef02fe Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 15:30:21 +0100 Subject: [PATCH 10/15] =?UTF-8?q?fix(transport):=20address=20second=20Cent?= =?UTF-8?q?aur=20review=20=E2=80=94=20all=206=20findings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add warn log for failed POSTs (was silent catch) - Retry reconnect POST on next EventSource reconnect if it fails - Flush pending sends AFTER reconnect POST (ordering guarantee) - Reattach detached sessions on reconnect (was deferred to send) - Fix comment to lead with client-side seq dedup, not event loop - Add test for SSE event delivery after reconnect POST failure Co-Authored-By: Claude Opus 4.6 --- .../src/__tests__/sse-connection.test.ts | 46 ++++++++++++++- packages/client/src/sse-connection.ts | 58 +++++++++++++------ server/__tests__/ws-handler-v2.test.ts | 6 +- server/ws-handler-v2.ts | 32 +++++++--- 4 files changed, 112 insertions(+), 30 deletions(-) diff --git a/packages/client/src/__tests__/sse-connection.test.ts b/packages/client/src/__tests__/sse-connection.test.ts index 88111d8a..be1be19a 100644 --- a/packages/client/src/__tests__/sse-connection.test.ts +++ b/packages/client/src/__tests__/sse-connection.test.ts @@ -363,7 +363,7 @@ describe('SseConnection', () => { expect(listener).toHaveBeenCalledWith({ type: '_open' }); }); - it('flushes pending sends immediately on reconnect welcome', () => { + it('flushes pending sends after reconnect POST completes', async () => { const postEndpoints: string[] = []; const mockFetch = vi.fn().mockImplementation((url: string) => { const endpoint = url.replace('https://localhost:3100/api/chat/', ''); @@ -380,10 +380,50 @@ describe('SseConnection', () => { conn.send({ type: 'send', prompt: 'queued msg', clientMsgId: 'q-1' }); postEndpoints.length = 0; - // Welcome — reconnect POST + queued send both fire immediately + // Welcome — reconnect POST fires, pending sends flush after it resolves lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - expect(postEndpoints).toEqual(['reconnect', 'send']); + // reconnect POST fires first + expect(postEndpoints).toEqual(['reconnect']); + + // After the reconnect POST resolves, pending sends flush + await vi.waitFor(() => { + expect(postEndpoints).toEqual(['reconnect', 'send']); + }); + }); + + it('SSE events still arrive after reconnect POST failure', async () => { + const mockFetch = vi.fn().mockImplementation((url: string) => { + if (url.includes('/reconnect')) { + return Promise.reject(new Error('network error')); + } + return Promise.resolve({ ok: true }); + }); + const conn = new SseConnection(createConfig({ fetch: mockFetch })); + const received: Array> = []; + conn.onMessage((msg) => received.push(msg)); + conn.connect(); + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); + conn.trackSeq('sess-1', 10); + + // Force reconnect — reconnect POST will fail + conn.checkAndReconnect(true); + + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); + + // Wait for reconnect POST to fail + await vi.waitFor(() => { + expect(mockFetch).toHaveBeenCalledWith( + expect.stringContaining('/reconnect'), + expect.any(Object), + ); + }); + + // SSE events still arrive via EventSource despite reconnect POST failure + lastES()._emit('message', { type: 'assistant', sessionId: 'sess-1', seq: 11 }); + expect(received).toContainEqual( + expect.objectContaining({ type: 'assistant', sessionId: 'sess-1', seq: 11 }), + ); }); it('does not emit _close when checkAndReconnect called while already disconnected', () => { diff --git a/packages/client/src/sse-connection.ts b/packages/client/src/sse-connection.ts index e60e67e6..f3e25418 100644 --- a/packages/client/src/sse-connection.ts +++ b/packages/client/src/sse-connection.ts @@ -37,6 +37,8 @@ export class SseConnection implements ChatConnection { private listener: ConnectionListener | null = null; private seqBySession = new Map(); private pendingSends: Array<{ endpoint: string; body: Record }> = []; + /** Sessions that need a reconnect POST — set when reconnect fails, retried on next welcome. */ + private _pendingReconnectSessions: Array<{ sessionId: string; lastSeq: number }> | null = null; private boundOnVisibility: (() => void) | null = null; private boundOnPageShow: ((e: PageTransitionEvent) => void) | null = null; private boundOnPageHide: (() => void) | null = null; @@ -81,7 +83,7 @@ export class SseConnection implements ChatConnection { if (!endpoint) return false; if (this._connected && this._connectionId) { - this.doPost(endpoint, msg); + this.doPost(endpoint, msg).catch(() => {}); return true; } @@ -143,7 +145,7 @@ export class SseConnection implements ChatConnection { // Try POST first if (this._connected && this._connectionId) { - this.doPost('suspend', { type: 'session_suspend', sessions }); + this.doPost('suspend', { type: 'session_suspend', sessions }).catch(() => {}); return; } @@ -204,20 +206,37 @@ export class SseConnection implements ChatConnection { } this._connectionId = msg.connectionId as string; - // Fire reconnect POST (fire-and-forget) if reconnecting with sessions. - // No need to defer _connected — handleSendV2 handles ownership on first + // Fire reconnect POST if reconnecting with sessions, or retry a + // previously failed reconnect. handleSendV2 handles ownership on first // message, and replayed events arrive via SSE regardless. - if (this._isReconnect && this.seqBySession.size > 0) { - this.doPost('reconnect', { - type: 'reconnect', - sessions: Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({ - sessionId, - lastSeq, - })), - }); - } + const sessions = + this._pendingReconnectSessions ?? + (this._isReconnect && this.seqBySession.size > 0 + ? Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({ + sessionId, + lastSeq, + })) + : null); this._connected = true; - this.flushPendingSends(); + if (sessions) { + this._pendingReconnectSessions = sessions; + this.doPost('reconnect', { type: 'reconnect', sessions }).then( + () => { + this._pendingReconnectSessions = null; + // Flush pending sends AFTER reconnect so the server processes + // handleReconnect (cursor reset, replay) before user messages. + this.flushPendingSends(); + }, + () => { + // doPost already logs the warning. Keep _pendingReconnectSessions + // so the next EventSource reconnect retries automatically. + // Still flush — handleSendV2 handles ownership independently. + this.flushPendingSends(); + }, + ); + } else { + this.flushPendingSends(); + } this.listener?.({ type: '_open' }); this._isReconnect = true; }); @@ -263,9 +282,14 @@ export class SseConnection implements ChatConnection { }, body: JSON.stringify(body), }); - } catch { + } catch (err) { // POST failures are non-fatal — the server may be temporarily - // unreachable. The SSE stream will reconnect and replay missed events. + // unreachable. SSE EventSource auto-reconnects and replays missed events + // from the EventStore. However, a failed reconnect POST means the server + // won't reset the cursor or re-send boot context until the next + // reconnect cycle. Client-side seq dedup prevents duplicate delivery. + console.warn(`[mitzo] ${endpoint} POST failed:`, err instanceof Error ? err.message : err); + throw err; } } @@ -274,7 +298,7 @@ export class SseConnection implements ChatConnection { const toFlush = this.pendingSends; this.pendingSends = []; for (const { endpoint, body } of toFlush) { - this.doPost(endpoint, body); + this.doPost(endpoint, body).catch(() => {}); } } diff --git a/server/__tests__/ws-handler-v2.test.ts b/server/__tests__/ws-handler-v2.test.ts index ebefe709..f122dd66 100644 --- a/server/__tests__/ws-handler-v2.test.ts +++ b/server/__tests__/ws-handler-v2.test.ts @@ -242,14 +242,14 @@ describe('handleReconnect', () => { ]); }); - it('does not reattach or rekey on reconnect (deferred to handleSendV2)', () => { + it('reattaches detached session on reconnect (owner connection)', () => { (reattachChat as ReturnType).mockClear(); (rekeyChat as ReturnType).mockClear(); const sessionReg = mockSessionRegistry(); sessionReg.findBySessionId.mockReturnValue({ clientId: 'c1:sess-1' }); sessionReg.isActive.mockReturnValue(true); - sessionReg.isAttached.mockReturnValue(false); // detached — but reconnect should NOT reattach + sessionReg.isAttached.mockReturnValue(false); // detached — reconnect should reattach const ctx = createContext({ sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], @@ -263,7 +263,7 @@ describe('handleReconnect', () => { ctx, ); - expect(reattachChat).not.toHaveBeenCalled(); + expect(reattachChat).toHaveBeenCalledWith('c1:sess-1', transport); expect(rekeyChat).not.toHaveBeenCalled(); }); diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index 744f4620..a47c7bea 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -252,15 +252,31 @@ export function handleReconnect( const newCursor = events.length > 0 ? events[events.length - 1].seq : entry.lastSeq; ctx.connRegistry.resetCursor(connectionId, entry.sessionId, newCursor); - // Ownership dance (reattach/rekey/zombie cleanup) is NOT done here. - // handleSendV2 handles all of that on the first user message — reconnect - // only needs to restore the event stream and boot context. + // Reattach detached sessions so the agent's transport is refreshed. + // Without this, a passively observing user would see events via + // watch/broadcast but the agent's transport stays stale. + const found = ctx.sessionRegistry.findBySessionId(entry.sessionId); + if (found && ctx.sessionRegistry.isActive(found.clientId)) { + const ownerConnection = found.clientId.split(':')[0]; + if ( + ownerConnection === connectionId && + !ctx.sessionRegistry.isAttached(found.clientId) + ) { + const transport = ctx.connRegistry.get(connectionId)?.transport; + if (transport) { + reattachChat(found.clientId, transport); + log.info('reattached detached session on reconnect', { + connectionId, + sessionId: entry.sessionId, + }); + } + } + } // If the session was suspended, clear suspend state. Don't replay // buffered events — they were already replayed from EventStore above // (sendOrBuffer appends to both stores, so EventStore covers the // suspend period). resume() just clears the suspend flag + buffer. - const found = ctx.sessionRegistry.findBySessionId(entry.sessionId); const running = found ? ctx.sessionRegistry.isActive(found.clientId) : false; let suspendReplayed = 0; if (found && running && ctx.sessionRegistry.isSuspended(found.clientId)) { @@ -541,9 +557,11 @@ export function handleSendV2( ctx.connRegistry.watch(connectionId, sessionId); // No resetCursor here — handleReconnect (fire-and-forget POST) sets // cursor to lastSeq when it arrives. Between watch and reconnect, - // broadcasts may deliver events the client already has, but Node's - // single-threaded event loop prevents true interleaving and client- - // side seq dedup handles any duplicates. + // broadcasts may deliver events the client already has. This is safe + // because client-side seq dedup (store.ts) drops events with seq <= + // lastProcessedSeq. The two HTTP requests (reconnect POST and send + // POST) can arrive as separate event loop ticks in any order, but + // duplicate delivery is always harmless thanks to seq dedup. ctx.connRegistry.setActive(connectionId, sessionId); sendToChat(activeClientId, prompt, msg.images, msg.contextBlocks, msg.clientMsgId); span.setAttribute('routing.decision', isOwner ? 'active' : 'takeover'); From 1737898d5d9b6791a1edb802fbe54ff3d4c86e63 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 15:38:23 +0100 Subject: [PATCH 11/15] style(server): fix prettier formatting in ws-handler-v2 Co-Authored-By: Claude Opus 4.6 --- server/ws-handler-v2.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index a47c7bea..947a69c4 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -258,10 +258,7 @@ export function handleReconnect( const found = ctx.sessionRegistry.findBySessionId(entry.sessionId); if (found && ctx.sessionRegistry.isActive(found.clientId)) { const ownerConnection = found.clientId.split(':')[0]; - if ( - ownerConnection === connectionId && - !ctx.sessionRegistry.isAttached(found.clientId) - ) { + if (ownerConnection === connectionId && !ctx.sessionRegistry.isAttached(found.clientId)) { const transport = ctx.connRegistry.get(connectionId)?.transport; if (transport) { reattachChat(found.clientId, transport); From 76c40403c5fe4ec688fedb87b8c0f978425fa8a4 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 15:45:25 +0100 Subject: [PATCH 12/15] =?UTF-8?q?fix(transport):=20address=20third=20Centa?= =?UTF-8?q?ur=20review=20=E2=80=94=204=20findings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Check res.ok in doPost (RED: HTTP 500 was treated as success) - Clear _pendingReconnectSessions on disconnect (YELLOW: stale retry) - Filter pending reconnect sessions against seqBySession (YELLOW: cleared sessions retried) - Reattach detached sessions when owner connection is gone (YELLOW: device restart gap) Co-Authored-By: Claude Opus 4.6 --- .../src/__tests__/sse-connection.test.ts | 66 +++++++++++++++++++ packages/client/src/sse-connection.ts | 13 +++- server/__tests__/ws-handler-v2.test.ts | 25 +++++++ server/ws-handler-v2.ts | 10 ++- 4 files changed, 111 insertions(+), 3 deletions(-) diff --git a/packages/client/src/__tests__/sse-connection.test.ts b/packages/client/src/__tests__/sse-connection.test.ts index be1be19a..357327f5 100644 --- a/packages/client/src/__tests__/sse-connection.test.ts +++ b/packages/client/src/__tests__/sse-connection.test.ts @@ -548,6 +548,72 @@ describe('SseConnection', () => { ); }); + it('treats HTTP 500 as failure and retries reconnect on next welcome', async () => { + let reconnectCount = 0; + const mockFetch = vi.fn().mockImplementation((url: string) => { + if (url.includes('/reconnect')) { + reconnectCount++; + return Promise.resolve({ ok: false, status: 500 }); + } + return Promise.resolve({ ok: true }); + }); + const conn = new SseConnection(createConfig({ fetch: mockFetch })); + conn.connect(); + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); + conn.trackSeq('sess-1', 10); + + // Force reconnect — reconnect POST will get HTTP 500 + conn.checkAndReconnect(true); + reconnectCount = 0; + + // First welcome — reconnect POST fires and gets 500 + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); + await vi.waitFor(() => expect(reconnectCount).toBe(1)); + + // Force another reconnect — should retry because 500 kept _pendingReconnectSessions + conn.checkAndReconnect(true); + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-ghi' }); + await vi.waitFor(() => expect(reconnectCount).toBe(2)); + }); + + it('disconnect clears pending reconnect sessions', async () => { + const mockFetch = vi.fn().mockImplementation((url: string) => { + if (url.includes('/reconnect')) { + return Promise.reject(new Error('network error')); + } + return Promise.resolve({ ok: true }); + }); + const conn = new SseConnection(createConfig({ fetch: mockFetch })); + conn.connect(); + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); + conn.trackSeq('sess-1', 10); + + // Force reconnect — reconnect POST will fail, setting _pendingReconnectSessions + conn.checkAndReconnect(true); + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); + await vi.waitFor(() => { + expect(mockFetch).toHaveBeenCalledWith( + expect.stringContaining('/reconnect'), + expect.any(Object), + ); + }); + + // Disconnect — should clear pending reconnect sessions + conn.disconnect(); + conn.clearSession('sess-1'); // Session no longer tracked after full disconnect + mockFetch.mockClear(); + + // Reconnect fresh — should NOT retry old sessions (pendingReconnect cleared, + // seqBySession empty, so no reconnect POST fires) + conn.connect(); + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-ghi' }); + + expect(mockFetch).not.toHaveBeenCalledWith( + expect.stringContaining('/reconnect'), + expect.any(Object), + ); + }); + it('sendSuspend is no-op with no tracked sessions', () => { const mockFetch = vi.fn().mockResolvedValue({ ok: true }); const conn = new SseConnection(createConfig({ fetch: mockFetch })); diff --git a/packages/client/src/sse-connection.ts b/packages/client/src/sse-connection.ts index f3e25418..e514e9a7 100644 --- a/packages/client/src/sse-connection.ts +++ b/packages/client/src/sse-connection.ts @@ -64,6 +64,7 @@ export class SseConnection implements ChatConnection { this.es = null; } this._connected = false; + this._pendingReconnectSessions = null; } /** @@ -209,8 +210,13 @@ export class SseConnection implements ChatConnection { // Fire reconnect POST if reconnecting with sessions, or retry a // previously failed reconnect. handleSendV2 handles ownership on first // message, and replayed events arrive via SSE regardless. + // Filter pending retries against current seqBySession — sessions may + // have been cleared (clearSession) since the retry was queued. + const pending = this._pendingReconnectSessions?.filter((s) => + this.seqBySession.has(s.sessionId), + ); const sessions = - this._pendingReconnectSessions ?? + (pending && pending.length > 0 ? pending : null) ?? (this._isReconnect && this.seqBySession.size > 0 ? Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({ sessionId, @@ -274,7 +280,7 @@ export class SseConnection implements ChatConnection { private async doPost(endpoint: string, body: Record): Promise { if (!this._connectionId) return; try { - await this.config.fetch(`${this.config.baseUrl}/api/chat/${endpoint}`, { + const res = await this.config.fetch(`${this.config.baseUrl}/api/chat/${endpoint}`, { method: 'POST', headers: { 'Content-Type': 'application/json', @@ -282,6 +288,9 @@ export class SseConnection implements ChatConnection { }, body: JSON.stringify(body), }); + if (!res.ok) { + throw new Error(`HTTP ${res.status}`); + } } catch (err) { // POST failures are non-fatal — the server may be temporarily // unreachable. SSE EventSource auto-reconnects and replays missed events diff --git a/server/__tests__/ws-handler-v2.test.ts b/server/__tests__/ws-handler-v2.test.ts index f122dd66..2d3c755e 100644 --- a/server/__tests__/ws-handler-v2.test.ts +++ b/server/__tests__/ws-handler-v2.test.ts @@ -267,6 +267,31 @@ describe('handleReconnect', () => { expect(rekeyChat).not.toHaveBeenCalled(); }); + it('reattaches detached session when owner connection is gone (device restart)', () => { + (reattachChat as ReturnType).mockClear(); + + const sessionReg = mockSessionRegistry(); + // Session owned by old connection 'c-old', but 'c-old' is no longer registered + sessionReg.findBySessionId.mockReturnValue({ clientId: 'c-old:sess-1' }); + sessionReg.isActive.mockReturnValue(true); + sessionReg.isAttached.mockReturnValue(false); + + const ctx = createContext({ + sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], + }); + const transport = mockTransport(); + // Register new connection 'c-new' — 'c-old' is NOT registered (gone) + ctx.connRegistry.register('c-new', transport); + + handleReconnect( + 'c-new', + { type: 'reconnect', sessions: [{ sessionId: 'sess-1', lastSeq: 0 }] }, + ctx, + ); + + expect(reattachChat).toHaveBeenCalledWith('c-old:sess-1', transport); + }); + it('resets cursor to client lastSeq immediately after watch (before replay)', () => { const eventStore = mockEventStore(); eventStore.getEventsAfter.mockReturnValue([ diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index 947a69c4..e97d6a9b 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -258,13 +258,21 @@ export function handleReconnect( const found = ctx.sessionRegistry.findBySessionId(entry.sessionId); if (found && ctx.sessionRegistry.isActive(found.clientId)) { const ownerConnection = found.clientId.split(':')[0]; - if (ownerConnection === connectionId && !ctx.sessionRegistry.isAttached(found.clientId)) { + // Reattach if: (a) same connection owns it, OR (b) old owner + // connection is gone (device restart gave us a new connectionId). + const ownerGone = + ownerConnection !== connectionId && !ctx.connRegistry.get(ownerConnection); + if ( + (ownerConnection === connectionId || ownerGone) && + !ctx.sessionRegistry.isAttached(found.clientId) + ) { const transport = ctx.connRegistry.get(connectionId)?.transport; if (transport) { reattachChat(found.clientId, transport); log.info('reattached detached session on reconnect', { connectionId, sessionId: entry.sessionId, + ownerGone, }); } } From 46c37505db750222de3758e4899f26fdbebd2364 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 18:28:12 +0100 Subject: [PATCH 13/15] =?UTF-8?q?fix(transport):=20address=20fourth=20Cent?= =?UTF-8?q?aur=20review=20=E2=80=94=203=20findings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Refresh lastSeq from current seqBySession on pending retry (YELLOW: stale seq causes unnecessary replay) - Guard reconnect POST callback against stale connectionId (YELLOW: double-flush on rapid reconnect) - Use getOwnerConnection() helper instead of .split(':')[0] (BLUE: consistency) Co-Authored-By: Claude Opus 4.6 --- packages/client/src/sse-connection.ts | 14 +++++++++++--- server/ws-handler-v2.ts | 2 +- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/packages/client/src/sse-connection.ts b/packages/client/src/sse-connection.ts index e514e9a7..88df10a0 100644 --- a/packages/client/src/sse-connection.ts +++ b/packages/client/src/sse-connection.ts @@ -212,9 +212,11 @@ export class SseConnection implements ChatConnection { // message, and replayed events arrive via SSE regardless. // Filter pending retries against current seqBySession — sessions may // have been cleared (clearSession) since the retry was queued. - const pending = this._pendingReconnectSessions?.filter((s) => - this.seqBySession.has(s.sessionId), - ); + // Refresh lastSeq from current map — SSE events may have advanced it + // since the original failure, avoiding unnecessary replay. + const pending = this._pendingReconnectSessions + ?.filter((s) => this.seqBySession.has(s.sessionId)) + .map((s) => ({ ...s, lastSeq: this.seqBySession.get(s.sessionId)! })); const sessions = (pending && pending.length > 0 ? pending : null) ?? (this._isReconnect && this.seqBySession.size > 0 @@ -226,14 +228,20 @@ export class SseConnection implements ChatConnection { this._connected = true; if (sessions) { this._pendingReconnectSessions = sessions; + // Capture connectionId to detect stale callbacks — if a new welcome + // arrives while this POST is in-flight, the callback should no-op + // to avoid double-flushing pending sends. + const postConnectionId = this._connectionId; this.doPost('reconnect', { type: 'reconnect', sessions }).then( () => { + if (this._connectionId !== postConnectionId) return; // stale callback this._pendingReconnectSessions = null; // Flush pending sends AFTER reconnect so the server processes // handleReconnect (cursor reset, replay) before user messages. this.flushPendingSends(); }, () => { + if (this._connectionId !== postConnectionId) return; // stale callback // doPost already logs the warning. Keep _pendingReconnectSessions // so the next EventSource reconnect retries automatically. // Still flush — handleSendV2 handles ownership independently. diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index e97d6a9b..bdefce3d 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -257,7 +257,7 @@ export function handleReconnect( // watch/broadcast but the agent's transport stays stale. const found = ctx.sessionRegistry.findBySessionId(entry.sessionId); if (found && ctx.sessionRegistry.isActive(found.clientId)) { - const ownerConnection = found.clientId.split(':')[0]; + const ownerConnection = getOwnerConnection(found.clientId); // Reattach if: (a) same connection owns it, OR (b) old owner // connection is gone (device restart gave us a new connectionId). const ownerGone = From d50296f41ab88a6296fde5cd570eee837e07e2d7 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 18:37:40 +0100 Subject: [PATCH 14/15] =?UTF-8?q?fix(transport):=20address=20fifth=20Centa?= =?UTF-8?q?ur=20review=20=E2=80=94=208=20findings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Defer _connected until reconnect POST succeeds (YELLOW: prevents sends from bypassing queue during in-flight window) - Don't flush pending sends on POST failure (YELLOW: server hasn't set up cursor/replay, sends stay queued for next attempt) - Add test for send queuing during reconnect POST in-flight (YELLOW) - Fix 4 stale "periodic sync" comments (BLUE: removed in P3) - Document cursor-at-0 bandwidth trade-off in handleSendV2 (YELLOW) - Document transport/clientId mismatch window on reattach (YELLOW) Co-Authored-By: Claude Opus 4.6 --- .../src/__tests__/sse-connection.test.ts | 72 ++++++++++++++++--- packages/client/src/sse-connection.ts | 14 ++-- .../__tests__/connection-registry.test.ts | 4 +- packages/harness/src/connection-registry.ts | 2 +- server/index.ts | 2 +- server/ws-handler-v2.ts | 19 +++-- 6 files changed, 89 insertions(+), 24 deletions(-) diff --git a/packages/client/src/__tests__/sse-connection.test.ts b/packages/client/src/__tests__/sse-connection.test.ts index 357327f5..184ee6f0 100644 --- a/packages/client/src/__tests__/sse-connection.test.ts +++ b/packages/client/src/__tests__/sse-connection.test.ts @@ -318,7 +318,7 @@ describe('SseConnection', () => { ); }); - it('marks connected immediately on reconnect welcome', () => { + it('marks connected after reconnect POST succeeds (not on welcome)', async () => { const mockFetch = vi.fn().mockResolvedValue({ ok: true }); const conn = new SseConnection(createConfig({ fetch: mockFetch })); const listener = vi.fn(); @@ -331,14 +331,19 @@ describe('SseConnection', () => { conn.checkAndReconnect(true); listener.mockClear(); - // New welcome — should be connected immediately (fire-and-forget) + // New welcome — NOT connected yet (reconnect POST in-flight) lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); + expect(conn.isConnected()).toBe(false); + expect(listener).not.toHaveBeenCalledWith({ type: '_open' }); - expect(conn.isConnected()).toBe(true); - expect(listener).toHaveBeenCalledWith({ type: '_open' }); + // After POST resolves — now connected + await vi.waitFor(() => { + expect(conn.isConnected()).toBe(true); + expect(listener).toHaveBeenCalledWith({ type: '_open' }); + }); }); - it('POST failure does not affect connection state', async () => { + it('POST failure keeps connection disconnected (sends stay queued)', async () => { const mockFetch = vi.fn().mockImplementation((url: string) => { if (url.includes('/reconnect')) { return Promise.reject(new Error('network error')); @@ -355,12 +360,58 @@ describe('SseConnection', () => { conn.checkAndReconnect(true); listener.mockClear(); - // Welcome — reconnect POST fires (and will fail), but connection is immediate + // Welcome — reconnect POST fires (and will fail) lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); - // Connected immediately regardless of POST outcome - expect(conn.isConnected()).toBe(true); - expect(listener).toHaveBeenCalledWith({ type: '_open' }); + // Wait for POST to fail + await vi.waitFor(() => { + expect(mockFetch).toHaveBeenCalledWith( + expect.stringContaining('/reconnect'), + expect.any(Object), + ); + }); + + // Still disconnected — sends stay queued until next successful reconnect + expect(conn.isConnected()).toBe(false); + expect(listener).not.toHaveBeenCalledWith({ type: '_open' }); + }); + + it('queues sends during reconnect POST in-flight window', async () => { + let resolveReconnect!: () => void; + const mockFetch = vi.fn().mockImplementation((url: string) => { + if (url.includes('/reconnect')) { + return new Promise<{ ok: boolean }>((resolve) => { + resolveReconnect = () => resolve({ ok: true }); + }); + } + return Promise.resolve({ ok: true }); + }); + const conn = new SseConnection(createConfig({ fetch: mockFetch })); + conn.connect(); + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); + conn.trackSeq('sess-1', 10); + + conn.checkAndReconnect(true); + mockFetch.mockClear(); + + // Welcome — reconnect POST starts (held open) + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); + + // Send during in-flight window — should queue, not send directly + const queued = conn.send({ type: 'send', prompt: 'during reconnect', clientMsgId: 'q-1' }); + expect(queued).toBe(true); + // Only reconnect POST should have been called, not the send + expect(mockFetch).toHaveBeenCalledTimes(1); + expect(mockFetch).toHaveBeenCalledWith( + expect.stringContaining('/reconnect'), + expect.any(Object), + ); + + // Resolve reconnect POST — sends should flush + resolveReconnect(); + await vi.waitFor(() => { + expect(mockFetch).toHaveBeenCalledWith(expect.stringContaining('/send'), expect.any(Object)); + }); }); it('flushes pending sends after reconnect POST completes', async () => { @@ -570,6 +621,9 @@ describe('SseConnection', () => { lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); await vi.waitFor(() => expect(reconnectCount).toBe(1)); + // Still disconnected after 500 + expect(conn.isConnected()).toBe(false); + // Force another reconnect — should retry because 500 kept _pendingReconnectSessions conn.checkAndReconnect(true); lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-ghi' }); diff --git a/packages/client/src/sse-connection.ts b/packages/client/src/sse-connection.ts index 88df10a0..d63b7574 100644 --- a/packages/client/src/sse-connection.ts +++ b/packages/client/src/sse-connection.ts @@ -225,33 +225,35 @@ export class SseConnection implements ChatConnection { lastSeq, })) : null); - this._connected = true; if (sessions) { this._pendingReconnectSessions = sessions; // Capture connectionId to detect stale callbacks — if a new welcome // arrives while this POST is in-flight, the callback should no-op // to avoid double-flushing pending sends. const postConnectionId = this._connectionId; + // Don't mark connected until POST succeeds — prevents send() from + // bypassing the pending queue and arriving before cursor setup. this.doPost('reconnect', { type: 'reconnect', sessions }).then( () => { if (this._connectionId !== postConnectionId) return; // stale callback this._pendingReconnectSessions = null; - // Flush pending sends AFTER reconnect so the server processes - // handleReconnect (cursor reset, replay) before user messages. + this._connected = true; this.flushPendingSends(); + this.listener?.({ type: '_open' }); }, () => { if (this._connectionId !== postConnectionId) return; // stale callback // doPost already logs the warning. Keep _pendingReconnectSessions // so the next EventSource reconnect retries automatically. - // Still flush — handleSendV2 handles ownership independently. - this.flushPendingSends(); + // Don't flush — server hasn't set up cursor/replay. Don't mark + // connected — sends stay queued until next successful reconnect. }, ); } else { + this._connected = true; this.flushPendingSends(); + this.listener?.({ type: '_open' }); } - this.listener?.({ type: '_open' }); this._isReconnect = true; }); diff --git a/packages/harness/__tests__/connection-registry.test.ts b/packages/harness/__tests__/connection-registry.test.ts index a5bc2f2f..20c7cbbb 100644 --- a/packages/harness/__tests__/connection-registry.test.ts +++ b/packages/harness/__tests__/connection-registry.test.ts @@ -219,7 +219,7 @@ describe('ConnectionRegistry', () => { registry.broadcast('sess-a', { type: 'msg1', seq: 5 }); registry.broadcast('sess-a', { type: 'msg2', seq: 10 }); - // Cursor should be at 10 now (can't inspect directly, but periodic sync will use it) + // Cursor should be at 10 now (can't inspect directly, but reconnect replay will use it) expect(t.send).toHaveBeenCalledTimes(2); }); @@ -278,7 +278,7 @@ describe('ConnectionRegistry', () => { // Client reconnects with lastSeq=50 (missed 51-100) registry.resetCursor('conn-1', 'sess-a', 50); - // Cursor should now be at 50 (verified by periodic sync behavior) + // Cursor should now be at 50 (verified by reconnect replay behavior) }); it('is a no-op for unknown connection', () => { diff --git a/packages/harness/src/connection-registry.ts b/packages/harness/src/connection-registry.ts index 94fd319d..9b7cb38f 100644 --- a/packages/harness/src/connection-registry.ts +++ b/packages/harness/src/connection-registry.ts @@ -110,7 +110,7 @@ export class ConnectionRegistry { * Send a message to all open connections watching a session. * Catches send errors to prevent one failing transport from * aborting the broadcast loop. Updates delivery cursor on success - * so periodic sync can retry failures. + * so reconnect replay covers the correct range. */ broadcast(sessionId: string, data: Record): void { const seq = data.seq as number | undefined; diff --git a/server/index.ts b/server/index.ts index fa6ae76b..37e86b2e 100644 --- a/server/index.ts +++ b/server/index.ts @@ -981,7 +981,7 @@ function shutdown(signal: string) { overviewEmitter.destroy(); sseRegistry.destroy(); chatSseRegistry.destroy(); - connRegistry.dispose(); // Stop periodic sync + clear state + connRegistry.dispose(); // Clear state registry.dispose(); for (const client of wss.clients) { client.close(1001, 'Server shutting down'); diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index bdefce3d..fd96c450 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -255,6 +255,14 @@ export function handleReconnect( // Reattach detached sessions so the agent's transport is refreshed. // Without this, a passively observing user would see events via // watch/broadcast but the agent's transport stays stale. + // + // Note: reattach refreshes the transport but does NOT rekey the + // clientId. If ownerGone=true (device restart), the clientId still + // references the old connectionId until the first send triggers + // rekeyChat via handleSendV2. During this window, getOwnerConnection() + // returns a stale value. This is safe — event delivery uses the + // transport (refreshed), and ownership checks on send/interrupt + // handle the rekey atomically. const found = ctx.sessionRegistry.findBySessionId(entry.sessionId); if (found && ctx.sessionRegistry.isActive(found.clientId)) { const ownerConnection = getOwnerConnection(found.clientId); @@ -562,11 +570,12 @@ export function handleSendV2( ctx.connRegistry.watch(connectionId, sessionId); // No resetCursor here — handleReconnect (fire-and-forget POST) sets // cursor to lastSeq when it arrives. Between watch and reconnect, - // broadcasts may deliver events the client already has. This is safe - // because client-side seq dedup (store.ts) drops events with seq <= - // lastProcessedSeq. The two HTTP requests (reconnect POST and send - // POST) can arrive as separate event loop ticks in any order, but - // duplicate delivery is always harmless thanks to seq dedup. + // the cursor starts at 0 (default), so broadcasts may deliver events + // the client already has. This is a bandwidth trade-off, not a + // correctness issue: client-side seq dedup (store.ts) drops events + // with seq <= lastProcessedSeq. The two HTTP requests (reconnect + // POST and send POST) can arrive as separate event loop ticks in any + // order, but duplicate delivery is always harmless thanks to seq dedup. ctx.connRegistry.setActive(connectionId, sessionId); sendToChat(activeClientId, prompt, msg.images, msg.contextBlocks, msg.clientMsgId); span.setAttribute('routing.decision', isOwner ? 'active' : 'takeover'); From c72a27ab8e9cfb1dbf372ffb347ffccc6289aa26 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 19:03:07 +0100 Subject: [PATCH 15/15] =?UTF-8?q?fix(transport):=20address=20sixth=20Centa?= =?UTF-8?q?ur=20review=20=E2=80=94=205=20findings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Throw on doPost when connectionId is null (YELLOW: silent success) - Add P4 TODO for cursor reset unification (YELLOW: dedup assumption) - Check EventStore state before reattaching on reconnect (YELLOW: zombie) - Extract getReconnectSessions() for readability (BLUE: dense logic) - Remove roadmap reference from WS reconnect comment (BLUE: stale) Co-Authored-By: Claude Opus 4.6 --- packages/client/src/sse-connection.ts | 40 ++++++++++++++------ server/ws-handler-v2.ts | 53 ++++++++++++++++++--------- 2 files changed, 63 insertions(+), 30 deletions(-) diff --git a/packages/client/src/sse-connection.ts b/packages/client/src/sse-connection.ts index d63b7574..c179121a 100644 --- a/packages/client/src/sse-connection.ts +++ b/packages/client/src/sse-connection.ts @@ -214,17 +214,7 @@ export class SseConnection implements ChatConnection { // have been cleared (clearSession) since the retry was queued. // Refresh lastSeq from current map — SSE events may have advanced it // since the original failure, avoiding unnecessary replay. - const pending = this._pendingReconnectSessions - ?.filter((s) => this.seqBySession.has(s.sessionId)) - .map((s) => ({ ...s, lastSeq: this.seqBySession.get(s.sessionId)! })); - const sessions = - (pending && pending.length > 0 ? pending : null) ?? - (this._isReconnect && this.seqBySession.size > 0 - ? Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({ - sessionId, - lastSeq, - })) - : null); + const sessions = this.getReconnectSessions(); if (sessions) { this._pendingReconnectSessions = sessions; // Capture connectionId to detect stale callbacks — if a new welcome @@ -288,7 +278,9 @@ export class SseConnection implements ChatConnection { } private async doPost(endpoint: string, body: Record): Promise { - if (!this._connectionId) return; + if (!this._connectionId) { + throw new Error('doPost called without connectionId'); + } try { const res = await this.config.fetch(`${this.config.baseUrl}/api/chat/${endpoint}`, { method: 'POST', @@ -354,6 +346,30 @@ export class SseConnection implements ChatConnection { } } + /** + * Build the session list for a reconnect POST. + * + * Priority: (1) pending retry sessions (filtered against current seqBySession, + * with refreshed lastSeq), (2) all tracked sessions if this is a reconnect. + * Returns null when there's nothing to reconnect. + */ + private getReconnectSessions(): Array<{ sessionId: string; lastSeq: number }> | null { + const pending = this._pendingReconnectSessions + ?.filter((s) => this.seqBySession.has(s.sessionId)) + .map((s) => ({ ...s, lastSeq: this.seqBySession.get(s.sessionId)! })); + + if (pending && pending.length > 0) return pending; + + if (this._isReconnect && this.seqBySession.size > 0) { + return Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({ + sessionId, + lastSeq, + })); + } + + return null; + } + // ─── Browser lifecycle ───────────────────────────────────────────────────── private addBrowserListeners(): void { diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index fd96c450..f3986d52 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -264,24 +264,38 @@ export function handleReconnect( // transport (refreshed), and ownership checks on send/interrupt // handle the rekey atomically. const found = ctx.sessionRegistry.findBySessionId(entry.sessionId); + const storeState = ctx.eventStore.getSessionState(entry.sessionId); + + // Skip reattach for zombie sessions — if EventStore says the session + // is ENDED or CLOSING, don't reattach. The session's query loop is + // finished and reattaching would resurrect a dead transport binding. if (found && ctx.sessionRegistry.isActive(found.clientId)) { - const ownerConnection = getOwnerConnection(found.clientId); - // Reattach if: (a) same connection owns it, OR (b) old owner - // connection is gone (device restart gave us a new connectionId). - const ownerGone = - ownerConnection !== connectionId && !ctx.connRegistry.get(ownerConnection); - if ( - (ownerConnection === connectionId || ownerGone) && - !ctx.sessionRegistry.isAttached(found.clientId) - ) { - const transport = ctx.connRegistry.get(connectionId)?.transport; - if (transport) { - reattachChat(found.clientId, transport); - log.info('reattached detached session on reconnect', { - connectionId, - sessionId: entry.sessionId, - ownerGone, - }); + if (storeState === 'ENDED' || storeState === 'CLOSING') { + log.info('skipping reattach for zombie session on reconnect', { + connectionId, + sessionId: entry.sessionId, + clientId: found.clientId, + storeState, + }); + } else { + const ownerConnection = getOwnerConnection(found.clientId); + // Reattach if: (a) same connection owns it, OR (b) old owner + // connection is gone (device restart gave us a new connectionId). + const ownerGone = + ownerConnection !== connectionId && !ctx.connRegistry.get(ownerConnection); + if ( + (ownerConnection === connectionId || ownerGone) && + !ctx.sessionRegistry.isAttached(found.clientId) + ) { + const transport = ctx.connRegistry.get(connectionId)?.transport; + if (transport) { + reattachChat(found.clientId, transport); + log.info('reattached detached session on reconnect', { + connectionId, + sessionId: entry.sessionId, + ownerGone, + }); + } } } } @@ -576,6 +590,9 @@ export function handleSendV2( // with seq <= lastProcessedSeq. The two HTTP requests (reconnect // POST and send POST) can arrive as separate event loop ticks in any // order, but duplicate delivery is always harmless thanks to seq dedup. + // TODO(P4): Unify cursor reset so watch() initializes the cursor from + // the client's lastSeq, removing the dependency on client-side seq dedup + // for the window between watch() and the reconnect POST arriving. ctx.connRegistry.setActive(connectionId, sessionId); sendToChat(activeClientId, prompt, msg.images, msg.contextBlocks, msg.clientMsgId); span.setAttribute('routing.decision', isOwner ? 'active' : 'takeover'); @@ -950,7 +967,7 @@ export async function dispatchV2Message( // Already handled at routing layer, ignore duplicate break; case 'reconnect': - // WS clients still send reconnect over WS (removed in P4). + // WS clients may also send reconnect over WS. handleReconnect(connectionId, msg, ctx); break; case 'watch':