From 459283ad2047afe4e911f5b9ccf3488faa073f8e Mon Sep 17 00:00:00 2001 From: dimakis Date: Fri, 26 Jun 2026 14:34:31 +0100 Subject: [PATCH] fix(server): report idle sessions as not-running on reconnect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit handleReconnect conflated "query loop alive" with "agent actively processing a turn". isActive() returns true for sessions whose query loop is alive but idle between turns (waiting in `for await`). This caused the reconnected message to report running=true, making the frontend queue the first user message behind a 5-second fallback timer instead of sending it immediately — the "two messages to get a response after reattach" bug. Fix: check lastSpeaker from EventStore. When lastSpeaker=assistant the agent completed its last turn and is idle, so report running=false. When lastSpeaker=user the agent hasn't answered yet, so running=true. Co-Authored-By: Claude Opus 4.6 --- server/__tests__/ws-handler-v2.test.ts | 98 ++++++++++++++++++++++++++ server/ws-handler-v2.ts | 16 +++++ 2 files changed, 114 insertions(+) diff --git a/server/__tests__/ws-handler-v2.test.ts b/server/__tests__/ws-handler-v2.test.ts index bad44d60..884e2192 100644 --- a/server/__tests__/ws-handler-v2.test.ts +++ b/server/__tests__/ws-handler-v2.test.ts @@ -1114,6 +1114,104 @@ describe('handleReconnect reconnected summary (P1)', () => { expect(reattachChat).not.toHaveBeenCalled(); }); + it('reports running=false when session is active but idle (lastSpeaker=assistant)', () => { + const sessionReg = mockSessionRegistry(); + sessionReg.findBySessionId.mockReturnValue({ clientId: 'driver-1' }); + sessionReg.isActive.mockReturnValue(true); + sessionReg.isAttached.mockReturnValue(false); + + const eventStore = mockEventStore(); + eventStore.getSession.mockReturnValue({ isActive: true, lastSpeaker: 'assistant' }); + + const ctx = createContext({ + sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], + eventStore: eventStore as unknown as V2HandlerContext['eventStore'], + }); + const transport = mockTransport(); + ctx.connRegistry.register('c1', transport); + + handleReconnect( + 'c1', + { type: 'reconnect', sessions: [{ sessionId: 'sess-idle', lastSeq: 0 }] }, + ctx, + ); + + const summary = transport.sent.find((m) => m.type === 'reconnected') as { + sessions: Array<{ sessionId: string; running: boolean }>; + }; + expect(summary.sessions[0].running).toBe(false); + }); + + it('reports running=true when session is active and mid-turn (lastSpeaker=user)', () => { + const sessionReg = mockSessionRegistry(); + sessionReg.findBySessionId.mockReturnValue({ clientId: 'driver-1' }); + sessionReg.isActive.mockReturnValue(true); + sessionReg.isAttached.mockReturnValue(false); + + const eventStore = mockEventStore(); + eventStore.getSession.mockReturnValue({ isActive: true, lastSpeaker: 'user' }); + + const ctx = createContext({ + sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], + eventStore: eventStore as unknown as V2HandlerContext['eventStore'], + }); + const transport = mockTransport(); + ctx.connRegistry.register('c1', transport); + + handleReconnect( + 'c1', + { type: 'reconnect', sessions: [{ sessionId: 'sess-busy', lastSeq: 0 }] }, + ctx, + ); + + const summary = transport.sent.find((m) => m.type === 'reconnected') as { + sessions: Array<{ sessionId: string; running: boolean }>; + }; + expect(summary.sessions[0].running).toBe(true); + }); + + it('handles mixed running states across multiple sessions', () => { + const sessionReg = mockSessionRegistry(); + sessionReg.findBySessionId + .mockReturnValueOnce({ clientId: 'driver-1' }) + .mockReturnValueOnce(null) + .mockReturnValueOnce({ clientId: 'driver-3' }); + sessionReg.isActive.mockReturnValueOnce(true).mockReturnValueOnce(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 }, + { sessionId: 'sess-2', lastSeq: 0 }, + { sessionId: 'sess-3', lastSeq: 0 }, + ], + }, + ctx, + ); + + const summary = transport.sent.find((m) => m.type === 'reconnected') as { + sessions: Array<{ sessionId: string; running: boolean }>; + }; + expect(summary.sessions).toHaveLength(3); + expect(summary.sessions[0]).toEqual( + expect.objectContaining({ sessionId: 'sess-1', running: true }), + ); + expect(summary.sessions[1]).toEqual( + expect.objectContaining({ sessionId: 'sess-2', running: false }), + ); + expect(summary.sessions[2]).toEqual( + expect.objectContaining({ sessionId: 'sess-3', running: false }), + ); + }); + it('replays multiple events in sequence order', () => { const eventStore = mockEventStore(); eventStore.getEventsAfter.mockReturnValue([ diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index 32d9d5a1..507edbf0 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -267,6 +267,22 @@ export function handleReconnect( }); ctx.sessionRegistry.remove(found!.clientId); } + // Distinguish "query loop alive but idle" from "agent actively + // processing a turn". lastSpeaker === 'assistant' means the agent + // completed its last turn and is waiting for input — report as + // not-running so the client sends messages immediately instead of + // queuing them behind a 5-second fallback timer. + if (running) { + const storeMeta = ctx.eventStore.getSession(entry.sessionId); + if (storeMeta?.lastSpeaker === 'assistant') { + running = false; + log.info('session alive but idle (last speaker: assistant)', { + connectionId, + sessionId: entry.sessionId, + clientId: found!.clientId, + }); + } + } if (found && running && !ctx.sessionRegistry.isAttached(found.clientId)) { const ownerConnection = getOwnerConnection(found.clientId); const ownerGone = !ctx.connRegistry.get(ownerConnection);