diff --git a/packages/client/src/__tests__/sse-connection.test.ts b/packages/client/src/__tests__/sse-connection.test.ts index e00e62ae..184ee6f0 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,16 +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 }); - }); + 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(); conn.onMessage(listener); @@ -337,173 +329,21 @@ describe('SseConnection', () => { // Force reconnect conn.checkAndReconnect(true); - mockFetch.mockClear(); listener.mockClear(); - // New welcome — reconnect POST fires but doesn't resolve yet + // New welcome — NOT connected yet (reconnect POST in-flight) 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 }); - }); - 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); - - // 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 - 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 }); + // After POST resolves — now connected + await vi.waitFor(() => { + expect(conn.isConnected()).toBe(true); + expect(listener).toHaveBeenCalledWith({ type: '_open' }); }); - 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 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')); @@ -520,101 +360,64 @@ describe('SseConnection', () => { conn.checkAndReconnect(true); listener.mockClear(); + // Welcome — reconnect POST fires (and will fail) 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 }); + // Wait for POST to fail + await vi.waitFor(() => { + expect(mockFetch).toHaveBeenCalledWith( + expect.stringContaining('/reconnect'), + expect.any(Object), + ); }); - 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(); + // Still disconnected — sends stay queued until next successful reconnect 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 - expect(conn.isConnected()).toBe(true); - expect(listener).toHaveBeenCalledWith({ type: '_open' }); + expect(listener).not.toHaveBeenCalledWith({ type: '_open' }); }); - it('dispatches SSE events to listener while reconnect POST is in-flight', async () => { - let resolveReconnect!: (v: { ok: boolean }) => void; + 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((resolve) => { - resolveReconnect = resolve; + return new Promise<{ ok: boolean }>((resolve) => { + resolveReconnect = () => 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); conn.checkAndReconnect(true); - listener.mockClear(); + mockFetch.mockClear(); - // Welcome — reconnect POST in-flight, _connected = false + // Welcome — reconnect POST starts (held open) 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' }), + // 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), ); - resolveReconnect({ ok: true }); - await vi.runAllTimersAsync(); + // Resolve reconnect POST — sends should flush + resolveReconnect(); + await vi.waitFor(() => { + expect(mockFetch).toHaveBeenCalledWith(expect.stringContaining('/send'), expect.any(Object)); + }); }); - it('queued sends survive POST failure and flush on successful retry', async () => { - let callCount = 0; + 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/', ''); - 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,82 +426,55 @@ 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 fires, pending sends flush after it resolves 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']); - }); + // reconnect POST fires first + expect(postEndpoints).toEqual(['reconnect']); - 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 }); + // After the reconnect POST resolves, pending sends flush + await vi.waitFor(() => { + expect(postEndpoints).toEqual(['reconnect', 'send']); }); - 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; + it('SSE events still arrive after reconnect POST failure', async () => { const mockFetch = vi.fn().mockImplementation((url: string) => { if (url.includes('/reconnect')) { - return new Promise((resolve) => { - resolveReconnect = resolve; - }); + 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 — creates ES2 + // Force reconnect — reconnect POST will fail 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(); + // Wait for reconnect POST to fail + await vi.waitFor(() => { + expect(mockFetch).toHaveBeenCalledWith( + expect.stringContaining('/reconnect'), + expect.any(Object), + ); + }); - // ES3 hasn't welcomed yet, so _connected must remain false - expect(conn.isConnected()).toBe(false); + // 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', () => { @@ -714,45 +490,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 })); @@ -862,6 +599,75 @@ 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)); + + // 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' }); + 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 8ca85ed8..c179121a 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,8 @@ 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; + /** 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; @@ -47,7 +47,6 @@ export class SseConnection implements ChatConnection { constructor(config: SseConnectionConfig) { this.config = { createEventSource: (url: string) => new EventSource(url), - reconnectDelayMs: 500, suspendUrl: '', ...config, }; @@ -60,15 +59,12 @@ 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; } this._connected = false; + this._pendingReconnectSessions = null; } /** @@ -88,12 +84,12 @@ 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; } - // 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(); } @@ -150,7 +146,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; } @@ -175,7 +171,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 +188,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,13 +207,38 @@ 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; - if (this._isReconnect && this.seqBySession.size > 0) { - this.doReconnectPost(welcomeConnectionId, welcomeEs); + // 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. + // Refresh lastSeq from current map — SSE events may have advanced it + // since the original failure, avoiding unnecessary replay. + const sessions = this.getReconnectSessions(); + 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; + 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. + // 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(); @@ -262,70 +277,12 @@ 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; + if (!this._connectionId) { + throw new Error('doPost called without connectionId'); + } 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', @@ -333,9 +290,17 @@ export class SseConnection implements ChatConnection { }, body: JSON.stringify(body), }); - } catch { + if (!res.ok) { + throw new Error(`HTTP ${res.status}`); + } + } 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; } } @@ -344,7 +309,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(() => {}); } } @@ -381,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/packages/harness/__tests__/connection-registry.test.ts b/packages/harness/__tests__/connection-registry.test.ts index 6c4cd851..20c7cbbb 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; @@ -230,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); }); @@ -289,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', () => { @@ -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..9b7cb38f 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; @@ -140,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; @@ -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/packages/protocol/__tests__/event-store.test.ts b/packages/protocol/__tests__/event-store.test.ts index 8630b286..776fe690 100644 --- a/packages/protocol/__tests__/event-store.test.ts +++ b/packages/protocol/__tests__/event-store.test.ts @@ -851,6 +851,60 @@ describe('EventStore', () => { }); }); + describe('symposium schema migration', () => { + it('sessions default to session_type=chat', () => { + store.upsertSession({ sessionId: 'chat-1', summary: 'Normal chat' }); + const session = store.getSession('chat-1'); + expect(session!.sessionType).toBe('chat'); + }); + + it('persists session_type=symposium on insert', () => { + store.upsertSession({ sessionId: 'symp-1', summary: 'Symposium', sessionType: 'symposium' }); + const session = store.getSession('symp-1'); + expect(session!.sessionType).toBe('symposium'); + }); + + it('updates session_type on existing session', () => { + store.upsertSession({ sessionId: 'symp-2', summary: 'Test' }); + expect(store.getSession('symp-2')!.sessionType).toBe('chat'); + + store.upsertSession({ sessionId: 'symp-2', sessionType: 'symposium' }); + expect(store.getSession('symp-2')!.sessionType).toBe('symposium'); + }); + + it('persists symposium_config as JSON string', () => { + const config = JSON.stringify({ + seats: [ + { name: 'A', model: 'claude-opus-4-6', systemPrompt: '', color: '#000' }, + { name: 'B', model: 'gemini-2.5-pro', systemPrompt: '', color: '#FFF' }, + ], + turnRules: { mode: 'round-robin', maxTurns: 10 }, + interceptMode: 'auto', + }); + store.upsertSession({ + sessionId: 'symp-3', + sessionType: 'symposium', + symposiumConfig: config, + }); + + const session = store.getSession('symp-3'); + expect(session!.symposiumConfig).toBe(config); + const parsed = JSON.parse(session!.symposiumConfig!); + expect(parsed.seats).toHaveLength(2); + }); + + it('defaults symposium_config to null', () => { + store.upsertSession({ sessionId: 'chat-2', summary: 'Chat' }); + expect(store.getSession('chat-2')!.symposiumConfig).toBeNull(); + }); + + it('events table supports seat_id in payload', () => { + store.append('symp-4', 'block_delta', { delta: 'hello', seatId: 'seat-0' }); + const events = store.getSessionEvents('symp-4'); + expect(events[0].payload.seatId).toBe('seat-0'); + }); + }); + describe('close', () => { it('is safe to call multiple times', () => { store.close(); diff --git a/packages/protocol/__tests__/types.test.ts b/packages/protocol/__tests__/types.test.ts index a1ddb945..93372a76 100644 --- a/packages/protocol/__tests__/types.test.ts +++ b/packages/protocol/__tests__/types.test.ts @@ -15,6 +15,12 @@ import type { Session, StoredEvent, SessionMeta, + SessionType, + TurnMode, + InterceptMode, + SeatConfig, + TurnRules, + SymposiumConfig, } from '../src/types.js'; describe('protocol types', () => { @@ -206,4 +212,108 @@ describe('protocol types', () => { expect(meta.promptCount).toBe(3); expect(meta.totalCostUsd).toBe(0.01); }); + + // --- Symposium types --- + + it('SessionType accepts valid values', () => { + const types: SessionType[] = ['chat', 'symposium']; + expect(types).toHaveLength(2); + }); + + it('TurnMode accepts valid values', () => { + const modes: TurnMode[] = ['round-robin', 'directed', 'budgeted']; + expect(modes).toHaveLength(3); + }); + + it('InterceptMode accepts valid values', () => { + const modes: InterceptMode[] = ['auto', 'manual']; + expect(modes).toHaveLength(2); + }); + + it('SeatConfig has required fields', () => { + const seat: SeatConfig = { + name: 'Architect', + model: 'claude-opus-4-6', + systemPrompt: 'You are a senior architect.', + color: '#3B82F6', + }; + expect(seat.name).toBe('Architect'); + expect(seat.model).toBe('claude-opus-4-6'); + }); + + it('TurnRules has required fields and optional budgetUsd', () => { + const rules: TurnRules = { + mode: 'round-robin', + maxTurns: 10, + }; + expect(rules.mode).toBe('round-robin'); + expect(rules.budgetUsd).toBeUndefined(); + + const budgeted: TurnRules = { + mode: 'budgeted', + maxTurns: 20, + budgetUsd: 2.0, + }; + expect(budgeted.budgetUsd).toBe(2.0); + }); + + it('SymposiumConfig composes seats, turnRules, and interceptMode', () => { + const config: SymposiumConfig = { + seats: [ + { name: 'Architect', model: 'claude-opus-4-6', systemPrompt: 'Design.', color: '#3B82F6' }, + { name: 'Critic', model: 'gemini-2.5-pro', systemPrompt: 'Challenge.', color: '#EF4444' }, + ], + turnRules: { mode: 'round-robin', maxTurns: 10 }, + interceptMode: 'manual', + }; + expect(config.seats).toHaveLength(2); + expect(config.turnRules.mode).toBe('round-robin'); + expect(config.interceptMode).toBe('manual'); + }); + + it('SessionMeta includes sessionType and symposiumConfig', () => { + const meta: SessionMeta = { + sessionId: 's1', + summary: 'symposium test', + branch: null, + cwd: null, + mode: 'agent', + isActive: true, + isHidden: false, + promptCount: 0, + manuallyRenamed: false, + initialPrompt: null, + wtId: null, + inputTokens: 0, + outputTokens: 0, + cacheReadTokens: 0, + cacheCreationTokens: 0, + totalCostUsd: 0, + numTurns: 0, + durationMs: 0, + durationApiMs: 0, + goalId: null, + telosTaskId: null, + closedBy: null, + lastSpeaker: null, + lastSpeakerAt: null, + state: null, + lastStateChange: null, + agentName: null, + bootContext: null, + sessionType: 'symposium', + symposiumConfig: JSON.stringify({ + seats: [ + { name: 'A', model: 'claude-opus-4-6', systemPrompt: '', color: '#000' }, + { name: 'B', model: 'gemini-2.5-pro', systemPrompt: '', color: '#FFF' }, + ], + turnRules: { mode: 'round-robin', maxTurns: 10 }, + interceptMode: 'auto', + }), + createdAt: Date.now(), + updatedAt: Date.now(), + }; + expect(meta.sessionType).toBe('symposium'); + expect(meta.symposiumConfig).not.toBeNull(); + }); }); diff --git a/packages/protocol/src/event-store.ts b/packages/protocol/src/event-store.ts index 74271737..ca99a947 100644 --- a/packages/protocol/src/event-store.ts +++ b/packages/protocol/src/event-store.ts @@ -81,6 +81,8 @@ interface SessionRow { last_state_change: number | null; agent_name: string | null; boot_context: string | null; + session_type: string; + symposium_config: string | null; created_at: number; updated_at: number; } @@ -157,6 +159,7 @@ export class EventStore { this.migrateAttentionTracking(db); this.migrateSessionState(db); this.migrateBootContext(db); + this.migrateSymposium(db); this.migrateUserMessageIndex(db); this.log.info('EventStore initialized', { dbPath }); @@ -337,6 +340,19 @@ export class EventStore { } } + private migrateSymposium(db: Database.Database): void { + const columns = db.prepare("PRAGMA table_info('sessions')").all() as Array<{ name: string }>; + const columnNames = new Set(columns.map((c) => c.name)); + if (!columnNames.has('session_type')) { + db.exec("ALTER TABLE sessions ADD COLUMN session_type TEXT NOT NULL DEFAULT 'chat'"); + this.log.info('migrated sessions table: added session_type'); + } + if (!columnNames.has('symposium_config')) { + db.exec('ALTER TABLE sessions ADD COLUMN symposium_config TEXT'); + this.log.info('migrated sessions table: added symposium_config'); + } + } + private migrateUserMessageIndex(db: Database.Database): void { db.exec( `CREATE INDEX IF NOT EXISTS idx_events_user_msg_dedup @@ -427,6 +443,14 @@ export class EventStore { fields.push('boot_context = ?'); values.push(meta.bootContext); } + if (meta.sessionType !== undefined) { + fields.push('session_type = ?'); + values.push(meta.sessionType); + } + if (meta.symposiumConfig !== undefined) { + fields.push('symposium_config = ?'); + values.push(meta.symposiumConfig); + } if (meta.updatedAt !== undefined) { fields.push('updated_at = ?'); values.push(meta.updatedAt); @@ -452,6 +476,8 @@ export class EventStore { 'closed_by', 'agent_name', 'boot_context', + 'session_type', + 'symposium_config', ]; const vals: unknown[] = [ meta.sessionId, @@ -467,6 +493,8 @@ export class EventStore { meta.closedBy ?? null, meta.agentName ?? null, meta.bootContext ?? null, + meta.sessionType ?? 'chat', + meta.symposiumConfig ?? null, ]; if (meta.updatedAt !== undefined) { cols.push('updated_at'); @@ -781,6 +809,8 @@ function rowToSession(row: SessionRow): SessionMeta { lastStateChange: row.last_state_change ?? null, agentName: row.agent_name ?? null, bootContext: row.boot_context ?? null, + sessionType: (row.session_type as SessionMeta['sessionType']) ?? 'chat', + symposiumConfig: row.symposium_config ?? null, createdAt: row.created_at, updatedAt: row.updated_at, }; diff --git a/packages/protocol/src/index.ts b/packages/protocol/src/index.ts index 7c1040f7..3d4296d5 100644 --- a/packages/protocol/src/index.ts +++ b/packages/protocol/src/index.ts @@ -49,6 +49,12 @@ export type { AgentMemoryConfig, AgentOutput, AgentOutputConventions, + SessionType, + TurnMode, + InterceptMode, + SeatConfig, + TurnRules, + SymposiumConfig, } from './types.js'; // Constants diff --git a/packages/protocol/src/types.ts b/packages/protocol/src/types.ts index f7f4f031..5975903b 100644 --- a/packages/protocol/src/types.ts +++ b/packages/protocol/src/types.ts @@ -275,6 +275,47 @@ export interface ServiceHealthPayload { checkedAt: number; } +// --- Session types --- + +export type SessionType = 'chat' | 'symposium'; + +// --- Symposium types --- + +export type TurnMode = 'round-robin' | 'directed' | 'budgeted'; +export type InterceptMode = 'auto' | 'manual'; + +/** Configuration for a single seat in a symposium. */ +export interface SeatConfig { + /** Display name for this seat (e.g. 'Architect', 'Critic'). */ + name: string; + /** Canonical model name (e.g. 'claude-opus-4-6'). */ + model: string; + /** System prompt for this seat. */ + systemPrompt: string; + /** CSS color for visual differentiation in the UI. */ + color: string; +} + +/** Rules governing turn progression in a symposium. */ +export interface TurnRules { + /** How turns are assigned to seats. */ + mode: TurnMode; + /** Maximum number of turns before the symposium auto-stops. */ + maxTurns: number; + /** Optional cost ceiling in USD. */ + budgetUsd?: number; +} + +/** Full configuration for a symposium session. */ +export interface SymposiumConfig { + /** Participating seats (v1: exactly 2). */ + seats: SeatConfig[]; + /** Turn progression rules. */ + turnRules: TurnRules; + /** Whether the user intercepts each response before delivery. */ + interceptMode: InterceptMode; +} + // --- Event store types --- /** Optional logger interface — keeps the protocol package free of server dependencies. */ @@ -328,6 +369,10 @@ export interface SessionMeta { agentName: string | null; /** Serialized JSON of the boot_context payload (sources, tokens, sections). */ bootContext: string | null; + /** Session type: 'chat' (default) or 'symposium'. */ + sessionType: SessionType; + /** Serialized JSON of the SymposiumConfig (null for chat sessions). */ + symposiumConfig: string | null; createdAt: number; updatedAt: number; } diff --git a/packages/protocol/src/ws-schemas-v2.ts b/packages/protocol/src/ws-schemas-v2.ts index 8fc07d35..648c3d5e 100644 --- a/packages/protocol/src/ws-schemas-v2.ts +++ b/packages/protocol/src/ws-schemas-v2.ts @@ -124,6 +124,8 @@ export const V2SetModeMessage = z.object({ // ─── 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, 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'), diff --git a/server/__tests__/ws-handler-v2.test.ts b/server/__tests__/ws-handler-v2.test.ts index bad44d60..2d3c755e 100644 --- a/server/__tests__/ws-handler-v2.test.ts +++ b/server/__tests__/ws-handler-v2.test.ts @@ -242,13 +242,14 @@ describe('handleReconnect', () => { ]); }); - it('reattaches detached session on reconnect', () => { + 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); + sessionReg.isAttached.mockReturnValue(false); // detached — reconnect should reattach const ctx = createContext({ sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], @@ -263,29 +264,32 @@ describe('handleReconnect', () => { ); expect(reattachChat).toHaveBeenCalledWith('c1:sess-1', transport); + expect(rekeyChat).not.toHaveBeenCalled(); }); - it('does not reattach if session is already attached', () => { + it('reattaches detached session when owner connection is gone (device restart)', () => { (reattachChat as ReturnType).mockClear(); const sessionReg = mockSessionRegistry(); - sessionReg.findBySessionId.mockReturnValue({ clientId: 'c1:sess-1' }); + // 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(true); // already attached + sessionReg.isAttached.mockReturnValue(false); const ctx = createContext({ sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], }); const transport = mockTransport(); - ctx.connRegistry.register('c1', transport); + // Register new connection 'c-new' — 'c-old' is NOT registered (gone) + ctx.connRegistry.register('c-new', transport); handleReconnect( - 'c1', + 'c-new', { type: 'reconnect', sessions: [{ sessionId: 'sess-1', lastSeq: 0 }] }, ctx, ); - expect(reattachChat).not.toHaveBeenCalled(); + expect(reattachChat).toHaveBeenCalledWith('c-old:sess-1', transport); }); it('resets cursor to client lastSeq immediately after watch (before replay)', () => { @@ -1086,17 +1090,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 +1111,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', () => { @@ -1574,7 +1575,7 @@ describe('dispatchV2Message', () => { expect(stopChat).toHaveBeenCalledWith('driver-1'); }); - it('routes reconnect messages and produces reconnected summary', async () => { + it('handles reconnect messages over WS (until P4 removes WS transport)', async () => { const ctx = createContext(); const transport = mockTransport(); ctx.connRegistry.register('c1', transport); @@ -1589,7 +1590,8 @@ describe('dispatchV2Message', () => { ctx, ); - expect(transport.sent).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 () => { @@ -2150,81 +2152,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 +2560,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', () => { @@ -2771,13 +2649,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({ @@ -2793,7 +2670,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', () => { @@ -2951,31 +2829,26 @@ describe('handleSessionSuspend', () => { // ─── handleReconnect — suspend resume ─────────────────────────────────────── describe('handleReconnect suspend resume', () => { - it('replays buffered events for suspended sessions', () => { + it('clears suspend state and sends session_resumed', () => { + (reattachChat as ReturnType).mockClear(); + 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 +2857,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/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) { diff --git a/server/index.ts b/server/index.ts index 06b7ae64..37e86b2e 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, '..', '..'); @@ -992,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'); @@ -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 32d9d5a1..f3986d52 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); @@ -247,50 +247,55 @@ 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). + // 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); 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', { + + // 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)) { + 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, - oldClientId: found.clientId, - newClientId, + ownerGone, }); } - log.info('reattached detached session on reconnect', { - connectionId, - sessionId: entry.sessionId, - clientId: newClientId, - ownerGone, - }); } } } @@ -299,6 +304,7 @@ export function handleReconnect( // 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 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); @@ -576,6 +582,17 @@ 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, + // 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. + // 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,6 +967,7 @@ export async function dispatchV2Message( // Already handled at routing layer, ignore duplicate break; case 'reconnect': + // WS clients may also send reconnect over WS. handleReconnect(connectionId, msg, ctx); break; case 'watch':