mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 00:03:52 +00:00
fix(native-chat): stop eviction losing the rows a closing provider emits
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.
This commit is contained in:
@@ -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'])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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<void> {
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user