diff --git a/src/main/acp/acp-structured-acquire.ts b/src/main/acp/acp-structured-acquire.ts index 7310f9687e52..bb4ca0bc6164 100644 --- a/src/main/acp/acp-structured-acquire.ts +++ b/src/main/acp/acp-structured-acquire.ts @@ -226,6 +226,7 @@ export async function acquireAcpStructuredSession(input: { await identity.onSpawned(pid) const process = await identity.read(pid) const child: AcpStructuredChild = { + events, sessionId, fence: acquire.fence, acquisitionGeneration: generation, diff --git a/src/main/acp/acp-structured-child.ts b/src/main/acp/acp-structured-child.ts index 78b3493e2d2f..665b5326510c 100644 --- a/src/main/acp/acp-structured-child.ts +++ b/src/main/acp/acp-structured-child.ts @@ -6,6 +6,7 @@ import type { AcpStructuredSession } from './acp-structured-session' export type AcpStructuredChild = Pick< AcpStructuredSession, + | 'events' | 'sessionId' | 'fence' | 'acquisitionGeneration' diff --git a/src/main/acp/acp-structured-host-lifecycle.test.ts b/src/main/acp/acp-structured-host-lifecycle.test.ts index d0e667548008..85b265bb8197 100644 --- a/src/main/acp/acp-structured-host-lifecycle.test.ts +++ b/src/main/acp/acp-structured-host-lifecycle.test.ts @@ -289,3 +289,26 @@ describe('closing or stopping a Grok chat while it starts', () => { expect(await rig.adapter.closeSession(SESSION)).toBe(true) }) }) + +describe('a Grok send whose acceptance could not be written', () => { + it('is Working while Grok holds it, and stops once Grok lets it go', async () => { + const { rig, host } = await openAttachedHostRig() + const journal = host.collaboratorsForTests().sessions.get(SESSION)!.journal + vi.spyOn(journal, 'acceptLate').mockRejectedValue(new Error('disk full')) + const sent = await send(host, 'hello') + const prompt = await rig.frame('session/prompt') + // Grok names the prompt only with its first frame: until then it holds the send. + await waitFor(() => expect(host.currentWork(SESSION)?.working()).toBe(true)) + + rig.child().agent.notify('session/update', replyChunk(promptIdOf(prompt), 'hi')) + rig.child().agent.reply(prompt, { stopReason: 'end_turn' }) + await rig.settle() + + await waitFor(() => expect(host.currentWork(SESSION)?.working()).toBe(false)) + // The acceptance never landed: the row is still the handover's alone. + expect( + (await host.journalSnapshot(SESSION)).submissions.find((row) => row.clientMessageId === sent) + ).toMatchObject({ dispatchState: 'pending' }) + await host.close(SESSION, 'user-close') + }) +}) diff --git a/src/main/acp/acp-structured-host.test-support.ts b/src/main/acp/acp-structured-host.test-support.ts index ce003faa81e5..95e5bd1e51cd 100644 --- a/src/main/acp/acp-structured-host.test-support.ts +++ b/src/main/acp/acp-structured-host.test-support.ts @@ -104,7 +104,12 @@ export async function openHostRig( onEvent: (event) => void hosted.host?.handleAdapterEvent(event), onChildWorkEvidence: (sessionId, evidence) => hosted.host?.publishChildWorkEvidence(sessionId, evidence), - onDispatchSettledLate: (settlement) => void hosted.host?.settleLateDispatch(settlement), + onDispatchSettledLate: (settlement) => + void hosted.host + ?.settleLateDispatch(settlement) + .catch((error: unknown) => + state.log.logger.warn('late settlement failed', { scope: 'late-settlement', error }) + ), now: () => HOST_TEST_NOW, readProcessStartTime: async () => 1_700_000_000_000 + ++generation, mintGeneration: () => `generation-${generation}`, diff --git a/src/main/acp/acp-structured-session-adapter.ts b/src/main/acp/acp-structured-session-adapter.ts index cdcb46ef5408..6a72b9dad6c7 100644 --- a/src/main/acp/acp-structured-session-adapter.ts +++ b/src/main/acp/acp-structured-session-adapter.ts @@ -295,6 +295,10 @@ export class AcpStructuredSessionAdapter implements StructuredAgentSessionAdapte } } + // Undefined until the session opens: the host holds what it handed a starting child. + heldSends = (sessionId: string): ReadonlySet | undefined => + this.children.sessions.get(sessionId)?.turns?.heldSends + stopBackgroundTasks: NonNullable = ( input ) => stopAcpChildren(this.live(input.sessionId), input.fence, input.taskIds, input.acknowledge) diff --git a/src/main/acp/acp-structured-session.ts b/src/main/acp/acp-structured-session.ts index 54cf8b8e555a..98e326d8b3fa 100644 --- a/src/main/acp/acp-structured-session.ts +++ b/src/main/acp/acp-structured-session.ts @@ -20,9 +20,11 @@ import type { AcpStructuredLane } from './acp-structured-lane' import type { AcpStructuredOptions } from './acp-structured-options' import type { AcpStructuredPrompts } from './acp-structured-prompts' import type { AcpStructuredTurns } from './acp-structured-turns' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import type { StructuredAgentSessionProviderOpenResult } from '../native-chat/agent-session-wire/structured-agent-session-provider-open-contract' export type AcpStructuredSession = { + events?: StructuredAgentSessionEventSink sessionId: string fence: number acquisitionGeneration: string @@ -101,6 +103,7 @@ export function closeAcpSessionJournal( if (session.journalClosed !== null) { return } + session.events?.observeExit?.(session.turns.heldSends) session.journalClosed = reason session.prompts.clear() session.turns.end(reason, cause) @@ -121,6 +124,7 @@ export function endAcpStructuredSession( return } session.ended = true + session.events?.observeExit?.(undefined, observedAt) session.exitObservedAt = observedAt const stderr = session.connection.stderrTail() // Read at the proven exit: a crash usually ends the agent's stdout before its exit is observed, diff --git a/src/main/acp/acp-structured-turns.ts b/src/main/acp/acp-structured-turns.ts index 3d076d3d741b..dc39195c326a 100644 --- a/src/main/acp/acp-structured-turns.ts +++ b/src/main/acp/acp-structured-turns.ts @@ -75,6 +75,11 @@ export class AcpStructuredTurns { ) } + /** The sends this agent holds: each from its dispatch until its settlement. */ + get heldSends(): ReadonlySet { + return this.unsettled + } + /** Steers wait behind the running prompt's cancel. */ get holdsSteers(): boolean { return this.steers.length > 0 diff --git a/src/main/claude/claude-command-lifecycle.test.ts b/src/main/claude/claude-command-lifecycle.test.ts index a58d0c840fe5..292d62ca3055 100644 --- a/src/main/claude/claude-command-lifecycle.test.ts +++ b/src/main/claude/claude-command-lifecycle.test.ts @@ -111,7 +111,6 @@ async function replayCapture( ) { const capture = loadCapture(name) const settlements: Settlement[] = [] - const idles: string[] = [] const turnStates = new Map() // The capture supplies every frame, startup proof included. const claude = fakeClaude({ initProof: 'none', replayUuid: null }) @@ -130,8 +129,7 @@ async function replayCapture( readProcessStartTime: async () => 1, now: () => 1_700_000_200_000, persistHandle: async () => {}, - onDispatchSettledLate: (settlement) => settlements.push(settlement), - onSessionIdle: ({ sessionId }) => idles.push(sessionId) + onDispatchSettledLate: (settlement) => settlements.push(settlement) }) await adapter.acquire({ identity: identityFor(), @@ -248,7 +246,7 @@ async function replayCapture( settlements .filter((settlement) => settlement.clientMessageId === clientMessageId) .map(({ sessionId: _sessionId, clientMessageId: _id, ...outcome }) => outcome) - return { settlementsFor, idles, turnStates, liveUuid, connection, adapter } + return { settlementsFor, turnStates, liveUuid, connection, adapter } } const WITHDRAWN = { @@ -574,7 +572,6 @@ describe('a send the CLI holds', () => { isLifecycleFrame(frame, 'cancelled') }) - expect(replay.idles).toEqual(['session-1']) expect(replay.adapter.holdsDispatch('session-1')).toBe(true) expect(replay.settlementsFor('client-A')).toEqual([]) }) @@ -621,29 +618,3 @@ describe('a send the CLI holds', () => { } ) }) - -describe('the CLI reporting its session idle', () => { - it.each(['interrupt-lost', 'cancel-async', 'batch-lead', 'auth-failed'])( - 'is reported once per idle frame (%s), never for running', - async (name) => { - const replay = await replayCapture(name) - - expect(replay.idles).toEqual(['session-1']) - } - ) - - it('is not reported from a child the session no longer holds', async () => { - const replay = await replayCapture('auth-failed') - await replay.adapter.closeSession('session-1') - - replay.connection.handlers.onMessage?.({ - type: 'system', - subtype: 'session_state_changed', - state: 'idle', - uuid: 'late-idle', - session_id: PROVIDER_SESSION_ID - }) - - expect(replay.idles).toEqual(['session-1']) - }) -}) diff --git a/src/main/claude/claude-command-lifecycle.ts b/src/main/claude/claude-command-lifecycle.ts index 4cfd512fe2ac..1b8c0bbd25d8 100644 --- a/src/main/claude/claude-command-lifecycle.ts +++ b/src/main/claude/claude-command-lifecycle.ts @@ -5,6 +5,7 @@ // queued. None ran. After `started`, 2.1.280 sends only `cancelled`, when an interrupt or a hard // failure ends the turn (measured), so the send may be in the conversation: any end is doubt. // An unstarted send's `cancelled` lands before the Stop's answer, so it settles if that is lost. +// `completed` with no `started` names a send its transcript already holds: doubt, never a verdict. import { agentSessionFailureFact, @@ -63,6 +64,13 @@ export function observeClaudeCommandLifecycle( agentSessionFailureFact(UNSTARTED_VERDICT[state]), onSettledLate ) + } else if (state === 'completed' && waiter.commandLifecycle !== 'started' && !waiter.retired) { + releaseClaudeDispatchWaitersInDoubt( + session, + [waiter], + DISPATCH_DOUBT_PROVIDER_ENDED_UNANSWERED, + onSettledLate + ) } } @@ -80,9 +88,18 @@ export function claudeStartedUnechoedSend(session: ClaudeSession): ClaudeStarted : null } +/** The sends the CLI holds: every waiter its echo, a lifecycle frame, idle or exit has not ended, + * and each one a frame ended until the host is told, so no reader sees it held by neither. */ +export function claudeHeldSends(session: ClaudeSession): ReadonlySet { + return new Set([ + ...session.dispatchWaiters.flatMap((waiter) => waiter.clientMessageId ?? []), + ...(session.unreportedSettlements ?? []) + ]) +} + /** A send the CLI took, until its echo, a terminal state or exit: the sweep must not rest its child. */ -export function claudeHoldsDispatch(session: ClaudeSession): boolean { - return session.dispatchWaiters.some((waiter) => waiter.commandLifecycle !== undefined) +export function claudeHoldsDispatch(session: ClaudeSession | undefined): boolean { + return session?.dispatchWaiters.some((waiter) => waiter.commandLifecycle !== undefined) ?? false } /** diff --git a/src/main/claude/claude-structured-dispatch.ts b/src/main/claude/claude-structured-dispatch.ts index 3604e1e86cf1..ef550a60624e 100644 --- a/src/main/claude/claude-structured-dispatch.ts +++ b/src/main/claude/claude-structured-dispatch.ts @@ -59,14 +59,8 @@ export function rejectClaudeDispatchWaiters( const retiredWaiters = session.retiredDispatchWaiters.filter((waiter) => named.has(waiter.sentUuid) ) - for (const waiter of activeWaiters) { - forgetWaiter(session, waiter) - waiter.resolve(null) - } - for (const waiter of retiredWaiters) { - forgetRetiredWaiter(session, waiter) - } - for (const waiter of [...activeWaiters, ...retiredWaiters]) { + // Each released as it is reported: one dropped before its report reads as held by nobody. + const report = (waiter: ClaudeDispatchWaiter): void => { if (waiter.clientMessageId) { onSettledLate?.({ clientMessageId: waiter.clientMessageId, @@ -75,6 +69,15 @@ export function rejectClaudeDispatchWaiters( }) } } + for (const waiter of activeWaiters) { + forgetWaiter(session, waiter) + waiter.resolve(null) + report(waiter) + } + for (const waiter of retiredWaiters) { + forgetRetiredWaiter(session, waiter) + report(waiter) + } } /** Releases sends the CLI took and let go without an echo or a verdict: each may have run, so each diff --git a/src/main/claude/claude-structured-session-acquisition.ts b/src/main/claude/claude-structured-session-acquisition.ts index 9b48879d000b..bd87eb66488f 100644 --- a/src/main/claude/claude-structured-session-acquisition.ts +++ b/src/main/claude/claude-structured-session-acquisition.ts @@ -27,7 +27,8 @@ import { type ClaudeSession, type ClaudeSessionExit, type ClaudeStructuredSessionAdapterDeps, - type ClaudeAcquireCallbacks + type ClaudeAcquireCallbacks, + type ClaudeLateDispatchOutcome } from './claude-structured-session-state' import { resolveClaudeAcquisitionError } from './claude-structured-session-close' import { withObservedProviderExit } from '../native-chat/agent-session-wire/structured-agent-session-failure-text' @@ -107,19 +108,16 @@ export async function acquireClaudeSession({ if (message.type === 'result' && sessions.get(sessionId) === liveSession) { persistClaudeTurnResumePoint(sessionId, liveSession, deps) } - // The CLI's idle releases its doubted sends; a late echo still accepts one it goes on to run. + // The CLI's idle wakes whatever waits for its turn to end. if (claudeSessionStateEndsTurn(message) && sessions.get(sessionId) === liveSession) { settleClaudeTurnEndWaiters(liveSession) - deps.onSessionIdle?.({ sessionId }) } } // Settled after the turn this echo opens is emitted: a send read as answered before its turn // lands reads as nothing running, and Stop and Working blink off in between. - const settlements: (() => void)[] = [] + const settlements: ClaudeLateDispatchOutcome[] = [] const turnOrigin = liveSession - ? resolveClaudeReplayTurn(liveSession, message, (settlement) => - settlements.push(() => deps.onDispatchSettledLate?.({ sessionId, ...settlement })) - ) + ? resolveClaudeReplayTurn(liveSession, message, (settlement) => settlements.push(settlement)) : null const startsTurn = turnOrigin !== null // Turn endpoints are stamped on the host clock, never the frame's own timestamp. @@ -127,29 +125,48 @@ export async function acquireClaudeSession({ startsTurn || message.type === 'result' ? { observedAt: deps.now?.() ?? Date.now() } : {} const requestedAt = turnOrigin?.requestedAt const startedSend = !startsTurn && liveSession ? claudeStartedUnechoedSend(liveSession) : null - callbacks.deliver(attempt, sessionId, () => { - if (modelChanged && liveSession && init?.model) { + // Still held until reported: the emit can publish at once, and the host holds a settled send + // only from its report, so dropping it any earlier reads it as held by nobody (doubt). + const session = liveSession + const reported = session?.unreportedSettlements + if (session && settlements.length > 0) { + session.unreportedSettlements = new Set([ + ...(reported ?? []), + ...settlements.map((settlement) => settlement.clientMessageId) + ]) + } + try { + callbacks.deliver(attempt, sessionId, () => { + if (modelChanged && liveSession && init?.model) { + callbacks.emit(liveSession, input.events, { + type: 'options-reported', + sessionId, + fence: liveSession.fence, + acquisitionGeneration: liveSession.acquisitionGeneration, + reportedOptions: { model: init.model } + }) + } callbacks.emit(liveSession, input.events, { - type: 'options-reported', + type: 'message', sessionId, - fence: liveSession.fence, - acquisitionGeneration: liveSession.acquisitionGeneration, - reportedOptions: { model: init.model } + message, + ...(startsTurn ? { startsTurn: true } : {}), + ...(requestedAt === null || requestedAt === undefined ? {} : { requestedAt }), + ...(turnOrigin?.clientMessageId ? { clientMessageId: turnOrigin.clientMessageId } : {}), + ...(startedSend ? { startedSend } : {}), + ...observedAt }) - } - callbacks.emit(liveSession, input.events, { - type: 'message', - sessionId, - message, - ...(startsTurn ? { startsTurn: true } : {}), - ...(requestedAt === null || requestedAt === undefined ? {} : { requestedAt }), - ...(turnOrigin?.clientMessageId ? { clientMessageId: turnOrigin.clientMessageId } : {}), - ...(startedSend ? { startedSend } : {}), - ...observedAt }) - }) - for (const settle of settlements) { - settle() + } finally { + try { + for (const settlement of settlements) { + deps.onDispatchSettledLate?.({ sessionId, ...settlement }) + } + } finally { + if (session) { + session.unreportedSettlements = reported + } + } } } const emit = (event: Parameters[2]): void => diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index 417bb364397a..21f4c8b6fbb3 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -7,7 +7,7 @@ import type { StructuredAgentSessionAdapter } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import { stopCurrentClaudeBackgroundTasks } from './claude-structured-control-actions' -import { claudeHoldsDispatch } from './claude-command-lifecycle' +import { claudeHeldSends, claudeHoldsDispatch } from './claude-command-lifecycle' import { releaseClaudeAcquisition } from './claude-structured-acquisition-release' import { acquireClaudeSession } from './claude-structured-session-acquisition' import { supportsClaudeStructuredLocation } from './claude-structured-location-support' @@ -217,9 +217,11 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda readCommands: NonNullable = (sessionId) => this.sessions.get(sessionId)?.commands.commands readonly atRestCommands: ClaudeStructuredSessionAdapterDeps['atRestCommands'] - holdsDispatch = (sessionId: string): boolean => { - const session = this.sessions.get(sessionId) - return session ? claudeHoldsDispatch(session) : false + holdsDispatch = (sessionId: string): boolean => claudeHoldsDispatch(this.sessions.get(sessionId)) + heldSends = (sessionId: string): ReadonlySet | undefined => { + const session = this.sessions.get(sessionId) ?? this.exits.get(sessionId)?.session + const held = session && claudeHeldSends(session) + return held ?? (this.settledExitErrors.has(sessionId) ? new Set() : undefined) } holdsLiveProviderProcess = (sessionId: string, acquisitionGeneration: string): boolean => { const session = this.sessions.get(sessionId) diff --git a/src/main/claude/claude-structured-session-close.ts b/src/main/claude/claude-structured-session-close.ts index 947b68fa847d..ec9f5be3899c 100644 --- a/src/main/claude/claude-structured-session-close.ts +++ b/src/main/claude/claude-structured-session-close.ts @@ -17,6 +17,7 @@ import type { ClaudeJournalTranslator } from './claude-journal-translator-contra import type { ClaudePromptRegistry } from './claude-structured-prompt-replies' import { closeProcessRegistry } from '@orca/process-host/close-process-registry' import { retireClaudeDispatchWaiters } from './claude-structured-dispatch' +import { claudeHeldSends } from './claude-command-lifecycle' import { settledClaudeTurnEndLeaf } from './claude-structured-resume-point' import { settleClaudeTurnEndWaiters } from './claude-request-end-wait' import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' @@ -100,7 +101,6 @@ async function finalizeClaudePublishedSession( input: CloseClaudePublishedSessionInput, session: ClaudeSession ): Promise { - retireClaudeDispatchWaiters(session) settleClaudeTurnEndWaiters(session) // Settle every in-flight permission callback so closing leaves no dangling promise; `null` // writes no response, and the SDK ignores any post-cleanup answer regardless. @@ -123,6 +123,10 @@ async function finalizeClaudePublishedSession( } rootExitVerdict = cleanupError } + // Held until the step that tells the host the child ended: released before the close's wait, a + // send reads as held by nobody, so in doubt, while the host still keeps the child. + session.events?.observeExit?.(claudeHeldSends(session)) + retireClaudeDispatchWaiters(session) // Queues the session's ending for the host's child records; the adapter delivers it after close. // Even a proven close is Orca losing contact, not a Stop the CLI accepted: what still ran ends // with no outcome anyone reported. @@ -144,6 +148,8 @@ async function finalizeClaudePublishedSession( } if (!session.closeEnded) { session.closeEnded = true + const observedAt = Date.now() + session.events?.observeExit?.(undefined, observedAt) const ended: ClaudeStructuredSessionEvent = { type: 'ended', sessionId: input.sessionId, @@ -152,7 +158,7 @@ async function finalizeClaudePublishedSession( cause: 'requested-close', fence: session.fence, acquisitionGeneration: session.acquisitionGeneration, - observedAt: Date.now() + observedAt } try { try { diff --git a/src/main/claude/claude-structured-session-exit-lifecycle.ts b/src/main/claude/claude-structured-session-exit-lifecycle.ts index 3936ed89cb19..ed4f6eb7027f 100644 --- a/src/main/claude/claude-structured-session-exit-lifecycle.ts +++ b/src/main/claude/claude-structured-session-exit-lifecycle.ts @@ -1,6 +1,7 @@ import { agentSessionFailureFact, providerDiagnosticOf } from '../../shared/agent-session-failure' import { providerExitObserved } from '../native-chat/agent-session-wire/structured-agent-session-failure-text' import { settledClaudeTurnEndLeaf } from './claude-structured-resume-point' +import { claudeHeldSends } from './claude-command-lifecycle' import { claudeRootExitObserved, settleClaudeExitedSession @@ -32,6 +33,8 @@ export function observeClaudeSessionExit( if (!session || session.connection !== attempt.connection) { return } + const observedAt = lifecycle.deps.now?.() ?? Date.now() + session.events?.observeExit?.(claudeHeldSends(session), observedAt) lifecycle.sessions.delete(sessionId) // Re-enter the provider's close ladder before publishing lifecycle recovery. // An exit callback is root evidence only; the retained tree proof must run @@ -41,6 +44,7 @@ export function observeClaudeSessionExit( connection: session.connection, session, error, + observedAt, closePromise } lifecycle.exits.set(sessionId, exit) @@ -93,7 +97,7 @@ export function settleClaudeUnexpectedExit( cause: 'unexpected-exit', fence: exit.session.fence, acquisitionGeneration: exit.session.acquisitionGeneration, - observedAt: deps.now?.() ?? Date.now() + observedAt: exit.observedAt } try { lifecycle.emit(exit.session, ended) diff --git a/src/main/claude/claude-structured-session-state.ts b/src/main/claude/claude-structured-session-state.ts index 33973db35288..c4f3268aa19c 100644 --- a/src/main/claude/claude-structured-session-state.ts +++ b/src/main/claude/claude-structured-session-state.ts @@ -95,8 +95,6 @@ export type ClaudeStructuredSessionAdapterDeps = { onEvent?: (event: ClaudeStructuredSessionEvent) => void /** Direct settlement path for provider-proven late dispatch outcomes. */ onDispatchSettledLate?: (input: { sessionId: string } & ClaudeLateDispatchOutcome) => void - /** The CLI reported `session_state_changed idle`: its turn is over. */ - onSessionIdle?: (input: { sessionId: string }) => void /** What the session's child work did, delivered after the journal handled the frame. */ /** `producer`: the reporting child's own sink, which writes what the report changes. */ onChildWorkEvidence?: ( @@ -159,6 +157,8 @@ export type ClaudeSession = { acquisitionGeneration: string prompts: ClaudePromptRegistry dispatchWaiters: ClaudeDispatchWaiter[] + /** Sends the frame being handled settled, until their settlements reach the host. */ + unreportedSettlements?: ReadonlySet /** Bounded identities for dispatches settled without an echo; a late replay still accepts one. */ retiredDispatchWaiters: ClaudeDispatchWaiter[] /** Once a retired waiter is evicted, legacy content-only replay matching is unsafe. */ @@ -233,6 +233,7 @@ export function mintClaudeAcquisitionGeneration(deps: ClaudeStructuredSessionAda * what the ladder observed, not an absence it would otherwise report as proven. */ export type ClaudeSessionExit = { + observedAt: number connection: ClaudeStreamJsonConnection /** Full session identity retained until the exit settles. */ session: ClaudeSession diff --git a/src/main/codex/codex-structured-session-close.ts b/src/main/codex/codex-structured-session-close.ts index d5a0a90ec5b9..ae477d33893d 100644 --- a/src/main/codex/codex-structured-session-close.ts +++ b/src/main/codex/codex-structured-session-close.ts @@ -27,6 +27,7 @@ export function handleCodexSessionExit(input: { return false } session.exitObservedAt ??= Date.now() + session.events?.observeExit?.(session.dispatchEchoes.heldSends(), session.exitObservedAt) // Before the admission check: the child is gone whether or not its end was admitted. session.turnOpenWaits.releaseAll() const event: StructuredAgentSessionEndedEvent = { diff --git a/src/main/codex/codex-structured-session-start.ts b/src/main/codex/codex-structured-session-start.ts index 86b6d924022c..9ed4b7a4875b 100644 --- a/src/main/codex/codex-structured-session-start.ts +++ b/src/main/codex/codex-structured-session-start.ts @@ -115,6 +115,7 @@ export async function openCodexStructuredSession( } const catalogAccess = codexAcquireCatalogAccess(deps, launch) const session: CodexSession = { + events: input.events, account: start.account, connection, ...codexSessionLifecycle(input.fence, generation), diff --git a/src/main/codex/codex-structured-session-state.ts b/src/main/codex/codex-structured-session-state.ts index 59e04d086bf2..7dfd1555e426 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -32,6 +32,7 @@ import type { } from '../native-chat/agent-model-catalog/agent-model-catalog-store' import type { StructuredAgentSessionProviderOpen } from '../native-chat/agent-session-wire/structured-agent-session-provider-open-contract' import type { NativeChatVisualsLaunch } from '../native-chat/native-chat-visuals-delivery' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' export type CodexSessionCatalogAccess = AgentModelCatalogSessionAccess @@ -133,6 +134,7 @@ export type CodexStructuredSessionAdapterDeps = { } export type CodexSession = { + events?: StructuredAgentSessionEventSink account?: AgentSessionAccountKind connection: CodexAppServerConnection ended: boolean 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 61dcbba56bcb..85967eef6604 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 @@ -369,8 +369,8 @@ export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop & /** The provider reported taking a send it has neither answered nor ended, as a queued follow-up * or a silent retry does. Derived from the live child; false with none. */ holdsDispatch?(sessionId: string): boolean - /** The sends the live child still holds as its own work, by client message id; undefined when - * this adapter cannot say, and the host reads an unanswered send as held. */ + /** The sends the live child still holds as its own work, by client message id; undefined with + * no live child, and the host then reads an unanswered send as held. */ heldSends?(sessionId: string): ReadonlySet | undefined /** The adapter's own child for this exact acquisition has a pid and its root exit has not been * seen: first-hand proof of life for lease renewal. Absent or false falls back to a PID probe. */ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-publish.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-publish.ts index d11c27786c26..5c7a2229c912 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-publish.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-publish.ts @@ -84,6 +84,7 @@ export async function publishAttachedChild( fence, phase: 'starting' as const, heldSends: () => context.deps.adapter.heldSends?.(sessionId), + exitObservation: eventSink.exitObservation, ...(startedFor === undefined ? {} : { startedFor }), ...(permissionMode === undefined ? {} : { permissionMode }) } 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 6c922fb24fbf..f611e164b075 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 @@ -40,6 +40,7 @@ import type { StructuredAgentSessionStartupAttempts } from './structured-agent-s /** How the child's root went: the close this host asked for, or a death of its own. */ export type StructuredAgentSessionChildExit = { + heldSends?: ReadonlySet expected: boolean /** Log text only; the chat's words come from `failure`. */ reason: string @@ -149,7 +150,8 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< const { expected } = exit const close = child.close // Receipt of the exit is the one end time the host may record for a running turn. - const observedAt = exit.observedAt ?? context.now() + const observation = child.exitObservation?.() + const observedAt = observation?.observedAt ?? exit.observedAt ?? context.now() // The host's own phase decides, so a provider that omits the flag still gets a start that // failed told as one: the row says so. const exitedDuringStartup = child.phase === 'starting' @@ -263,7 +265,8 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< unfinishedWork, session.journal, observedAt, - watched + watched, + observation?.heldSends ?? exit.heldSends )), ...(startFailure ? { exitFailure: startFailure } : {}), ...(startFailed && child.generation diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-opening-send-hold.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-opening-send-hold.test.ts index 62f983e9ddd8..a968e2e4c6f8 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-opening-send-hold.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-opening-send-hold.test.ts @@ -9,6 +9,7 @@ import { join } from 'node:path' import { afterEach, beforeEach, expect, it, vi } from 'vitest' import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import { agentSessionSendSubmission } from '../../../shared/agent-session-wire' import { projectStructuredAgentSessionMessages } from '../../../shared/structured-agent-session-message-projection' import { projectNativeChatTranscriptMessages } from '../../../shared/native-chat-transcript-projection' import { nativeChatRowsInDrawOrder } from '../../../shared/native-chat-turn-grouping' @@ -158,6 +159,20 @@ async function drawn(): Promise<{ text: string; turn: string | undefined }[]> { return [...nativeChatRowsInDrawOrder(keyed, drawOrder)] } +it('reads a send as accepted across its echo, never in doubt, so a worker start keeps its worker', async () => { + const connection = claude.connections[0]! + const first = await send('FIRST prompt') + await eventually(() => expect(written(connection, 'FIRST prompt')).toBeDefined()) + // What `orchestration worker-start` waits on; a doubt here closed the new worker (STA-11160). + const answer = host.waitForSendSettlement(SESSION, first, { budgetMs: 5_000 }) + frame(connection, { + ...written(connection, 'FIRST prompt')!, + isReplay: true, + parent_tool_use_id: null + }) + expect(agentSessionSendSubmission((await answer)?.value)?.dispatchState).toBe('accepted') +}) + it('holds a follow-up until the echo opens the turn, then folds it in, drawn inside that turn', async () => { const connection = claude.connections[0]! const first = await send('FIRST prompt') @@ -306,13 +321,27 @@ function stop() { }) } -// Claude's interrupt names no turn, but its Stop ends the CLI, and that end settles the send. +// Claude's interrupt names no turn, but its Stop ends the CLI, and that end settles the send. Until +// the close proves it, the CLI still holds the send: never doubt while the child is being stopped. it('settles a send a Stop took before its echo, and the next message goes out', async () => { const connection = claude.connections[0]! const first = await send('FIRST prompt') await eventually(() => expect(written(connection, 'FIRST prompt')).toBeDefined()) + const close = connection.close + let finishClose: (() => void) | undefined + connection.close = () => + new Promise((resolve) => (finishClose = () => void close().then(resolve))) - expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + const stopped = stop() + await eventually(() => expect(finishClose).toBeDefined()) + expect( + host + .currentWork(SESSION) + ?.restatedSubmissions() + .find((entry) => entry.clientMessageId === first)?.dispatchState + ).toBe('pending') + finishClose?.() + expect(await stopped).toMatchObject({ ok: true, value: { cancelled: true } }) await eventually(() => expect(connection.closeCount).toBe(1)) await eventually(async () => expect( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-stop-exit-ends-record.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-stop-exit-ends-record.test.ts index 454bcf0bf89e..1b4a6da24838 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-stop-exit-ends-record.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-stop-exit-ends-record.test.ts @@ -42,10 +42,13 @@ let claude: ReturnType let log: ReturnType let persistHandle: ReturnType Promise>> let childWork: string[] +let now: number +let initialAttach: ReturnType beforeEach(async () => { root = await mkdtemp(join(tmpdir(), 'orca-claude-stop-exit-ends-record-')) resetHostTestOperationIds() + now = NOW log = recordingStructuredAgentSessionLogger() persistHandle = vi.fn(async () => undefined) childWork = [] @@ -80,7 +83,7 @@ beforeEach(async () => { }, openConnection: claude.openConnection, readProcessStartTime: async () => 1_700_000_000_000, - now: () => NOW + now: () => now }) store = await openTestAgentSessionRecordStore(root) host = new StructuredAgentSessionHost({ @@ -93,20 +96,21 @@ beforeEach(async () => { logger: log.logger, // Ticked by hand only, and with no idle window: a tick reaps whatever is at rest. idleSweep: { intervalMs: 3_600_000, idleMs: 0 }, - now: () => NOW + now: () => now }) - const params = hostTestAttachParams(null, { + initialAttach = 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 }) + expect(await host.attach(CALLER, initialAttach)).toMatchObject({ ok: true }) await vi.waitFor(() => expect(host.readStatusSummary(SESSION)?.hostExecutionPhase).toBe('ready')) await Promise.all(lifecycle) }) afterEach(async () => { + vi.restoreAllMocks() await adapter.closeAll() await host.flushAllStreamedEvents() await rm(root, { recursive: true, force: true }) @@ -524,17 +528,70 @@ it('has the idle reaper join a close still unverifiable, ending the record with expect(host.hasSession(SESSION)).toBe(false) }) -it("ends a Claude journal-sink failure as Orca's own fault, never a quiet rest", async () => { +it('retains exit ownership through kept reattach and final-tail failure recovery', async () => { const connection = claude.connections[0]! await openTurn(connection) - - host['eventRecovery'].recoverAfterSinkFailure(SESSION, new Error('journal write failed')) - - await eventually(() => expect(child()).toBeNull()) + const pending = await send('also held at exit') + await eventually(() => expect(wrote(connection, 'also held at exit')).toBe(true)) + const exiting = child()! + const originalSink = host['runtimeState'].currentEventSink(SESSION)! + const lateEvent = deferred() + const handle = host.handleAdapterEvent.bind(host) + vi.spyOn(host, 'handleAdapterEvent').mockImplementation((event) => + event.type === 'ended' ? lateEvent.promise.then(() => handle(event)) : handle(event) + ) + const journal = host['sessions'].get(SESSION)!.journal + const append = journal.appendResolvedItem.bind(journal) + vi.spyOn(journal, 'appendResolvedItem').mockImplementationOnce(async (resolve, options) => { + await append(resolve, options) + now += 30_000 + throw new Error('final journal write acknowledgement lost') + }) + const reserved = vi.fn() + const continueAttach = deferred() + const reserve = store.reserveOwner.bind(store) + vi.spyOn(store, 'reserveOwner').mockImplementationOnce(async (request) => { + const result = await reserve(request) + reserved() + await continueAttach.promise + return result + }) + const recovery = vi.spyOn(host['eventRecovery'], 'recoverAfterSinkFailure') + // A repeated attach replays its original operation; a new one cannot take a live lease. + const reattaching = host.attach(CALLER, initialAttach) + try { + await eventually(() => expect(reserved).toHaveBeenCalledOnce()) + connection.exitVerdict = { root: 'exited', tree: 'unverifiable' } + connection.handlers.onExit?.(new Error('claude exited')) + await eventually(() => expect(recovery).toHaveBeenCalledOnce()) + expect(originalSink.state().failed).toBe(true) + expect(adapter.heldSends(SESSION)?.size).toBe(0) + continueAttach.resolve() + await expect(reattaching).resolves.toMatchObject({ ok: true }) + await eventually(() => expect(child()).toBeNull()) + } finally { + continueAttach.resolve() + lateEvent.resolve() + } expect(host['sessions'].get(SESSION)?.lastEndedChild).toMatchObject({ cause: 'exit', failure: { kind: 'hostFault' } }) + expect(exiting.exitObservation?.()).toMatchObject({ observedAt: NOW }) + expect(exiting.exitObservation?.().heldSends).toContain(pending) + expect(exiting.exitObservation).toBe(originalSink.exitObservation) + expect(claude.connections).toHaveLength(1) + const snapshot = await host.journalSnapshot(SESSION) + expect(snapshot.items).toContainEqual( + expect.objectContaining({ + body: expect.objectContaining({ kind: 'status', failure: { kind: 'hostFault' } }) + }) + ) + expect(snapshot.items).toContainEqual( + expect.objectContaining({ + body: expect.objectContaining({ kind: 'turn', state: 'interrupted', completedAt: NOW }) + }) + ) expect(connection.closeCount).toBe(1) }) 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 cc1cbfc8fc9e..628d449cc4fe 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 @@ -95,22 +95,28 @@ export class StructuredAgentSessionEventRecovery { endExitedStructuredAgentSessionChildUnderSerialize(this.exitContext, sessionId, child, exit) recoverAfterSinkFailure(sessionId: string, error: unknown): void { - if (this.sinkFailures.has(sessionId)) { + const child = this.context.sessions.get(sessionId)?.child + const heldAtFailure = new Set(child?.heldSends?.()) + if (!child || this.sinkFailures.has(sessionId)) { return } this.sinkFailures.add(sessionId) void this.context .serialize(sessionId, async () => { - 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 ( + this.context.sessions.get(sessionId)?.child !== child || + !stop || + !(await stopAgentSessionProviderRoot(() => stop(sessionId))) + ) { 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, + heldSends: heldAtFailure, reason: `journal sink failure: ${error instanceof Error ? error.message : String(error)}`, failure: agentSessionFailureFact('hostFault') }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts index eab76ae0143c..461cf3730f56 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts @@ -17,6 +17,7 @@ import { type StructuredAgentSessionTransitionSink } from './structured-agent-session-transition' import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' +import { StructuredAgentSessionExitObservation } from './structured-agent-session-exit-observation' export type StructuredAgentSessionSinkAdmission = | { accepted: true } @@ -85,6 +86,7 @@ export type StructuredAgentSessionRevisionResolver = ( export type StructuredAgentSessionRevisionOptions = StructuredAgentSessionItemAppendOptions export type StructuredAgentSessionEventSink = StructuredAgentSessionTransitionSink & { + observeExit?: StructuredAgentSessionExitObservation['record'] appendItem( identity: AgentJournalItemIdentity, body: AgentJournalItemBody, @@ -171,6 +173,7 @@ export type StructuredAgentSessionEventTarget = { export type DeferredStructuredAgentSessionEventSink = { sink: StructuredAgentSessionEventSink + exitObservation(): Pick bind(target: StructuredAgentSessionEventTarget): void unbind(): void close(): void @@ -216,6 +219,7 @@ export function createDeferredStructuredAgentSessionEventSink(deps: { readingControl?: StructuredAgentSessionReadingControl onBackpressureChange?: (backpressured: boolean, state: StructuredAgentSessionSinkState) => void }): DeferredStructuredAgentSessionEventSink { + const exit = new StructuredAgentSessionExitObservation() const watermarks = { ...DEFAULT_WATERMARKS, ...deps.watermarks } const failed = (error: unknown): void => { deps.logger.error('writing provider events to the chat journal failed', { @@ -285,10 +289,10 @@ export function createDeferredStructuredAgentSessionEventSink(deps: { ) return { + exitObservation: () => exit, sink: { - appendItem: (identity, body, options) => { - appendItem(identity, body, options) - }, + observeExit: exit.record, + appendItem: (identity, body, options) => void appendItem(identity, body, options), tryAppendItem: appendItem, ...resolvedAppend, ...createStructuredAgentSessionTransitionMembers(queue), @@ -307,9 +311,7 @@ export function createDeferredStructuredAgentSessionEventSink(deps: { return admission }, tryAppendLifecycleBatch: appendLifecycleBatch, - bindReadingControl: (control) => { - return queue.bindReadingControl(control) - }, + bindReadingControl: (control) => queue.bindReadingControl(control), appendTombstone: (identity, options = {}) => { queue.submit( { @@ -327,9 +329,7 @@ export function createDeferredStructuredAgentSessionEventSink(deps: { }, options ), - publish: (options = {}) => { - publish(options) - }, + publish: (options = {}) => void publish(options), written: queue.written, setActivity: (activity) => { queue.submit({ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-exit-observation.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-exit-observation.ts new file mode 100644 index 000000000000..fcf1aa26e3ea --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-exit-observation.ts @@ -0,0 +1,9 @@ +export class StructuredAgentSessionExitObservation { + heldSends?: ReadonlySet + observedAt?: number + + record = (heldSends?: ReadonlySet, observedAt?: number): void => { + this.heldSends ??= heldSends && new Set(heldSends) + this.observedAt ??= observedAt + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handing-over.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handing-over.ts index 932ab39621da..6cb2af015e01 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handing-over.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handing-over.ts @@ -1,16 +1,20 @@ // The sends the host is handing to its child right now: their `dispatch{pending}` row is written // before the adapter is called, so the child holds them before it can report so (`heldSends`). +// Also each send whose late settlement is being written: the child let it go before its row lands. import type { AgentSessionJournal } from '../agent-session-journal/journal-store' /** The chat's journal, which identifies the conversation for as long as it is held. */ type HandoverJournal = Pick -const handingOver = new WeakMap>() -const NONE: ReadonlySet = new Set() +// Counted per send: a late settlement's write can open inside the same send's handover. +const handingOver = new WeakMap>() +const NONE: ReadonlyMap = new Map() -/** The sends this journal's chat is handing over now. */ -export function structuredAgentSessionHandingOver(journal: HandoverJournal): ReadonlySet { +/** The sends this journal's chat is handing over now, each with its open windows. */ +export function structuredAgentSessionHandingOver( + journal: HandoverJournal +): ReadonlyMap { return handingOver.get(journal) ?? NONE } @@ -38,15 +42,20 @@ export function beginStructuredAgentSessionHandover( ): () => void { let sends = handingOver.get(journal) if (!sends) { - sends = new Set() + sends = new Map() handingOver.set(journal, sends) } - sends.add(clientMessageId) + sends.set(clientMessageId, (sends.get(clientMessageId) ?? 0) + 1) let held = true return (): void => { if (held) { held = false - sends.delete(clientMessageId) + const count = sends.get(clientMessageId) ?? 1 + if (count > 1) { + sends.set(clientMessageId, count - 1) + } else { + sends.delete(clientMessageId) + } workChanged?.() } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutation-wiring.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutation-wiring.ts index 0794f53fa685..5989ca262e70 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutation-wiring.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutation-wiring.ts @@ -33,6 +33,7 @@ export function createStructuredAgentSessionHostMutationContext( conversation: input.lifetime.conversation, readChildWork: input.clientDelivery.readChildWork, publishChildWork: input.clientDelivery.publishChildWork, + publishWorkChanged: (sessionId) => input.clientDelivery.publishGenerationEnded(sessionId), captureChildWorkStops: input.clientDelivery.captureChildWorkStops, childWorkOpen: input.clientDelivery.childWorkOpen, childWorkIncomplete: input.clientDelivery.childWorkIncomplete, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts index 411106d73c44..a19a73cf5c46 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts @@ -62,7 +62,6 @@ import { type StructuredAgentSessionOptionsReader } from './structured-agent-session-options-read' import { settleStructuredAgentSessionLateDispatch } from './structured-agent-session-late-dispatch' -import { releaseStructuredAgentSessionUnansweredDispatches } from './structured-agent-session-unanswered-dispatch-release' export function sendStructuredAgentSessionTurn( context: StructuredAgentSessionMutationContext, @@ -304,9 +303,6 @@ export function structuredAgentSessionMutationDelegates( reader: StructuredAgentSessionOptionsReader = IN_PROCESS_OPTIONS_READER ) => readStructuredAgentSessionOptions(context(), sessionId, reader), settleLateDispatch: (input: Parameters[1]) => - settleStructuredAgentSessionLateDispatch(context(), input), - releaseUnansweredDispatches: ( - input: Parameters[1] - ) => releaseStructuredAgentSessionUnansweredDispatches(context(), input) + settleStructuredAgentSessionLateDispatch(context(), input) } } 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 d7a5e9765890..66e226c5c36a 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 @@ -13,6 +13,7 @@ import type { AgentSessionRecoveryCapsule } from '../../runtime/agent-session-re import type { AgentSessionSpawnTokenScan } from '../../runtime/agent-session-spawn-token-process-scan' import type { JournalHostDatabase } from '../agent-session-journal/journal-host-database' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import type { DeferredStructuredAgentSessionEventSink } from './structured-agent-session-event-sink' import type { JournalStopSettle } from '../agent-session-journal/queued-message-pause' import type { StructuredAgentSessionAdapter, @@ -80,6 +81,7 @@ export type StructuredAgentSessionProviderChild = StructuredAgentSessionProvider readonly permissionMode?: AgentChatPermissionMode /** The adapter's `heldSends` for this conversation, read at each decision. */ readonly heldSends?: () => ReadonlySet | undefined + readonly exitObservation?: DeferredStructuredAgentSessionEventSink['exitObservation'] close?: StructuredAgentSessionChildClose } 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 0a87b81610d3..d5141b4256d7 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 @@ -344,7 +344,6 @@ export class StructuredAgentSessionHost { subscribe = (input: AgentSessionSubscribeInput) => this.backgroundTasks.subscribe(input) settleLateDispatch = this.mutations.settleLateDispatch - releaseUnansweredDispatches = this.mutations.releaseUnansweredDispatches /** What the live child holds changed with no row: every reader re-derives current work. */ publishWorkChanged: (sessionId: string) => void = this.clientDelivery.publishGenerationEnded diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts index 0bbc45bd51ec..d926eee872bf 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts @@ -6,9 +6,11 @@ import type { AgentJournalDispatchRejection } from '../../../shared/agent-sessio import type { ResolveDispatchInput } from '../agent-session-journal/journal-store-contracts' import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' +import { whileStructuredAgentSessionHandingOver } from './structured-agent-session-handing-over' /** Settle a provider-proven outcome independently of an in-flight client mutation. `unknown` is - * released doubt: the provider let the send go unanswered, so nothing re-sends it. */ + * released doubt: the provider let the send go unanswered, so nothing re-sends it. The send reads + * held until its row settles; a failed write leaves it unanswered, which clients learn at once. */ export async function settleStructuredAgentSessionLateDispatch( context: StructuredAgentSessionMutationContext, input: { @@ -52,8 +54,19 @@ export async function settleStructuredAgentSessionLateDispatch( ...(input.answeredInTurn ? { answeredInTurn: input.answeredInTurn } : {}), fence } - // The journal queue drains before close; the host queue would defer this past teardown. - await (settlement.state === 'accepted' - ? session.journal.acceptLate(settlement) - : session.journal.resolveDispatch(settlement)) + try { + // The journal queue drains before close; the host queue would defer this past teardown. + await whileStructuredAgentSessionHandingOver( + session.journal, + input.clientMessageId, + async () => { + await (settlement.state === 'accepted' + ? session.journal.acceptLate(settlement) + : session.journal.resolveDispatch(settlement)) + } + ) + } catch (error) { + context.publishWorkChanged(input.sessionId) + throw error + } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts index 4b7d40382e04..41dbf4fa6ce6 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts @@ -36,6 +36,8 @@ export type StructuredAgentSessionMutationContext = { /** The session's child records, as the strip reads them; what command admission decides on. */ readChildWork: (sessionId: string) => AgentChildWorkView[] | undefined publishChildWork: (sessionId: string, evidence: AgentChildWorkEvidence[]) => void + /** What the live child holds changed with no row (`publishGenerationEnded`). */ + publishWorkChanged: (sessionId: string) => void captureChildWorkStops: (sessionId: string, handleIds: readonly string[]) => CapturedChildWorkStops /** Some of the session's running child work has no record. */ childWorkOpen: (sessionId: string) => boolean diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts deleted file mode 100644 index 9a27415dc711..000000000000 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts +++ /dev/null @@ -1,105 +0,0 @@ -import { describe, expect, it, vi } from 'vitest' -import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' -import { projectStructuredAgentSessionStatus } from '../../../shared/structured-agent-session-projection' -import { releaseStructuredAgentSessionUnansweredDispatches } from './structured-agent-session-unanswered-dispatch-release' - -const FENCE = 7 - -function submission(over: Partial): AgentJournalSubmission { - return { - clientMessageId: 'm-1', - fence: FENCE, - payloadFingerprint: 'fp', - dispatchState: 'unknown', - providerItemId: null, - reason: 'provider_write_outcome_unknown: timeout', - submittedAt: 1, - resolvedAt: 2, - ...over - } -} - -type ReleaseContext = Parameters[0] -type ReleaseSession = NonNullable> -type ResolveDispatchInput = Parameters[0] - -function contextWith(submissions: AgentJournalSubmission[]) { - const resolved: ResolveDispatchInput[] = [] - const journal = { - submissions: () => submissions, - resolveDispatch: vi.fn(async (input: ResolveDispatchInput) => { - resolved.push(input) - return { epoch: 'e', sequence: 1 } - }) - } - // The mutation reads only `journal.submissions`, `journal.resolveDispatch` and the record fence. - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: all three are supplied here; the rest of the session and the record is unreachable from this mutation. - const context = { - sessions: new Map([['s-1', { journal }]]), - deps: { store: { getRecord: () => ({ lease: { runtimeFence: FENCE } }) } } - } as unknown as ReleaseContext - return { context, resolved, journal } -} - -describe('releasing dispatches the provider can no longer answer', () => { - it('retires a live unknown so the session stops reading working', async () => { - const live = [submission({})] - // POSITIVE CONTROL: this is the latch being dissolved. - expect(projectStructuredAgentSessionStatus([], live, FENCE)).toBe('working') - const { context, resolved } = contextWith(live) - - await releaseStructuredAgentSessionUnansweredDispatches(context, { - sessionId: 's-1', - reason: 'provider_idle_before_acknowledgement' - }) - - expect(resolved).toEqual([ - { - clientMessageId: 'm-1', - state: 'unknown', - // The sharper earlier fact survives, exactly as restart recovery keeps it. - reason: 'provider_write_outcome_unknown: timeout', - fence: FENCE, - recovered: true - } - ]) - expect(projectStructuredAgentSessionStatus([], [submission({ recovered: true })], FENCE)).toBe( - 'idle' - ) - }) - - it('never touches a pending send, whose dispatch may still be in flight', async () => { - const { context, resolved } = contextWith([ - submission({ clientMessageId: 'm-2', dispatchState: 'pending', reason: null }) - ]) - - await releaseStructuredAgentSessionUnansweredDispatches(context, { - sessionId: 's-1', - reason: 'provider_idle_before_acknowledgement' - }) - - expect(resolved).toEqual([]) - }) - - it('leaves an already recovered unknown alone', async () => { - const { context, resolved } = contextWith([submission({ recovered: true })]) - - await releaseStructuredAgentSessionUnansweredDispatches(context, { - sessionId: 's-1', - reason: 'provider_idle_before_acknowledgement' - }) - - expect(resolved).toEqual([]) - }) - - it('falls back to the supplied reason when the submission named none', async () => { - const { context, resolved } = contextWith([submission({ reason: null })]) - - await releaseStructuredAgentSessionUnansweredDispatches(context, { - sessionId: 's-1', - reason: 'provider_idle_before_acknowledgement' - }) - - expect(resolved[0]).toMatchObject({ reason: 'provider_idle_before_acknowledgement' }) - }) -}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.ts deleted file mode 100644 index 62078ef853c6..000000000000 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.ts +++ /dev/null @@ -1,43 +0,0 @@ -import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' -import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types' -import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' - -/** - * Releases doubted sends once the provider reports it has stopped running. - * - * A dispatch whose RPC timed out is recorded `unknown` — doubt, never proof of - * non-delivery — and a live `unknown` reads as work still owed, so the session - * shows working until something re-derives it. The provider reporting its thread - * not running, with no turn open, IS that re-derivation. - * - * `pending` is deliberately untouched: that send's dispatch has not returned yet - * and may be in flight right now. And `recovered` only retires the obligation — - * it never makes a send re-deliverable, because the provider may well have run it. - */ -export async function releaseStructuredAgentSessionUnansweredDispatches( - context: Pick & { - deps: { store: Pick } - }, - input: { sessionId: string; reason: string } -): Promise { - const session = context.sessions.get(input.sessionId) - if (!session) { - return - } - const stranded = session.journal - .submissions() - .filter((entry) => entry.dispatchState === 'unknown' && entry.recovered !== true) - if (stranded.length === 0) { - return - } - for (const entry of stranded) { - await session.journal.resolveDispatch({ - clientMessageId: entry.clientMessageId, - state: 'unknown', - // The earlier reason names a sharper fact than this one does. - reason: entry.reason ?? input.reason, - fence: structuredAgentSessionConversationFence(context.deps.store, input.sessionId), - recovered: true - }) - } -} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts index 62cbe6260072..17c244821050 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts @@ -343,7 +343,12 @@ describe('provider-exit settlement', () => { it('settles a submission the dead child never acknowledged', async () => { const appendLifecycleBatch = vi.fn(async () => ({ epoch: 'epoch-1', sequence: 1 })) const session: StructuredAgentSessionChildExitSession = { - child: { generation: GENERATION, fence: 7, phase: 'ready' }, + child: { + generation: GENERATION, + fence: 7, + phase: 'ready', + exitObservation: () => ({ heldSends: new Set(['client-1']) }) + }, journal: exitJournalDouble({ items: () => [], appendLifecycleBatch, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unfinished-work.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unfinished-work.ts index e0a98e7d7716..9a90800184ac 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unfinished-work.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unfinished-work.ts @@ -9,6 +9,7 @@ import { type StructuredAgentSessionWatchedExit } from './structured-agent-session-stale-turn-verdict' import type { UnopenedSendJournal } from './structured-agent-session-unopened-send-withdrawal' +import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission' export type DeadGenerationJournal = UnopenedSendJournal & { appendPlannedLifecycleBatch: AgentSessionJournal['appendPlannedLifecycleBatch'] @@ -47,11 +48,20 @@ export function unfinishedStructuredAgentSessionWorkWasInterrupted( before: StructuredAgentSessionUnfinishedWork, journal: DeadGenerationJournal, observedExitAt: number, - exit?: StructuredAgentSessionWatchedExit + exit?: StructuredAgentSessionWatchedExit, + heldSends?: ReadonlySet ): boolean { const currentSnapshot = journal.snapshot() if ( - hasUnsettledSubmission(journal) || + journal + .submissions?.() + .some( + (send) => + heldSends?.has(send.clientMessageId) && + send.fence === exit?.ownerFence && + !isQueuedAgentJournalSubmission(send) && + (send.dispatchState === 'pending' || send.dispatchState === 'unknown') + ) || currentSnapshot.items.some(isInProgressStructuredAgentSessionItem) ) { return true diff --git a/src/main/pi/rpc-location.test.ts b/src/main/pi/rpc-location.test.ts index b3c500ba45ff..b298fff9045b 100644 --- a/src/main/pi/rpc-location.test.ts +++ b/src/main/pi/rpc-location.test.ts @@ -31,7 +31,6 @@ describe('Pi execution host support', () => { }, onLifecycle: vi.fn(), onSettled: vi.fn(), - onIdle: vi.fn(), logger: { warn: vi.fn(), error: vi.fn() } }) expect(structuredAgentRuntimeRegistration('pi')?.supportsLocation(local)).toBe(proof) diff --git a/src/main/pi/rpc-message-dispatch.ts b/src/main/pi/rpc-message-dispatch.ts index b3afb19c5767..807f52b37ae9 100644 --- a/src/main/pi/rpc-message-dispatch.ts +++ b/src/main/pi/rpc-message-dispatch.ts @@ -60,7 +60,7 @@ export async function dispatchPiRpcMessage( const model = input.executionOptions?.model ?? session.initialOptions.model if (!input.executionOptions) { input.signal?.throwIfAborted() - return session.turns.submit( + return session.submit( input.clientMessageId, input.requestedAt ?? Date.now(), { ...prompt }, @@ -108,7 +108,7 @@ export async function dispatchPiRpcMessage( if (!isCurrent() || session.connection.closed) { return exited() } - return session.turns.submit( + return session.submit( input.clientMessageId, input.requestedAt ?? Date.now(), { ...prompt }, diff --git a/src/main/pi/rpc-runtime-registration.ts b/src/main/pi/rpc-runtime-registration.ts index a86a3fdf5ffa..aa20e80f636d 100644 --- a/src/main/pi/rpc-runtime-registration.ts +++ b/src/main/pi/rpc-runtime-registration.ts @@ -35,7 +35,6 @@ function createPiRpcAdapter(context: StructuredAgentAdapterContext): StructuredA ...(outcome.state === 'accepted' ? { providerIdentity: outcome.providerIdentity } : outcome) }) }, - onIdle: context.followUps.releaseUnansweredDispatches, logger: deps.logger }) } diff --git a/src/main/pi/rpc-session-adapter.test-fixture.ts b/src/main/pi/rpc-session-adapter.test-fixture.ts index b5d1d48dc400..52a50e5c2af3 100644 --- a/src/main/pi/rpc-session-adapter.test-fixture.ts +++ b/src/main/pi/rpc-session-adapter.test-fixture.ts @@ -127,7 +127,6 @@ export async function setup( const connections: FakeConnection[] = [] const lifecycle = vi.fn() const settled = vi.fn() - const idle = vi.fn() const onSpawned = vi.fn(async (_process: AgentSessionProcessIdentity) => { expect(connections.at(-1)?.requests).toEqual([]) }) @@ -164,7 +163,6 @@ export async function setup( openConnection, onLifecycle: lifecycle, onSettled: settled, - onIdle: idle, logger: { warn: vi.fn(), error: vi.fn() } }) const acquired = await adapter.acquire(input) @@ -193,7 +191,6 @@ export async function setup( onSpawned, lifecycle, settled, - idle, input, resolveLaunch, openConnection, diff --git a/src/main/pi/rpc-session-adapter.test.ts b/src/main/pi/rpc-session-adapter.test.ts index f3b7153f59dd..a0416a9ed56f 100644 --- a/src/main/pi/rpc-session-adapter.test.ts +++ b/src/main/pi/rpc-session-adapter.test.ts @@ -163,6 +163,27 @@ describe('Pi RPC session ownership and delivery', () => { expect(h.connection.sent[0]).toMatchObject({ type: 'prompt', message: 'hello' }) }) + it('holds each send from its handover until it settles, and none whose write failed', async () => { + const h = await setup() + const send = (clientMessageId: string) => + h.adapter.dispatch({ + sessionId, + fence: 7, + clientMessageId, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: clientMessageId }] } + }) + vi.spyOn(h.connection, 'send').mockRejectedValueOnce(new Error('stdin closed')) + await expect(send('lost')).resolves.toMatchObject({ state: 'unknown' }) + expect(h.adapter.heldSends(sessionId)).toEqual(new Set()) + + await send('first') + expect(h.adapter.heldSends(sessionId)).toEqual(new Set(['first'])) + h.connection.receive({ type: 'agent_start' }) + await send('steer') + // The first is accepted; the steer waits behind its running turn. + expect(h.adapter.heldSends(sessionId)).toEqual(new Set(['steer'])) + }) + it('applies each sent model before effort and before prompting Pi', async () => { const h = await setup() h.connection.requestOverride = (command) => diff --git a/src/main/pi/rpc-session-adapter.ts b/src/main/pi/rpc-session-adapter.ts index 8afdfbd1cb41..324422f384f5 100644 --- a/src/main/pi/rpc-session-adapter.ts +++ b/src/main/pi/rpc-session-adapter.ts @@ -225,6 +225,9 @@ export class PiRpcSessionAdapter implements StructuredAgentSessionAdapter { holdsDispatch(id: string): boolean { return this.sessions.get(id)?.turns.holdsDispatch ?? false } + heldSends(id: string): ReadonlySet | undefined { + return this.sessions.get(id)?.heldSends + } holdsLiveProviderProcess(id: string, generation: string): boolean { const session = this.sessions.get(id) return session?.generation === generation && !session.connection.closed diff --git a/src/main/pi/rpc-session.ts b/src/main/pi/rpc-session.ts index ddab7c619861..4f934fe98570 100644 --- a/src/main/pi/rpc-session.ts +++ b/src/main/pi/rpc-session.ts @@ -52,7 +52,6 @@ export type PiRpcSessionDeps = { fence: number outcome: AgentSessionDispatchOutcome }) => void - onIdle: (input: { sessionId: string }) => void logger: StructuredAgentSessionLogger } @@ -63,6 +62,8 @@ export class PiRpcSession { readonly turns: PiRpcTurns readonly dialogs: PiRpcDialogCallbacks readonly selected = new Map() + /** The sends this child holds: each from its handover until its settlement. */ + readonly heldSends = new Set() readonly optionProof = { current: new Map(), writes: new Map() } readonly startup: PiRpcSessionStartup initialOptions: Readonly> = {} @@ -107,10 +108,7 @@ export class PiRpcSession { send: (frame) => this.connection.send(frame), request: (command, params, options) => this.connection.request(command, params, options), settled: (id, outcome) => this.settle(id, outcome), - idle: () => { - this.dialogs.cancelAll() - deps.onIdle({ sessionId: input.identity.sessionId }) - }, + idle: () => this.dialogs.cancelAll(), failed: (error) => this.fail(error), diagnostic: (error) => deps.logger.warn('Pi context usage could not be read', { @@ -158,6 +156,21 @@ export class PiRpcSession { ) } + async submit(...args: Parameters): ReturnType { + const [clientMessageId] = args + this.heldSends.add(clientMessageId) + let outcome: AgentSessionDispatchOutcome | undefined + try { + outcome = await this.turns.submit(...args) + return outcome + } finally { + // Admitted, it stays held until its settlement; anything else was answered here. + if (outcome?.state !== 'admitted') { + this.heldSends.delete(clientMessageId) + } + } + } + async open(signal: AbortSignal): Promise { const result = await this.startup.run(this.resolvedLaunch, signal) this.initialOptions = { @@ -241,13 +254,15 @@ export class PiRpcSession { return } this.publishedExit = true + const observedAt = Date.now() + this.input.events?.observeExit?.(this.heldSends, observedAt) this.failedCause ??= error this.dialogs.cancelAll() this.turns.end() // The lane is sealed once finalized, so the end event travels with the final transfer. this.lane.finalize({ type: 'session.ended', - verdict: { state: 'interrupted', completedAt: Date.now() } + verdict: { state: 'interrupted', completedAt: observedAt } }) let timer: ReturnType | undefined try { @@ -273,7 +288,7 @@ export class PiRpcSession { sessionId: this.input.identity.sessionId, fence: this.input.fence, acquisitionGeneration: this.generation, - observedAt: Date.now(), + observedAt, cause: this.requestedClose ? 'requested-close' : 'unexpected-exit', reason: cause.message, failure: agentSessionFailureFact('providerExited', { @@ -288,6 +303,7 @@ export class PiRpcSession { } private settle(clientMessageId: string, outcome: AgentSessionDispatchOutcome): void { + this.heldSends.delete(clientMessageId) this.deps.onSettled({ sessionId: this.input.identity.sessionId, clientMessageId, diff --git a/src/main/runtime/claude-structured-idle-releases-unknown.test.ts b/src/main/runtime/claude-structured-idle-releases-unknown.test.ts index 9fa9cbfcfa6a..2541f004e00a 100644 --- a/src/main/runtime/claude-structured-idle-releases-unknown.test.ts +++ b/src/main/runtime/claude-structured-idle-releases-unknown.test.ts @@ -1,6 +1,6 @@ -// A Claude send whose write ended in doubt is recorded `unknown`, and a live `unknown` reads as -// work still owed. The CLI reporting `session_state_changed idle` retires the doubt; a send whose -// dispatch is still `pending` is left alone unless the CLI had started it. +// A Claude chat is Working only while the CLI holds a send. One whose write ended in doubt is +// recorded `unknown` and stops Working at once; one the CLI took and let go without an echo is +// released doubt; one it has only queued stays held through an idle that may precede its start. // Against the production runtime, adapter, record store and host, with only the CLI scripted. import { afterEach, describe, expect, it, vi } from 'vitest' @@ -14,7 +14,10 @@ import { import { hasUnansweredStructuredAgentSessionDispatch } from '../../shared/structured-agent-session-unanswered-dispatch' import { hostTestMessage } from '../native-chat/agent-session-wire/structured-agent-session-host-test-data' import type { StructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-host' -import { createScriptedClaudeRuntime } from './structured-claude-scripted-runtime-test-support' +import { + createScriptedClaudeRuntime, + scriptedClaudeExitError +} from './structured-claude-scripted-runtime-test-support' const SESSION = 'claude-idle-release' const PROVIDER_SESSION = claudeSessionIdForOrcaSession(SESSION) @@ -67,48 +70,64 @@ function sessionState(state: 'running' | 'idle'): Record { } } -describe('a live Claude chat whose CLI reports idle', () => { - it('releases a send whose write ended in doubt, and leaves a pending one alone', async () => { +describe('a live Claude chat whose send could not be written', () => { + it('releases doubt at once and stays quiet on idle exit after a later successful turn', async () => { const host = await claude.install() await expect(host.attach(CALLER, claude.attachParams(SESSION, null))).resolves.toMatchObject({ ok: true }) - const child = claude.child(SESSION) // Handed to the SDK, then the write's outcome was lost. + const child = claude.child(SESSION) child.connection.send = async (_message, beforeDispatch) => { await beforeDispatch?.() throw new Error('stdin write stalled') } const doubted = await send(host, 'first') await vi.waitFor(async () => - expect(await submission(host, doubted)).toMatchObject({ dispatchState: 'unknown' }) + expect(await submission(host, doubted)).toMatchObject({ + dispatchState: 'unknown', + reason: `${DISPATCH_DOUBT_WRITE_OUTCOME_UNKNOWN}: stdin write stalled` + }) ) - // Written, never echoed: its dispatch has returned, but nothing has settled it. - child.connection.send = async (_message, beforeDispatch) => { + + // Nothing will rewrite the row, so nothing may wait on it. + await vi.waitFor(() => expect(host.currentWork(SESSION)?.working()).toBe(false)) + const raw = await submission(host, doubted) + expect(raw?.recovered).toBeUndefined() + const published = raw && host.currentWork(SESSION)?.publishedSubmission(raw) + expect(published).toMatchObject({ dispatchState: 'unknown', recovered: true }) + expect(published && hasUnansweredStructuredAgentSessionDispatch([published], fence(host))).toBe( + false + ) + + child.connection.send = async (message, beforeDispatch) => { await beforeDispatch?.() + child.handlers.onMessage?.({ ...message, session_id: PROVIDER_SESSION }) + child.handlers.onMessage?.({ + type: 'result', + subtype: 'success', + is_error: false, + uuid: 'later-result', + session_id: PROVIDER_SESSION + }) } - const pending = await send(host, 'second') + const later = await send(host, 'second succeeds') await vi.waitFor(async () => - expect(await submission(host, pending)).toMatchObject({ dispatchState: 'pending' }) + expect(await submission(host, later)).toMatchObject({ dispatchState: 'accepted' }) ) - - child.handlers.onMessage?.(sessionState('running')) - await new Promise((resolve) => setTimeout(resolve, 50)) - // POSITIVE CONTROL: only idle releases it. - expect((await submission(host, doubted))?.recovered).toBeUndefined() - - child.handlers.onMessage?.(sessionState('idle')) - - await vi.waitFor(async () => - expect(await submission(host, doubted)).toMatchObject({ - dispatchState: 'unknown', - recovered: true, - // The write's own doubt is the sharper fact, so it survives the release. - reason: `${DISPATCH_DOUBT_WRITE_OUTCOME_UNKNOWN}: stdin write stalled` - }) + await vi.waitFor(() => expect(host.currentWork(SESSION)?.working()).toBe(false)) + child.exit(scriptedClaudeExitError('claude stream-json exited (code 137)')) + await vi.waitFor(() => + expect(host.deps.store.getRecord(SESSION)?.lease.claimStatus).toBe('released') + ) + const snapshot = await host.journalSnapshot(SESSION) + expect(snapshot.items.some((item) => item.body.kind === 'status' && item.body.failure)).toBe( + false ) - expect(await submission(host, pending)).toMatchObject({ dispatchState: 'pending' }) - expect((await submission(host, pending))?.recovered).toBeUndefined() + expect( + snapshot.items.some((item) => item.body.kind === 'turn' && item.body.state === 'completed') + ).toBe(true) + expect(await submission(host, doubted)).toMatchObject({ dispatchState: 'unknown' }) }) }) @@ -124,9 +143,11 @@ function lifecycle(commandUuid: string, state: string): Record describe('a live Claude chat whose CLI took a send and let it go without an echo', () => { it.each([ - ['its turn ended cancelled', 'cancelled', DISPATCH_DOUBT_PROVIDER_ENDED_UNANSWERED], - ['it went idle with no terminal state', 'idle', DISPATCH_DOUBT_PROVIDER_IDLE] - ])('records released doubt when %s', async (_label, ending, reason) => { + ['its turn was cancelled', ['started', 'cancelled'], DISPATCH_DOUBT_PROVIDER_ENDED_UNANSWERED], + ['it went idle with no terminal state', ['started', 'idle'], DISPATCH_DOUBT_PROVIDER_IDLE], + // Synthesized: no captured transcript has `completed` without `started`. + ['it completed with no start', ['completed'], DISPATCH_DOUBT_PROVIDER_ENDED_UNANSWERED] + ])('records released doubt when %s', async (_label, frames, reason) => { const host = await claude.install() await expect(host.attach(CALLER, claude.attachParams(SESSION, null))).resolves.toMatchObject({ ok: true @@ -142,12 +163,14 @@ describe('a live Claude chat whose CLI took a send and let it go without an echo await vi.waitFor(() => expect(sentUuid).not.toBe('')) expect(await submission(host, taken)).toMatchObject({ dispatchState: 'pending' }) child.handlers.onMessage?.(lifecycle(sentUuid, 'queued')) - child.handlers.onMessage?.(lifecycle(sentUuid, 'started')) expect(host.deps.adapter.holdsDispatch?.(SESSION)).toBe(true) + expect(host.currentWork(SESSION)?.working()).toBe(true) - child.handlers.onMessage?.( - ending === 'idle' ? sessionState('idle') : lifecycle(sentUuid, ending) - ) + for (const frame of frames) { + child.handlers.onMessage?.( + frame === 'idle' ? sessionState('idle') : lifecycle(sentUuid, frame) + ) + } await vi.waitFor(async () => expect(await submission(host, taken)).toMatchObject({ @@ -157,6 +180,7 @@ describe('a live Claude chat whose CLI took a send and let it go without an echo }) ) expect(host.deps.adapter.holdsDispatch?.(SESSION)).toBe(false) + expect(host.currentWork(SESSION)?.working()).toBe(false) // Released doubt: the chat stops reading working, and nothing re-sends it. expect( hasUnansweredStructuredAgentSessionDispatch((await host.journalSnapshot(SESSION)).submissions) @@ -185,6 +209,7 @@ describe('a live Claude chat whose CLI took a send and let it go without an echo await new Promise((resolve) => setTimeout(resolve, 50)) expect(host.deps.adapter.holdsDispatch?.(SESSION)).toBe(true) + expect(host.currentWork(SESSION)?.working()).toBe(true) expect(await submission(host, queued)).toMatchObject({ dispatchState: 'pending' }) expect((await submission(host, queued))?.recovered).toBeUndefined() }) diff --git a/src/main/runtime/structured-agent-runtime-registrations.ts b/src/main/runtime/structured-agent-runtime-registrations.ts index fd6b22376c3f..d3b7f1ebe738 100644 --- a/src/main/runtime/structured-agent-runtime-registrations.ts +++ b/src/main/runtime/structured-agent-runtime-registrations.ts @@ -118,7 +118,6 @@ function createClaudeAdapter( onChildWorkEvidence: (sessionId, evidence, producer) => host()?.publishChildWorkEvidence(sessionId, evidence, producer), onDispatchSettledLate: followUps.onDispatchSettledLate, - onSessionIdle: followUps.releaseUnansweredDispatches, ...(deps.openClaudeConnection ? { openClaudeConnection: deps.openClaudeConnection } : {}), ...(deps.readProcessStartTime ? { readProcessStartTime: deps.readProcessStartTime } : {}) }) diff --git a/src/main/runtime/structured-agent-session-codex-held-sends.test.ts b/src/main/runtime/structured-agent-session-codex-held-sends.test.ts index 5f84e4d4bf26..33578d298399 100644 --- a/src/main/runtime/structured-agent-session-codex-held-sends.test.ts +++ b/src/main/runtime/structured-agent-session-codex-held-sends.test.ts @@ -245,6 +245,24 @@ describe('a Codex send whose turn completes without taking it', () => { ) const published = lastPublished(blocked) expect(published && hasUnansweredStructuredAgentSessionDispatch([published], fence)).toBe(false) + + const later = await send('second succeeds') + await vi.waitFor(() => expect(answers).toBe(2)) + turns.start() + turns.echo(later) + turns.end('completed') + await vi.waitFor(async () => + expect((await rawSubmission(later)).dispatchState).toBe('accepted') + ) + handlers?.onExit?.(new Error('codex exited')) + await vi.waitFor(() => + expect(host.deps.store.getRecord(SESSION)?.lease.claimStatus).toBe('released') + ) + expect( + (await host.journalSnapshot(SESSION)).items.some( + (item) => item.body.kind === 'status' && item.body.failure + ) + ).toBe(false) }) it('keeps a send Working while Codex holds it, and its echo still settles it', async () => { diff --git a/src/main/runtime/structured-agent-session-dispatch-followups.ts b/src/main/runtime/structured-agent-session-dispatch-followups.ts index dbc5a31b0d1f..92906bbd86c8 100644 --- a/src/main/runtime/structured-agent-session-dispatch-followups.ts +++ b/src/main/runtime/structured-agent-session-dispatch-followups.ts @@ -1,15 +1,10 @@ -// What the runtime does when a provider settles a dispatch after the fact: a late answer to a send, -// or the provider going idle with sends it never answered. Both run off the request that caused -// them, so their failure has nobody to return to and is logged instead. +// What the runtime does when a provider settles a dispatch after the fact. It runs off the request +// that caused it, so its failure has nobody to return to and is logged instead. -import { DISPATCH_DOUBT_PROVIDER_IDLE } from '../native-chat/agent-session-journal/journal-dispatch-doubt-reasons' import type { StructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-host' import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' -type DispatchFollowUpHost = Pick< - StructuredAgentSessionHost, - 'settleLateDispatch' | 'releaseUnansweredDispatches' -> +type DispatchFollowUpHost = Pick export function createStructuredAgentSessionDispatchFollowUps(input: { /** Null until the host is built: the adapters that call these are built first. */ @@ -19,7 +14,6 @@ export function createStructuredAgentSessionDispatchFollowUps(input: { onDispatchSettledLate: ( settlement: Parameters[0] ) => void - releaseUnansweredDispatches: (input: { sessionId: string }) => void } { const { logger } = input return { @@ -34,19 +28,6 @@ export function createStructuredAgentSessionDispatchFollowUps(input: { error }) ) - }, - // The provider going idle is what re-derives a doubted send; a late echo still accepts it. - releaseUnansweredDispatches: ({ sessionId }) => { - void input - .host() - ?.releaseUnansweredDispatches({ sessionId, reason: DISPATCH_DOUBT_PROVIDER_IDLE }) - .catch((error: unknown) => - logger.warn('releasing sends the idle provider never answered failed', { - scope: 'unanswered-dispatch', - sessionId, - error - }) - ) } } } diff --git a/src/main/runtime/structured-claude-runtime-adapter.ts b/src/main/runtime/structured-claude-runtime-adapter.ts index 0e008888a231..b8cc1e7cda59 100644 --- a/src/main/runtime/structured-claude-runtime-adapter.ts +++ b/src/main/runtime/structured-claude-runtime-adapter.ts @@ -37,7 +37,6 @@ export type StructuredClaudeRuntimeAdapterDeps = { readProcessStartTime?: ClaudeStructuredSessionAdapterDeps['readProcessStartTime'] onLifecycleEvent: (event: StructuredAgentSessionLifecycleEvent) => void onDispatchSettledLate?: ClaudeStructuredSessionAdapterDeps['onDispatchSettledLate'] - onSessionIdle?: ClaudeStructuredSessionAdapterDeps['onSessionIdle'] onChildWorkEvidence?: ClaudeStructuredSessionAdapterDeps['onChildWorkEvidence'] logger?: ClaudeStructuredSessionAdapterDeps['logger'] } @@ -128,7 +127,6 @@ export function createStructuredClaudeRuntimeAdapter( } }, ...(deps.onDispatchSettledLate ? { onDispatchSettledLate: deps.onDispatchSettledLate } : {}), - ...(deps.onSessionIdle ? { onSessionIdle: deps.onSessionIdle } : {}), ...(deps.onChildWorkEvidence ? { onChildWorkEvidence: deps.onChildWorkEvidence } : {}), ...(deps.logger ? { logger: deps.logger } : {}), ...openClaudeConnectionOf(deps),