From 5da53f1ecd2c62efec187932a56b34873fcb506d Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 19 Aug 2026 12:08:08 -0700 Subject: [PATCH] fix(native-chat): stop eviction losing the rows a closing provider emits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three defects in the teardown this PR added, all found in review. ORDER. The sink was torn down before the provider child stopped. Closing a Codex session is not silent: `closeCodexPublishedSession` fires its `ended` event, flushes coalesced text and disposes the translator AFTER the child exits, and those are the rows that clear the running-turn marker. A closed sink drops them, so closing a chat could leave the durable journal claiming the agent is still working — worse than the leak this teardown exists to fix. The child now stops first and the sink outlives it. UNPROVEN EXIT. `closeSession` returns false to mean the child's exit could not be proven, and the adapter deliberately keeps the session indexed so a retry can reach it. Eviction ignored that, forgot the session anyway, and reported success: the child was stranded and the second close was a no-op against an already-forgotten session. Every step now aborts the rest, which is what makes the next close a real retry. CACHED SINK. The runtime caches one sink per session id and hands the same instance to the next attach, so closing without discarding left a reopened chat bound to a permanently closed sink — it accepts every provider event and publishes none. Attach's own failure path already paired close with discard. Tests: the closing-rows and reattach cases are the load-bearing ones. Restoring the old order reddens five, ignoring the false return reddens the retry case, and neutering the discard reddens the reattach case. --- .../structured-agent-session-eviction.test.ts | 108 +++++++++++++++--- .../structured-agent-session-eviction.ts | 46 +++++--- .../structured-agent-session-host.ts | 3 +- 3 files changed, 126 insertions(+), 31 deletions(-) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts index b0be080fe0f..849611e9fea 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts @@ -5,6 +5,7 @@ import { STRUCTURED_AGENT_SESSION_EVICTION_STEPS, type StructuredAgentSessionEvictionContext } from './structured-agent-session-eviction' +import { StructuredAgentSessionHostRuntimeState } from './structured-agent-session-host-runtime-state' function context(): StructuredAgentSessionEvictionContext & { order: string[] } { const order: string[] = [] @@ -23,31 +24,88 @@ function context(): StructuredAgentSessionEvictionContext & { order: string[] } order.push('closeSession') }) } as unknown as StructuredAgentSessionEvictionContext['adapter'], - forget: vi.fn(() => order.push('forget')) + forget: vi.fn(() => order.push('forget')), + discardSink: vi.fn(() => order.push('discardSink')) } } +function runtimeState(): StructuredAgentSessionHostRuntimeState { + return new StructuredAgentSessionHostRuntimeState({ + store: {} as never, + adapter: {} as never + } as never) +} + describe('structured agent session eviction', () => { - // Order is the correctness property: draining after the child is gone discards rows the provider - // already sent, and forgetting before it stops strands the process. - it('drains what the session produced, then stops the child, then forgets it', async () => { + it('stops the child before it lets the sink go, then forgets the session', async () => { const ctx = context() await evictStructuredAgentSession(ctx) - expect(ctx.order).toEqual(['unbind', 'drained', 'close', 'closeSession', 'forget']) + expect(ctx.order).toEqual([ + 'closeSession', + 'drained', + 'unbind', + 'close', + 'discardSink', + 'forget' + ]) }) it('names every step, so a half-finished eviction says which one failed', () => { expect(STRUCTURED_AGENT_SESSION_EVICTION_STEPS.map((step) => step.name)).toEqual([ - 'stop-publishing', - 'drain-published', - 'close-sink', 'stop-provider-child', + 'drain-published', + 'stop-publishing', + 'close-sink', + 'discard-sink', 'forget-session' ]) }) +}) - // A child that refuses to stop must not also leave the session wired to a dead sink. - it('runs the remaining steps when one fails, and reports which failed', async () => { +// Closing the codex child is not silent: the adapter emits its `ended` event and flushes coalesced +// text as it shuts down, and those rows are what clear the running-turn marker. If the sink is +// already closed the journal keeps claiming the agent is working, forever. +describe('rows the provider emits while closing', () => { + it('still reach the journal', async () => { + const state = runtimeState() + const sessionId = 'session-closing-rows' + const sink = state.eventSinkFor(sessionId) + const published: string[] = [] + sink.bind({ journal: {} as never, fence: 1, publish: () => published.push('final-flush') }) + + await evictStructuredAgentSession({ + sessionId, + eventSink: sink, + adapter: { + closeSession: async () => { + // What codex-structured-session-close does on its way out. + sink.sink.publish() + } + } as never, + forget: () => {}, + discardSink: () => state.discardEventSink(sessionId) + }) + + expect(published).toEqual(['final-flush']) + }) +}) + +// `closeSession` returning false means the adapter could not prove the child exited and has kept +// the session indexed on purpose so a retry can reach it. +describe('a child that will not stop', () => { + it('aborts without forgetting the session, so the next close is a real retry', async () => { + const ctx = context() + ctx.adapter.closeSession = vi.fn(async () => false) + + await expect(evictStructuredAgentSession(ctx)).rejects.toMatchObject({ + step: 'stop-provider-child' + }) + expect(ctx.forget).not.toHaveBeenCalled() + expect(ctx.discardSink).not.toHaveBeenCalled() + expect(ctx.order).toEqual([]) + }) + + it('reports the failing step and leaves the sink usable for the retry', async () => { const ctx = context() ctx.adapter.closeSession = vi.fn(async () => { throw new Error('child would not stop') @@ -56,10 +114,32 @@ describe('structured agent session eviction', () => { await expect(evictStructuredAgentSession(ctx)).rejects.toBeInstanceOf( StructuredAgentSessionEvictionError ) - await expect(evictStructuredAgentSession(ctx)).rejects.toMatchObject({ - step: 'stop-provider-child', - sessionId: 'session-1' + expect(ctx.eventSink.close).not.toHaveBeenCalled() + expect(ctx.forget).not.toHaveBeenCalled() + }) +}) + +// The runtime caches ONE sink per session id and hands the same instance to the next attach, so an +// eviction that closes without discarding leaves a reopened chat wired to a permanently closed +// sink — it accepts every provider event and publishes none. +describe('eviction against the real sink cache', () => { + it('lets the session publish again after it is evicted and reattached', async () => { + const state = runtimeState() + const sessionId = 'session-reattach' + await evictStructuredAgentSession({ + sessionId, + eventSink: state.eventSinkFor(sessionId), + adapter: { closeSession: async () => {} } as never, + forget: () => {}, + discardSink: () => state.discardEventSink(sessionId) }) - expect(ctx.forget).toHaveBeenCalled() + + const published: string[] = [] + const reattached = state.eventSinkFor(sessionId) + reattached.bind({ journal: {} as never, fence: 2, publish: () => published.push('published') }) + reattached.sink.publish() + await reattached.drained() + + expect(published).toEqual(['published']) }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts index 1d32381ef66..93953b6835b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts @@ -5,11 +5,17 @@ // (app quit, handoff to a TUI, and error cleanup). Closing a chat was never wired to any of them, // so a provider child outlived the chat that owned it for the whole app session. // -// A list makes the next way a session can end cheap to support — add a caller, not a fourth -// teardown — and makes a half-finished eviction say WHICH step failed instead of failing opaquely. -// Order matters and is asserted by name in the tests: drain what the session already produced -// before stopping its child, and stop the child before forgetting it. Draining after the child is -// gone discards rows the provider already sent; forgetting first strands the process. +// ORDER. The provider child stops FIRST. Closing it is not silent: the codex adapter emits its +// `ended` event and flushes coalesced text as part of shutting down, and those are the rows that +// clear the running-turn marker. Draining or closing the sink ahead of that drops them, which +// leaves the durable journal claiming the agent is still working — a worse outcome than the leak +// this teardown exists to fix. So: stop the child, drain what it emitted on its way out, then let +// the sink go. +// +// FAILURE. A step that fails ABORTS the rest. `closeSession` returning false means the child's +// exit was not proven and the adapter has deliberately kept the session indexed so a retry can +// reach it; forgetting it anyway stranded the process forever and reported success. Leaving the +// session in place is what makes the next close a real retry instead of a no-op. import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' import type { DeferredStructuredAgentSessionEventSink } from './structured-agent-session-event-sink' @@ -19,6 +25,8 @@ export type StructuredAgentSessionEvictionContext = { eventSink: DeferredStructuredAgentSessionEventSink adapter: StructuredAgentSessionAdapter forget: () => void + /** Drops the cached sink so a later attach mints a fresh one. */ + discardSink: () => void } export type StructuredAgentSessionEvictionStep = { @@ -28,15 +36,24 @@ export type StructuredAgentSessionEvictionStep = { export const STRUCTURED_AGENT_SESSION_EVICTION_STEPS: readonly StructuredAgentSessionEvictionStep[] = [ - { name: 'stop-publishing', run: (context) => context.eventSink.unbind() }, - { name: 'drain-published', run: (context) => context.eventSink.drained() }, - { name: 'close-sink', run: (context) => context.eventSink.close() }, { name: 'stop-provider-child', run: async (context) => { - await context.adapter.closeSession?.(context.sessionId) + // An adapter with no close has nothing to stop; anything else must PROVE the exit. + const stopped = await context.adapter.closeSession?.(context.sessionId) + if (stopped === false) { + throw new Error('provider child exit was not proven') + } } }, + { name: 'drain-published', run: (context) => context.eventSink.drained() }, + { name: 'stop-publishing', run: (context) => context.eventSink.unbind() }, + { name: 'close-sink', run: (context) => context.eventSink.close() }, + // Why: the runtime caches one sink per session id and hands the SAME instance to the next + // attach. Closing without discarding leaves a reopened chat bound to a closed sink, which + // accepts every provider event and publishes none. Attach's own failure path already pairs + // these two; eviction has to as well. + { name: 'discard-sink', run: (context) => context.discardSink() }, { name: 'forget-session', run: (context) => context.forget() } ] @@ -52,22 +69,19 @@ export class StructuredAgentSessionEvictionError extends Error { } /** - * Runs every eviction step in order. A failing step is reported with its name and does NOT skip the - * rest: a child that refuses to stop must not also leave the session wired to a dead sink. + * Runs the eviction steps in order, stopping at the first failure. The step name travels with the + * error because the caller's only useful response is to retry, and a retry is only safe when the + * session is still indexed — which is exactly what aborting preserves. */ export async function evictStructuredAgentSession( context: StructuredAgentSessionEvictionContext, steps: readonly StructuredAgentSessionEvictionStep[] = STRUCTURED_AGENT_SESSION_EVICTION_STEPS ): Promise { - let failure: StructuredAgentSessionEvictionError | null = null for (const step of steps) { try { await step.run(context) } catch (error) { - failure ??= new StructuredAgentSessionEvictionError(step.name, context.sessionId, error) + throw new StructuredAgentSessionEvictionError(step.name, context.sessionId, error) } } - if (failure) { - throw failure - } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 67256242733..65468968d52 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -145,7 +145,8 @@ export class StructuredAgentSessionHost { sessionId, eventSink: this.runtimeState.eventSinkFor(sessionId), adapter: this.deps.adapter, - forget: () => this.sessions.delete(sessionId) + forget: () => this.sessions.delete(sessionId), + discardSink: () => this.runtimeState.discardEventSink(sessionId) } await evictStructuredAgentSession(context) })