diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index 2b065bb5639e..466b707aa7b6 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -26,16 +26,14 @@ import { type ClaudeStructuredSessionAdapterDeps, type ClaudeStructuredSessionEvent } from './claude-structured-session-state' -import { - closeAllClaudeSessions, - closeClaudeSession, - finishClaudeCloseAfterExit -} from './claude-structured-session-close' +import { closeAllClaudeSessions, closeClaudeSession } from './claude-structured-session-close' import { claudeStoppedRequestEndWait } from './claude-request-end-wait' import { drainClaudeObservedExits, + finishClaudeExitedClose, observeClaudeSessionExit, settleClaudeUnexpectedExit, + withClaudeSessionOrRejected, type ClaudeExitLifecycle } from './claude-structured-session-exit-lifecycle' import type { AgentSessionBackgroundTaskState } from '../../shared/agent-session-wire' @@ -66,7 +64,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda private readonly sessions = new Map() private readonly acquisitions = new ClaudeAcquisitionRegistry() private readonly exits = new Map() - private readonly settledExitErrors = new Map() + private readonly settledExits: ClaudeExitLifecycle['settledExits'] = new Map() private readonly exitLifecycle: ClaudeExitLifecycle constructor(private readonly deps: ClaudeStructuredSessionAdapterDeps) { @@ -74,7 +72,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda this.exitLifecycle = { sessions: this.sessions, exits: this.exits, - settledExitErrors: this.settledExitErrors, + settledExits: this.settledExits, deps, emit: (session, event) => this.emit(session, event) } @@ -83,7 +81,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda supportsLocation = supportsClaudeStructuredLocation acquire = (input: StructuredAgentSessionAcquireInput): Promise => { - this.settledExitErrors.delete(input.identity.sessionId) + this.settledExits.delete(input.identity.sessionId) return acquireClaudeSession({ input, deps: this.deps, @@ -96,13 +94,9 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda handleExit: (sessionId, attempt, error) => observeClaudeSessionExit(this.exitLifecycle, sessionId, attempt, error), finishClose: (sessionId, attempt) => - finishClaudeCloseAfterExit({ - sessions: this.sessions, - sessionId, - connection: attempt.connection, - deps: this.deps, - afterClose: (close) => this.afterClose(sessionId, close) - }), + finishClaudeExitedClose(this.exitLifecycle, sessionId, attempt.connection, (close) => + this.afterClose(sessionId, close) + ), settleExit: (sessionId, exit) => settleClaudeUnexpectedExit(this.exitLifecycle, sessionId, exit) } @@ -172,10 +166,14 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda } dispatch: StructuredAgentSessionAdapter['dispatch'] = (input) => - dispatchClaudeTurn(this.session(input.sessionId), input, input.beforeDispatch) + withClaudeSessionOrRejected(this.exitLifecycle, input.sessionId, (session) => + dispatchClaudeTurn(session, input, input.beforeDispatch) + ) compact: NonNullable = (input) => - dispatchClaudeCommand(this.session(input.sessionId), input.command) + withClaudeSessionOrRejected(this.exitLifecycle, input.sessionId, (session) => + dispatchClaudeCommand(session, input.command) + ) cancelTurn: StructuredAgentSessionAdapter['cancelTurn'] = (request) => cancelClaudeStructuredTurn({ @@ -272,7 +270,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda closeSession = (sessionId: string): Promise => // After the close, not before: releasing an exit still settling settles it on the way. - this.closeSessionProcess(sessionId).finally(() => this.settledExitErrors.delete(sessionId)) + this.closeSessionProcess(sessionId).finally(() => this.settledExits.delete(sessionId)) private closeSessionProcess(sessionId: string): Promise { // An exit seen first settles as that exit, whoever asked for the close after it. @@ -307,7 +305,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda // A child that just exited is named by its own diagnostic, not by its absence. throw ( this.exits.get(sessionId)?.error ?? - this.settledExitErrors.get(sessionId) ?? + this.settledExits.get(sessionId)?.error ?? new Error(`no live claude stream-json session for ${sessionId}`) ) } diff --git a/src/main/claude/claude-structured-session-exit-blame.test.ts b/src/main/claude/claude-structured-session-exit-blame.test.ts index 79430131e15c..8f992df7efd0 100644 --- a/src/main/claude/claude-structured-session-exit-blame.test.ts +++ b/src/main/claude/claude-structured-session-exit-blame.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from 'vitest' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { structuredAgentSessionCommandTurn } from '../native-chat/agent-session-wire/structured-agent-session-command-turn' import type { ClaudeStructuredSessionEvent } from './claude-structured-session-adapter' import { adapterFor, @@ -63,4 +64,41 @@ describe('what a started Claude session says ended it', () => { }) expect(connection.closeCount).toBeGreaterThanOrEqual(1) }) + + it('rejects a message or a compaction for a child it no longer holds, with why it ended', async () => { + const claude = fakeClaude() + const adapter = adapterFor(claude, {}, []) + await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) + const connection = claude.connections[0] + // The close cannot prove the child gone, so the adapter lets go of it and keeps its exit. + connection.close = async () => false + connection.handlers.onExit?.(new Error('claude stream-json exited (code 1): killed')) + await adapter.drainObservedExits() + + const rejected = { + state: 'rejected', + rejection: { kind: 'providerExited' } + } + await expect( + adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-1', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] }, + fence: 7 + }) + ).resolves.toMatchObject(rejected) + const turn = structuredAgentSessionCommandTurn('client-2') + await expect( + adapter.compact({ + sessionId: 'session-1', + fence: 7, + command: { + clientMessageId: 'client-2', + ...turn, + running: { kind: 'turn', turnId: turn.turnId, state: 'running' } + } + }) + ).resolves.toMatchObject(rejected) + expect(connection.sent).toEqual([]) + }) }) diff --git a/src/main/claude/claude-structured-session-exit-lifecycle.ts b/src/main/claude/claude-structured-session-exit-lifecycle.ts index 854646490a8c..d536a22d71f5 100644 --- a/src/main/claude/claude-structured-session-exit-lifecycle.ts +++ b/src/main/claude/claude-structured-session-exit-lifecycle.ts @@ -1,11 +1,19 @@ -import { agentSessionFailureFact, providerDiagnosticOf } from '../../shared/agent-session-failure' +import { + agentSessionFailureFact, + providerDiagnosticOf, + type SubmissionRejectionFact +} from '../../shared/agent-session-failure' import { providerExitObserved, providerStartupFailureFact } from '../native-chat/agent-session-wire/structured-agent-session-failure-text' +import type { AgentSessionDispatchOutcome } from '../native-chat/agent-session-wire/structured-agent-session-adapter' +import { claudeDispatchRejection } from './claude-structured-dispatch-content' +import type { ClaudeStreamJsonConnection } from './claude-stream-json-connection' import { settledClaudeTurnEndLeaf } from './claude-structured-resume-point' import { claudeRootExitObserved, + finishClaudeCloseAfterExit, settleClaudeExitedSession } from './claude-structured-session-close' import { failClaudeStartup } from './claude-structured-session-startup-state' @@ -17,12 +25,15 @@ import type { ClaudeStructuredSessionEvent } from './claude-structured-session-state' +export type ClaudeSettledExit = { error: Error; failure: SubmissionRejectionFact } + export type ClaudeExitLifecycle = { sessions: Map exits: Map - /** A settled exit's diagnostic, kept for a send admitted before the host heard of the exit. */ - settledExitErrors: Map - deps: Pick + /** A settled exit's diagnostic and why it ended, kept for a send admitted before the host heard + * of the exit. */ + settledExits: Map + deps: Pick emit: (session: ClaudeSession, event: ClaudeStructuredSessionEvent) => void } @@ -34,6 +45,7 @@ export function observeClaudeSessionExit( ): void { const session = lifecycle.sessions.get(sessionId) if (!session || session.connection !== attempt.connection) { + settleRetainedClaudeRootExit(lifecycle, sessionId, attempt.connection) return } lifecycle.sessions.delete(sessionId) @@ -55,6 +67,7 @@ export function observeClaudeSessionExit( .then((proven) => { // A first-hand root exit is final like a proven one: the owner releases the lease on it. if (!proven && !claudeRootExitObserved(session.connection)) { + reportClaudeUnprovenEnd(lifecycle, sessionId, exit) return undefined } return settleClaudeUnexpectedExit(lifecycle, sessionId, exit) @@ -62,6 +75,73 @@ export function observeClaudeSessionExit( .catch(() => undefined) } +/** The root exited after a close began: that close settles it, or one the adapter began after + * letting the session go ends the exit it kept. */ +export function finishClaudeExitedClose( + lifecycle: ClaudeExitLifecycle, + sessionId: string, + connection: ClaudeStreamJsonConnection | null, + afterClose: (close: () => Promise) => Promise +): void { + settleRetainedClaudeRootExit(lifecycle, sessionId, connection) + finishClaudeCloseAfterExit({ + sessions: lifecycle.sessions, + sessionId, + connection, + deps: lifecycle.deps, + afterClose + }) +} + +/** The root of an exit whose close could not prove it gone has now exited: the end it reported + * unproven is final. Only the retained exit's own connection reports it, and only once that + * exit's first close has answered, which publishes the end itself when it saw the root go. */ +function settleRetainedClaudeRootExit( + lifecycle: ClaudeExitLifecycle, + sessionId: string, + connection: ClaudeStreamJsonConnection | null +): void { + const exit = lifecycle.exits.get(sessionId) + if (!exit || exit.connection !== connection) { + return + } + const published = exit.publication ?? Promise.resolve() + exit.publication = published.then(() => + lifecycle.exits.get(sessionId) === exit && claudeRootExitObserved(exit.connection) + ? settleClaudeUnexpectedExit(lifecycle, sessionId, exit).catch(() => undefined) + : undefined + ) +} + +/** Never silence: the host keeps the child on record, closing, until its exit is proven or seen. */ +function reportClaudeUnprovenEnd( + lifecycle: ClaudeExitLifecycle, + sessionId: string, + exit: ClaudeSessionExit +): void { + if (lifecycle.exits.get(sessionId) !== exit) { + return + } + lifecycle.deps.onEvent?.({ + type: 'end-unproven', + sessionId, + reason: exit.error.message, + failure: claudeExitFailure(exit), + fence: exit.session.fence, + acquisitionGeneration: exit.session.acquisitionGeneration + }) +} + +/** A start that never landed says why it failed. After it landed, only the child's own exit blames + * the provider; an Orca fault that closed it is Orca's. */ +function claudeExitFailure(exit: Pick) { + return exit.session.startup.state !== 'proven' + ? providerStartupFailureFact(exit.session.startup.failure ?? exit.error) + : providerExitObserved(exit.error) + ? agentSessionFailureFact('providerExited', { detail: providerDiagnosticOf(exit.error) }) + : agentSessionFailureFact('hostFault') +} + /** 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( @@ -87,21 +167,12 @@ export function settleClaudeUnexpectedExit( return } exits.delete(sessionId) - lifecycle.settledExitErrors.set(sessionId, exit.error) + lifecycle.settledExits.set(sessionId, { error: exit.error, failure: claudeExitFailure(exit) }) const ended: ClaudeStructuredSessionEvent = { type: 'ended', sessionId, reason: exit.error.message, - // A start that never landed says why it failed. After it landed, only the child's own exit - // blames the provider; an Orca fault that closed it is Orca's. - failure: - exit.session.startup.state !== 'proven' - ? providerStartupFailureFact(exit.session.startup.failure ?? exit.error) - : providerExitObserved(exit.error) - ? agentSessionFailureFact('providerExited', { - detail: providerDiagnosticOf(exit.error) - }) - : agentSessionFailureFact('hostFault'), + failure: claudeExitFailure(exit), cause: 'unexpected-exit', fence: exit.session.fence, acquisitionGeneration: exit.session.acquisitionGeneration, @@ -117,6 +188,29 @@ export function settleClaudeUnexpectedExit( return exit.settlementPromise } +/** A message for a child this adapter no longer serves was never written, so it is rejected with + * why that child ended, never left in doubt. */ +function rejectClaudeDetachedDispatch( + lifecycle: Pick, + sessionId: string +): AgentSessionDispatchOutcome { + const exit = lifecycle.exits.get(sessionId) + const failure = exit + ? claudeExitFailure(exit) + : (lifecycle.settledExits.get(sessionId)?.failure ?? agentSessionFailureFact('providerExited')) + return { state: 'rejected', ...claudeDispatchRejection(failure) } +} + +/** Runs `run` on the live session, or rejects what was meant for a child this adapter let go of. */ +export async function withClaudeSessionOrRejected( + lifecycle: Pick, + sessionId: string, + run: (session: ClaudeSession) => Promise +): Promise { + const session = lifecycle.sessions.get(sessionId) + return session ? run(session) : rejectClaudeDetachedDispatch(lifecycle, sessionId) +} + /** Wait for each first-hand exit's publication, including exits observed while waiting. */ export async function drainClaudeObservedExits( exits: Map diff --git a/src/main/claude/claude-structured-session-state.ts b/src/main/claude/claude-structured-session-state.ts index 876090c07e4d..6c8287a0c8fe 100644 --- a/src/main/claude/claude-structured-session-state.ts +++ b/src/main/claude/claude-structured-session-state.ts @@ -6,7 +6,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 { + StructuredAgentSessionEndUnprovenEvent, + StructuredAgentSessionStartedEvent +} from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { ClaudeStreamJsonConnection, openClaudeStreamJsonConnection @@ -67,6 +70,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 + | StructuredAgentSessionEndUnprovenEvent | { type: 'ended' sessionId: string diff --git a/src/main/codex/codex-structured-journal-translation.ts b/src/main/codex/codex-structured-journal-translation.ts index 20fd8bf58df3..d154eb717066 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) => { + // Host lifecycle only: the journal records this child's end once its exit is seen. + if (event.type === 'end-unproven') { + return CODEX_JOURNAL_ADMITTED + } if (event.type === 'ended') { const streamAdmission = flushStreams() if (!streamAdmission.accepted) { diff --git a/src/main/codex/codex-structured-session-adapter-lifecycle.test.ts b/src/main/codex/codex-structured-session-adapter-lifecycle.test.ts index 3066b5613364..b996c1059752 100644 --- a/src/main/codex/codex-structured-session-adapter-lifecycle.test.ts +++ b/src/main/codex/codex-structured-session-adapter-lifecycle.test.ts @@ -188,12 +188,29 @@ describe('CodexStructuredSessionAdapter lifecycle', () => { body: USER_MESSAGE, fence: 7 }) - ).rejects.toThrow('no live codex app-server') + // Never written to a child it no longer serves: rejected, never left in doubt. + ).resolves.toMatchObject({ state: 'rejected', rejection: { kind: 'providerExited' } }) expect(adapter.backgroundTaskStops('session-1')).toBeDefined() await expect(adapter.closeSession('session-1')).resolves.toBe(false) expect(events.filter((event) => event.type === 'ended')).toHaveLength(1) }) + it('rejects a message for a connection that broke while the child may still run, without saying Codex stopped', async () => { + const codex = fakeCodex() + const adapter = await acquired(codex) + codex.connections[0].closed = true + + await expect( + adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-1', + body: USER_MESSAGE, + fence: 7 + }) + ).resolves.toMatchObject({ state: 'rejected', rejection: { kind: 'writeFailed' } }) + expect(codex.connections[0].calls.some((call) => call.method === 'turn/start')).toBe(false) + }) + it('keeps the live session when a child it already replaced dies', async () => { const codex = fakeCodex() const events: CodexStructuredSessionEvent[] = [] diff --git a/src/main/codex/codex-structured-session-adapter.test.ts b/src/main/codex/codex-structured-session-adapter.test.ts index c9e3c14cac91..9dc5fedbb315 100644 --- a/src/main/codex/codex-structured-session-adapter.test.ts +++ b/src/main/codex/codex-structured-session-adapter.test.ts @@ -269,7 +269,8 @@ describe('CodexStructuredSessionAdapter.acquire', () => { body: USER_MESSAGE, fence: 7 }) - ).rejects.toThrow('no live codex app-server') + // Never written to a child it no longer serves: rejected, never left in doubt. + ).resolves.toMatchObject({ state: 'rejected', rejection: { kind: 'providerExited' } }) }) it('classifies launch validation failure as pre-spawn without opening a child', async () => { diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index 11d9a2ffca31..fe67e8b9cee8 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -197,6 +197,10 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap requestedAt?: number beforeDispatch?: () => Promise }): Promise { + const detached = this.detachedRejection(input.sessionId) + if (detached) { + return detached + } const session = this.session(input.sessionId) session.dispatchPending = true try { @@ -207,6 +211,20 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap } } + /** A closed connection writes nothing, so what was meant for it was never sent: rejected, not + * doubt. Only a seen end says Codex stopped; a closing or broken connection may still have it. */ + private detachedRejection(sessionId: string): AgentSessionDispatchOutcome | null { + const session = this.sessions.get(sessionId) + if (session && !session.ended && !session.connection.closed) { + return null + } + const gone = !session || session.ended + return { + state: 'rejected', + ...codexDispatchRejection(agentSessionFailureFact(gone ? 'providerExited' : 'writeFailed')) + } + } + cancelTurn: StructuredAgentSessionAdapter['cancelTurn'] = (request) => cancelCodexStructuredTurn({ request, @@ -227,6 +245,10 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap /** The ack is Codex's receipt; the translator ends the command's turn from the turn it opens. */ compact: NonNullable = async (input) => { + const detached = this.detachedRejection(input.sessionId) + if (detached) { + return detached + } const session = this.session(input.sessionId) session.translator?.beginCommand(input.command) try { diff --git a/src/main/codex/codex-structured-session-state.ts b/src/main/codex/codex-structured-session-state.ts index 5773b659f5e1..87a23501b7fa 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -20,7 +20,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, + StructuredAgentSessionEndUnprovenEvent +} from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { CodexStructuredPermissionPolicy } from './codex-structured-permission-policy' import type { AgentModelCatalogSessionAccess, @@ -69,6 +72,7 @@ export type CodexStructuredSessionEvent = promptKey: string } | StructuredAgentSessionEndedEvent + | StructuredAgentSessionEndUnprovenEvent /** 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/codex/codex-structured-session-teardown.ts b/src/main/codex/codex-structured-session-teardown.ts index e7395fb3c845..de427305c092 100644 --- a/src/main/codex/codex-structured-session-teardown.ts +++ b/src/main/codex/codex-structured-session-teardown.ts @@ -4,6 +4,7 @@ // session owns are cleared exactly once, and only when the child was actually // proven stopped — a refused close leaves the session indexed for a retry. +import { agentSessionFailureFact } from '../../shared/agent-session-failure' import { AgentSessionAcquisitionRootExitObservedError } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' import { @@ -81,7 +82,20 @@ export class CodexStructuredSessionTeardown { expectedFence: fence, expectedAcquisitionGeneration: acquisitionGeneration, unexpectedReason: reason - }).then((closed) => this.settled(sessionId, closed)) + }).then((closed) => { + // Never silence: the host keeps the child, closing, until its exit is seen or proven. + if (!closed && this.deps.sessions.get(sessionId) === session && !session.ended) { + this.deps.onEvent?.({ + type: 'end-unproven', + sessionId, + reason: reason.message, + failure: agentSessionFailureFact('hostFault'), + fence, + acquisitionGeneration + }) + } + return this.settled(sessionId, closed) + }) } closeAll = (): Promise => 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 f94a3b134201..72c54447f043 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 @@ -217,8 +217,21 @@ export type StructuredAgentSessionStartedEvent = { restoreSkippedOptions: readonly string[] } +/** The adapter no longer serves this child, and its stop could not prove the root gone: it may + * still be running. The adapter keeps it for a retry, and its exit, whenever seen, is `ended`. */ +export type StructuredAgentSessionEndUnprovenEvent = { + type: 'end-unproven' + sessionId: string + /** Log text only. */ + reason: string + failure?: SubmissionRejectionFact + fence: number + acquisitionGeneration: string +} + export type StructuredAgentSessionLifecycleEvent = | StructuredAgentSessionEndedEvent + | StructuredAgentSessionEndUnprovenEvent | StructuredAgentSessionStartedEvent /** Whether the provider child behind an acquisition has proven its start. A publish-first diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts index c606a73e310f..48e21e2c861c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts @@ -9,6 +9,7 @@ import { structuredAgentSessionFailureWordsContext } from './structured-agent-se import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import type { StructuredAgentSessionEndedEvent } from './structured-agent-session-adapter' import type { + StructuredAgentSessionEndedChild, StructuredAgentSessionHostSession, StructuredAgentSessionProviderChild } from './structured-agent-session-host-types' @@ -99,8 +100,11 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< if (!session || session.child !== child) { return } - const { expected } = exit const close = child.close + // An end the adapter reported but could not prove is settled as that end, however its exit came. + const reported = close?.reported + const expected = exit.expected && !reported + const failure = reported ? reported.failure : exit.failure // Receipt of the exit is the one end time the host may record for a running turn. const observedAt = exit.observedAt ?? context.now() // The host's own phase decides, so a provider that omits the flag still gets a start that @@ -110,14 +114,15 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< endProviderChild(session, { generation: child.generation, fence: child.fence, - // A close keeps the cause of the stop that asked for it, and ends where it was asked. - cause: expected ? (close?.cause ?? 'evict') : 'exit', - reason: expected ? (close?.reason ?? null) : exit.reason, - ...(!expected && exit.failure ? { failure: exit.failure } : {}), + // A close keeps the cause of the stop that asked for it, and ends where it was asked. A + // reported end keeps a Stop or close asked since; none asked, it was the child's own end. + cause: expected ? (close?.cause ?? 'evict') : reported ? reportedEndCause(close) : 'exit', + reason: expected ? (close?.reason ?? null) : (reported?.reason ?? exit.reason), + ...(!expected && failure ? { failure } : {}), duringStartup: exitedDuringStartup, // The adapter publishes an exit only once it saw the root go, first-hand or proven. rootGone: true, - ...(expected && close ? { endedAt: close.requestedAt } : {}) + ...((expected || reported) && close ? { endedAt: close.requestedAt } : {}) }) context.publishStatus?.(sessionId) } @@ -158,12 +163,14 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< showUnexpectedExitOutcome: !expected && (exitedDuringStartup || - unfinishedStructuredAgentSessionWorkWasInterrupted( - unfinishedWork, - session.journal, - observedAt - )), - ...(!expected && exit.failure ? { exitFailure: exit.failure } : {}), + (reported + ? reported.interruptedWork + : unfinishedStructuredAgentSessionWorkWasInterrupted( + unfinishedWork, + session.journal, + observedAt + ))), + ...(!expected && failure ? { exitFailure: failure } : {}), ...(!expected && exitedDuringStartup && child.generation ? { exitedDuringStartup: { generation: child.generation } } : {}) @@ -209,6 +216,14 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< } } +/** A Stop, close or quit asked for since an unproven end keeps its cause; with none, the end was + * the child's own. */ +function reportedEndCause( + close: StructuredAgentSessionProviderChild['close'] +): StructuredAgentSessionEndedChild['cause'] { + return close && close.cause !== 'host-stop' ? close.cause : 'exit' +} + function logExitFailure( context: Pick, sessionId: string, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-unproven-fault-send.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-unproven-fault-send.test.ts new file mode 100644 index 000000000000..2438c2ba5093 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-unproven-fault-send.test.ts @@ -0,0 +1,499 @@ +// A Claude child ended by an Orca fault whose close could not prove it gone, then the user's next +// messages, on the shipping adapter and host. The child may still be running, so the host keeps it +// on record, closing: a message joins that close and retries the kill, and is refused readably while +// the exit stays unproven, never reaching the old child and never written off as `unknown`. The old +// child's exit ends the record as the fault it was, and the next message goes to a new child. + +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 type { AgentStatusStructuredSessionSubject } from '../../../shared/agent-status-subject' +import { AgentHookServer } from '../../agent-hooks/server' +import { ClaudeStructuredSessionAdapter } from '../../claude/claude-structured-session-adapter' +import { + fakeClaude, + PROVIDER_SESSION_ID, + type FakeConnection +} from '../../claude/claude-structured-session-test-support' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +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 { claudeAndCodexAgents } from './structured-agent-session-adapter-router-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' } +const CAPABILITIES = ['interrupt_receipt_v1', 'interrupt_cancel_queued_v1', 'msg_lifecycle_v1'] + +let root: string +let host: StructuredAgentSessionHost +let adapter: ClaudeStructuredSessionAdapter +let store: AgentSessionRecordStore +let claude: ReturnType +let server: AgentHookServer +let statusSubject: AgentStatusStructuredSessionSubject | undefined + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-claude-unproven-fault-send-')) + resetHostTestOperationIds() + server = new AgentHookServer() + statusSubject = undefined + claude = fakeClaude({ + replayUuid: null, + routes: { interrupt: () => ({ still_queued: [], cancelled: [] }) } + }) + const lifecycle: Promise[] = [] + adapter = new ClaudeStructuredSessionAdapter({ + resolveLaunch: async () => ({ + pathToClaudeCodeExecutable: 'claude', + options: {}, + cwd: root, + claudeConfigDir: join(root, 'claude-home'), + providerSessionId: PROVIDER_SESSION_ID, + resumeLeafUuid: null, + resumesTranscript: (store.getRecord(SESSION)?.providerHandleChain.length ?? 0) > 0, + continuesChain: (store.getRecord(SESSION)?.providerHandleChain.length ?? 0) > 0 + }), + onEvent: (event) => { + const mapped = structuredClaudeLifecycleEvent(event) + if (mapped) { + lifecycle.push(host.handleAdapterEvent(mapped)) + } + }, + onDispatchSettledLate: (settlement) => void host.settleLateDispatch(settlement), + onChildWorkEvidence: (sessionId, evidence) => + host.publishChildWorkEvidence(sessionId, evidence), + openConnection: claude.openConnection, + readProcessStartTime: async () => 1_700_000_000_000, + now: () => NOW + }) + store = await openTestAgentSessionRecordStore(root) + host = new StructuredAgentSessionHost({ + agents: claudeAndCodexAgents(adapter), + store, + adapter: Object.assign(adapter, { supportsCreate: () => true }), + journalDatabase: openTestJournalHostDatabase(root), + claimKeyId: 'key-1', + mintSpawnToken: () => 'spawn-a', + logger: recordingStructuredAgentSessionLogger().logger, + idleSweep: { intervalMs: 1_000_000_000, idleMs: 0 }, + statusSink: { + publish: (summary, subject) => { + statusSubject = subject + server.ingestStructuredStatus(summary, subject) + }, + forget: (subject) => server.dropStructuredStatus(subject), + publishChildWork: (subject, evidence, provider) => + server.ingestStructuredChildWork(subject, evidence, provider), + readChildWork: (subject) => server.getStructuredChildWorkViews(subject) + }, + now: () => NOW + }) + const params = hostTestAttachParams(null, { + provider: 'claude', + agent: 'claude', + accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: join(root, 'claude-home') }, + providerHandle: { kind: 'claude', sessionId: PROVIDER_SESSION_ID, leafUuid: null } + }) + expect(await host.attach(CALLER, params)).toMatchObject({ ok: true }) + await adapter.awaitStarted(SESSION) + await Promise.all(lifecycle) +}) + +afterEach(async () => { + // Every child the test left unproven has exited by now. + for (const connection of claude.connections) { + connection.exitVerdict = { root: 'exited', tree: 'exited' } + } + await adapter.closeAll() + await host.flushAllStreamedEvents() + await rm(root, { recursive: true, force: true }) +}) + +function eventually(assertion: () => T | Promise): Promise { + return vi.waitFor(assertion, { timeout: 10_000 }) +} + +function frame(connection: FakeConnection, message: Record): void { + connection.handlers.onMessage?.({ session_id: PROVIDER_SESSION_ID, ...message }) +} + +function wrote(connection: FakeConnection, text: string): boolean { + return connection.sent.some((message) => JSON.stringify(message).includes(text)) +} + +function envelope( + method: 'agentSession.send' | 'agentSession.cancel', + fields: Parameters[0]['fields'] +) { + return { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: store.getRecord(SESSION)!.lease.runtimeFence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method, + sessionId: SESSION, + fields + }) + } +} + +async function send(text: string): Promise { + const body = hostTestMessage(text) + const sent = await host.send(CALLER, { envelope: envelope('agentSession.send', { body }), body }) + if (!sent.ok) { + throw new Error(`send refused: ${JSON.stringify(sent.refusal)}`) + } + return sent.value.clientMessageId +} + +function stop() { + return host.cancel(CALLER, { envelope: envelope('agentSession.cancel', {}) }) +} + +async function submission(clientMessageId: string) { + await host.flushStreamedEvents(SESSION) + return (await host.journalSnapshot(SESSION)).submissions.find( + (entry) => entry.clientMessageId === clientMessageId + ) +} + +function laneDrained(): Promise { + return host['tasks'].serialize(SESSION, async () => {}) +} + +function hostSession() { + return host['sessions'].get(SESSION) +} + +function liveChildRecords() { + const records = statusSubject ? server.getStructuredChildWorkViews(statusSubject) : [] + return records.filter((record) => record.membership === 'live') +} + +/** The lead's turn ends with a background task still running, as a real Claude leaves it. */ +async function turnWithBackgroundTask(connection: FakeConnection): Promise { + const text = 'Start a background agent.' + await send(text) + await eventually(() => expect(wrote(connection, text)).toBe(true)) + frame(connection, { + type: 'system', + subtype: 'init', + uuid: 'init-1', + model: 'claude-sonnet-5', + capabilities: CAPABILITIES + }) + const written = connection.sent.at(-1)! + frame(connection, { ...written, uuid: written.uuid }) + frame(connection, { + type: 'assistant', + uuid: 'leaf-1', + parent_tool_use_id: null, + message: { id: 'msg-1', role: 'assistant', content: [{ type: 'text', text: 'Started.' }] } + }) + frame(connection, { + type: 'system', + subtype: 'task_started', + uuid: 'task-start', + task_id: 'background-1', + task_type: 'local_agent', + is_backgrounded: true, + description: 'bg agent' + }) + frame(connection, { type: 'result', subtype: 'success', is_error: false, uuid: 'result-1' }) + await host.flushStreamedEvents(SESSION) +} + +/** As a loaded machine leaves it: no close can prove the child gone; it still runs. */ +function closesUnproven(connection: FakeConnection, pending?: Promise): void { + connection.exitVerdict = { root: 'live', tree: 'unverifiable' } + connection.close = async () => { + connection.closeCount += 1 + connection.closed = true + if (connection.closeCount > 1) { + await pending + } + return connection.exitVerdict.root === 'exited' + } +} + +/** The journal refuses a background-task row, and the adapter's own fault path ends the child. */ +async function faultTheJournal(connection: FakeConnection): Promise { + const sink = host['runtimeState'].eventSinkFor(SESSION).sink + vi.spyOn(sink, 'tryAppendResolvedItemAndPublish').mockReturnValueOnce({ + accepted: false, + reason: 'failed' + }) + frame(connection, { + type: 'system', + subtype: 'task_started', + uuid: 'task-start-2', + task_id: 'background-2', + task_type: 'local_bash', + is_backgrounded: true, + description: 'npm run build' + }) +} + +/** The old child's own exit, as the real connection reports it: inside the close it began. */ +function exits(connection: FakeConnection): void { + connection.exitVerdict = { root: 'exited', tree: 'exited' } + connection.handlers.onExit?.(new Error('claude stream-json exited (code 0)'), { expected: true }) +} + +/** The fault ended the child, its close could not prove it, and the host was told. */ +async function faultedUnproven(connection: FakeConnection, pending?: Promise): Promise { + closesUnproven(connection, pending) + await faultTheJournal(connection) + await adapter.drainObservedExits() + await laneDrained() +} + +const REFUSED = "Couldn't stop Claude from before." +const FAULT_ROW = "Orca ran into a problem, so this didn't go through. Try again." + +async function faultRows() { + await host.flushStreamedEvents(SESSION) + return (await host.journalSnapshot(SESSION)).items.flatMap((item) => + item.body.kind === 'status' && item.body.failure?.kind === 'hostFault' ? [item.body.text] : [] + ) +} + +async function settled(clientMessageId: string) { + return eventually(async () => { + const entry = await submission(clientMessageId) + expect(entry?.dispatchState).not.toBe('pending') + return entry + }) +} + +it('refuses messages readably after a journal fault whose close was unproven, never sending them or leaving them unknown, and sends the next once the old child exits', async () => { + const connection = claude.connections[0]! + await turnWithBackgroundTask(connection) + await faultedUnproven(connection) + const sentBefore = connection.sent.length + + // Before: the dispatch threw on a child the adapter no longer held, and the send read `unknown`. + const held = await send('Hello again after the fault.') + expect(await settled(held)).toMatchObject({ dispatchState: 'rejected' }) + expect((await submission(held))?.reason).toContain(REFUSED) + expect(connection.sent.length).toBe(sentBefore) + // The child may still run: the host keeps it on record, closing, with the fault it reported. + expect(hostSession()?.child).toMatchObject({ + phase: 'ready', + close: { cause: 'host-stop', reported: { failure: { kind: 'hostFault' } } } + }) + const closes = connection.closeCount + + // Stop at rest joins the close and tries the kill again. + await expect(stop()).resolves.toMatchObject({ ok: true }) + expect(connection.closeCount).toBe(closes + 1) + expect(hostSession()?.child?.close).toBeDefined() + + // The old child exits: the record ends with the fault, with no sweep tick, and work goes on. The + // Stop asked for since is the user's. + exits(connection) + await eventually(() => expect(hostSession()?.child).toBeNull()) + expect(hostSession()?.lastEndedChild).toMatchObject({ + cause: 'user-stop', + failure: { kind: 'hostFault' } + }) + expect(await faultRows()).toEqual([FAULT_ROW]) + expect(liveChildRecords().map((record) => record.providerId)).not.toContain('background-1') + + const next = await send('Carry on.') + const resumed = await eventually(() => { + const started = claude.connections.at(-1)! + expect(started).not.toBe(connection) + expect(wrote(started, 'Carry on.')).toBe(true) + return started + }) + const echoed = resumed.sent.at(-1)! + frame(resumed, { ...echoed, uuid: echoed.uuid }) + await eventually(async () => expect((await submission(next))?.dispatchState).toBe('accepted')) + expect(wrote(connection, 'Carry on.')).toBe(false) + + await expect(host['lifetime'].idleSweep.tick()).resolves.toBeUndefined() +}) + +it('rejects, never leaves in doubt, a message dispatched after the adapter let go of a child its close has not yet answered for', async () => { + const connection = claude.connections[0]! + await turnWithBackgroundTask(connection) + const closing = Promise.withResolvers() + closesUnproven(connection, closing.promise) + await faultTheJournal(connection) + // The fault's first close gave up; the exit path's own close is still running. + await eventually(() => expect(connection.closeCount).toBe(2)) + const sentBefore = connection.sent.length + + const raced = await send('Sent while the close runs.') + expect(await settled(raced)).toMatchObject({ + dispatchState: 'rejected', + rejection: { kind: 'hostFault' } + }) + expect(connection.sent.length).toBe(sentBefore) + + closing.resolve() + await adapter.drainObservedExits() + await eventually(() => expect(hostSession()?.child?.close?.reported).toBeDefined()) +}) + +it('keeps the child of a journal sink failure whose stop was unproven on record, closing, with the fault', async () => { + const connection = claude.connections[0]! + await turnWithBackgroundTask(connection) + closesUnproven(connection) + // The host's own recovery: the journal write fails for good, and Orca stops the provider. + const journal = hostSession()!.journal + vi.spyOn(journal, 'appendItem').mockRejectedValueOnce(new Error('disk unavailable')) + frame(connection, { + type: 'assistant', + uuid: 'leaf-2', + parent_tool_use_id: null, + message: { id: 'msg-2', role: 'assistant', content: [{ type: 'text', text: 'Lost.' }] } + }) + + await eventually(() => + expect(hostSession()?.child?.close).toMatchObject({ + cause: 'host-stop', + reported: { failure: { kind: 'hostFault' } } + }) + ) + expect(connection.closeCount).toBeGreaterThan(0) + host['runtimeState'].eventSinkFor(SESSION) +}) + +/** A turn still running when the fault comes. */ +async function turnRunning(connection: FakeConnection): Promise { + const text = 'Do a long thing.' + await send(text) + await eventually(() => expect(wrote(connection, text)).toBe(true)) + frame(connection, { + type: 'system', + subtype: 'init', + uuid: 'init-1', + model: 'claude-sonnet-5', + capabilities: CAPABILITIES + }) + const written = connection.sent.at(-1)! + frame(connection, { ...written, uuid: written.uuid }) + frame(connection, { + type: 'assistant', + uuid: 'leaf-1', + parent_tool_use_id: null, + message: { id: 'msg-1', role: 'assistant', content: [{ type: 'text', text: 'Working...' }] } + }) + await host.flushStreamedEvents(SESSION) +} + +it.each([ + ['proven at once', false], + ['proven only by the late exit', true] +])('tells why a turn a journal fault cut short stopped, its close %s', async (_name, unproven) => { + const connection = claude.connections[0]! + await turnRunning(connection) + if (unproven) { + await faultedUnproven(connection) + expect(await faultRows()).toEqual([]) + exits(connection) + } else { + connection.exitVerdict = { root: 'exited', tree: 'exited' } + await faultTheJournal(connection) + await adapter.drainObservedExits() + } + await eventually(() => expect(hostSession()?.child).toBeNull()) + + // The same end reads the same however long its proof took. + expect(await turnRows()).toEqual([{ state: 'interrupted', outcome: undefined }]) + expect(await faultRows()).toEqual([FAULT_ROW]) + expect(hostSession()?.lastEndedChild).toMatchObject({ + cause: 'exit', + failure: { kind: 'hostFault' } + }) +}) + +async function turnRows() { + await host.flushStreamedEvents(SESSION) + return (await host.journalSnapshot(SESSION)).items.flatMap((item) => + item.body.kind === 'turn' ? [{ state: item.body.state, outcome: item.body.outcome }] : [] + ) +} + +it.each([ + ['a Stop', 'user-stop'], + ['a chat close', 'user-close'] +] as const)( + 'still tells why a turn a fault cut short stopped when %s comes before the late exit, as the fault alone would', + async (_name, cause) => { + const connection = claude.connections[0]! + await turnRunning(connection) + await faultedUnproven(connection) + expect(hostSession()?.child?.close?.reported).toBeDefined() + + // The turn still reads running, so the user acts; that close cannot prove the exit either. + await (cause === 'user-stop' + ? expect(stop()).resolves.toMatchObject({ ok: true }) + : expect(host.close(SESSION, 'user-close')).rejects.toThrow()) + await laneDrained() + exits(connection) + await eventually(() => expect(hostSession()?.child).toBeNull()) + + // The fault ended the turn, so it reads interrupted, never the user's cancellation, with the + // fault's one row; the end stays the user's. + expect(await turnRows()).toEqual([{ state: 'interrupted', outcome: undefined }]) + expect(await faultRows()).toEqual([FAULT_ROW]) + expect(hostSession()?.lastEndedChild).toMatchObject({ cause, failure: { kind: 'hostFault' } }) + } +) + +it("closes a message accepted while a fault's close was unproven when the chat close that takes it over is proven later", async () => { + const connection = claude.connections[0]! + await turnWithBackgroundTask(connection) + await faultedUnproven(connection) + const closing = hostSession()!.child! + const faultAt = closing.close!.requestedAt + + // A message is accepted, then the chat is closed, then the exit is proven, all before the + // message's delivery step runs. + const held = Promise.withResolvers() + const holding = host['tasks'].serialize(SESSION, () => held.promise) + const sent = send('Sent before the close.') + // The close's own rejection of what is queued fails, so only where its end stands closes it. + vi.spyOn(hostSession()!.journal, 'rejectQueuedSubmissions').mockRejectedValueOnce( + new Error('disk full') + ) + const closed = host.close(SESSION, 'user-close') + const ended = host['tasks'].serialize(SESSION, () => + host['eventRecovery'].endExitedChildUnderSerialize(SESSION, closing, { + expected: true, + reason: 'claude session closed' + }) + ) + held.resolve() + await holding + const message = await sent + await expect(closed).rejects.toThrow() + await ended + + // The close took the fault's close over where it was asked, after the message: the end stands + // there, so the message is closed with the chat and starts no new agent. + expect(await settled(message)).toMatchObject({ + dispatchState: 'rejected', + rejection: { kind: 'chatClosed' } + }) + const end = hostSession()!.lastEndedChild! + expect(end).toMatchObject({ cause: 'user-close', failure: { kind: 'hostFault' } }) + expect(end.endedAt.sequence).toBeGreaterThan(faultAt.sequence) + expect(claude.connections).toHaveLength(1) + expect(wrote(connection, 'Sent before the close.')).toBe(false) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-codex-forced-close-unproven.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-codex-forced-close-unproven.test.ts new file mode 100644 index 000000000000..df652508f0f9 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-codex-forced-close-unproven.test.ts @@ -0,0 +1,340 @@ +// A Codex app-server Orca force-closed after a frame it could not record, whose close could not prove +// it gone, on the shipping adapter and host. The child may still be running, so the host keeps it on +// record, closing: a message sent meanwhile joins that close and is refused readably, never left in +// doubt, and the old app-server's exit ends the record as the fault it was. + +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 { claudeAndCodexAgents } from './structured-agent-session-adapter-router-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 adapter: CodexStructuredSessionAdapter +let store: AgentSessionRecordStore +let codex: ReturnType +let events: CodexStructuredSessionEvent[] +const unexited: (() => void)[] = [] + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-codex-forced-close-unproven-')) + resetHostTestOperationIds() + events = [] + codex = fakeCodex() + let generation = 0 + 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.nativeId ?? null + } + }, + // As the runtime routes them. + 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({ + agents: claudeAndCodexAgents(adapter), + 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 () => { + // Every app-server a test left running has exited by now. + for (const exit of unexited.splice(0)) { + exit() + } + await adapter.closeAll() + await host.flushAllStreamedEvents() + await rm(root, { recursive: true, force: true }) +}) + +type CodexConnection = (typeof codex.connections)[number] + +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)}`) + } + return sent.value.clientMessageId +} + +async function submission(clientMessageId: string) { + await host.flushStreamedEvents(SESSION) + return (await host.journalSnapshot(SESSION)).submissions.find( + (entry) => entry.clientMessageId === clientMessageId + ) +} + +function laneDrained(): Promise { + return host['tasks'].serialize(SESSION, async () => {}) +} + +function hostSession() { + return host['sessions'].get(SESSION) +} + +function startedTurnWith(connection: CodexConnection, text: string): boolean { + return connection.calls.some( + (call) => call.method === 'turn/start' && JSON.stringify(call.params).includes(text) + ) +} + +/** As the real connection: a close proves nothing until the app-server's exit is observed. */ +function closesOnlyOnceExited(connection: CodexConnection): { exit: () => void } { + let exited = false + connection.close = async () => { + connection.closeCount += 1 + connection.closed = true + return exited + } + // As the real connection reports it: inside the close Orca began. + const exit = (): void => { + if (!exited) { + exited = true + connection.handlers.onExit?.(new Error('codex app-server exited'), { expected: true }) + } + } + unexited.push(exit) + return { exit } +} + +async function resumedWith(connection: CodexConnection, text: string): Promise { + return vi.waitFor(() => { + const next = codex.connections.at(-1)! + expect(next).not.toBe(connection) + expect(startedTurnWith(next, text)).toBe(true) + return next + }) +} + +const FAULT_ROW = "Orca ran into a problem, so this didn't go through. Try again." + +async function faultRows() { + await host.flushStreamedEvents(SESSION) + return (await host.journalSnapshot(SESSION)).items.flatMap((item) => + item.body.kind === 'status' && item.body.failure?.kind === 'hostFault' ? [item.body.text] : [] + ) +} + +/** A frame the journal cannot record: Orca force-closes the app-server as an unexpected end. */ +function forceCloseOnAnUnrecordableFrame(connection: CodexConnection): void { + const sink = host['runtimeState'].eventSinkFor(SESSION).sink + vi.spyOn(sink, 'tryAppendItem').mockReturnValueOnce({ accepted: false, reason: 'failed' }) + connection.handlers.onUnhandledFrame?.('frame:unknown-method', { method: 'mystery/event' }) +} + +function turnRunning(connection: CodexConnection): Promise { + const threadId = adapter['sessions'].get(SESSION)!.threadId + connection.handlers.onNotification?.('turn/started', { threadId, turn: { id: 'turn-1' } }) + return host.flushStreamedEvents(SESSION) +} + +it('keeps an app-server Orca force-closed unproven on record, refuses the next message readably, and sends the one after its exit', async () => { + const connection = codex.connections[0]! + const old = closesOnlyOnceExited(connection) + forceCloseOnAnUnrecordableFrame(connection) + + await vi.waitFor(() => expect(events.some((event) => event.type === 'end-unproven')).toBe(true)) + await laneDrained() + expect(hostSession()?.child).toMatchObject({ + generation: 'generation-1', + close: { cause: 'host-stop', reported: { failure: { kind: 'hostFault' } } } + }) + + const held = await send('Carry on.') + await vi.waitFor(async () => expect((await submission(held))?.dispatchState).toBe('rejected')) + expect((await submission(held))?.reason).toContain("Couldn't stop Codex from before.") + expect(startedTurnWith(connection, 'Carry on.')).toBe(false) + expect(codex.connections).toHaveLength(1) + + old.exit() + await vi.waitFor(() => expect(hostSession()?.child).toBeNull()) + await send('Go on.') + const resumed = await resumedWith(connection, 'Go on.') + + expect(startedTurnWith(connection, 'Go on.')).toBe(false) + expect(resumed.closeCount).toBe(0) +}) + +it("drops an exit from an app-server that is no longer the chat's", async () => { + const connection = codex.connections[0]! + const old = closesOnlyOnceExited(connection) + forceCloseOnAnUnrecordableFrame(connection) + await vi.waitFor(() => expect(hostSession()?.child?.close?.reported).toBeDefined()) + old.exit() + await vi.waitFor(() => expect(hostSession()?.child).toBeNull()) + await send('Carry on.') + const resumed = await resumedWith(connection, 'Carry on.') + const reports = events.filter((event) => event.type === 'ended').length + + connection.handlers.onExit?.(new Error('codex app-server exited again'), { expected: true }) + await laneDrained() + + expect(events.filter((event) => event.type === 'ended')).toHaveLength(reports) + expect(resumed.closeCount).toBe(0) + expect(hostSession()?.child?.generation).toBe('generation-2') +}) + +it.each([ + ['proven at once', false], + ['proven only by the late exit', true] +])('tells why a turn a forced close cut short stopped, its close %s', async (_name, unproven) => { + const connection = codex.connections[0]! + const old = unproven ? closesOnlyOnceExited(connection) : null + await turnRunning(connection) + forceCloseOnAnUnrecordableFrame(connection) + if (old) { + await vi.waitFor(() => expect(hostSession()?.child?.close?.reported).toBeDefined()) + expect(await faultRows()).toEqual([]) + old.exit() + } + await vi.waitFor(() => expect(hostSession()?.child).toBeNull()) + + // The same end reads the same however long its proof took. + await vi.waitFor(async () => expect(await faultRows()).toEqual([FAULT_ROW])) + expect(await turnRows()).toEqual([{ state: 'interrupted', outcome: undefined }]) + expect(hostSession()?.lastEndedChild).toMatchObject({ + cause: 'exit', + failure: { kind: 'hostFault' } + }) +}) + +async function turnRows() { + await host.flushStreamedEvents(SESSION) + return (await host.journalSnapshot(SESSION)).items.flatMap((item) => + item.body.kind === 'turn' ? [{ state: item.body.state, outcome: item.body.outcome }] : [] + ) +} + +it.each([ + ['the chat is closed', 'user-close'], + ['Orca evicts it', 'evict'] +] as const)( + 'still tells why a turn a forced close cut short stopped when %s before the late exit, as the fault alone would', + async (_name, cause) => { + const connection = codex.connections[0]! + const old = closesOnlyOnceExited(connection) + await turnRunning(connection) + forceCloseOnAnUnrecordableFrame(connection) + await vi.waitFor(() => expect(hostSession()?.child?.close?.reported).toBeDefined()) + const journal = hostSession()!.journal + const stopEvents = vi.spyOn(journal, 'appendStopEvent') + + // That close cannot prove the exit either. + await expect(host.close(SESSION, cause)).rejects.toThrow() + old.exit() + await vi.waitFor(() => expect(hostSession()?.child).toBeNull()) + + await vi.waitFor(async () => expect(await faultRows()).toEqual([FAULT_ROW])) + // The fault ended the turn: it reads interrupted, never the user's cancellation. + expect(await turnRows()).toEqual([{ state: 'interrupted', outcome: undefined }]) + expect(stopEvents).not.toHaveBeenCalled() + expect(hostSession()?.lastEndedChild).toMatchObject({ cause, failure: { kind: 'hostFault' } }) + } +) + +it('closes a message accepted while a forced close was unproven when the chat close that takes it over is proven later', async () => { + const connection = codex.connections[0]! + closesOnlyOnceExited(connection) + forceCloseOnAnUnrecordableFrame(connection) + await vi.waitFor(() => expect(hostSession()?.child?.close?.reported).toBeDefined()) + const closing = hostSession()!.child! + const faultAt = closing.close!.requestedAt + + // A message is accepted, then the chat is closed, then the exit is proven, all before the + // message's delivery step runs. + const held = Promise.withResolvers() + const holding = host['tasks'].serialize(SESSION, () => held.promise) + const sent = send('Sent before the close.') + // The close's own rejection of what is queued fails, so only where its end stands closes it. + vi.spyOn(hostSession()!.journal, 'rejectQueuedSubmissions').mockRejectedValueOnce( + new Error('disk full') + ) + const closed = host.close(SESSION, 'user-close') + const ended = host['tasks'].serialize(SESSION, () => + host['eventRecovery'].endExitedChildUnderSerialize(SESSION, closing, { + expected: true, + reason: 'codex session closed' + }) + ) + held.resolve() + await holding + const message = await sent + await expect(closed).rejects.toThrow() + await ended + + // The end stands where the close took over, after the message, so it is closed with the chat. + await vi.waitFor(async () => + expect(await submission(message)).toMatchObject({ + dispatchState: 'rejected', + rejection: { kind: 'chatClosed' } + }) + ) + const end = hostSession()!.lastEndedChild! + expect(end).toMatchObject({ cause: 'user-close', failure: { kind: 'hostFault' } }) + expect(end.endedAt.sequence).toBeGreaterThan(faultAt.sequence) + expect(codex.connections).toHaveLength(1) + expect(startedTurnWith(connection, 'Sent before the close.')).toBe(false) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-command-turn.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-command-turn.ts index e51b8e672f96..54fc2ab9a868 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-command-turn.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-command-turn.ts @@ -28,6 +28,7 @@ import { type AgentJournalSubmission } from '../../../shared/agent-session-journal-types' import type { AgentSessionConversationCommand } from '../../../shared/agent-session-conversation-command' +import { classifyDispatchRejection } from '../../../shared/structured-agent-session-dispatch-rejection' import type { AgentSessionRecord } from '../../../shared/agent-session-record' import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' import { agentSessionRefusalReference } from '../../../shared/agent-session-wire-refusals' @@ -226,12 +227,18 @@ export async function handOverStructuredAgentSessionCommand( return } if (admission.state === 'rejected') { - // The provider refused the compaction itself: its row reads as the compaction failing. + // A child that was already gone, or never started, keeps its own reason (the adapter had let + // it go); any other refusal reads as the compaction failing. + const { category, kind } = classifyDispatchRejection(admission) + const agentGone = + category === 'undelivered' || category === 'startFailed' || kind === 'hostFault' await settleUnsentCommand( ctx, clientMessageId, admission, - agentSessionFailureFact('compactionFailed', { detail: admission.rejection.detail }) + agentGone + ? undefined + : agentSessionFailureFact('compactionFailed', { detail: admission.rejection.detail }) ) } else if (admission.state !== 'admitted') { // An unknown write leaves the turn to the provider's end or the child's: it may have run. 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 398d04873d41..352deffd98d8 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 @@ -15,6 +15,10 @@ import { type StructuredAgentSessionChildExitContext } from './structured-agent-session-child-exit' import type { StructuredAgentSessionHostRuntimeState } from './structured-agent-session-host-runtime-state' +import { + recordUnprovenStructuredAgentSessionEnd, + recordUnprovenStructuredAgentSessionEndUnderSerialize +} from './structured-agent-session-unproven-end' export class StructuredAgentSessionEventRecovery { private readonly sinkFailures = new Set() @@ -64,16 +68,32 @@ export class StructuredAgentSessionEventRecovery { const child = this.context.sessions.get(sessionId)?.child const stop = this.context.deps.adapter.forceCloseSession ?? this.context.deps.adapter.closeSession - if (!child || !stop || !(await stopAgentSessionProviderRoot(() => stop(sessionId)))) { + if (!child || !stop) { return } - // Ended in the same step as the stop, so no report of that close can end it first as a - // quiet rest: Orca stopped the provider because its own journal failed. - await this.endExitedChildUnderSerialize(sessionId, child, { - expected: false, + // Orca stopped the provider because its own journal failed. + const end = { reason: `journal sink failure: ${error instanceof Error ? error.message : String(error)}`, failure: agentSessionFailureFact('hostFault') - }) + } + const stopped = await stopAgentSessionProviderRoot(() => stop(sessionId)).catch( + (stopError: unknown) => { + this.context.deps.logger.warn('stopping a provider after its journal failed', { + scope: 'sink-failure-recovery', + sessionId, + error: stopError + }) + return false + } + ) + if (!stopped) { + // Never silent: the child stays on record, closing, and its exit is settled as this end. + recordUnprovenStructuredAgentSessionEndUnderSerialize(this.context, sessionId, child, end) + return + } + // Ended in the same step as the stop, so no report of that close can end it first as a + // quiet rest. + await this.endExitedChildUnderSerialize(sessionId, child, { expected: false, ...end }) }) .catch((error: unknown) => this.context.deps.logger.warn( @@ -95,6 +115,9 @@ export class StructuredAgentSessionEventRecovery { if (event.type === 'started') { return settleStructuredAgentSessionProviderStarted(this.context, event) } + if (event.type === 'end-unproven') { + return recordUnprovenStructuredAgentSessionEnd(this.context, event) + } await settleStructuredAgentSessionChildExit(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 5e8d7e2097f9..eff75083782f 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 @@ -94,7 +94,9 @@ export async function abandonQueuedStructuredAgentSessionMessages( * already begun, and waits for the exit's proof as long as a caller may. Still unproven, it throws * and the close keeps running: a later proof, or the next asker's attempt, ends the record. * `ending` is how the child's end is told: a user's Stop, the host stopping it for a cause (with - * its text), or an eviction the conversation's close follows. The first stop's ending decides. + * its text), or an eviction the conversation's close follows. The first stop's ending decides; a + * fault's unproven end is no one's stop, so the first stop after it takes over that close with its + * own cause, keeping the fault's details and writing no event. */ export async function stopStructuredAgentSessionAgentUnderSerialize( context: StructuredAgentSessionLifetimeContext, @@ -107,17 +109,22 @@ export async function stopStructuredAgentSessionAgentUnderSerialize( return } const cause = 'recorded' in ending ? ending.recorded : ending.cause - if (!child.close) { + // An end the adapter reported unproven began no one's stop: the first stop asked since decides. + const reported = child.close?.cause === 'host-stop' ? child.close.reported : undefined + if (!child.close || (reported && cause !== 'host-stop')) { // Judged before the kill: a stop that ends nothing writes nothing. Its event is issued before - // the kill and never awaited by it; the journal writes rows in order. - const recorded = (await stopEndsWork(context, sessionId, session, ending)) - ? recordStopEvent(context, sessionId, session, ending) - : Promise.resolve(null) + // the kill and never awaited by it; the journal writes rows in order. The fault already ended + // the work a reported end's taker would name, so like any stop that joins a close it writes none. + const recorded = + !reported && (await stopEndsWork(context, sessionId, session, ending)) + ? recordStopEvent(context, sessionId, session, ending) + : Promise.resolve(null) child.close = { cause, reason: ('reason' in ending ? ending.reason : undefined) ?? null, recorded, - requestedAt: session.journal.cursor() + requestedAt: session.journal.cursor(), + ...(reported ? { reported } : {}) } } else if (child.close.cause === cause) { // The same stop asked again, such as a second close of the chat, closes what came since, and diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts index 243e2d8bea2f..904d1fbd77d9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts @@ -58,6 +58,17 @@ export type StructuredAgentSessionChildClose = { /** Where the journal stood when that stop was asked for: the child's end is ordered there, so a * message accepted while the exit was being proven came after it. A repeated ask moves it. */ requestedAt: AgentJournalCursor + /** Set when no stop began this close: the child ended (a fault, an exit) and its own close could + * not prove it gone. The exit that ends the record is settled as this end. In memory only. */ + readonly reported?: StructuredAgentSessionReportedEnd +} + +/** An end the adapter reported but could not prove, kept until its exit is seen or proven. */ +export type StructuredAgentSessionReportedEnd = { + reason: string + failure?: SubmissionRejectionFact + /** Work was in flight when the end was reported, so its settlement shows why it stopped. */ + interruptedWork: boolean } /** The provider process behind a conversation. Written only in diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unproven-end.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unproven-end.test.ts new file mode 100644 index 000000000000..073f0e4809c9 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unproven-end.test.ts @@ -0,0 +1,118 @@ +// An end the adapter could not prove is judged by the same rule an observed exit is: only work that +// was in flight reads as interrupted. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, expect, it } from 'vitest' +import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { openAgentSessionJournal } from '../agent-session-journal/journal-store-factory' +import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' +import { + captureUnfinishedStructuredAgentSessionWork, + settleStructuredAgentSessionDeadGeneration, + unfinishedStructuredAgentSessionWorkWasInterrupted +} from './structured-agent-session-dead-generation-settlement' +import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' +import { recordUnprovenStructuredAgentSessionEndUnderSerialize } from './structured-agent-session-unproven-end' + +const SESSION = 'session-unproven-end' +const THREAD = 'thread-1' +const NOW = 1_000 +const FAULT = agentSessionFailureFact('hostFault') + +let root: string +let journal: AgentSessionJournal + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-unproven-end-')) + journal = await openAgentSessionJournal({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { transport: 'codex-app-server', agent: 'codex', nativeId: THREAD } + }, + database: openTestJournalHostDatabase(root), + now: () => NOW + }) + // The turn finished; only an approval card the agent asked for is left open. + await journal.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 1 }, + { + kind: 'approval', + title: 'Run command?', + detail: null, + options: [{ id: 'yes', label: 'Allow' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + }, + { fence: 7, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + await journal.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 2 }, + { kind: 'turn', turnId: 'turn-1', state: 'completed', startedAt: 900, completedAt: 950 }, + { fence: 7, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) +}) + +afterEach(async () => { + await journal.close() + await rm(root, { recursive: true, force: true }) +}) + +function statusRows() { + return journal.snapshot().items.filter((item) => item.body.kind === 'status') +} + +it('shows no fault row for an idle open approval when the exit was observed', async () => { + const before = captureUnfinishedStructuredAgentSessionWork(journal) + const show = unfinishedStructuredAgentSessionWorkWasInterrupted(before, journal, NOW) + + await settleStructuredAgentSessionDeadGeneration({ + journal, + sessionId: SESSION, + fence: 7, + settlementId: `provider-exit:${SESSION}:7:generation-1`, + pendingSubmissionReason: 'provider_exited_before_acknowledgement', + verdict: { state: 'interrupted', completedAt: NOW }, + showUnexpectedExitOutcome: show, + exitFailure: FAULT + }) + + expect(show).toBe(false) + expect(statusRows()).toEqual([]) +}) + +function reportOn(journalOf: AgentSessionJournal) { + const session: Pick = { + child: { generation: 'generation-1', fence: 7, phase: 'ready' }, + journal: journalOf + } + recordUnprovenStructuredAgentSessionEndUnderSerialize( + { sessions: new Map([[SESSION, session]]), now: () => NOW }, + SESSION, + { generation: 'generation-1', fence: 7 }, + { reason: 'journal sink failure', failure: FAULT } + ) + return session.child?.close +} + +it('judges the same idle approval not interrupted when the end could not be proven', () => { + expect(reportOn(journal)).toMatchObject({ + cause: 'host-stop', + reported: { reason: 'journal sink failure', failure: FAULT, interruptedWork: false } + }) +}) + +it('judges a turn still running when the end was reported interrupted', async () => { + await journal.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-2', ordinal: 1 }, + { kind: 'turn', turnId: 'turn-2', state: 'running', startedAt: 990 }, + { fence: 7, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + + expect(reportOn(journal)?.reported).toMatchObject({ interruptedWork: true }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unproven-end.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unproven-end.ts new file mode 100644 index 000000000000..560c2657d10d --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unproven-end.ts @@ -0,0 +1,73 @@ +// An end the adapter could not prove. The child may still be running, so the host keeps it on record +// and closing: nothing writes to it or starts beside it, and every later send, start or Stop joins +// that close and retries the kill. The exit that finally ends the record is settled as the end that +// was reported, not as a stop anyone asked for. + +import type { StructuredAgentSessionEndUnprovenEvent } from './structured-agent-session-adapter' +import { + captureUnfinishedStructuredAgentSessionWork, + unfinishedStructuredAgentSessionWorkWasInterrupted +} from './structured-agent-session-dead-generation-settlement' +import type { + StructuredAgentSessionHostSession, + StructuredAgentSessionProviderChildIdentity, + StructuredAgentSessionReportedEnd +} from './structured-agent-session-host-types' +import { sameProviderChild } from './structured-agent-session-provider-child' + +type UnprovenEndSession = Pick + +export type StructuredAgentSessionUnprovenEndContext = { + sessions: Map + now: () => number + publishStatus?: (sessionId: string) => void + serialize: (sessionId: string, task: () => Promise) => Promise +} + +/** For a caller inside the session's serialize. A report for any other child is stale. */ +export function recordUnprovenStructuredAgentSessionEndUnderSerialize( + context: Omit, + sessionId: string, + ended: StructuredAgentSessionProviderChildIdentity, + report: Pick +): void { + const session = context.sessions.get(sessionId) + const child = session?.child + if (!session || !child || !sameProviderChild(child, ended)) { + return + } + // A close a stop already began keeps its cause: it is that stop's to end. + if (!child.close) { + child.close = { + cause: 'host-stop', + reason: report.reason, + recorded: Promise.resolve(null), + requestedAt: session.journal.cursor(), + reported: { + reason: report.reason, + ...(report.failure ? { failure: report.failure } : {}), + // The rule an observed exit uses: an approval or question left open is not interrupted. + interruptedWork: unfinishedStructuredAgentSessionWorkWasInterrupted( + captureUnfinishedStructuredAgentSessionWork(session.journal), + session.journal, + context.now() + ) + } + } + } + context.publishStatus?.(sessionId) +} + +export function recordUnprovenStructuredAgentSessionEnd( + context: StructuredAgentSessionUnprovenEndContext, + event: StructuredAgentSessionEndUnprovenEvent +): Promise { + return context.serialize(event.sessionId, async () => + recordUnprovenStructuredAgentSessionEndUnderSerialize( + context, + event.sessionId, + { generation: event.acquisitionGeneration, fence: event.fence }, + { reason: event.reason, ...(event.failure ? { failure: event.failure } : {}) } + ) + ) +} diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-compaction.test.ts b/src/main/native-chat/agent-session-wire/structured-conversation-compaction.test.ts index 6113b17ddec6..9d67984f2dae 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-compaction.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-conversation-compaction.test.ts @@ -364,6 +364,31 @@ it('refuses the command for an agent that does not declare compaction, whatever expect(await commandTurn(params.envelope.clientOperationId)).toBeUndefined() }) +it('keeps the reason when the agent the compaction was meant for had already gone', async () => { + await attach() + // As the adapter answers for a child it let go after a fault Orca could not prove ended. + compact.mockResolvedValue({ + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('hostFault'), { surface: 'rejection' }) + }) + await state.host.conversationCommand(CALLER, compactParams()) + + await vi.waitFor(async () => + expect( + (await journal()).items + .filter((item) => item.body.kind === 'status' && item.body.tone === 'error') + .map((item) => item.body) + ).toEqual([ + { + kind: 'status', + text: "Orca ran into a problem, so this didn't go through. Try again.", + failure: { kind: 'hostFault' }, + tone: 'error' + } + ]) + ) +}) + it('refuses the command at handover when the provider opened a turn meanwhile (B3)', async () => { await attach() const events = state.acquire.mock.calls.at(-1)?.[0].events diff --git a/src/main/runtime/claude-structured-startup-fault-is-not-an-exit.test.ts b/src/main/runtime/claude-structured-startup-fault-is-not-an-exit.test.ts index 863cbaeaebf9..f79ab0ed75dc 100644 --- a/src/main/runtime/claude-structured-startup-fault-is-not-an-exit.test.ts +++ b/src/main/runtime/claude-structured-startup-fault-is-not-an-exit.test.ts @@ -136,4 +136,38 @@ describe('a Claude start that Orca fails while the CLI is still running', () => ) expect(await failureRows(host)).toEqual([{ text: STOPPED_TEXT, kind: 'providerStartFailed' }]) }) + + it('rejects every message held on a start Orca failed with that start, even when its close could not prove the CLI gone', async () => { + claude.behave(SESSION, { initHangs: true, initNamesForeignSession: true, closeUnproven: true }) + const host = await claude.install() + await expect(host.attach(CALLER, claude.attachParams(SESSION, null))).resolves.toMatchObject({ + ok: true + }) + const first = await send(host, 'hello') + const second = await send(host, 'and this') + + claude.child(SESSION).answerInit() + await waitForStructuredAgentSessionRecovery() + // The CLI may still run, so the chat waits on it; once it exits, the start that failed decides. + await vi.waitFor(() => + expect(host['sessions'].get(SESSION)?.child?.close?.reported).toBeDefined() + ) + claude.child(SESSION).exit(scriptedClaudeExitError(DIAGNOSTIC)) + await released(host) + + for (const held of [first, second]) { + await vi.waitFor(async () => + expect(await submission(host, held)).toMatchObject({ + dispatchState: 'rejected', + reason: expect.stringMatching(/^Claude couldn't start\./), + rejection: { kind: 'startFailed' } + }) + ) + } + expect(claude.child(SESSION).calls).not.toContain('send') + expect(host['sessions'].get(SESSION)?.lastEndedChild).toMatchObject({ + cause: 'exit', + failure: { kind: 'startFailed' } + }) + }) }) diff --git a/src/main/runtime/structured-agent-runtime-registrations.ts b/src/main/runtime/structured-agent-runtime-registrations.ts index 5db941bb8d88..d7edca446fe1 100644 --- a/src/main/runtime/structured-agent-runtime-registrations.ts +++ b/src/main/runtime/structured-agent-runtime-registrations.ts @@ -21,6 +21,7 @@ import type { createStructuredAgentSessionDispatchFollowUps } from './structured import type { StructuredAgentSessionRuntimeDeps } from './structured-agent-session-runtime' import type { createStructuredAgentEnvironmentResolvers } from './structured-agent-shell-environment' import { createStructuredClaudeRuntimeAdapter } from './structured-claude-runtime-adapter' +import { structuredCodexLifecycleEvent } from './structured-codex-lifecycle-event' import { resolveStructuredClaudeAccountHomePath, resolveStructuredCodexAccountHomePath, @@ -99,9 +100,9 @@ function createCodexAdapter(context: StructuredAgentAdapterContext): StructuredA onPrimaryThreadStoppedRunning: followUps.releaseUnansweredDispatches, logger: deps.logger, onEvent: (event) => { - // Every exit, expected or not: the host ends that child's record. - if (event.type === 'ended' && 'cause' in event) { - context.deliverLifecycle(event) + const lifecycleEvent = structuredCodexLifecycleEvent(event) + if (lifecycleEvent) { + context.deliverLifecycle(lifecycleEvent) } } }) diff --git a/src/main/runtime/structured-agent-session-lifecycle-delivery.test.ts b/src/main/runtime/structured-agent-session-lifecycle-delivery.test.ts index b236c81d594b..fd05b2cd9628 100644 --- a/src/main/runtime/structured-agent-session-lifecycle-delivery.test.ts +++ b/src/main/runtime/structured-agent-session-lifecycle-delivery.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import type { StructuredAgentSessionLifecycleEvent } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import { recordingStructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support' import { createStructuredAgentSessionLifecycleDelivery } from './structured-agent-session-lifecycle-delivery' @@ -46,4 +46,40 @@ describe('provider exits reaching the host', () => { await delivery.drain() expect(handled).toEqual(['closed', 'crashed']) }) + + it("keeps an unproven end 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 === 'end-unproven') { + await sessionLane.promise + } + handled.push(`${event.type}:${event.sessionId}`) + }, + logger: recordingStructuredAgentSessionLogger().logger, + drainObservedExits: async () => {} + }) + + delivery.deliver({ + type: 'end-unproven', + sessionId: 'session-a', + reason: 'journal fault', + fence: 1, + acquisitionGeneration: 'a-1' + }) + delivery.deliver(ended('session-b', 'unexpected-exit')) + 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', 'end-unproven: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 ba307c2cfb0a..3144afa041f5 100644 --- a/src/main/runtime/structured-agent-session-lifecycle-delivery.ts +++ b/src/main/runtime/structured-agent-session-lifecycle-delivery.ts @@ -3,13 +3,19 @@ // 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` and the end of a close the host asked for stay off -// it: each 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`, an end whose close is unproven and the end of a +// close the host asked for 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', + 'end-unproven': 'lifecycle-end-unproven' +} as const satisfies Record + export function createStructuredAgentSessionLifecycleDelivery(input: { handle: (event: StructuredAgentSessionLifecycleEvent) => Promise | undefined logger: StructuredAgentSessionLogger @@ -27,7 +33,7 @@ export function createStructuredAgentSessionLifecycleDelivery(input: { 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 }) @@ -35,7 +41,7 @@ export function createStructuredAgentSessionLifecycleDelivery(input: { } return { deliver: (event) => { - if (event.type === 'started' || event.cause === 'requested-close') { + if (event.type !== 'ended' || event.cause === 'requested-close') { // Called now, so the step is queued on its own session's lane, behind nothing of another's. const settling = settle(event) settlingStarts.add(settling) diff --git a/src/main/runtime/structured-claude-runtime-adapter.ts b/src/main/runtime/structured-claude-runtime-adapter.ts index 8b4b70f8bc2b..53adbcce4bb0 100644 --- a/src/main/runtime/structured-claude-runtime-adapter.ts +++ b/src/main/runtime/structured-claude-runtime-adapter.ts @@ -46,7 +46,7 @@ export type StructuredClaudeRuntimeAdapterDeps = { export function structuredClaudeLifecycleEvent( event: ClaudeStructuredSessionEvent ): StructuredAgentSessionLifecycleEvent | null { - if (event.type === 'started') { + if (event.type === 'started' || event.type === 'end-unproven') { return event } // Every exit of a child with an identity, expected or not: the host ends that child's record. 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..4773ce65bbb6 --- /dev/null +++ b/src/main/runtime/structured-codex-lifecycle-event.ts @@ -0,0 +1,12 @@ +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: every exit, expected or not, and + * an end whose close could not prove the exit. */ +export function structuredCodexLifecycleEvent( + event: CodexStructuredSessionEvent +): StructuredAgentSessionLifecycleEvent | null { + return event.type === 'end-unproven' || (event.type === 'ended' && 'cause' in event) + ? event + : null +}