From 2e45031b74b14e52587249caf78696b1f0ed65ff Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 4 Jul 2026 10:56:08 +0100 Subject: [PATCH 01/10] 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 b1839750d36e5902c8fa07026f89bf8bf3a253a0 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 4 Jul 2026 12:04:39 +0100 Subject: [PATCH 02/10] 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 77d0bc7e..fd4e65a4 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 82fa0fcd947824bc99cbacfa42b62a1cb74a8266 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 4 Jul 2026 12:08:07 +0100 Subject: [PATCH 03/10] 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 ec58a674549e88f0bc047eae21aa8f7374db07eb Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 4 Jul 2026 12:08:29 +0100 Subject: [PATCH 04/10] 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 4e04c2f0e57375b546c368ad50a3b1db558e19be Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 4 Jul 2026 12:12:57 +0100 Subject: [PATCH 05/10] 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 c00e233477136ec78e9b823d67519e8368ee9b2c Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 15:34:13 +0100 Subject: [PATCH 06/10] =?UTF-8?q?refactor(client):=20remove=20client-side?= =?UTF-8?q?=20message=20queueing=20=E2=80=94=20server=20decides=20(P2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove wasRunning/pendingSend/PENDING_SEND_TIMEOUT_MS routing heuristics from the client. Messages are now always sent immediately; the server deduplicates via clientMsgId (storeAndEchoIfNew) and queues follow-ups in the SDK's inputQueue. Co-Authored-By: Claude Opus 4.6 --- .../client/__tests__/protocol-parser.test.ts | 26 ++--- packages/client/__tests__/store.test.ts | 72 +----------- packages/client/src/protocol-parser.ts | 19 ---- packages/client/src/store.ts | 106 ++++-------------- 4 files changed, 35 insertions(+), 188 deletions(-) diff --git a/packages/client/__tests__/protocol-parser.test.ts b/packages/client/__tests__/protocol-parser.test.ts index b6e5dd1d..320db51b 100644 --- a/packages/client/__tests__/protocol-parser.test.ts +++ b/packages/client/__tests__/protocol-parser.test.ts @@ -5,7 +5,6 @@ import type { ProtocolCallbacks, ProtocolParserState } from '../src/protocol-par function makeState(overrides?: Partial): ProtocolParserState { return { currentSessionId: undefined, - pendingSend: [], ...overrides, }; } @@ -157,21 +156,12 @@ describe('session lifecycle', () => { ]); }); - it('session_end dequeues first pending send and queues it', () => { - const state = makeState({ - pendingSend: [ - { type: 'send', prompt: 'follow-up' }, - { type: 'send', prompt: 'second' }, - ], - }); + it('session_end dispatches SESSION_END action', () => { const cb = makeCallbacks(); - const r = parseServerMessage({ type: 'session_end', sessionId: 'sid' }, state, cb, POOL_KEY); + const r = parseServerMessage({ type: 'session_end', sessionId: 'sid' }, makeState(), cb, POOL_KEY); expect(r.messagesActions).toContainEqual({ type: 'SESSION_END', sessionId: 'sid' }); - // Optimistic running=true when draining pending send - expect(r.messagesActions).toContainEqual({ type: 'SESSION_STATE_CHANGED', state: 'running' }); - expect(cb.sendQueued).toHaveBeenCalledWith(POOL_KEY, { type: 'send', prompt: 'follow-up' }); - // Second message stays queued - expect(state.pendingSend).toEqual([{ type: 'send', prompt: 'second' }]); + // No client-side pending send drain — server handles message queueing + expect(cb.sendQueued).not.toHaveBeenCalled(); }); }); @@ -378,10 +368,10 @@ describe('error handling', () => { expect(r.messagesActions).toEqual([{ type: 'ERROR', error: 'Something broke' }]); }); - it('error clears pendingSend queue', () => { - const state = makeState({ pendingSend: [{ type: 'send', prompt: 'test' }] }); - parseServerMessage({ type: 'error', error: 'fail' }, state, makeCallbacks(), POOL_KEY); - expect(state.pendingSend).toEqual([]); + it('error does not require pendingSend cleanup (removed in P2)', () => { + const state = makeState(); + const r = parseServerMessage({ type: 'error', error: 'fail' }, state, makeCallbacks(), POOL_KEY); + expect(r.messagesActions).toContainEqual({ type: 'ERROR', error: 'fail' }); }); }); diff --git a/packages/client/__tests__/store.test.ts b/packages/client/__tests__/store.test.ts index 2fba7283..2f2bb803 100644 --- a/packages/client/__tests__/store.test.ts +++ b/packages/client/__tests__/store.test.ts @@ -402,89 +402,23 @@ describe('sendMessage', () => { ); }); - it('queues second message while first turn is running', async () => { + it('sends second message immediately while first turn is running (server dedup)', async () => { const store = createReadyStore(); store.getState().sendMessage('first'); lastWs.simulateMessage({ type: 'session_id', sessionId: 'sess-q' }); - // Turn is still running — second send should queue + // Turn is still running — P2: second send goes immediately (server deduplicates) const sentBefore = lastWs.sent.length; store.getState().sendMessage('second'); - // Should NOT have sent yet - const newSendsImmediate = lastWs.sent.slice(sentBefore).map((s) => JSON.parse(s)); - expect(newSendsImmediate.filter((m) => m.type === 'send')).toHaveLength(0); - - // session_end triggers flush of queued message - lastWs.simulateMessage({ type: 'session_end', sessionId: 'sess-q' }); - const newSends = lastWs.sent.slice(sentBefore).map((s) => JSON.parse(s)); expect(newSends).toContainEqual(expect.objectContaining({ type: 'send', prompt: 'second' })); - }); - - it('queues second message as pendingSend while first turn is active', () => { - const store = createReadyStore(); - - store.getState().sendMessage('first'); - lastWs.simulateMessage({ type: 'session_id', sessionId: 'sess-pend' }); - - const sentBefore = lastWs.sent.length; - store.getState().sendMessage('second'); - // Second message should be queued, not immediately sent - const immediateSends = lastWs.sent - .slice(sentBefore) - .filter((s) => JSON.parse(s).type === 'send'); - expect(immediateSends).toHaveLength(0); - - // But the optimistic user message should appear in the store + // Optimistic user message should appear in the store const userMsgs = store.getState().messages.messages.filter((m) => m.role === 'user'); expect(userMsgs).toHaveLength(2); }); - - it('cancels pending timeout when session_end arrives in time', () => { - vi.useFakeTimers(); - try { - const store = createMitzoStore(makeOptions()); - lastWs.completeHandshake(); - - store.getState().sendMessage('first'); - lastWs.simulateMessage({ type: 'session_id', sessionId: 'sess-ok' }); - - store.getState().sendMessage('second'); - lastWs.simulateMessage({ type: 'session_end', sessionId: 'sess-ok' }); - - const sentAfterEnd = lastWs.sent.length; - vi.advanceTimersByTime(6_000); - - expect(lastWs.sent.length).toBe(sentAfterEnd); - } finally { - vi.useRealTimers(); - } - }); - - it('cancels pending timeout on newSession', () => { - vi.useFakeTimers(); - try { - const store = createMitzoStore(makeOptions()); - lastWs.completeHandshake(); - - store.getState().sendMessage('first'); - lastWs.simulateMessage({ type: 'session_id', sessionId: 'sess-new' }); - - store.getState().sendMessage('second'); - const sentBefore = lastWs.sent.length; - - store.getState().newSession(); - vi.advanceTimersByTime(6_000); - - const flushed = lastWs.sent.slice(sentBefore).filter((s) => JSON.parse(s).type === 'send'); - expect(flushed).toHaveLength(0); - } finally { - vi.useRealTimers(); - } - }); }); describe('WS → store wiring', () => { diff --git a/packages/client/src/protocol-parser.ts b/packages/client/src/protocol-parser.ts index bf86fb19..bd5a1e85 100644 --- a/packages/client/src/protocol-parser.ts +++ b/packages/client/src/protocol-parser.ts @@ -50,9 +50,6 @@ export interface ProtocolCallbacks { /** @deprecated v1 only — called to send a queued message after session_end. */ sendQueued?(poolKey: string, msg: unknown): void; - /** v2: Called when a queued message should be sent after session_end. */ - onSendQueued?(msg: Record): void; - /** v2: Called with token data from session_switched response. */ onTokensHydrated?(tokens: Record): void; @@ -65,9 +62,6 @@ export interface ProtocolCallbacks { export interface ProtocolParserState { /** Currently tracked session ID (used for expiry detection). */ currentSessionId: string | undefined; - - /** Queued messages to send after current session ends (FIFO). */ - pendingSend: Record[]; } // ─── Parser result ─────────────────────────────────────────────────────────── @@ -364,18 +358,6 @@ export function parseServerMessage( if (msg.sessionId && !state.currentSessionId) { callbacks.onSessionAssigned(msg.sessionId as string); } - // Drain first queued message. Optimistic running=true avoids UI flicker - // between the send and the server's session_state_changed confirmation. - const pending = state.pendingSend.shift(); - if (pending) { - result.messagesActions.push({ type: 'SESSION_STATE_CHANGED', state: 'running' }); - if (callbacks.onSendQueued) { - callbacks.onSendQueued(pending); - } else { - callbacks.setWsRunning?.(poolKey, true); - callbacks.sendQueued?.(poolKey, pending); - } - } break; } @@ -450,7 +432,6 @@ export function parseServerMessage( const errorMsg = msg.error as string; callbacks.setWsRunning?.(poolKey, false); - state.pendingSend = []; result.messagesActions.push({ type: 'ERROR', error: errorMsg || 'Unknown error', diff --git a/packages/client/src/store.ts b/packages/client/src/store.ts index acc99340..0a13f82d 100644 --- a/packages/client/src/store.ts +++ b/packages/client/src/store.ts @@ -172,8 +172,6 @@ function removeTaskFromTree(tasks: Task[], id: string): Task[] { }); } -const PENDING_SEND_TIMEOUT_MS = 5_000; - // ─── Factory ───────────────────────────────────────────────────────────────── export function createMitzoStore(options: MitzoStoreOptions): StoreApi { @@ -182,11 +180,8 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi; - } = { + const parserState: ProtocolParserState = { currentSessionId: undefined, - pendingSend: [], }; let recoveryInFlight = false; @@ -219,13 +214,6 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi((set, get) => ({ // ── Initial state ──────────────────────────────────────────────────── @@ -260,8 +248,6 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi ({ @@ -292,8 +278,6 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi => { - const msg: Record = { - type: 'send', - sessionId: parserState.currentSessionId ?? null, - prompt: text, - clientMsgId, - }; - const model = opts?.model ?? get().config.modelId; - const mode = opts?.mode ?? get().config.mode; - if (model) msg.model = model; - if (mode) msg.mode = mode; - if (opts?.contextBlocks?.length) msg.contextBlocks = opts.contextBlocks; - if (opts?.images?.length) { - msg.images = opts.images.map((img) => ({ data: img.data, mediaType: img.mediaType })); - } - if (opts?.cwd) msg.cwd = opts.cwd; - if (opts?.extraTools) msg.extraTools = opts.extraTools; - if (opts?.isolation !== undefined) msg.isolation = opts.isolation; - if (opts?.telosTaskId !== undefined) msg.telosTaskId = opts.telosTaskId; - if (opts?.agentName !== undefined) msg.agentName = opts.agentName; - return msg; + const msg: Record = { + type: 'send', + sessionId: parserState.currentSessionId ?? null, + prompt: text, + clientMsgId, }; + const model = opts?.model ?? get().config.modelId; + const mode = opts?.mode ?? get().config.mode; + if (model) msg.model = model; + if (mode) msg.mode = mode; + if (opts?.contextBlocks?.length) msg.contextBlocks = opts.contextBlocks; + if (opts?.images?.length) { + msg.images = opts.images.map((img) => ({ data: img.data, mediaType: img.mediaType })); + } + if (opts?.cwd) msg.cwd = opts.cwd; + if (opts?.extraTools) msg.extraTools = opts.extraTools; + if (opts?.isolation !== undefined) msg.isolation = opts.isolation; + if (opts?.telosTaskId !== undefined) msg.telosTaskId = opts.telosTaskId; + if (opts?.agentName !== undefined) msg.agentName = opts.agentName; set((s) => ({ messages: messagesReducer(s.messages, { @@ -348,39 +328,11 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi ({ - messages: messagesReducer(s.messages, { - type: 'SESSION_STATE_CHANGED', - state: 'running', - }), - })); - connection.send(pending); - // Reschedule for remaining queued messages - if (parserState.pendingSend.length > 0) { - parserState.pendingSendTimer = setTimeout(drainOne, PENDING_SEND_TIMEOUT_MS); - } else { - parserState.pendingSendTimer = undefined; - } - }, PENDING_SEND_TIMEOUT_MS); - } else { - const sent = connection.send(msg); - if (!sent) { - set({ sendError: 'Not connected. Message will be sent when reconnected.' }); - } + // Always send immediately — server deduplicates via clientMsgId + // (storeAndEchoIfNew) and queues in the SDK's inputQueue. + const sent = connection.send(msg); + if (!sent) { + set({ sendError: 'Not connected. Message will be sent when reconnected.' }); } }, @@ -679,10 +631,6 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi) { - connection.send(msg); - }, - onReconnected() { const activeId = parserState.currentSessionId; if (activeId) fetchAndRestoreMessages(activeId); @@ -747,12 +695,6 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi ({ messages: messagesReducer(s.messages, action), From c99a9da1e75722e86836b3dfe65d9ba5473f1124 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 15:42:28 +0100 Subject: [PATCH 07/10] =?UTF-8?q?fix(transport):=20address=20Centaur=20rev?= =?UTF-8?q?iew=20=E2=80=94=20stale=20comment,=20dead=20code,=20test=20gap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix stale comment in server/index.ts (periodic sync removed) - Remove dead `case 'reconnect'` in ws-handler-v2.ts (schema rejects before switch) - Add test: flushPendingSends runs even when reconnect POST rejects Co-Authored-By: Claude Opus 4.6 --- .../src/__tests__/sse-connection.test.ts | 27 +++++++++++++++++++ server/index.ts | 2 +- server/ws-handler-v2.ts | 4 +-- 3 files changed, 29 insertions(+), 4 deletions(-) diff --git a/packages/client/src/__tests__/sse-connection.test.ts b/packages/client/src/__tests__/sse-connection.test.ts index 88111d8a..605953d2 100644 --- a/packages/client/src/__tests__/sse-connection.test.ts +++ b/packages/client/src/__tests__/sse-connection.test.ts @@ -386,6 +386,33 @@ describe('SseConnection', () => { expect(postEndpoints).toEqual(['reconnect', 'send']); }); + it('flushes pending sends even when reconnect POST rejects', () => { + 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 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 — queue a send while disconnected + conn.checkAndReconnect(true); + conn.send({ type: 'send', prompt: 'queued msg', clientMsgId: 'q-1' }); + postEndpoints.length = 0; + + // Welcome — reconnect POST fires (will reject), but queued send still flushes + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); + + // Both POSTs fired: reconnect (fire-and-forget) + queued send + expect(postEndpoints).toEqual(['reconnect', 'send']); + }); + it('does not emit _close when checkAndReconnect called while already disconnected', () => { const conn = new SseConnection(createConfig()); const listener = vi.fn(); diff --git a/server/index.ts b/server/index.ts index fd4e65a4..06617d00 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 connection 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 eb32690e..0c526d0c 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -912,9 +912,7 @@ 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; + // 'reconnect' removed — handled via REST POST; schema rejects it before reaching here. case 'watch': handleWatch(connectionId, msg, ctx); break; From fdfc04b0a724062aa88a2c733a93085d95ef246c Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 16:03:19 +0100 Subject: [PATCH 08/10] fix(transport): address second Centaur review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix stale JSDoc in connection-registry.ts (periodic sync → reconnect replay) - Remove dead sendQueued v1 callback from ProtocolCallbacks - Add test: SSE events dispatch correctly when reconnect POST fails Co-Authored-By: Claude Opus 4.6 --- .../client/__tests__/protocol-parser.test.ts | 3 -- .../src/__tests__/sse-connection.test.ts | 33 +++++++++++++++++++ packages/client/src/protocol-parser.ts | 3 -- packages/harness/src/connection-registry.ts | 2 +- 4 files changed, 34 insertions(+), 7 deletions(-) diff --git a/packages/client/__tests__/protocol-parser.test.ts b/packages/client/__tests__/protocol-parser.test.ts index 320db51b..d48b6462 100644 --- a/packages/client/__tests__/protocol-parser.test.ts +++ b/packages/client/__tests__/protocol-parser.test.ts @@ -16,7 +16,6 @@ function makeCallbacks(overrides?: Partial): ProtocolCallback onMessagesRestored: vi.fn(), onSessionRenamed: vi.fn(), setWsRunning: vi.fn(), - sendQueued: vi.fn(), ...overrides, }; } @@ -160,8 +159,6 @@ describe('session lifecycle', () => { const cb = makeCallbacks(); const r = parseServerMessage({ type: 'session_end', sessionId: 'sid' }, makeState(), cb, POOL_KEY); expect(r.messagesActions).toContainEqual({ type: 'SESSION_END', sessionId: 'sid' }); - // No client-side pending send drain — server handles message queueing - expect(cb.sendQueued).not.toHaveBeenCalled(); }); }); diff --git a/packages/client/src/__tests__/sse-connection.test.ts b/packages/client/src/__tests__/sse-connection.test.ts index 605953d2..b16776fb 100644 --- a/packages/client/src/__tests__/sse-connection.test.ts +++ b/packages/client/src/__tests__/sse-connection.test.ts @@ -413,6 +413,39 @@ describe('SseConnection', () => { expect(postEndpoints).toEqual(['reconnect', 'send']); }); + it('dispatches SSE events to listener even when reconnect POST fails', 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 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(); + + // Reconnect welcome — POST will fail, but SSE events should still dispatch + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); + + // Simulate an SSE event arriving after the failed reconnect POST + lastES()._emit('message', { + type: 'session_state_changed', + sessionId: 'sess-1', + state: 'running', + }); + + expect(listener).toHaveBeenCalledWith(expect.objectContaining({ type: '_open' })); + expect(listener).toHaveBeenCalledWith( + expect.objectContaining({ type: 'session_state_changed', state: 'running' }), + ); + }); + it('does not emit _close when checkAndReconnect called while already disconnected', () => { const conn = new SseConnection(createConfig()); const listener = vi.fn(); diff --git a/packages/client/src/protocol-parser.ts b/packages/client/src/protocol-parser.ts index bd5a1e85..e45914b5 100644 --- a/packages/client/src/protocol-parser.ts +++ b/packages/client/src/protocol-parser.ts @@ -47,9 +47,6 @@ export interface ProtocolCallbacks { /** @deprecated v1 only — called to mark the WS pool entry as running/not-running. */ setWsRunning?(poolKey: string, running: boolean): void; - /** @deprecated v1 only — called to send a queued message after session_end. */ - sendQueued?(poolKey: string, msg: unknown): void; - /** v2: Called with token data from session_switched response. */ onTokensHydrated?(tokens: Record): void; diff --git a/packages/harness/src/connection-registry.ts b/packages/harness/src/connection-registry.ts index 94fd319d..5b2d9341 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 can cover gaps. */ broadcast(sessionId: string, data: Record): void { const seq = data.seq as number | undefined; From 5214e33113d210c26cfc04e32468b275b80d36f9 Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 16:12:33 +0100 Subject: [PATCH 09/10] fix(transport): add seq-based SSE dedup and eager reattach on reconnect SSE onmessage now skips events with seq <= lastSeq to prevent duplicate BLOCK_DELTA concatenation after reconnect POST failure. handleReconnect eagerly reattaches detached sessions to cancel the detach TTL timer. Co-Authored-By: Claude Opus 4.6 --- .../src/__tests__/sse-connection.test.ts | 68 +++++++++++++++++++ packages/client/src/sse-connection.ts | 7 ++ server/ws-handler-v2.ts | 21 ++++-- 3 files changed, 92 insertions(+), 4 deletions(-) diff --git a/packages/client/src/__tests__/sse-connection.test.ts b/packages/client/src/__tests__/sse-connection.test.ts index b16776fb..08973ad7 100644 --- a/packages/client/src/__tests__/sse-connection.test.ts +++ b/packages/client/src/__tests__/sse-connection.test.ts @@ -446,6 +446,74 @@ describe('SseConnection', () => { ); }); + it('deduplicates SSE events by seq (skips already-seen seq numbers)', () => { + const conn = new SseConnection(createConfig()); + const listener = vi.fn(); + conn.onMessage(listener); + conn.connect(); + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' }); + listener.mockClear(); + + // First delivery — should pass through + lastES()._emit('message', { + type: 'block_delta', + sessionId: 'sess-1', + seq: 5, + delta: 'hello', + }); + expect(listener).toHaveBeenCalledTimes(1); + + // Duplicate delivery (same seq) — should be skipped + lastES()._emit('message', { + type: 'block_delta', + sessionId: 'sess-1', + seq: 5, + delta: 'hello', + }); + expect(listener).toHaveBeenCalledTimes(1); + + // Old seq — should also be skipped + lastES()._emit('message', { + type: 'block_delta', + sessionId: 'sess-1', + seq: 3, + delta: 'old', + }); + expect(listener).toHaveBeenCalledTimes(1); + + // New seq — should pass through + lastES()._emit('message', { + type: 'block_delta', + sessionId: 'sess-1', + seq: 6, + delta: 'world', + }); + expect(listener).toHaveBeenCalledTimes(2); + }); + + it('reconnect POST fires before flushPendingSends (ordering guarantee)', () => { + const callOrder: string[] = []; + const mockFetch = vi.fn().mockImplementation((url: string) => { + const endpoint = url.replace('https://localhost:3100/api/chat/', ''); + callOrder.push(endpoint); + 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); + conn.send({ type: 'send', prompt: 'queued', clientMsgId: 'q-1' }); + callOrder.length = 0; + + lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-def' }); + + // reconnect MUST fire before the queued send + expect(callOrder[0]).toBe('reconnect'); + expect(callOrder[1]).toBe('send'); + }); + it('does not emit _close when checkAndReconnect called while already disconnected', () => { const conn = new SseConnection(createConfig()); const listener = vi.fn(); diff --git a/packages/client/src/sse-connection.ts b/packages/client/src/sse-connection.ts index e60e67e6..3e8566cc 100644 --- a/packages/client/src/sse-connection.ts +++ b/packages/client/src/sse-connection.ts @@ -232,7 +232,14 @@ export class SseConnection implements ChatConnection { return; } + // Seq-based dedup: if the reconnect POST fails, the server may re-deliver + // events using a stale cursor. The frontend reducer is not block-level + // idempotent (BLOCK_DELTA concatenates), so skip already-seen events. if (typeof msg.seq === 'number' && typeof msg.sessionId === 'string') { + const lastSeq = this.seqBySession.get(msg.sessionId as string); + if (lastSeq !== undefined && (msg.seq as number) <= lastSeq) { + return; // Already seen — skip duplicate + } this.seqBySession.set(msg.sessionId as string, msg.seq as number); } diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index 0c526d0c..a49cd2b7 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -252,15 +252,28 @@ 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. + // Eagerly reattach detached sessions to cancel the detach TTL timer. + // Without this, the TTL keeps ticking until the first user message + // triggers handleSendV2's reattach — if the user reconnects but doesn't + // send a message quickly, the session could be aborted despite being + // connected and watching events. + const found = ctx.sessionRegistry.findBySessionId(entry.sessionId); + if (found) { + const transport = ctx.connRegistry.get(connectionId)?.transport; + if (transport && !ctx.sessionRegistry.isAttached(found.clientId)) { + reattachChat(found.clientId, transport); + log.info('reattached detached session on reconnect', { + connectionId, + sessionId: entry.sessionId, + clientId: found.clientId, + }); + } + } // 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)) { From 1d244af6c5fd37e3f43f122b6af367931f26090b Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 25 Jul 2026 18:27:32 +0100 Subject: [PATCH 10/10] style(client): fix Prettier formatting in protocol-parser test Co-Authored-By: Claude Opus 4.6 --- packages/client/__tests__/protocol-parser.test.ts | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/packages/client/__tests__/protocol-parser.test.ts b/packages/client/__tests__/protocol-parser.test.ts index d48b6462..6b22f19a 100644 --- a/packages/client/__tests__/protocol-parser.test.ts +++ b/packages/client/__tests__/protocol-parser.test.ts @@ -157,7 +157,12 @@ describe('session lifecycle', () => { it('session_end dispatches SESSION_END action', () => { const cb = makeCallbacks(); - const r = parseServerMessage({ type: 'session_end', sessionId: 'sid' }, makeState(), cb, POOL_KEY); + const r = parseServerMessage( + { type: 'session_end', sessionId: 'sid' }, + makeState(), + cb, + POOL_KEY, + ); expect(r.messagesActions).toContainEqual({ type: 'SESSION_END', sessionId: 'sid' }); }); }); @@ -367,7 +372,12 @@ describe('error handling', () => { it('error does not require pendingSend cleanup (removed in P2)', () => { const state = makeState(); - const r = parseServerMessage({ type: 'error', error: 'fail' }, state, makeCallbacks(), POOL_KEY); + const r = parseServerMessage( + { type: 'error', error: 'fail' }, + state, + makeCallbacks(), + POOL_KEY, + ); expect(r.messagesActions).toContainEqual({ type: 'ERROR', error: 'fail' }); }); });