diff --git a/src/main/acp/acp-structured-acquire.ts b/src/main/acp/acp-structured-acquire.ts index 3db29603d75..8ee3ceaf293 100644 --- a/src/main/acp/acp-structured-acquire.ts +++ b/src/main/acp/acp-structured-acquire.ts @@ -281,6 +281,7 @@ export async function acquireAcpStructuredSession(input: { ...takeover } session = { + events, sessionId, fence: acquire.fence, acquisitionGeneration: generation, @@ -305,16 +306,13 @@ export async function acquireAcpStructuredSession(input: { exitObservedAt: null } const reading = liveLane - const unbind = events?.bindReadingControl?.({ + session.unbindReadingControl = events?.bindReadingControl?.({ pauseReading: () => connection.pauseReading(), resumeReading: () => { connection.resumeReading() reading.retry() } }) - if (unbind) { - session.unbindReadingControl = unbind - } if (connection.exited || connection.closed) { throw new Error(connection.stderrTail() || `${spec.command} exited while starting`) } diff --git a/src/main/acp/acp-structured-host-lifecycle.test.ts b/src/main/acp/acp-structured-host-lifecycle.test.ts index a1486cd3263..106fa29398f 100644 --- a/src/main/acp/acp-structured-host-lifecycle.test.ts +++ b/src/main/acp/acp-structured-host-lifecycle.test.ts @@ -285,3 +285,29 @@ 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 + const write = journal.resolveDispatch.bind(journal) + vi.spyOn(journal, 'resolveDispatch').mockImplementation((input, hook) => + input.state === 'accepted' ? Promise.reject(new Error('disk full')) : write(input, hook) + ) + 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 75b474314ac..0f14e2cdde9 100644 --- a/src/main/acp/acp-structured-host.test-support.ts +++ b/src/main/acp/acp-structured-host.test-support.ts @@ -103,7 +103,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 cfad5c45310..64a6cc71586 100644 --- a/src/main/acp/acp-structured-session-adapter.ts +++ b/src/main/acp/acp-structured-session-adapter.ts @@ -246,6 +246,9 @@ export class AcpStructuredSessionAdapter implements StructuredAgentSessionAdapte readCommands = (sessionId: string) => this.sessions.get(sessionId)?.options.readCommands() + heldSends = (sessionId: string): ReadonlySet | undefined => + this.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 d0ed4bc3e56..4f907bfff77 100644 --- a/src/main/acp/acp-structured-session.ts +++ b/src/main/acp/acp-structured-session.ts @@ -20,8 +20,10 @@ 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' export type AcpStructuredSession = { + events?: StructuredAgentSessionEventSink sessionId: string fence: number acquisitionGeneration: string @@ -100,6 +102,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) @@ -120,6 +123,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 3d076d3d741..dc39195c326 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 88aa6d4a12f..84a968d7260 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 4cfd512fe2a..89bfe89f4e7 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,6 +88,11 @@ 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. */ +export function claudeHeldSends(session: ClaudeSession): ReadonlySet { + return new Set(session.dispatchWaiters.flatMap((waiter) => waiter.clientMessageId ?? [])) +} + /** 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) diff --git a/src/main/claude/claude-structured-session-acquisition.ts b/src/main/claude/claude-structured-session-acquisition.ts index 749bc825849..df56c1cf448 100644 --- a/src/main/claude/claude-structured-session-acquisition.ts +++ b/src/main/claude/claude-structured-session-acquisition.ts @@ -112,10 +112,9 @@ 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 diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index 84b4804cd69..c382366121c 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -8,7 +8,7 @@ import type { } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import { stopCurrentClaudeBackgroundTasks } from './claude-structured-control-actions' import { dispatchClaudeTurn } from './claude-structured-dispatch' -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,6 +217,11 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda const session = this.sessions.get(sessionId) return session ? claudeHoldsDispatch(session) : false } + 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) return ( diff --git a/src/main/claude/claude-structured-session-close.ts b/src/main/claude/claude-structured-session-close.ts index 316d90b0a50..8fc7ec3ad2d 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' @@ -95,6 +96,7 @@ async function finalizeClaudePublishedSession( input: CloseClaudePublishedSessionInput, session: ClaudeSession ): Promise { + session.events?.observeExit?.(claudeHeldSends(session)) retireClaudeDispatchWaiters(session) settleClaudeTurnEndWaiters(session) // Settle every in-flight permission callback so closing leaves no dangling promise; `null` @@ -139,6 +141,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, @@ -147,7 +151,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 3936ed89cb1..ed4f6eb7027 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 44895a4b027..2622ae8ad6d 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?: ( @@ -226,6 +224,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-acquire.ts b/src/main/codex/codex-structured-session-acquire.ts index 7d400dab145..773efbae033 100644 --- a/src/main/codex/codex-structured-session-acquire.ts +++ b/src/main/codex/codex-structured-session-acquire.ts @@ -227,6 +227,7 @@ export async function acquireCodexStructuredSession(input: { // Where this session's child work goes: the host's records, after each frame is journaled. const sink = codexChildWorkSink(sessionId, deps, acquireInput.events) const session: CodexSession = { + events: acquireInput.events, account, connection, ...codexSessionLifecycle(acquireInput.fence, acquired.acquisitionGeneration as string), diff --git a/src/main/codex/codex-structured-session-close.ts b/src/main/codex/codex-structured-session-close.ts index be214a25d59..1dc3c75dfb1 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-state.ts b/src/main/codex/codex-structured-session-state.ts index 7a253003bd9..dbd1084b855 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -29,6 +29,7 @@ import type { AgentModelCatalogStore } from '../native-chat/agent-model-catalog/agent-model-catalog-store' 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 @@ -126,6 +127,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 f33e42ce186..97d48bed6e4 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-orchestration.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts index 4a45cb42979..59ae69b1c5e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts @@ -247,14 +247,15 @@ async function runAttachUnderAbort( ) attempt.candidate = { sink: eventSink, - child: { + child: (owner === 'kept' && current) || { generation: acquisitionGeneration ?? current?.generation ?? null, fence, // An acquire reports its phase; a re-attach keeps what its child already proved. phase: owner === 'acquired' ? providerChildPhase : (current?.phase ?? 'ready'), ...(startedFor === undefined ? {} : { startedFor }), ...(permissionMode === undefined ? {} : { permissionMode }), - heldSends: () => context.deps.adapter.heldSends?.(sessionId) + heldSends: () => context.deps.adapter.heldSends?.(sessionId), + exitObservation: eventSink.exitObservation } } // Only a proven child has the protocol session recovery asks; a starting one recovers in 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 0fa4b7c7709..0763085ead6 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 @@ -38,6 +38,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 @@ -131,7 +132,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 = exit.startupUnproven === true || child.phase === 'starting' @@ -237,7 +239,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-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 14adfb0f2b6..740e27f4bc3 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 }) @@ -522,17 +526,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 a5d755b5c57..b695435fd31 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 @@ -96,22 +96,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 1f48f068058..a533b38f22c 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 00000000000..fcf1aa26e3e --- /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 64cb1fe0616..e16a997e387 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 } @@ -24,15 +28,20 @@ export async function whileStructuredAgentSessionHandingOver( ): Promise { 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 const released = (): 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-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts index 31de6353827..4aae2cf0d08 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 @@ -59,7 +59,6 @@ import { } from './structured-agent-session-options-read' import { writeStructuredAgentSessionOption } from './structured-agent-session-message-options' import { settleStructuredAgentSessionLateDispatch } from './structured-agent-session-late-dispatch' -import { releaseStructuredAgentSessionUnansweredDispatches } from './structured-agent-session-unanswered-dispatch-release' export function sendStructuredAgentSessionTurn( context: StructuredAgentSessionMutationContext, @@ -302,9 +301,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 28c3951e5dc..2a91d0532b1 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 @@ -12,6 +12,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, @@ -79,6 +80,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 9fbe6f601e4..ae00088856f 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 @@ -273,6 +273,7 @@ export class StructuredAgentSessionHost { conversation: this.lifetime.conversation, readChildWork: this.clientDelivery.readChildWork, publishChildWork: this.clientDelivery.publishChildWork, + publishWorkChanged: (sessionId) => this.clientDelivery.publishGenerationEnded(sessionId), captureChildWorkStops: this.clientDelivery.captureChildWorkStops, childWorkOpen: this.clientDelivery.childWorkOpen, childWorkIncomplete: this.clientDelivery.childWorkIncomplete, @@ -336,7 +337,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 3d38494fea9..dfbe671efa4 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 @@ -3,11 +3,14 @@ import type { AgentJournalItemIdentity } from '../../../shared/agent-session-journal-types' import type { AgentJournalDispatchRejection } from '../../../shared/agent-session-failure-words' +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: { @@ -27,8 +30,7 @@ export async function settleStructuredAgentSessionLateDispatch( return } const fence = structuredAgentSessionConversationFence(context.deps.store, input.sessionId) - // The journal queue drains before close; the host queue would defer this past teardown. - await session.journal.resolveDispatch( + const settlement: ResolveDispatchInput = 'providerIdentity' in input ? { clientMessageId: input.clientMessageId, @@ -52,5 +54,13 @@ export async function settleStructuredAgentSessionLateDispatch( ...(input.answeredInTurn ? { answeredInTurn: input.answeredInTurn } : {}), fence } - ) + try { + // The journal queue drains before close; the host queue would defer this past teardown. + await whileStructuredAgentSessionHandingOver(session.journal, input.clientMessageId, () => + 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 44268403248..0e028ae2c2a 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 @@ -34,6 +34,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[] diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts index 3dd914f647f..31bbdb0ed01 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts @@ -743,12 +743,11 @@ describe('an unexpected provider exit', () => { expect(hostErrors).toContainEqual(expect.objectContaining({ message: 'journal failed' })) const history = await host.history({ sessionId: SESSION, direction: 'tail' }) expect(history.ok && history.page.submissions[0]?.dispatchState).toBe('unknown') - // A send whose delivery outcome is unknown IS work in progress, so the reassuring outcome is - // written — its failure fact beside it, and never the old bare `Provider exited: ` row. + // Raw doubt alone is not evidence that the exited child still held the input. const statuses = history.ok ? history.page.items.flatMap((item) => (item.body.kind === 'status' ? [item.body.text] : [])) : [] - expect(statuses).toEqual([UNEXPECTED_PROVIDER_EXIT_OUTCOME]) + expect(statuses).toEqual([]) expect(statuses.some((text) => text.startsWith('Provider exited'))).toBe(false) dispatch.mockResolvedValueOnce({ 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 9a27415dc71..00000000000 --- 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 62078ef853c..00000000000 --- 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 0891fa6f170..3fcf83906f2 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 e0a98e7d771..f6e4705ee68 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,18 @@ 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-configured-default.test.ts b/src/main/pi/rpc-configured-default.test.ts index 467417c8ead..d5567acceb6 100644 --- a/src/main/pi/rpc-configured-default.test.ts +++ b/src/main/pi/rpc-configured-default.test.ts @@ -193,7 +193,6 @@ async function piHost() { }, onLifecycle: vi.fn(), onSettled: vi.fn(), - onIdle: vi.fn(), logger: { warn: vi.fn(), error: vi.fn() } }) adapters.push(adapter) diff --git a/src/main/pi/rpc-location.test.ts b/src/main/pi/rpc-location.test.ts index b3c500ba45f..b298fff9045 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-runtime-registration.ts b/src/main/pi/rpc-runtime-registration.ts index 762da44ac13..b1bf7d5b608 100644 --- a/src/main/pi/rpc-runtime-registration.ts +++ b/src/main/pi/rpc-runtime-registration.ts @@ -33,7 +33,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.ts b/src/main/pi/rpc-session-adapter.test.ts index 8c74e65eea1..b87d2dd7d3b 100644 --- a/src/main/pi/rpc-session-adapter.test.ts +++ b/src/main/pi/rpc-session-adapter.test.ts @@ -114,8 +114,7 @@ async function setup( const rig = await openProviderTimelineRig({ agent: 'pi', sessionId }) const connections: FakeConnection[] = [] const lifecycle = vi.fn(), - settled = vi.fn(), - idle = vi.fn() + settled = vi.fn() const onSpawned = vi.fn(async (_process: AgentSessionProcessIdentity) => { expect(connections.at(-1)?.requests).toEqual([]) }) @@ -152,7 +151,6 @@ async function setup( openConnection, onLifecycle: lifecycle, onSettled: settled, - onIdle: idle, logger: { warn: vi.fn(), error: vi.fn() } }) const acquired = await adapter.acquire(input) @@ -173,7 +171,6 @@ async function setup( onSpawned, lifecycle, settled, - idle, input, resolveLaunch, openConnection, @@ -342,6 +339,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('rejects stale fences and a Stop for a different turn before sending abort', async () => { const h = await setup() await expect( diff --git a/src/main/pi/rpc-session-adapter.ts b/src/main/pi/rpc-session-adapter.ts index 139268fe0a7..360631f8993 100644 --- a/src/main/pi/rpc-session-adapter.ts +++ b/src/main/pi/rpc-session-adapter.ts @@ -146,7 +146,7 @@ export class PiRpcSessionAdapter implements StructuredAgentSessionAdapter { ) } } - return session.turns.submit( + return session.submit( input.clientMessageId, input.requestedAt ?? Date.now(), { ...prompt }, @@ -234,6 +234,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 + } /** Started with no model listed: Pi keeps a placeholder, so every prompt this child takes fails * as not signed in, even after a sign-in, which only a new Pi reads. */ startUnavailable(id: string): AgentSessionUnavailable | undefined { diff --git a/src/main/pi/rpc-session.ts b/src/main/pi/rpc-session.ts index ae693653df4..c6ccafa6f49 100644 --- a/src/main/pi/rpc-session.ts +++ b/src/main/pi/rpc-session.ts @@ -51,7 +51,6 @@ export type PiRpcSessionDeps = { fence: number outcome: AgentSessionDispatchOutcome }) => void - onIdle: (input: { sessionId: string }) => void logger: StructuredAgentSessionLogger } @@ -62,6 +61,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() private startCatalogListing?: AgentModelCatalogLiveListing readonly skipped: string[] = [] commands?: AgentSessionSlashCommand[] @@ -105,10 +106,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', { @@ -147,6 +145,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 start(): Promise { const state = piRpcStateSchema.parse(await this.connection.request('get_state')) this.lane.apply(this.turns.context.setModel(state.model ?? undefined, Date.now())) @@ -256,11 +269,13 @@ export class PiRpcSession { return } this.publishedExit = true + const observedAt = Date.now() + this.input.events?.observeExit?.(this.heldSends, observedAt) this.dialogs.cancelAll() this.turns.end() this.lane.finalize() this.lane.apply([ - { type: 'session.ended', verdict: { state: 'interrupted', completedAt: Date.now() } } + { type: 'session.ended', verdict: { state: 'interrupted', completedAt: observedAt } } ]) this.lane.flush() let timer: ReturnType | undefined @@ -286,13 +301,14 @@ 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: (this.failedCause ?? error).message }) } 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 9fa9cbfcfa6..9f66a050932 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,62 @@ 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' }) - ) - // Written, never echoed: its dispatch has returned, but nothing has settled it. - child.connection.send = async (_message, beforeDispatch) => { - await beforeDispatch?.() - } - const pending = await send(host, 'second') - await vi.waitFor(async () => - expect(await submission(host, pending)).toMatchObject({ dispatchState: 'pending' }) - ) - - 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` }) ) - expect(await submission(host, pending)).toMatchObject({ dispatchState: 'pending' }) - expect((await submission(host, pending))?.recovered).toBeUndefined() + + // 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 later = await send(host, 'second succeeds') + await vi.waitFor(async () => + expect(await submission(host, later)).toMatchObject({ dispatchState: 'accepted' }) + ) + 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( + snapshot.items.some((item) => item.body.kind === 'turn' && item.body.state === 'completed') + ).toBe(true) + expect(await submission(host, doubted)).toMatchObject({ dispatchState: 'unknown' }) }) }) @@ -124,9 +141,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 +161,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 +178,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 +207,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 20a31d05197..d0034331268 100644 --- a/src/main/runtime/structured-agent-runtime-registrations.ts +++ b/src/main/runtime/structured-agent-runtime-registrations.ts @@ -107,7 +107,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 2138b219eb2..c82970f8117 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 dbc5a31b0d1..92906bbd86c 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-agent-session-failure-logging.test.ts b/src/main/runtime/structured-agent-session-failure-logging.test.ts index d155df5d48d..527dda54409 100644 --- a/src/main/runtime/structured-agent-session-failure-logging.test.ts +++ b/src/main/runtime/structured-agent-session-failure-logging.test.ts @@ -42,9 +42,6 @@ function rejectingHost(error: Error) { return { settleLateDispatch: vi.fn(async () => { throw error - }), - releaseUnansweredDispatches: vi.fn(async () => { - throw error }) } } @@ -76,25 +73,6 @@ describe('failures the desktop host used to drop', () => { ) }) - it('logs releasing the sends an idle provider never answered when that fails', async () => { - const log = recordingStructuredAgentSessionLogger() - const error = new Error('release write failed') - const followUps = createStructuredAgentSessionDispatchFollowUps({ - host: () => rejectingHost(error), - logger: log.logger - }) - - followUps.releaseUnansweredDispatches({ sessionId: SESSION }) - - await vi.waitFor(() => - expect(log.entries).toEqual([ - expect.objectContaining({ - fields: { scope: 'unanswered-dispatch', sessionId: SESSION, error } - }) - ]) - ) - }) - it('logs a provider start or exit the host could not take', async () => { const log = recordingStructuredAgentSessionLogger() const error = new Error('host refused the event') diff --git a/src/main/runtime/structured-claude-runtime-adapter.ts b/src/main/runtime/structured-claude-runtime-adapter.ts index 0e008888a23..b8cc1e7cda5 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),