diff --git a/packages/client/runtime/src/client/index.ts b/packages/client/runtime/src/client/index.ts index f697fd3f1a..3b8e02b5a9 100644 --- a/packages/client/runtime/src/client/index.ts +++ b/packages/client/runtime/src/client/index.ts @@ -154,6 +154,12 @@ export function apply(ctx: Context): void { workspaces.handleConnected() ctx.emit('connection/reset') }, + onStateChange: (state) => { + // Generation death fires before any next-generation frame can arrive + // (reconnect replays flow from stream open, ahead of onConnected): + // the only safe moment to drop generation-scoped interaction state. + if (state === 'reconnecting') sessions.handleDisconnected() + }, }) ctx.effect(() => () => { loop.stop() }, 'runtime: connection stream loop') } diff --git a/packages/client/runtime/src/client/sessions/manager.ts b/packages/client/runtime/src/client/sessions/manager.ts index 62a0241df5..396c0be6e7 100644 --- a/packages/client/runtime/src/client/sessions/manager.ts +++ b/packages/client/runtime/src/client/sessions/manager.ts @@ -443,14 +443,32 @@ export class SessionManager { } } - /** After each connection generation: refresh the session baseline and rebuild opened windows. */ - handleConnected(): void { - // Approvals resolved while disconnected send no frame: drop the bits and - // let the mux-open replay re-add every still-pending question. + /** + * The moment a connection generation dies (before any next-generation frame + * can arrive — onConnected waits for the readiness handshake while replayed + * frames flow from stream open, so clearing there would race the replay): + * drop generation-scoped live state. Approvals resolved while disconnected + * send no frame, so the stale bits and the buffered answerable frames must + * not survive into the next generation — the mux-open replay re-adds every + * still-pending question with its live rpcId. + */ + handleDisconnected(): void { if (this.waitingApprovals.size > 0) { this.waitingApprovals.clear() this.notifier.markDirty() } + for (const [sessionId, buffer] of [...this.pendingBuffers]) { + const kept = buffer.filter(item => + item.payload.type !== 'approval/requested' && item.payload.type !== 'approval/resolved' + && item.payload.type !== 'question/requested' && item.payload.type !== 'question/resolved') + if (kept.length === buffer.length) continue + if (kept.length === 0) this.pendingBuffers.delete(sessionId) + else this.pendingBuffers.set(sessionId, kept) + } + } + + /** After each connection generation: refresh the session baseline and rebuild opened windows. */ + handleConnected(): void { void this.refreshList() for (const session of this.sessions.values()) void session.resync() } diff --git a/packages/client/runtime/src/client/sessions/service.ts b/packages/client/runtime/src/client/sessions/service.ts index 27965c2902..71067d6330 100644 --- a/packages/client/runtime/src/client/sessions/service.ts +++ b/packages/client/runtime/src/client/sessions/service.ts @@ -294,6 +294,11 @@ export class SessionsService implements ISessions { this.manager.handleConnected() } + /** Drop generation-scoped live interaction state the moment a connection generation dies. */ + handleDisconnected(): void { + this.manager.handleDisconnected() + } + /** * Create a session on the host. Resolution guarantee: by the time the * promise resolves, the created session is in the list store and diff --git a/packages/client/runtime/tests/manager.spec.ts b/packages/client/runtime/tests/manager.spec.ts index 90ab196059..33b46538d9 100644 --- a/packages/client/runtime/tests/manager.spec.ts +++ b/packages/client/runtime/tests/manager.spec.ts @@ -406,12 +406,32 @@ describe('waiting-approval list bit', () => { expect(manager.getListSnapshot().items).toHaveLength(0) }) - it('drops stale bits on reconnect — the reopen replay re-adds still-pending questions', () => { + it('drops stale bits at generation death — BEFORE the reopen replay re-adds still-pending questions', () => { const manager = new SessionManager(new FakeApiClient()) manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } }) manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } }) expect(manager.getListSnapshot().items[0]?.waitingApproval).toBe(true) - manager.handleConnected() // resolved-while-disconnected questions send no frame + // Generation death clears (resolved-while-disconnected questions send no frame)… + manager.handleDisconnected() expect(manager.getListSnapshot().items[0]?.waitingApproval).toBe(false) + // …and a replayed frame arriving before onConnected (stream open precedes + // the readiness handshake) survives the later handleConnected untouched. + manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } }) + manager.handleConnected() + expect(manager.getListSnapshot().items[0]?.waitingApproval).toBe(true) + }) + + it('generation death drops buffered answerable frames (a dead generation cannot be answered)', () => { + const manager = new SessionManager(new FakeApiClient()) + manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } }) + // Buffered pre-instantiation: an approval pair and a queued row. + manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } }) + manager.handleMuxEnvelope({ rpcId: 'q1' as never, payload: { type: 'question/requested', sessionId: S1, questions: [] } }) + manager.handleDisconnected() + // Instantiate after the death sweep: no zombie interaction replays (the + // pendingBuffers held only dead-generation rpcIds), so the session mints + // no pending waits. + const session = manager.get(S1) + expect(session.getSnapshot().pending).toEqual([]) }) })