diff --git a/src/main/claude/claude-stream-json-connection-close.test.ts b/src/main/claude/claude-stream-json-connection-close.test.ts index 22076f3600ea..8d813b549366 100644 --- a/src/main/claude/claude-stream-json-connection-close.test.ts +++ b/src/main/claude/claude-stream-json-connection-close.test.ts @@ -248,4 +248,57 @@ describe('Claude stream-json close ordering', () => { await expect(closing).resolves.toBe(true) expect(child.stdin.writableEnded).toBe(true) }) + + it('reports a root exit seen after a close gave up, and not one a running close sees', async () => { + mocks.refresh.mockReset() + mocks.proveClaudeChildExit.mockReset() + mocks.refresh.mockResolvedValue(undefined) + const open = async (child: ChildProcessWithoutNullStreams, reports: string[]) => { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: This injected query exercises only the async iterator used by the connection. + const queryImpl = ((params: Parameters[0]) => { + params.options?.spawnClaudeCodeProcess?.({ + command: 'claude', + args: [], + env: {}, + signal: new AbortController().signal + }) + return { + [Symbol.asyncIterator]: () => ({ + next: async (): Promise>> => ({ + value: undefined, + done: true + }) + }) + } + }) as unknown as typeof query + return openClaudeStreamJsonConnection( + { pathToClaudeCodeExecutable: 'claude', options: {}, cwd: '/work/repo' }, + { + onExit: () => reports.push('exit'), + onExitAfterClose: () => reports.push('exit-after-close') + }, + () => child, + queryImpl + ) + } + + const gaveUp: string[] = [] + const lateChild = fakeChild() + mocks.proveClaudeChildExit.mockResolvedValueOnce(false) + const late = await open(lateChild, gaveUp) + await expect(late.close()).resolves.toBe(false) + lateChild.emit('exit', 0, null) + expect(gaveUp).toEqual(['exit-after-close']) + + const running: string[] = [] + const runningChild = fakeChild() + const proof = Promise.withResolvers() + mocks.proveClaudeChildExit.mockReturnValueOnce(proof.promise) + const closing = (await open(runningChild, running)).close() + await vi.waitFor(() => expect(mocks.proveClaudeChildExit).toHaveBeenCalledTimes(2)) + runningChild.emit('exit', 0, null) + proof.resolve(true) + await expect(closing).resolves.toBe(true) + expect(running).toEqual([]) + }) }) diff --git a/src/main/claude/claude-stream-json-connection.ts b/src/main/claude/claude-stream-json-connection.ts index 6f67e04b36ea..7a9fffe6f2f4 100644 --- a/src/main/claude/claude-stream-json-connection.ts +++ b/src/main/claude/claude-stream-json-connection.ts @@ -66,6 +66,8 @@ export type ClaudeStreamJsonConnectionHandlers = { /** A transport/process fault that is not itself first-hand root exit proof. */ onFault?: (error: Error) => void onExit?: (error: Error) => void + /** The root exited after a close gave up proving it; no close is running to see that. */ + onExitAfterClose?: () => void } /** @@ -212,6 +214,9 @@ export async function openClaudeStreamJsonConnection( exitStatus = { code, signal } markExited() handleUnexpectedEnd() + if (closing && !closePromise) { + handlers.onExitAfterClose?.() + } }) const handleUnexpectedEnd = (cause?: Error): void => { diff --git a/src/main/claude/claude-structured-init-proof.ts b/src/main/claude/claude-structured-init-proof.ts index ae2ee5896a8b..fbf57f1636c4 100644 --- a/src/main/claude/claude-structured-init-proof.ts +++ b/src/main/claude/claude-structured-init-proof.ts @@ -1,5 +1,5 @@ import { CLAUDE_DEFAULT_SETTING_SOURCES } from './claude-structured-launch-resolution' -import type { ClaudeAuthDiagnostic } from './claude-structured-session-state' +import type { ClaudeAuthDiagnostic, ClaudeSession } from './claude-structured-session-state' import { AgentSessionAcquisitionRefusal } from '../native-chat/agent-session-wire/structured-agent-session-adapter' export type ClaudeInitObservation = { @@ -67,6 +67,17 @@ export function readClaudeCapabilities( return observed } +/** Every turn opens with an init frame naming the model the CLI is actually running; set_model + * answers success for a model it never resolves, so this report is the session's only adoption + * evidence. */ +export function observeClaudeInitFacts(session: ClaudeSession, init: ClaudeInitObservation): void { + if (init.model) { + session.reportedOptions.model = init.model + session.reportedModelMutation = session.optionMutationSequence + } + session.capabilities = readClaudeCapabilities(session.capabilities, init.message) +} + export function claudeInitializationAuthError( initialization: unknown ): AgentSessionAcquisitionRefusal | null { diff --git a/src/main/claude/claude-structured-session-acquisition.ts b/src/main/claude/claude-structured-session-acquisition.ts index 404787620e95..d4bb89f30ee6 100644 --- a/src/main/claude/claude-structured-session-acquisition.ts +++ b/src/main/claude/claude-structured-session-acquisition.ts @@ -11,7 +11,7 @@ import { resolveClaudeReplayTurn } from './claude-replay-turn-resolution' import { claudeSessionStateEndsTurn } from './claude-session-state-turn-over' import { settleClaudeTurnEndWaiters } from './claude-request-end-wait' import { - readClaudeCapabilities, + observeClaudeInitFacts, readClaudeFrameString, readClaudeInit } from './claude-structured-init-proof' @@ -100,15 +100,8 @@ export async function acquireClaudeSession({ } if (init) { initProof.resolve(init) - // Every turn opens with an init frame naming the model the CLI is actually - // running; set_model answers success for a model it never resolves, so this - // report is the session's only adoption evidence. - if (liveSession && init.model) { - liveSession.reportedOptions.model = init.model - liveSession.reportedModelMutation = liveSession.optionMutationSequence - } if (liveSession) { - liveSession.capabilities = readClaudeCapabilities(liveSession.capabilities, init.message) + observeClaudeInitFacts(liveSession, init) } } observedLeafUuid = readClaudeTranscriptEntryUuid(message) ?? observedLeafUuid @@ -206,7 +199,8 @@ export async function acquireClaudeSession({ childEnded ??= error initProof.reject(error) callbacks.handleExit(sessionId, attempt, error) - } + }, + onExitAfterClose: () => callbacks.handleExitAfterClose(sessionId, attempt) } ) ) diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index 746d603308f1..1dc4a7a900d3 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -31,6 +31,7 @@ import { claudeStoppedRequestEndWait } from './claude-request-end-wait' import { drainClaudeObservedExits, observeClaudeSessionExit, + reportClaudeExitAfterClose, settleClaudeUnexpectedExit, type ClaudeExitLifecycle } from './claude-structured-session-exit-lifecycle' @@ -95,6 +96,8 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda emit: (session, _events, event) => this.emit(session, event), handleExit: (sessionId, attempt, error) => observeClaudeSessionExit(this.exitLifecycle, sessionId, attempt, error), + handleExitAfterClose: (sessionId, attempt) => + reportClaudeExitAfterClose(this.exitLifecycle, sessionId, attempt), settleExit: (sessionId, exit) => settleClaudeUnexpectedExit(this.exitLifecycle, sessionId, exit) } diff --git a/src/main/claude/claude-structured-session-exit-lifecycle.ts b/src/main/claude/claude-structured-session-exit-lifecycle.ts index 854646490a8c..be4457d885e6 100644 --- a/src/main/claude/claude-structured-session-exit-lifecycle.ts +++ b/src/main/claude/claude-structured-session-exit-lifecycle.ts @@ -22,7 +22,7 @@ export type ClaudeExitLifecycle = { exits: Map /** A settled exit's diagnostic, kept for a send admitted before the host heard of the exit. */ settledExitErrors: Map - deps: Pick + deps: Pick emit: (session: ClaudeSession, event: ClaudeStructuredSessionEvent) => void } @@ -62,6 +62,25 @@ export function observeClaudeSessionExit( .catch(() => undefined) } +/** A close that gave up on this child has since seen its root exit. The host's retry of that close + * settles it; this only tells the host the retry can land now. */ +export function reportClaudeExitAfterClose( + lifecycle: ClaudeExitLifecycle, + sessionId: string, + attempt: ClaudeAcquisitionAttempt +): void { + const session = lifecycle.sessions.get(sessionId) + if (!session || session.connection !== attempt.connection) { + return + } + lifecycle.deps.onEvent?.({ + type: 'exitAfterClose', + sessionId, + fence: session.fence, + acquisitionGeneration: session.acquisitionGeneration + }) +} + /** Lifecycle recovery is published only after the close ladder ran and proved the tree gone or * observed the root's own exit. */ export function settleClaudeUnexpectedExit( diff --git a/src/main/claude/claude-structured-session-state.ts b/src/main/claude/claude-structured-session-state.ts index d61df04ac75c..63c2aa0b831b 100644 --- a/src/main/claude/claude-structured-session-state.ts +++ b/src/main/claude/claude-structured-session-state.ts @@ -5,7 +5,10 @@ import type { AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' -import type { StructuredAgentSessionStartedEvent } from '../native-chat/agent-session-wire/structured-agent-session-adapter' +import type { + StructuredAgentSessionExitAfterCloseEvent, + StructuredAgentSessionStartedEvent +} from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { ClaudeStreamJsonConnection, openClaudeStreamJsonConnection @@ -66,6 +69,7 @@ export type ClaudeStructuredSessionEvent = | { type: 'auth-diagnostic'; sessionId: string; diagnostic: ClaudeAuthDiagnostic } /** Startup facts applied and saved options restored; held prompts are about to be written. */ | StructuredAgentSessionStartedEvent + | StructuredAgentSessionExitAfterCloseEvent | { type: 'ended' sessionId: string @@ -345,5 +349,7 @@ export type ClaudeAcquireCallbacks = { event: ClaudeStructuredSessionEvent ) => void handleExit: (sessionId: string, attempt: ClaudeAcquisitionAttempt, error: Error) => void + /** The child's root exited after a close of it gave up proving that. */ + handleExitAfterClose: (sessionId: string, attempt: ClaudeAcquisitionAttempt) => void settleExit: (sessionId: string, exit: ClaudeSessionExit) => Promise } diff --git a/src/main/codex/codex-app-server-connection-types.ts b/src/main/codex/codex-app-server-connection-types.ts index df3f40ac39a5..0a16ec465ed3 100644 --- a/src/main/codex/codex-app-server-connection-types.ts +++ b/src/main/codex/codex-app-server-connection-types.ts @@ -9,6 +9,8 @@ export type CodexAppServerConnectionHandlers = { onServerRequest?: (request: CodexAppServerServerRequest) => void onUnhandledFrame?: (kind: string, payload: unknown) => void onExit?: (error: Error) => void + /** The child exited after a close gave up proving it; no close is running to see that. */ + onExitAfterClose?: () => void /** Awaited once the child has a pid and before the handshake; a rejection reaps the child. */ onSpawned?: (pid: number) => Promise } diff --git a/src/main/codex/codex-app-server-connection.test.ts b/src/main/codex/codex-app-server-connection.test.ts index 3d95e7ce380f..cb8bf421ae39 100644 --- a/src/main/codex/codex-app-server-connection.test.ts +++ b/src/main/codex/codex-app-server-connection.test.ts @@ -526,6 +526,44 @@ describe('openCodexAppServerConnection', () => { await expect(connection.close()).resolves.toBe(true) }) + it('reports an exit seen after a close gave up, and not one a running or proven close sees', async () => { + vi.useFakeTimers() + const late = stubChild({ exitOnStdinEnd: false }) + answerInitialize(late.child) + const gaveUp: string[] = [] + const lateConnection = await openCodexAppServerConnection( + { command: 'codex', args: ['app-server'] }, + { + onExit: () => gaveUp.push('exit'), + onExitAfterClose: () => gaveUp.push('exit-after-close') + }, + late.spawnImpl + ) + const first = lateConnection.close() + await vi.advanceTimersByTimeAsync(GRACEFUL_EXIT_MS + 3_500) + await expect(first).resolves.toBe(false) + late.child.emit('exit', 0, null) + expect(gaveUp).toEqual(['exit-after-close']) + + const running = stubChild({ exitOnStdinEnd: false }) + answerInitialize(running.child) + const seen: string[] = [] + const runningConnection = await openCodexAppServerConnection( + { command: 'codex', args: ['app-server'] }, + { + onExit: () => seen.push('exit'), + onExitAfterClose: () => seen.push('exit-after-close') + }, + running.spawnImpl + ) + const closing = runningConnection.close() + running.child.emit('exit', 0, null) + await expect(closing).resolves.toBe(true) + // Stdio 'close' can land after the proof settled; the close still proved the exit. + running.child.emit('close', 0, null) + expect(seen).toEqual([]) + }) + it.each([1_090_188, 2_900_090])( 'accepts a realistic %i-byte escaped command completion and keeps processing', async (frameBytes) => { diff --git a/src/main/codex/codex-app-server-connection.ts b/src/main/codex/codex-app-server-connection.ts index 3e7d16ea1bb0..53fd74b64c40 100644 --- a/src/main/codex/codex-app-server-connection.ts +++ b/src/main/codex/codex-app-server-connection.ts @@ -129,6 +129,9 @@ export async function openCodexAppServerConnection( if (exitObserved && !closing && !exitReported) { exitReported = true handlers.onExit?.(terminalError) + } else if (exitObserved && closing && !exitProof.holdsAttempt && !exitReported) { + exitReported = true + handlers.onExitAfterClose?.() } } diff --git a/src/main/codex/codex-structured-journal-translation.ts b/src/main/codex/codex-structured-journal-translation.ts index 20fd8bf58df3..d090667692f7 100644 --- a/src/main/codex/codex-structured-journal-translation.ts +++ b/src/main/codex/codex-structured-journal-translation.ts @@ -140,6 +140,10 @@ export function createCodexJournalTranslator( }) }, handle: (event) => { + // Evidence for the host's retry of a close, never routed here: nothing for the journal. + if (event.type === 'exitAfterClose') { + return CODEX_JOURNAL_ADMITTED + } if (event.type === 'ended') { const streamAdmission = flushStreams() if (!streamAdmission.accepted) { diff --git a/src/main/codex/codex-structured-session-acquire.ts b/src/main/codex/codex-structured-session-acquire.ts index 37c34a4bf3a1..1bea693887e5 100644 --- a/src/main/codex/codex-structured-session-acquire.ts +++ b/src/main/codex/codex-structured-session-acquire.ts @@ -21,7 +21,8 @@ import { buildCodexStructuredChildEnvironment } from './codex-structured-child-e import { openCodexThread } from './codex-structured-thread-open' import { closeCodexPublishedSession, - handleCodexSessionExit + handleCodexSessionExit, + reportCodexExitAfterClose } from './codex-structured-session-close' import { restoredCodexSessionOptions } from './codex-structured-session-options' import { @@ -176,7 +177,14 @@ export async function acquireCodexStructuredSession(input: { } finally { notificationRetries.clear(sessionId, acquisition.connection) } - } + }, + onExitAfterClose: () => + reportCodexExitAfterClose({ + sessions, + sessionId, + connection: acquisition.connection, + ...(deps.onEvent ? { onEvent: deps.onEvent } : {}) + }) } ) acquisition.connection = connection diff --git a/src/main/codex/codex-structured-session-close.ts b/src/main/codex/codex-structured-session-close.ts index fe6c0d676735..e50e4f79384d 100644 --- a/src/main/codex/codex-structured-session-close.ts +++ b/src/main/codex/codex-structured-session-close.ts @@ -64,6 +64,26 @@ export function handleCodexSessionExit(input: { return true } +/** A close that gave up on this child has since seen it exit. The host's retry of that close + * settles it; this only tells the host the retry can land now. */ +export function reportCodexExitAfterClose(input: { + sessions: Map + sessionId: string + connection: CodexAppServerConnection | null + onEvent?: (event: CodexStructuredSessionEvent) => void +}): void { + const session = input.sessions.get(input.sessionId) + if (!session || session.connection !== input.connection || session.ended) { + return + } + input.onEvent?.({ + type: 'exitAfterClose', + sessionId: input.sessionId, + fence: session.fence, + acquisitionGeneration: session.acquisitionGeneration + }) +} + export async function closeCodexPublishedSession( sessions: Map, sessionId: string, diff --git a/src/main/codex/codex-structured-session-state.ts b/src/main/codex/codex-structured-session-state.ts index a47cb5a78fe2..cbe0ad76e1be 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -18,7 +18,10 @@ import type { CodexDispatchEchoes } from './codex-structured-dispatch-echo' import type { AgentChildWorkEvidence } from '../../shared/agent-status-child-work-evidence' import type { CodexBackgroundTaskTracker } from './codex-background-task-tracker' import type { CodexJournalTranslator } from './codex-structured-journal-translation' -import type { StructuredAgentSessionEndedEvent } from '../native-chat/agent-session-wire/structured-agent-session-adapter' +import type { + StructuredAgentSessionEndedEvent, + StructuredAgentSessionExitAfterCloseEvent +} from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { CodexStructuredPermissionPolicy } from './codex-structured-permission-policy' import type { AgentModelCatalogSessionAccess, @@ -67,6 +70,7 @@ export type CodexStructuredSessionEvent = promptKey: string } | StructuredAgentSessionEndedEvent + | StructuredAgentSessionExitAfterCloseEvent /** Translator-only compatibility for callers that do not participate in host recovery. */ | { type: 'ended'; sessionId: string; reason: string; observedAt?: number } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index 318bada1fe52..36d93b981e52 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -212,9 +212,19 @@ export type StructuredAgentSessionStartedEvent = { restoreSkippedOptions: readonly string[] } +/** A close that gave up proving this child's exit has since seen its root exit. Evidence to retry + * the stop owed for this child, not a verdict: that stop's own proof decides. */ +export type StructuredAgentSessionExitAfterCloseEvent = { + type: 'exitAfterClose' + sessionId: string + fence: number + acquisitionGeneration: string +} + export type StructuredAgentSessionLifecycleEvent = | StructuredAgentSessionEndedEvent | StructuredAgentSessionStartedEvent + | StructuredAgentSessionExitAfterCloseEvent /** Whether the provider child behind an acquisition has proven its start. A publish-first * acquire hands over a `starting` child and the `started` lifecycle event flips it. */ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-unproven-stop-send.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-unproven-stop-send.test.ts index b86e25674641..17bb74cfa896 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-unproven-stop-send.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-unproven-stop-send.test.ts @@ -9,6 +9,7 @@ import { afterEach, beforeEach, expect, it, vi } from 'vitest' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-live-turn' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import { claudeUnwrittenUserMessageError } from '../../claude/claude-agent-sdk-user-message-queue' import { ClaudeStructuredSessionAdapter } from '../../claude/claude-structured-session-adapter' import { @@ -20,6 +21,7 @@ import type { AgentSessionRecordStore } from '../../runtime/agent-session-record import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' import { structuredClaudeLifecycleEvent } from '../../runtime/structured-claude-runtime-adapter' import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' +import type * as EvictionDeadline from './structured-agent-session-eviction-deadline' import { StructuredAgentSessionHost } from './structured-agent-session-host' import { recordingStructuredAgentSessionLogger } from './structured-agent-session-logger-test-support' import { @@ -31,6 +33,18 @@ import { resetHostTestOperationIds } from './structured-agent-session-host-test-data' +// Shortened only by the test of a stop that proves the exit after its deadline. +const stepDeadline = vi.hoisted((): { ms: number | undefined } => ({ ms: undefined })) +vi.mock('./structured-agent-session-eviction-deadline', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + withStructuredAgentSessionEvictionDeadline: ( + ...[steps, timeoutMs]: Parameters + ) => actual.withStructuredAgentSessionEvictionDeadline(steps, stepDeadline.ms ?? timeoutMs) + } +}) + const CALLER = { callerKey: 'client-1' } const CAPABILITIES = ['interrupt_receipt_v1', 'interrupt_cancel_queued_v1', 'msg_lifecycle_v1'] @@ -94,6 +108,7 @@ beforeEach(async () => { }) afterEach(async () => { + stepDeadline.ms = undefined await adapter.closeAll() await host.flushAllStreamedEvents() await rm(root, { recursive: true, force: true }) @@ -567,3 +582,147 @@ it('keeps an option change at rest when the Stop proved the exit and only its bo expect(connection.closeCount).toBe(1) expect(claude.connections).toHaveLength(1) }) + +it('sends a held message as soon as the old child exits after its retry gave up, with no sweep tick', async () => { + const connection = await heldAfterStop() + + // The old process exits now: the connection reports it, and the retry that follows proves it. + connection.handlers.onExitAfterClose?.() + const resumed = await resumedWith(connection, 'Carry on.') + + expect(connection.closeCount).toBe(3) + expect(resumed.closed).toBe(false) + expect(owedWindDown()).toBeUndefined() +}) + +it("keeps the person's Stop as the cut turn's end when the old child exits late, writing no Stop of its own", async () => { + const connection = claude.connections[0]! + await openTurn(connection) + closeUnprovenFor(connection, 2) + // No interrupted result: only the stop that lands ends the turn the Stop cut. + await expect(stop()).resolves.toMatchObject({ ok: true, value: { cancelled: true } }) + await send('Carry on.') + await eventually(async () => expect(await waitRows()).toHaveLength(1)) + await laneDrained() + const journal = host['sessions'].get(SESSION)!.journal + const personStop = journal.stopMarks.latest() + expect(personStop).toMatchObject({ event: { reason: 'user-stop' } }) + + connection.handlers.onExitAfterClose?.() + await resumedWith(connection, 'Carry on.') + + expect(journal.stopMarks.latest()).toEqual(personStop) + await host.flushStreamedEvents(SESSION) + const { items } = await host.journalSnapshot(SESSION) + const cut = readAgentJournalTurn(items.find((item) => item.body.kind === 'turn')?.body) + expect(cut).toMatchObject({ state: 'interrupted', outcome: 'cancellation' }) +}) + +it('finishes an owed stop as soon as the old child exits with nothing waiting, starting no agent', async () => { + const connection = await stopWithUnprovenClose(1) + + connection.handlers.onExitAfterClose?.() + await eventually(() => expect(owedWindDown()).toBeUndefined()) + await commitSettled() + + expect(connection.closeCount).toBe(2) + expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('released') + expect(claude.connections).toHaveLength(1) +}) + +it('sends a held message as soon as a stop that ran past its deadline proves the exit late', async () => { + const connection = claude.connections[0]! + await openTurn(connection) + stepDeadline.ms = 200 + // Every close of the old child waits on its exit, which comes after both stop passes gave up. + const exited = Promise.withResolvers() + const close = connection.close + connection.close = async () => { + await exited.promise + return close() + } + await expect(stop()).resolves.toMatchObject({ ok: true, value: { cancelled: true } }) + frame(connection, INTERRUPTED_RESULT) + await send('Carry on.') + await eventually(async () => expect(await waitRows()).toHaveLength(1)) + await laneDrained() + expect(owedWindDown()).toBeDefined() + expect(claude.connections).toHaveLength(1) + + // Both passes already gave up; the retry the late proof starts gets the full deadline. + stepDeadline.ms = undefined + exited.resolve() + const resumed = await resumedWith(connection, 'Carry on.') + + expect(resumed.closed).toBe(false) + expect(owedWindDown()).toBeUndefined() +}) + +it('retries an owed stop once when several passes gave up on one hung close and its exit lands', async () => { + const connection = claude.connections[0]! + await openTurn(connection) + stepDeadline.ms = 200 + const exited = Promise.withResolvers() + const close = connection.close + connection.close = async () => { + await exited.promise + return close() + } + // The Stop, the held send's retry and two sweep ticks each give up on the same close. + await expect(stop()).resolves.toMatchObject({ ok: true, value: { cancelled: true } }) + frame(connection, INTERRUPTED_RESULT) + await send('Carry on.') + await eventually(async () => expect(await waitRows()).toHaveLength(1)) + await laneDrained() + await host['lifetime'].idleSweep.tick() + await host['lifetime'].idleSweep.tick() + await laneDrained() + const releases = vi + .spyOn(store, 'transitionHandoff') + .mockRejectedValueOnce(new Error('record store busy')) + + stepDeadline.ms = undefined + exited.resolve() + await eventually(() => expect(releases).toHaveBeenCalled()) + await laneDrained() + await laneDrained() + + // The one retry failed; the sweep or the next action retries it, not every pass that gave up. + expect(releases).toHaveBeenCalledOnce() + expect(owedWindDown()).toBeDefined() + expect(claude.connections).toHaveLength(1) +}) + +it('ignores an exit report naming another child than the one whose stop is owed', async () => { + const connection = await heldAfterStop() + const session = host['sessions'].get(SESSION)! + + await host.handleAdapterEvent({ + type: 'exitAfterClose', + sessionId: SESSION, + fence: session.child!.fence, + acquisitionGeneration: 'an-earlier-child' + }) + await laneDrained() + + expect(connection.closeCount).toBe(2) + expect(owedWindDown()).toBeDefined() + expect(claude.connections).toHaveLength(1) +}) + +it("never lets an old child's late exit report retry a newer child's owed stop", async () => { + const old = await heldAfterStop() + old.handlers.onExitAfterClose?.() + const resumed = await resumedWith(old, 'Carry on.') + closeUnprovenFor(resumed, 1) + await expect(host.close(SESSION, 'user-close')).rejects.toThrow() + expect(resumed.closeCount).toBe(1) + expect(owedWindDown()).toBeDefined() + + // The first child's connection reports its exit again: it names no child whose stop is owed. + old.handlers.onExitAfterClose?.() + await laneDrained() + + expect(resumed.closeCount).toBe(1) + expect(owedWindDown()).toBeDefined() +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-codex-late-exit.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-codex-late-exit.test.ts new file mode 100644 index 000000000000..311b4b98ccbe --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-codex-late-exit.test.ts @@ -0,0 +1,184 @@ +// A Codex chat whose close could not prove the app-server gone, then the user's next message, on the +// shipping adapter and host. The message waits on that close; when the old app-server is later seen +// to exit, the close is retried at once and the message goes out, with no idle-sweep tick. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import { agentSessionProviderHandleChainHead } from '../../../shared/agent-session-provider-handle' +import { fakeCodex } from '../../codex/codex-structured-session-adapter-fixture' +import { + CodexStructuredSessionAdapter, + type CodexStructuredSessionEvent +} from '../../codex/codex-structured-session-adapter' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' +import { structuredCodexLifecycleEvent } from '../../runtime/structured-codex-lifecycle-event' +import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { recordingStructuredAgentSessionLogger } from './structured-agent-session-logger-test-support' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + hostTestAttachParams, + hostTestMessage, + hostTestOperationId, + resetHostTestOperationIds +} from './structured-agent-session-host-test-data' + +const CALLER = { callerKey: 'client-1' } + +let root: string +let host: StructuredAgentSessionHost +let store: AgentSessionRecordStore +let codex: ReturnType +let events: CodexStructuredSessionEvent[] + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-codex-late-exit-')) + resetHostTestOperationIds() + events = [] + codex = fakeCodex() + let generation = 0 + const adapter = new CodexStructuredSessionAdapter({ + // As the shipping resolver: a restart resumes the thread the record names. + resolveLaunch: async () => { + const head = agentSessionProviderHandleChainHead( + store.getRecord(SESSION)?.providerHandleChain ?? [] + ) + return { + command: 'codex', + args: ['app-server'], + cwd: root, + codexHome: null, + resumeThreadId: head?.handle.provider === 'codex' ? head.handle.threadId : null + } + }, + // As the runtime routes them: what the host's lifecycle handler consumes reaches it. + onEvent: (event) => { + events.push(event) + const lifecycleEvent = structuredCodexLifecycleEvent(event) + if (lifecycleEvent) { + void host.handleAdapterEvent(lifecycleEvent) + } + }, + openConnection: codex.openConnection, + readProcessStartTime: async () => 1_700_000_000_000, + now: () => NOW, + mintAcquisitionGeneration: () => `generation-${++generation}` + }) + store = await openTestAgentSessionRecordStore(root) + host = new StructuredAgentSessionHost({ + store, + adapter: Object.assign(adapter, { supportsCreate: () => true }), + journalDatabase: openTestJournalHostDatabase(root), + claimKeyId: 'key-1', + mintSpawnToken: () => 'spawn-1', + logger: recordingStructuredAgentSessionLogger().logger, + now: () => NOW + }) + expect(await host.attach(CALLER, hostTestAttachParams(null))).toMatchObject({ ok: true }) +}) + +afterEach(async () => { + await host.flushAllStreamedEvents() + await rm(root, { recursive: true, force: true }) +}) + +async function send(text: string): Promise { + const body = hostTestMessage(text) + const sent = await host.send(CALLER, { + envelope: { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: store.getRecord(SESSION)!.lease.runtimeFence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.send', + sessionId: SESSION, + fields: { body } + }) + }, + body + }) + if (!sent.ok) { + throw new Error(`send refused: ${JSON.stringify(sent.refusal)}`) + } +} + +async function waitRows() { + await host.flushStreamedEvents(SESSION) + return (await host.journalSnapshot(SESSION)).items.filter( + (item) => item.body.kind === 'status' && item.body.failure?.kind === 'previousExitUnverifiable' + ) +} + +type CodexConnection = (typeof codex.connections)[number] + +function startedTurnWith(connection: CodexConnection, text: string): boolean { + return connection.calls.some( + (call) => call.method === 'turn/start' && JSON.stringify(call.params).includes(text) + ) +} + +/** Closes the chat with an app-server whose close cannot prove the exit, then holds a message. */ +async function heldAfterUnprovenClose(): Promise { + const connection = codex.connections[0]! + const close = connection.close + let unproven = 2 + connection.close = async () => { + if (unproven === 0) { + return close() + } + unproven -= 1 + connection.closeCount += 1 + return false + } + await expect(host.close(SESSION, 'user-close')).rejects.toThrow() + expect(host['sessions'].get(SESSION)?.owesProviderChildWindDown).toMatchObject({ + cause: 'user-close' + }) + await send('Carry on.') + await vi.waitFor(async () => expect(await waitRows()).toHaveLength(1)) + expect(connection.closeCount).toBe(2) + expect(codex.connections).toHaveLength(1) + return connection +} + +function exitReports(): number { + return events.filter((event) => event.type === 'exitAfterClose').length +} + +it('sends a held message as soon as the old app-server exits after its close gave up', async () => { + const connection = await heldAfterUnprovenClose() + + connection.handlers.onExitAfterClose?.() + const resumed = await vi.waitFor(() => { + const next = codex.connections.at(-1)! + expect(next).not.toBe(connection) + expect(startedTurnWith(next, 'Carry on.')).toBe(true) + return next + }) + + expect(connection.closeCount).toBe(3) + expect(startedTurnWith(connection, 'Carry on.')).toBe(false) + expect(resumed.closeCount).toBe(0) + expect(host['sessions'].get(SESSION)?.owesProviderChildWindDown).toBeUndefined() +}) + +it("drops a late exit report from an app-server that is no longer the chat's", async () => { + const connection = await heldAfterUnprovenClose() + connection.handlers.onExitAfterClose?.() + const resumed = await vi.waitFor(() => { + const next = codex.connections.at(-1)! + expect(startedTurnWith(next, 'Carry on.')).toBe(true) + return next + }) + expect(exitReports()).toBe(1) + + connection.handlers.onExitAfterClose?.() + + expect(exitReports()).toBe(1) + expect(resumed.closeCount).toBe(0) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts index 19af6d048d45..93ed1dd35462 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts @@ -15,13 +15,17 @@ import type { StructuredAgentSessionConversations } from './structured-agent-ses import { abandonQueuedStructuredAgentSessionMessages, closeStructuredAgentSessionConversationUnderSerialize, + finishOwedStructuredAgentSessionWindDownAfterExitUnderSerialize, finishOwedStructuredAgentSessionWindDownUnderSerialize, stopStructuredAgentSessionAgentUnderSerialize, type StructuredAgentSessionCloseCause, type StructuredAgentSessionLifetimeContext, type StructuredAgentSessionStopEnding } from './structured-agent-session-host-lifetime' -import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' +import type { + StructuredAgentSessionHostSession, + StructuredAgentSessionProviderChildIdentity +} from './structured-agent-session-host-types' import { StructuredAgentSessionIdleSweep } from './structured-agent-session-idle-sweep' import { AGENT_SESSION_NOT_ATTACHED } from './structured-agent-session-mutation-admission' import { adapterSupportsRecord } from './structured-agent-session-provider-support' @@ -102,6 +106,26 @@ export function createStructuredAgentSessionConversationLifetime(host: { return { idleSweep, stopAgent, + /** The child a stop gave up on was seen to exit; quit's own eviction finishes any after dispose. */ + finishStopAfterExit: ( + sessionId: string, + child: StructuredAgentSessionProviderChildIdentity + ): Promise => + serialize(sessionId, async () => { + if (!disposed) { + await finishOwedStructuredAgentSessionWindDownAfterExitUnderSerialize( + host.context(), + sessionId, + child + ) + } + }).catch((error: unknown) => + deps().logger.warn('finishing a stop after its child exited failed', { + scope: 'stop-after-exit', + sessionId, + error + }) + ), /** Quit has begun: nothing opens a conversation or sweeps one after this. */ dispose: (): void => { disposed = true diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts index 2b9340859f42..99df13044f81 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts @@ -3,7 +3,8 @@ import type { StructuredAgentSessionLifecycleEvent } from './structured-agent-se import { stopAgentSessionProviderRoot } from './structured-agent-session-provider-exit-proof' import type { StructuredAgentSessionHostDeps, - StructuredAgentSessionHostSession + StructuredAgentSessionHostSession, + StructuredAgentSessionProviderChildIdentity } from './structured-agent-session-host-types' import type { StructuredAgentSessionSinkBarrier } from './structured-agent-session-event-sink' import { settleStructuredAgentSessionProviderStarted } from './structured-agent-session-provider-started' @@ -22,6 +23,10 @@ export class StructuredAgentSessionEventRecovery { publishStatus?: (sessionId: string) => void serialize: (sessionId: string, task: () => Promise) => Promise now: () => number + finishStopAfterExit: ( + sessionId: string, + child: StructuredAgentSessionProviderChildIdentity + ) => Promise } ) {} @@ -78,6 +83,12 @@ export class StructuredAgentSessionEventRecovery { if (event.type === 'started') { return settleStructuredAgentSessionProviderStarted(this.context, event) } + if (event.type === 'exitAfterClose') { + return this.context.finishStopAfterExit(event.sessionId, { + generation: event.acquisitionGeneration, + fence: event.fence + }) + } await settleUnexpectedStructuredAgentSessionExit(this.exitContext, event) } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts index 60c20cb768d9..f5c64e97361f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts @@ -50,6 +50,11 @@ export type StructuredAgentSessionLifetimeContext = { publishStatus?: (sessionId: string) => void /** Hands the delivery loop what is queued; for a caller inside the session's serialize. */ wakeDelivery?: (sessionId: string) => void + /** Retries, on the session's own lane, the stop owed for `child` once that child is seen gone. */ + finishStopAfterExit?: ( + sessionId: string, + child: StructuredAgentSessionProviderChildIdentity + ) => Promise /** Quit-only snapshot taken immediately before the provider child is stopped. */ restartWitness?: { beforeStop: (sessionId: string) => void @@ -150,6 +155,9 @@ export async function stopStructuredAgentSessionAgentUnderSerialize( const owed = owedStop(session, cause, ending.retry === true) session.owesProviderChildWindDown = owed const stopping = session.child + // A stop step past its deadline still runs; proving the exit after this pass gave up, it has no + // caller left to finish the stop. + let gaveUp = false const eviction: StructuredAgentSessionEvictionContext = { sessionId, // The retry must not re-stop a child the adapter already proved gone, so this stays honest. @@ -164,7 +172,7 @@ export async function stopStructuredAgentSessionAgentUnderSerialize( // Host state must not disagree with the adapter for the steps in between. onProviderChildStopped: (verdict) => { if (stopping) { - endProviderChild(session, { + const ended = endProviderChild(session, { generation: stopping.generation, fence: stopping.fence, cause, @@ -174,6 +182,10 @@ export async function stopStructuredAgentSessionAgentUnderSerialize( ...(owed ? { endedAt: owed.requestedAt } : {}), ...verdict }) + // Passes that joined one hung close all prove it; only the one that ended the child asks. + if (ended && gaveUp) { + void context.finishStopAfterExit?.(sessionId, stopping) + } } context.restartWitness?.stopped(sessionId) }, @@ -229,6 +241,7 @@ export async function stopStructuredAgentSessionAgentUnderSerialize( withStructuredAgentSessionEvictionDeadline(STRUCTURED_AGENT_SESSION_EVICTION_STEPS) ) } catch (error) { + gaveUp = true if (session.owesProviderChildWindDown === owed && owed) { session.owesProviderChildWindDown = { ...owed, failedAt: session.journal.cursor() } } @@ -271,6 +284,21 @@ export async function finishOwedStructuredAgentSessionWindDownUnderSerialize( return context.sessions.get(sessionId)?.owesProviderChildWindDown === undefined } +/** The stop owed for `child`, retried once that child was seen to exit after its stop gave up, so + * what waited on it goes out now rather than at the sweep's next tick. A report naming any other + * child is stale and touches nothing. */ +export async function finishOwedStructuredAgentSessionWindDownAfterExitUnderSerialize( + context: StructuredAgentSessionLifetimeContext, + sessionId: string, + child: StructuredAgentSessionProviderChildIdentity +): Promise { + const session = context.sessions.get(sessionId) + const owed = session && pendingProviderChildWindDown(session) + if (owed && sameProviderChild(owed, child)) { + await finishOwedStructuredAgentSessionWindDownUnderSerialize(context, sessionId) + } +} + /** A close's cause: the user closing this chat, or the host evicting it (quit, idle, teardown). */ export type StructuredAgentSessionCloseCause = Extract< StructuredAgentSessionStopCause, 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 44a576e2f781..ccc808a54eb2 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 @@ -10,7 +10,6 @@ import type { AgentJournalSnapshot } from '../../../shared/agent-session-journal import type { AgentSessionExecutionLocation } from '../../../shared/agent-session-record' import type * as SessionWire from '../../../shared/agent-session-wire' import type { AgentSessionAttachParams } from './structured-agent-session-attach' -import { createRestartReconciler } from './structured-agent-session-restart-reconcile' import type { AgentSessionSubscribeInput } from './structured-agent-session-subscribers' import { StructuredAgentSessionTaskQueue } from './structured-agent-session-task-queue' import * as providerSupport from './structured-agent-session-provider-support' @@ -85,9 +84,6 @@ export class StructuredAgentSessionHost { private readonly subscribers = this.clientDelivery.subscribers private readonly tasks = new StructuredAgentSessionTaskQueue() private readonly runtimeState: StructuredAgentSessionHostRuntimeState - private readonly reconcileLeases: ( - sessionId: string - ) => Promise private readonly restore: ReturnType private readonly lifetime: StructuredAgentSessionConversationLifetime private readonly conversationDelivery: ReturnType< @@ -112,12 +108,6 @@ export class StructuredAgentSessionHost { this.runtimeState = new StructuredAgentSessionHostRuntimeState(deps, (sessionId, error) => this.eventRecovery.recoverAfterSinkFailure(sessionId, error) ) - this.reconcileLeases = createRestartReconciler({ - store: deps.store, - probe: (record) => this.runtimeState.probeRecord(record), - ...(deps.probeOwners ? { probeMany: deps.probeOwners } : {}), - now: () => this.now() - }) this.conversationDelivery = createStructuredAgentSessionConversationDelivery({ deps, sessions: this.sessions, @@ -137,7 +127,8 @@ export class StructuredAgentSessionHost { flushStreamedEvents: (sessionId) => this.flushStreamedEvents(sessionId) }) this.restore = createStructuredAgentSessionHostRestore(deps, { - reconcileLeases: this.reconcileLeases, + probe: (record) => this.runtimeState.probeRecord(record), + now: () => this.now(), resolveRecovery: (sessionId) => this.runtimeState.resolveRecovery(sessionId), serialize: (sessionId, task) => this.serialize(sessionId, task), hasSession: this.hasSession, @@ -158,7 +149,8 @@ export class StructuredAgentSessionHost { ), publishStatus: this.clientDelivery.publishStatusAndSettlement, serialize: (sessionId, task) => this.tasks.trackAttach(this.serialize(sessionId, task)), - now: () => this.now() + now: () => this.now(), + finishStopAfterExit: (id, child) => this.lifetime.finishStopAfterExit(id, child) }) this.restartResume = createStructuredAgentSessionRestartResume(deps, this.sessions, { ...structuredAgentSessionRestartResumeSurfaces(this, this.now), @@ -193,7 +185,8 @@ export class StructuredAgentSessionHost { sessions: this.sessions, now: () => this.now(), publishStatus: this.clientDelivery.publishStatus, - wakeDelivery: (sessionId: string) => this.conversationDelivery.loop.wake(sessionId) + wakeDelivery: (sessionId: string) => this.conversationDelivery.loop.wake(sessionId), + finishStopAfterExit: (id, child) => this.lifetime.finishStopAfterExit(id, child) } satisfies StructuredAgentSessionLifetimeContext } @@ -203,7 +196,7 @@ export class StructuredAgentSessionHost { ...this.lifetimeContext(), subscribers: this.subscribers, tasks: this.tasks, - reconcileLeases: (sessionId) => this.reconcileLeases(sessionId), + reconcileLeases: (sessionId) => this.restore.reconcileLeases(sessionId), serialize: (sessionId, task) => this.serialize(sessionId, task), openConversation: this.conversationDelivery.open } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts index 3046aa971f67..90aa3f16e23b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts @@ -10,6 +10,8 @@ // a send does. And a journal it cannot open is not a refusal — the chat shows that failure with a // Retry, so the tab is worth publishing either way. +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire' import { agentSessionRefusalError } from '../../../shared/agent-session-wire-refusals' import { adapterSupportsRecord } from './structured-agent-session-provider-support' @@ -17,6 +19,7 @@ import { StructuredAgentSessionReadableRestorer } from './structured-agent-sessi import { StructuredAgentSessionRestartRestoreGate } from './structured-agent-session-restart-restore-gate' import { createReaderReconcile, + createRestartReconciler, reportEachFailureOnce } from './structured-agent-session-restart-reconcile' import type { @@ -57,21 +60,30 @@ export async function revealStructuredAgentSession( } /** The host's startup readable-restore sweep: reconcile, resolve, then open each chat's journal. - * Its lease bookkeeping is a reader's, which never fails a read or startup; startup shares it. */ + * Its lease bookkeeping is a reader's, which never fails a read or startup; startup shares it. + * `reconcileLeases` is the one restart reconciler, which attach and start await strictly. */ export function createStructuredAgentSessionHostRestore( deps: StructuredAgentSessionHostDeps, wiring: Omit< ConstructorParameters[0], 'openDeps' | 'supportsRecord' | 'reconcile' | 'resolveRecovery' > & { - reconcileLeases: (sessionId: string) => Promise + probe: (record: AgentSessionRecord) => Promise + now: () => number resolveRecovery: (sessionId: string) => Promise } ): { + reconcileLeases: (sessionId: string) => Promise reconcileRestartLeases: () => Promise restoreReadableSessions: (sessionIds?: readonly string[]) => Promise } { - const { reconcileLeases, resolveRecovery, ...rest } = wiring + const { probe, now, resolveRecovery, ...rest } = wiring + const reconcileLeases = createRestartReconciler({ + store: deps.store, + probe, + ...(deps.probeOwners ? { probeMany: deps.probeOwners } : {}), + now + }) const failures = reportEachFailureOnce(deps.logger) const reconcile = createReaderReconcile(reconcileLeases, failures) const restorer = new StructuredAgentSessionReadableRestorer({ @@ -91,6 +103,7 @@ export function createStructuredAgentSessionHostRestore( }) const gate = new StructuredAgentSessionRestartRestoreGate() return { + reconcileLeases, reconcileRestartLeases: async () => { await reconcile('startup') }, diff --git a/src/main/runtime/structured-agent-session-failure-logging.test.ts b/src/main/runtime/structured-agent-session-failure-logging.test.ts index 9a8d9edb96f1..3d73f0e4b20a 100644 --- a/src/main/runtime/structured-agent-session-failure-logging.test.ts +++ b/src/main/runtime/structured-agent-session-failure-logging.test.ts @@ -94,7 +94,7 @@ describe('failures the desktop host used to drop', () => { ) }) - it('logs a provider start or exit the host could not take', async () => { + it('logs a provider start, exit or late exit report the host could not take', async () => { const log = recordingStructuredAgentSessionLogger() const error = new Error('host refused the event') const lifecycle = createStructuredAgentSessionLifecycleDelivery({ @@ -121,10 +121,17 @@ describe('failures the desktop host used to drop', () => { fence: 1, acquisitionGeneration: 'generation-1' }) + lifecycle.deliver({ + type: 'exitAfterClose', + sessionId: SESSION, + fence: 1, + acquisitionGeneration: 'generation-1' + }) await lifecycle.drain() expect(log.entries.map((entry) => entry.fields)).toEqual([ { scope: 'lifecycle-started', sessionId: SESSION, error }, + { scope: 'lifecycle-exit-after-close', sessionId: SESSION, error }, { scope: 'lifecycle-exit', sessionId: SESSION, error } ]) }) diff --git a/src/main/runtime/structured-agent-session-lifecycle-delivery.test.ts b/src/main/runtime/structured-agent-session-lifecycle-delivery.test.ts new file mode 100644 index 000000000000..8d69af1189f2 --- /dev/null +++ b/src/main/runtime/structured-agent-session-lifecycle-delivery.test.ts @@ -0,0 +1,45 @@ +import { expect, it, vi } from 'vitest' +import { recordingStructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support' +import { createStructuredAgentSessionLifecycleDelivery } from './structured-agent-session-lifecycle-delivery' + +it("keeps a late exit report off the exit chain, so its wait on its own session holds up no other session's exit, and drain still waits for it", async () => { + const sessionLane = Promise.withResolvers() + const handled: string[] = [] + const delivery = createStructuredAgentSessionLifecycleDelivery({ + handle: async (event) => { + if (event.type === 'exitAfterClose') { + await sessionLane.promise + } + handled.push(`${event.type}:${event.sessionId}`) + }, + logger: recordingStructuredAgentSessionLogger().logger, + drainObservedExits: async () => {} + }) + + delivery.deliver({ + type: 'exitAfterClose', + sessionId: 'session-a', + fence: 1, + acquisitionGeneration: 'a-1' + }) + delivery.deliver({ + type: 'ended', + sessionId: 'session-b', + reason: 'exited', + cause: 'unexpected-exit', + fence: 1, + acquisitionGeneration: 'b-1' + }) + await vi.waitFor(() => expect(handled).toEqual(['ended:session-b'])) + + let drained = false + const draining = delivery.drain().then(() => { + drained = true + }) + await new Promise((resolve) => setImmediate(resolve)) + expect(drained).toBe(false) + + sessionLane.resolve() + await draining + expect(handled).toEqual(['ended:session-b', 'exitAfterClose:session-a']) +}) diff --git a/src/main/runtime/structured-agent-session-lifecycle-delivery.ts b/src/main/runtime/structured-agent-session-lifecycle-delivery.ts index ccf026de359c..ce9a023783d1 100644 --- a/src/main/runtime/structured-agent-session-lifecycle-delivery.ts +++ b/src/main/runtime/structured-agent-session-lifecycle-delivery.ts @@ -3,12 +3,18 @@ // Exit recovery runs on one chain so teardown can drain it: exit callbacks arrive from child // process tasks, and a fire-and-forget one could otherwise append after the host flushed and // removed its journal directory. That chain orders nothing across sessions, and a recovery on it -// can run a whole reacquisition, so `started` stays off it: it takes only its own session's -// serialized step, and is tracked here so the same drain still waits for it. +// can run a whole reacquisition, so `started` and `exitAfterClose` stay off it: each takes only its +// own session's serialized step, and is tracked here so the same drain still waits for it. import type { StructuredAgentSessionLifecycleEvent } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' +const LOG_SCOPES = { + started: 'lifecycle-started', + ended: 'lifecycle-exit', + exitAfterClose: 'lifecycle-exit-after-close' +} as const satisfies Record + export function createStructuredAgentSessionLifecycleDelivery(input: { handle: (event: StructuredAgentSessionLifecycleEvent) => Promise | undefined logger: StructuredAgentSessionLogger @@ -20,13 +26,13 @@ export function createStructuredAgentSessionLifecycleDelivery(input: { drain: () => Promise } { let recoveryChain = Promise.resolve() - const settlingStarts = new Set>() + const settlingOffChain = new Set>() const settle = async (event: StructuredAgentSessionLifecycleEvent): Promise => { try { await input.handle(event) } catch (error) { input.logger.warn(`delivering a provider ${event.type} event to the host failed`, { - scope: event.type === 'started' ? 'lifecycle-started' : 'lifecycle-exit', + scope: LOG_SCOPES[event.type], sessionId: event.sessionId, error }) @@ -34,11 +40,11 @@ export function createStructuredAgentSessionLifecycleDelivery(input: { } return { deliver: (event) => { - if (event.type === 'started') { + if (event.type !== 'ended') { // Called now, so the step is queued on its session ahead of any later exit of that child. const settling = settle(event) - settlingStarts.add(settling) - void settling.finally(() => settlingStarts.delete(settling)) + settlingOffChain.add(settling) + void settling.finally(() => settlingOffChain.delete(settling)) return } recoveryChain = recoveryChain.then(() => settle(event)) @@ -50,8 +56,8 @@ export function createStructuredAgentSessionLifecycleDelivery(input: { await input.drainObservedExits() const observed = recoveryChain await observed - if (settlingStarts.size > 0) { - await Promise.all(settlingStarts) + if (settlingOffChain.size > 0) { + await Promise.all(settlingOffChain) continue } if (observed === recoveryChain) { diff --git a/src/main/runtime/structured-agent-session-runtime.ts b/src/main/runtime/structured-agent-session-runtime.ts index 3785fdeaf5ce..0fdbee3c5b43 100644 --- a/src/main/runtime/structured-agent-session-runtime.ts +++ b/src/main/runtime/structured-agent-session-runtime.ts @@ -52,6 +52,7 @@ import { createStructuredAgentEnvironmentResolvers } from './structured-agent-sh import type { ClaudeStructuredAuthPolicy } from '../claude-accounts/claude-structured-auth-policy' import { createStructuredClaudeRuntimeAdapter } from './structured-claude-runtime-adapter' import { createStructuredAgentSessionLifecycleDelivery } from './structured-agent-session-lifecycle-delivery' +import { structuredCodexLifecycleEvent } from './structured-codex-lifecycle-event' import { createStructuredAgentSessionDispatchFollowUps } from './structured-agent-session-dispatch-followups' import { agentModelCatalogStore } from '../native-chat/agent-model-catalog/agent-model-catalog-store' import { @@ -271,8 +272,9 @@ async function installOnJournal( onDispatchSettledLate, onPrimaryThreadStoppedRunning: releaseUnansweredDispatches, onEvent: (event) => { - if (event.type === 'ended' && 'cause' in event && event.cause === 'unexpected-exit') { - lifecycle.deliver(event) + const lifecycleEvent = structuredCodexLifecycleEvent(event) + if (lifecycleEvent) { + lifecycle.deliver(lifecycleEvent) } } }) diff --git a/src/main/runtime/structured-claude-runtime-adapter.ts b/src/main/runtime/structured-claude-runtime-adapter.ts index 9024c65b503d..b640ae2dec23 100644 --- a/src/main/runtime/structured-claude-runtime-adapter.ts +++ b/src/main/runtime/structured-claude-runtime-adapter.ts @@ -44,7 +44,7 @@ export type StructuredClaudeRuntimeAdapterDeps = { export function structuredClaudeLifecycleEvent( event: ClaudeStructuredSessionEvent ): StructuredAgentSessionLifecycleEvent | null { - if (event.type === 'started') { + if (event.type === 'started' || event.type === 'exitAfterClose') { return event } if ( diff --git a/src/main/runtime/structured-codex-lifecycle-event.ts b/src/main/runtime/structured-codex-lifecycle-event.ts new file mode 100644 index 000000000000..fbf7a63b91ec --- /dev/null +++ b/src/main/runtime/structured-codex-lifecycle-event.ts @@ -0,0 +1,15 @@ +import type { CodexStructuredSessionEvent } from '../codex/codex-structured-session-state' +import type { StructuredAgentSessionLifecycleEvent } from '../native-chat/agent-session-wire/structured-agent-session-adapter' + +/** The Codex adapter events the host's lifecycle handler consumes. A close the host asked for and + * saw land is settled by that close, so its `ended` stays here. */ +export function structuredCodexLifecycleEvent( + event: CodexStructuredSessionEvent +): StructuredAgentSessionLifecycleEvent | null { + if (event.type === 'exitAfterClose') { + return event + } + return event.type === 'ended' && 'cause' in event && event.cause === 'unexpected-exit' + ? event + : null +} diff --git a/src/shared/child-process/retryable-process-exit-proof.test.ts b/src/shared/child-process/retryable-process-exit-proof.test.ts index a7bc57976452..73959d19bf28 100644 --- a/src/shared/child-process/retryable-process-exit-proof.test.ts +++ b/src/shared/child-process/retryable-process-exit-proof.test.ts @@ -13,6 +13,7 @@ describe('RetryableProcessExitProof', () => { await expect(Promise.all([first, concurrent])).resolves.toEqual([true, true]) await expect(proof.run(proveExit)).resolves.toBe(true) expect(proveExit).toHaveBeenCalledOnce() + expect(proof.holdsAttempt).toBe(true) }) it('permits another attempt after exit was not proven', async () => { @@ -20,6 +21,7 @@ describe('RetryableProcessExitProof', () => { const proveExit = vi.fn().mockResolvedValueOnce(false).mockResolvedValueOnce(true) await expect(proof.run(proveExit)).resolves.toBe(false) + expect(proof.holdsAttempt).toBe(false) await expect(proof.run(proveExit)).resolves.toBe(true) expect(proveExit).toHaveBeenCalledTimes(2) }) diff --git a/src/shared/child-process/retryable-process-exit-proof.ts b/src/shared/child-process/retryable-process-exit-proof.ts index f8c3a89de316..d5538360ea75 100644 --- a/src/shared/child-process/retryable-process-exit-proof.ts +++ b/src/shared/child-process/retryable-process-exit-proof.ts @@ -1,6 +1,11 @@ export class RetryableProcessExitProof { private inFlight: Promise | null = null + /** False before any attempt and after one gave up; a proven attempt stays held. */ + get holdsAttempt(): boolean { + return this.inFlight !== null + } + run(proveExit: () => Promise): Promise { if (this.inFlight) { return this.inFlight