diff --git a/config/scripts/vitest-sqlite-runtime-files.mjs b/config/scripts/vitest-sqlite-runtime-files.mjs index 018f2c472cda..a7aa6e50bb1c 100644 --- a/config/scripts/vitest-sqlite-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-runtime-files.mjs @@ -123,6 +123,7 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/native-chat/agent-session-wire/structured-agent-session-late-proof-tool-ends.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-late-settlement.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-exit-delivery.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-ended-prompt-cancel.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-legacy-handoff-record.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-live-turn-window.test.ts', @@ -158,6 +159,7 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/native-chat/agent-session-wire/structured-agent-session-replaced-server-turn.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-resend-ledger-refusal.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-resolved-append.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-resumed-child-owner.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-restart-offer-newer-orca.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-restart-offer-withdrawal.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-restart-provider-output.test.ts', diff --git a/src/main/acp/acp-structured-acquire.ts b/src/main/acp/acp-structured-acquire.ts index 29f483c5201f..18177694139e 100644 --- a/src/main/acp/acp-structured-acquire.ts +++ b/src/main/acp/acp-structured-acquire.ts @@ -185,7 +185,7 @@ export async function acquireAcpStructuredSession(input: { dialect: spec.dialect, now, childStops: () => childStops, - ...acpLaneChildWorkDelivery(deps, sessionId), + ...acpLaneChildWorkDelivery(deps, sessionId, events), logger: deps.logger ?? createStructuredAgentSessionLogger(), onInputAccepted: (clientMessageId) => session?.turns.accept(clientMessageId), onFailed: () => input.forceClose(sessionId) diff --git a/src/main/acp/acp-structured-host-crash.test.ts b/src/main/acp/acp-structured-host-crash.test.ts index 53465c822765..39032c7a86c6 100644 --- a/src/main/acp/acp-structured-host-crash.test.ts +++ b/src/main/acp/acp-structured-host-crash.test.ts @@ -38,6 +38,21 @@ async function midReply() { batches.push(batch) return append(batch) }) + // A settlement plans its batch at its own turn in the write queue: recorded as it is planned, and + // only when it writes something (an empty plan writes no row). + const appendPlanned = journal.appendPlannedLifecycleBatch + vi.spyOn(journal, 'appendPlannedLifecycleBatch').mockImplementation((input) => + appendPlanned({ + ...input, + plan: () => { + const planned = input.plan() + if (planned.mutations.length > 0 || planned.dispatches.length > 0) { + batches.push({ ...input, mutations: planned.mutations }) + } + return planned + } + }) + ) child.stderr = LAST_WORDS return { ...rig, child, batches } } diff --git a/src/main/acp/acp-structured-lane.ts b/src/main/acp/acp-structured-lane.ts index b21c0405cd5e..97e8dd17435e 100644 --- a/src/main/acp/acp-structured-lane.ts +++ b/src/main/acp/acp-structured-lane.ts @@ -2,6 +2,7 @@ // the shared assembler writes them. Events apply strictly in order; one the sink refuses under // backpressure holds every later one until the sink drains, so nothing is reordered or dropped. +import type { StructuredAgentSessionChildWorkProducer } from '../native-chat/agent-session-wire/structured-agent-session-subagent-rows' import { parseAgentJournalItemKey } from '../../shared/agent-session-journal-item-key' import { createProviderTimelineAssembler, @@ -304,10 +305,11 @@ export class AcpStructuredLane { export function acpLaneChildWorkDelivery( deps: AcpStructuredSessionAdapterDeps, - sessionId: string + sessionId: string, + producer?: StructuredAgentSessionChildWorkProducer ): Pick { return { - onChildWorkEvidence: (evidence) => deps.onChildWorkEvidence?.(sessionId, evidence), + onChildWorkEvidence: (evidence) => deps.onChildWorkEvidence?.(sessionId, evidence, producer), onChildWorkFailure: (error) => deps.logger?.error('publishing ACP child work failed', { scope: 'acp-child-work', diff --git a/src/main/acp/acp-structured-session-adapter-deps.ts b/src/main/acp/acp-structured-session-adapter-deps.ts index ec2802e38f1b..0de78c9bf20f 100644 --- a/src/main/acp/acp-structured-session-adapter-deps.ts +++ b/src/main/acp/acp-structured-session-adapter-deps.ts @@ -1,3 +1,4 @@ +import type { StructuredAgentSessionChildWorkProducer } from '../native-chat/agent-session-wire/structured-agent-session-subagent-rows' import type { StructuredAgentSessionLifecycleEvent } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' import type { @@ -34,7 +35,12 @@ export type AcpStructuredSessionAdapterDeps = { readProcessStartTime?: (pid: number) => Promise /** Every exit, expected or not: the host ends that child's record. */ onEvent?: (event: StructuredAgentSessionLifecycleEvent) => void - onChildWorkEvidence?: (sessionId: string, evidence: AgentChildWorkEvidence[]) => void + /** `producer`: the reporting child's own sink, which writes what the report changes. */ + onChildWorkEvidence?: ( + sessionId: string, + evidence: AgentChildWorkEvidence[], + producer?: StructuredAgentSessionChildWorkProducer + ) => void /** A send this adapter admitted, settled once the agent answered for it. */ onDispatchSettledLate?: ( settlement: { sessionId: string; clientMessageId: string } & ( diff --git a/src/main/claude/claude-background-task-row-journal.ts b/src/main/claude/claude-background-task-row-journal.ts index 1419729c5cf1..85e85d07e1d0 100644 --- a/src/main/claude/claude-background-task-row-journal.ts +++ b/src/main/claude/claude-background-task-row-journal.ts @@ -144,8 +144,10 @@ export function writeClaudeBackgroundTaskRow( } beforeAppend?.() const identity = claudeBackgroundTaskIdentity(id, row.generation) + // A task row a later run reuses (`resolveClaudeBackgroundTaskIdentity`) is that run's work now. const appendOptions = { turnScope: turnScope(), + executes: true as const, ...(lifecycle ? { lifecycle: true } : {}) } const publishOptions = lifecycle ? { lifecycle: true } : {} diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index bbc1d6fea64e..8ac3e3c16a45 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -155,7 +155,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda ): void { const evidence = drainClaudeChildWork(session, message, this.deps.now?.() ?? Date.now()) if (evidence.length > 0) { - this.deps.onChildWorkEvidence?.(sessionId, evidence) + this.deps.onChildWorkEvidence?.(sessionId, evidence, session?.events) } } diff --git a/src/main/claude/claude-structured-session-state.ts b/src/main/claude/claude-structured-session-state.ts index f5d7b67777b2..41e7d76c0a77 100644 --- a/src/main/claude/claude-structured-session-state.ts +++ b/src/main/claude/claude-structured-session-state.ts @@ -1,3 +1,4 @@ +import type { StructuredAgentSessionChildWorkProducer } from '../native-chat/agent-session-wire/structured-agent-session-subagent-rows' import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' import type { AgentJournalDispatchRejection } from '../../shared/agent-session-failure-words' import type { SubmissionRejectionFact } from '../../shared/agent-session-failure' @@ -103,7 +104,12 @@ export type ClaudeStructuredSessionAdapterDeps = { /** 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. */ - onChildWorkEvidence?: (sessionId: string, evidence: AgentChildWorkEvidence[]) => void + /** `producer`: the reporting child's own sink, which writes what the report changes. */ + onChildWorkEvidence?: ( + sessionId: string, + evidence: AgentChildWorkEvidence[], + producer?: StructuredAgentSessionChildWorkProducer + ) => void openConnection?: typeof openClaudeStreamJsonConnection readProcessStartTime?: (pid: number) => Promise mintLinkId?: () => string diff --git a/src/main/claude/claude-subagent-permission-request.test.ts b/src/main/claude/claude-subagent-permission-request.test.ts index 95bfb84cb2b8..43f96eadcaad 100644 --- a/src/main/claude/claude-subagent-permission-request.test.ts +++ b/src/main/claude/claude-subagent-permission-request.test.ts @@ -23,7 +23,7 @@ import type { AgentStatusStructuredSessionSubject } from '../../shared/agent-sta import { AgentHookServer } from '../agent-hooks/server' import type { AgentSessionJournal } from '../native-chat/agent-session-journal/journal-store' import { createTrackedJournalOpener } from '../native-chat/agent-session-journal/journal-host-database-test-support' -import { settleStructuredAgentSessionDeadGeneration } from '../native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement' +import { settleObservedExitForTest } from '../native-chat/agent-session-wire/structured-agent-session-observed-exit.test-fixture' import { StructuredAgentSessionCurrentWork } from '../native-chat/agent-session-wire/structured-agent-session-current-work' import { createDeferredStructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { testEventSinkLogging } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support' @@ -633,7 +633,7 @@ describe("a Claude subagent's permission request", () => { let settled: Promise | undefined hooks.onEvent = (event) => { if (event.type === 'ended') { - settled = settleStructuredAgentSessionDeadGeneration({ + settled = settleObservedExitForTest({ journal: run.journal, sessionId: SESSION, fence: FENCE, diff --git a/src/main/codex/codex-background-task-tracker.ts b/src/main/codex/codex-background-task-tracker.ts index 2ce5cf5a4de6..69e0f11e4d54 100644 --- a/src/main/codex/codex-background-task-tracker.ts +++ b/src/main/codex/codex-background-task-tracker.ts @@ -1,3 +1,4 @@ +import type { StructuredAgentSessionChildWorkProducer } from '../native-chat/agent-session-wire/structured-agent-session-subagent-rows' import type { AgentSessionBackgroundTask, AgentSessionBackgroundTaskState @@ -23,10 +24,11 @@ export type CodexChildWorkSink = { export function codexChildWorkSink( sessionId: string, - deps: Pick + deps: Pick, + producer?: StructuredAgentSessionChildWorkProducer ): CodexChildWorkSink { return { - deliver: (evidence) => deps.onChildWorkEvidence?.(sessionId, evidence), + deliver: (evidence) => deps.onChildWorkEvidence?.(sessionId, evidence, producer), now: () => deps.now?.() ?? Date.now() } } diff --git a/src/main/codex/codex-structured-journal-sink.ts b/src/main/codex/codex-structured-journal-sink.ts index 9087449584f4..2af85c59b3d1 100644 --- a/src/main/codex/codex-structured-journal-sink.ts +++ b/src/main/codex/codex-structured-journal-sink.ts @@ -6,7 +6,7 @@ import type { import type { StructuredAgentSessionItemAppendOptions, StructuredAgentSessionEventSink, - StructuredAgentSessionLifecycleIdentityResolver, + StructuredAgentSessionIdentityResolver, StructuredAgentSessionSinkAdmission } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { @@ -129,7 +129,7 @@ export function appendCodexLifecycleTransition( sink: StructuredAgentSessionEventSink, identitySizeBound: AgentJournalItemIdentity, body: AgentJournalItemBody, - resolveIdentity: StructuredAgentSessionLifecycleIdentityResolver, + resolveIdentity: StructuredAgentSessionIdentityResolver, attribution: AgentJournalRowAttribution ): CodexJournalTranslationAdmission { if (sink.tryAppendLifecycleTransition) { diff --git a/src/main/codex/codex-structured-session-acquire.ts b/src/main/codex/codex-structured-session-acquire.ts index 2cbc854e54d1..92695c45ceea 100644 --- a/src/main/codex/codex-structured-session-acquire.ts +++ b/src/main/codex/codex-structured-session-acquire.ts @@ -224,7 +224,7 @@ export async function acquireCodexStructuredSession(input: { assertCodexConnectionOpen(connection, sessionId) acquisitions.deleteIfCurrent(sessionId, attempt) // Where this session's child work goes: the host's records, after each frame is journaled. - const sink = codexChildWorkSink(sessionId, deps) + const sink = codexChildWorkSink(sessionId, deps, acquireInput.events) const session: CodexSession = { account, connection, diff --git a/src/main/codex/codex-structured-session-state.ts b/src/main/codex/codex-structured-session-state.ts index 0e0560c1ccfc..de3f46d53ba1 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -1,3 +1,4 @@ +import type { StructuredAgentSessionChildWorkProducer } from '../native-chat/agent-session-wire/structured-agent-session-subagent-rows' import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' import type { AgentSessionAccountKind } from '../../shared/agent-session-availability' import type { @@ -95,7 +96,12 @@ export type CodexStructuredSessionAdapterDeps = { /** Where bookkeeping a close or exit does after the child is gone reports a failure. */ logger?: StructuredAgentSessionLogger /** What the session's child work did, delivered after the journal handled the frame. */ - onChildWorkEvidence?: (sessionId: string, evidence: AgentChildWorkEvidence[]) => void + /** `producer`: the reporting child's own sink, which writes what the report changes. */ + onChildWorkEvidence?: ( + sessionId: string, + evidence: AgentChildWorkEvidence[], + producer?: StructuredAgentSessionChildWorkProducer + ) => void /** A send admitted earlier: its identity once Codex echoes it, or its rejection when the turn * Codex answered it into ended without taking it. */ onDispatchSettledLate?: ( diff --git a/src/main/native-chat/agent-session-journal/journal-atomic-terminal-settlement.spec.ts b/src/main/native-chat/agent-session-journal/journal-atomic-terminal-settlement.spec.ts index e54e98275db8..64d5a6e47867 100644 --- a/src/main/native-chat/agent-session-journal/journal-atomic-terminal-settlement.spec.ts +++ b/src/main/native-chat/agent-session-journal/journal-atomic-terminal-settlement.spec.ts @@ -9,10 +9,8 @@ import { import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' import { AgentSessionSubscribers } from '../agent-session-wire/structured-agent-session-subscribers' -import { - settleStaleStructuredAgentSessionState, - settleStructuredAgentSessionDeadGeneration -} from '../agent-session-wire/structured-agent-session-dead-generation-settlement' +import { settleStaleStructuredAgentSessionState } from '../agent-session-wire/structured-agent-session-dead-generation-settlement' +import { settleObservedExitForTest } from '../agent-session-wire/structured-agent-session-observed-exit.test-fixture' import { createTrackedJournalOpener, openTestJournalHostDatabase, @@ -97,7 +95,7 @@ export function registerJournalAtomicTerminalSettlementTests(): void { } function settle() { - return settleStructuredAgentSessionDeadGeneration({ + return settleObservedExitForTest({ journal, sessionId: SESSION, fence: 7, @@ -398,12 +396,12 @@ export function registerJournalAtomicTerminalSettlementTests(): void { WHEN NEW.session_id = '${SESSION}' AND json_extract(NEW.row_json, '$.recovered') = 1 AND json_extract(NEW.row_json, '$.kind') = 'item' BEGIN SELECT RAISE(ABORT, 'item rejected after submission'); END`) - expect(await settleStructuredAgentSessionDeadGeneration(input)).toMatchObject({ ok: false }) + expect(await settleObservedExitForTest(input)).toMatchObject({ ok: false }) expect(journal.snapshot()).toEqual(before) expect(rows()).toEqual(diskBefore) expect(published).toEqual([]) database.db.exec('DROP TRIGGER reject_submission_settlement_item') - expect(await settleStructuredAgentSessionDeadGeneration(input)).toEqual({ ok: true }) + expect(await settleObservedExitForTest(input)).toMatchObject({ ok: true }) expect(journal.submission('unanswered')).toMatchObject({ dispatchState: 'rejected', recovered: true, @@ -412,7 +410,7 @@ export function registerJournalAtomicTerminalSettlementTests(): void { const committed = rows().slice(diskBefore.length) expect(committed.filter((row) => row.kind === 'dispatch')).toHaveLength(1) expect(published).toEqual([journal.snapshot()]) - expect(await settleStructuredAgentSessionDeadGeneration(input)).toEqual({ ok: true }) + expect(await settleObservedExitForTest(input)).toMatchObject({ ok: true }) expect(rows().slice(diskBefore.length)).toEqual(committed) expect(published).toHaveLength(1) } diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts index 17b9f5e3b956..a77f72f1e9d8 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts @@ -41,6 +41,8 @@ export type JournalReplacementItem = AgentJournalProducerLinkage & { /** Carried from the epoch it came from (`JournalItemRow.ownerFence`), never the replacing * writer's fence: replacing history does not make it the current generation's work. */ ownerFence: number + /** Carried the same way (`JournalItemRow.executionOwner`). */ + executionOwner?: number } export function replaceJournalEpoch(input: { @@ -75,6 +77,7 @@ export function replaceJournalEpoch(input: { seq: state.lastSequence + 1, fence: input.fence, ownerFence: item.ownerFence, + executionOwner: item.executionOwner, ts: item.observedAt ?? input.now(), linkage: item, turnScope: item.turnScope ?? state.derivedTurnScope.scopeFor(item.body) diff --git a/src/main/native-chat/agent-session-journal/journal-item-fold.ts b/src/main/native-chat/agent-session-journal/journal-item-fold.ts index c45d8975b25d..c23d0547a1d0 100644 --- a/src/main/native-chat/agent-session-journal/journal-item-fold.ts +++ b/src/main/native-chat/agent-session-journal/journal-item-fold.ts @@ -29,8 +29,8 @@ export function upsertJournalItem( revision: number, next: AgentJournalRenderItem, /** The write's fence, and the provenance it states (`JournalItemRow.ownerFence`), which only - * creation reads: a revision keeps the item's. */ - fences: { fence: number; ownerFence?: number } + * creation reads: a revision keeps the item's. A stated `executionOwner` moves on any write. */ + fences: { fence: number; ownerFence?: number; executionOwner?: number } ): void { const tombstoned = state.tombstones.get(itemId) if (tombstoned !== undefined && revision <= tombstoned) { @@ -41,6 +41,9 @@ export function upsertJournalItem( return } state.derivedTurnScope.observe(itemId, isRootAgentJournalItem(next), existing?.body, next.body) + if (fences.executionOwner !== undefined) { + state.itemExecutionOwners.set(itemId, fences.executionOwner) + } if (!existing) { state.items.set(itemId, next) state.itemFences.set(itemId, fences.ownerFence ?? fences.fence) @@ -100,4 +103,5 @@ export function removeJournalItem( state.tombstones.set(itemId, revision) state.items.delete(itemId) state.itemFences.delete(itemId) + state.itemExecutionOwners.delete(itemId) } diff --git a/src/main/native-chat/agent-session-journal/journal-item-provenance.ts b/src/main/native-chat/agent-session-journal/journal-item-provenance.ts index afb1df346d1e..6e5a56e75ee2 100644 --- a/src/main/native-chat/agent-session-journal/journal-item-provenance.ts +++ b/src/main/native-chat/agent-session-journal/journal-item-provenance.ts @@ -1,9 +1,29 @@ // Execution provenance: the generation whose execution produced an item (`JournalItemRow.ownerFence`), // kept apart from each row's writer fence. The write that creates an item states it; every revision -// keeps it. +// keeps it. A reusable child's row (a subagent, a Claude background task) a later run resumes also +// names that run (`JournalItemRow.executionOwner`), which is whose work the item is now. import type { JournalReducerState } from './journal-reducer' +/** Whose execution the item is now: the run that last stated it executes it, else its creator. */ +export function journalItemExecutionFence( + state: Pick, + itemId: string +): number | undefined { + return state.itemExecutionOwners.get(itemId) ?? state.itemFences.get(itemId) +} + +/** The item's creator and, when one said, the run executing it, each as stored. */ +export function journalItemProvenance( + state: Pick, + itemId: string +): { ownerFence?: number; executionOwner?: number } { + return { + ownerFence: state.itemFences.get(itemId), + executionOwner: state.itemExecutionOwners.get(itemId) + } +} + /** An existing item keeps its owner whoever revises it: a newer run's start revises the newest saved * turn. Creation states the writer's claim (an observation, carried history), else its fence. */ export function journalItemOwnerFence( diff --git a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-appender.ts b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-appender.ts index f906fadfbed4..dd1c548a4c0e 100644 --- a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-appender.ts +++ b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-appender.ts @@ -16,6 +16,7 @@ import { } from './journal-row-builders' import type { JournalLifecycleBatchInput, + JournalPlannedLifecycleBatchInput, JournalResolvedLifecycleBatchInput } from './journal-store-contracts' import type { JournalRow } from './journal-row-schema' @@ -58,10 +59,28 @@ export class JournalLifecycleBatchAppender { ...this.planMutations(input, input.mutations) ] }) - .then((rows) => { - const last = rows.at(-1) - return last ? { epoch: last.epoch, sequence: last.seq } : this.deps.cursor() + .then((rows) => this.cursorAfter(rows)) + } + + /** `append`, with every row chosen at the batch's own turn in the queue. */ + appendPlanned(input: JournalPlannedLifecycleBatchInput): Promise { + return this.deps + .enqueueRows(() => { + if (this.wasApplied(input.settlementId)) { + return [] + } + const { mutations, dispatches } = input.plan() + return [ + ...dispatches.map((dispatch) => journalDispatchRowBuilder(this.deps.state, dispatch)), + ...this.planMutations(input, mutations) + ] }) + .then((rows) => this.cursorAfter(rows)) + } + + private cursorAfter(rows: readonly JournalRow[]): AgentJournalCursor { + const last = rows.at(-1) + return last ? { epoch: last.epoch, sequence: last.seq } : this.deps.cursor() } /** The rows a resolved settlement writes, planned at its own turn in the queue: its mutations, diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index ecb9858b0bc9..cbc87dc9e433 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -32,6 +32,7 @@ import { acceptSubmissionFromProviderItem, applyJournalSubmission } from './jour import { applyJournalDispatchRow } from './journal-dispatch-reducer' import { isWriteFailureSubmission } from '../../../shared/structured-agent-session-dispatch-rejection' import { projectJournalStopNote } from './journal-stop-note-projection' +import { journalItemExecutionFence } from './journal-item-provenance' import { retireRewoundJournalSubmission } from './journal-rewind-submission-retirement' import { latestAgentSessionContextClearSequence } from '../../../shared/agent-session-context-clear' import { @@ -54,6 +55,8 @@ export type JournalReducerState = { /** Each item's execution provenance (`JournalItemRow.ownerFence`): the generation whose work it * is, which decides whether it is current. */ itemFences: Map + /** The run executing a reusable child's item now (`JournalItemRow.executionOwner`), when one said. */ + itemExecutionOwners: Map /** Latest saved output per owner, rebuilt during the existing row fold. */ providerActivityAt: Map /** Revision of a removed item, so a late lower revision cannot resurrect it. */ @@ -86,6 +89,7 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou highestFence: 0, items: new Map(), itemFences: new Map(), + itemExecutionOwners: new Map(), providerActivityAt: new Map(), tombstones: new Map(), removedTurnIds: new Set(), @@ -122,9 +126,9 @@ export function applyJournalRow( itemId, row.revision, journalRenderItem(itemId, row.revision, row.body, row, statedOrDerivedTurnScope(state, row)), - { fence: row.fence, ownerFence: row.ownerFence } + { fence: row.fence, ownerFence: row.ownerFence, executionOwner: row.executionOwner } ) - const fence = state.itemFences.get(itemId) ?? row.fence + const fence = journalItemExecutionFence(state, itemId) ?? row.fence observeJournalProviderActivity(state, row, row.itemId, row.body, savedAt, fence) return } @@ -158,7 +162,7 @@ export function applyJournalRow( fence: row.fence, ownerFence: mutation.ownerFence }) - const fence = state.itemFences.get(itemId) ?? row.fence + const fence = journalItemExecutionFence(state, itemId) ?? row.fence observeJournalProviderActivity(state, row, mutation.itemId, body, savedAt, fence) } else { removeJournalItem(state, resolveItemId(state, mutation.itemId), mutation.revision) diff --git a/src/main/native-chat/agent-session-journal/journal-row-builders.ts b/src/main/native-chat/agent-session-journal/journal-row-builders.ts index 4b5ce901bfab..88574cd829d3 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-builders.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-builders.ts @@ -55,6 +55,7 @@ export function journalItemRowBuilder( options: AgentJournalRowAttribution & { fence: number ownerFence?: number + executionOwner?: number observedAt?: number recovered?: true }, @@ -68,6 +69,7 @@ export function journalItemRowBuilder( seq, fence: options.fence, ownerFence: options.ownerFence, + executionOwner: options.executionOwner, ts: options.observedAt ?? ts, recovered: options.recovered, linkage: options, @@ -240,6 +242,7 @@ export function buildJournalItemRow( fence: number /** The writer's provenance claim (`journalItemOwnerFence`); absent for host bookkeeping. */ ownerFence?: number + executionOwner?: number ts: number recovered?: true linkage?: AgentJournalProducerLinkage @@ -260,6 +263,7 @@ export function buildJournalItemRow( body, ...journalRowBase(input.state.epoch, input.seq, input.fence, input.ts, [body]), ownerFence: journalItemOwnerFence(input.state, resolved, input), + ...(input.executionOwner === undefined ? {} : { executionOwner: input.executionOwner }), ...(input.recovered ? { recovered: input.recovered } : {}), turnScope: input.turnScope, ...agentJournalLinkageFields(input.linkage) diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index 91b3eb43a8e4..88cde342d37e 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -83,7 +83,13 @@ export type AgentJournalEpochReason = (typeof AGENT_JOURNAL_EPOCH_REASONS)[numbe /** The generation whose execution produced an item: execution provenance, distinct from the * row's `fence`, which is writer authority. Rides the row like linkage, with no `v` bump: an older * host ignores it. Absent on rows from before it: the item's creating fence stands in. */ -type JournalItemProvenance = { ownerFence?: number } +type JournalItemProvenance = { + ownerFence?: number + /** The fence of the child process whose write this is, on a reusable child's row (a subagent, a + * Claude background task) that a later run can resume. Moves with each such write; absent, the + * item's `ownerFence` stands in. */ + executionOwner?: number +} export type JournalItemRow = JournalRowBase & JournalItemProvenance & { diff --git a/src/main/native-chat/agent-session-journal/journal-row-unusable-annotations.ts b/src/main/native-chat/agent-session-journal/journal-row-unusable-annotations.ts index e6d25b21df9e..53de64ace507 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-unusable-annotations.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-unusable-annotations.ts @@ -60,9 +60,11 @@ function dropUnusableTurnScope(record: Record): void { /** Provenance this build cannot read, removed like linkage: the item's creating fence stands in. */ function dropUnusableOwnerFence(record: Record): void { - const value = record.ownerFence - if (value !== undefined && !(Number.isSafeInteger(value) && Number(value) >= 0)) { - delete record.ownerFence + for (const key of ['ownerFence', 'executionOwner'] as const) { + const value = record[key] + if (value !== undefined && !(Number.isSafeInteger(value) && Number(value) >= 0)) { + delete record[key] + } } } diff --git a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts index c0d637fdef70..c9f2feae47b1 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts @@ -67,6 +67,9 @@ export type JournalItemAppendOptions = AgentJournalRowAttribution & { /** The generation whose execution produced the item: set by a provider's observation, and by * nothing else. Host bookkeeping leaves it out and keeps the item's (`journalItemOwnerFence`). */ ownerFence?: number + /** A reusable child's write: the fence of the child process that produced it + * (`JournalItemRow.executionOwner`). */ + executionOwner?: number observedAt?: number recovered?: true } @@ -96,6 +99,18 @@ export type JournalLifecycleBatchInput = { rejectsQueued?: AgentJournalDispatchRejection } +/** A lifecycle batch whose rows are chosen at its own turn in the write queue, so a write queued + * ahead of it (an answer, a Stop) is what it plans from. */ +export type JournalPlannedLifecycleBatchInput = Pick< + JournalLifecycleBatchInput, + 'settlementId' | 'fence' | 'recovered' +> & { + plan: () => { + mutations: readonly JournalLifecycleMutationInput[] + dispatches: readonly ResolveDispatchInput[] + } +} + export type JournalResolvedLifecycleBatchInput = Omit< JournalLifecycleBatchInput, 'mutations' | 'rejectsQueued' | 'dispatches' diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index bbe6fa475303..4dca9355e0ef 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -45,6 +45,7 @@ import type { JournalItemAppendOptions, JournalItemLinkageVisitor, JournalLifecycleBatchInput, + JournalPlannedLifecycleBatchInput, JournalReadSince, JournalSubmissionConsume, JournalSubmissionInput, @@ -66,6 +67,7 @@ import type { import type { JournalSubmissionWriter } from './journal-submission-writer' import type { JournalEpochController } from './journal-epoch-controller' import { JournalWriteQueue } from './journal-write-queue' +import { journalItemExecutionFence, journalItemProvenance } from './journal-item-provenance' import { createJournalStoreCollaborators } from './journal-store-collaborators' import type { JournalItemAppender } from './journal-item-appender' import type { JournalLifecycleBatchAppender } from './journal-lifecycle-batch-appender' @@ -249,8 +251,12 @@ export class AgentSessionJournal { /** Includes revisions and completion tombstones, whose timestamps disappear from render items. */ lastActivityAt = (): number => this.state.lastActivityAt - /** The item's execution provenance (`JournalItemRow.ownerFence`), while it is in the timeline. */ - itemFence = (itemId: string): number | undefined => this.state.itemFences.get(itemId) + /** Whose execution the item is now, while it is in the timeline: the run a reusable child's row + * last named (`JournalItemRow.executionOwner`), else its creator (`JournalItemRow.ownerFence`). */ + itemFence = (itemId: string): number | undefined => journalItemExecutionFence(this.state, itemId) + + /** Both provenance facts as stored, for a rebuild that writes them again. */ + itemProvenance = (itemId: string) => journalItemProvenance(this.state, itemId) /** Latest saved output at this fence, excluding client actions and recovery bookkeeping. */ lastProviderActivityAt = (fence: number) => this.state.providerActivityAt.get(fence) @@ -336,6 +342,10 @@ export class AgentSessionJournal { return this.lifecycleBatchAppender.append(input) } + /** A lifecycle batch planned at its own turn in the queue (`JournalPlannedLifecycleBatchInput`). */ + appendPlannedLifecycleBatch = (input: JournalPlannedLifecycleBatchInput) => + this.lifecycleBatchAppender.appendPlanned(input) + /** Several writes as one turn in the queue; see `JournalStepWriter`. */ appendSteps: JournalStepWriter['append'] = (steps) => this.stepWriter.append(steps) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-agent-start.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-agent-start.ts index 2c180c121b33..f61746cb05ee 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-agent-start.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-agent-start.ts @@ -146,7 +146,8 @@ async function startStructuredAgentSessionAgent( } const params = structuredAgentSessionResumeParams( record, - structuredAgentSessionResumeOperationId(context.now()) + structuredAgentSessionResumeOperationId(context.now()), + context.sessions.get(sessionId)?.lastEndedChild ) if (!params) { return record.lease.unreconciled 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 e9fcdf063094..c6bcc3f887b9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts @@ -9,7 +9,6 @@ import { } from '../../../shared/agent-session-failure' import { PROVIDER_EXIT_ROW_PREFIX } from '../../../shared/agent-session-stop-row-identity' import { structuredAgentSessionFailureWordsContext } from './structured-agent-session-send-preparation' -import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import type { StructuredAgentSessionEndedEvent } from './structured-agent-session-adapter' import type { StructuredAgentSessionHostSession, @@ -21,7 +20,12 @@ import { type StructuredAgentSessionLeaseStore } from './structured-agent-session-lease-release' import type { StructuredAgentSessionSinkBarrier } from './structured-agent-session-event-sink' -import { settleStructuredAgentSessionDeadGeneration } from './structured-agent-session-dead-generation-settlement' +import { + settleStructuredAgentSessionLeftovers, + type StructuredAgentSessionExitSettlement, + type StructuredAgentSessionLeftoverStore +} from './structured-agent-session-leftover-settlement' +import type { StaleStructuredAgentSessionStateJournal } from './structured-agent-session-stale-state-settlement' import { captureUnfinishedStructuredAgentSessionWork, type DeadGenerationJournal, @@ -47,7 +51,7 @@ export type StructuredAgentSessionChildExit = { export type StructuredAgentSessionChildExitSession = Pick< StructuredAgentSessionHostSession, 'child' | 'lastEndedChild' -> & { journal: DeadGenerationJournal & Pick } +> & { journal: DeadGenerationJournal & StaleStructuredAgentSessionStateJournal } /** The one edge every way a generation ends goes through * (`StructuredAgentSessionClientDelivery.publishGenerationEnded`). */ @@ -59,7 +63,7 @@ export type StructuredAgentSessionGenerationEnded = ( export type StructuredAgentSessionChildExitContext< TSession extends StructuredAgentSessionChildExitSession = StructuredAgentSessionHostSession > = { - store: StructuredAgentSessionLeaseStore + store: StructuredAgentSessionLeaseStore & StructuredAgentSessionLeftoverStore sessions: Map flushLifecycle: (sessionId: string) => Promise publishStatus?: (sessionId: string) => void @@ -184,6 +188,7 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< context.wakeDelivery?.(sessionId) return } + let settlement: StructuredAgentSessionExitSettlement | null = null try { // The exited child's own writes land first: its dead generation is settled from all of them. try { @@ -207,13 +212,11 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< // Folded before the fallback's end is built, so the end reads it (`turnEndAfterStop`). await close?.recorded const generation = child.generation ?? 'unknown' - // The exit proves this child gone, as the record's death evidence later says: what it left - // `unverifiable` is revised now, not at the next open. + // The exit proves this child gone, as the record's death evidence says once released: what it + // left `unverifiable` is revised now. const watched = { ownerFence: child.fence, observedAt } - const settled = await settleStructuredAgentSessionDeadGeneration({ - journal: session.journal, - sessionId, - fence: child.fence, + settlement = { + ownerFence: child.fence, settlementId: `${expected ? 'expected-close:' : PROVIDER_EXIT_ROW_PREFIX}${sessionId}:${child.fence}:${generation}`, pendingSubmissionReason: expected ? 'provider_closed_before_acknowledgement' @@ -238,15 +241,25 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< : {}), exit: watched, ...(unrunRejection ? { unrunRejection } : {}) - }) - if (!settled.ok) { - logExitFailure(context, sessionId, 'exit-settlement', settled.error) } } finally { - // The root's exit was observed, so the owner is released even when terminal settlement could - // not be durably accepted. Bare cause: whatever this settlement could not write is settled from - // it later (the settle recording it queues, or the next open or acquire). + // What the exited child left is settled first, at its own fence, as its own closing tail lands. + // The root's exit was observed, so the owner is released even when that settlement could not + // be durably accepted: whatever it could not write is settled at the next open or acquire. let released = false + if (settlement) { + const settled = await settleStructuredAgentSessionLeftovers({ + store: context.store, + sessionId, + journal: session.journal, + exit: settlement, + // Ended whether or not the release lands: the host saw the root go. + ended: { fence: child.fence, rootGone: true } + }) + if (!settled.ok) { + logExitFailure(context, sessionId, 'exit-settlement', settled.error) + } + } try { await releaseStoredStructuredAgentSessionOwnerAfterExit({ store: context.store, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts index eadb81b98121..e2dd8b3d85d2 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts @@ -63,8 +63,9 @@ beforeEach(async () => { claudeConfigDir: join(root, 'claude-home'), providerSessionId: PROVIDER_SESSION_ID, resumeLeafUuid: null, - resumesTranscript: false, - continuesChain: false + // A later child resumes the transcript the first one started. + resumesTranscript: (store.getRecord(SESSION)?.providerHandleChain.length ?? 0) > 0, + continuesChain: (store.getRecord(SESSION)?.providerHandleChain.length ?? 0) > 0 }), onEvent: (event) => { const mapped = structuredClaudeLifecycleEvent(event) @@ -75,8 +76,8 @@ beforeEach(async () => { onDispatchSettledLate: (settlement) => void host.settleLateDispatch(settlement), persistHandle: async () => undefined, logger: log.logger, - onChildWorkEvidence: (sessionId, evidence) => - host.publishChildWorkEvidence(sessionId, evidence), + onChildWorkEvidence: (sessionId, evidence, producer) => + host.publishChildWorkEvidence(sessionId, evidence, producer), openConnection: claude.openConnection, readProcessStartTime: async () => 1_700_000_000_000, now: () => clock @@ -298,3 +299,39 @@ it('settles a helper still running when its session closes as unverifiable, clai ) expect((await savedRows())[0]).not.toHaveProperty('outcomeBasis') }) + +it('keeps a helper the next child resumes while it is still starting working, past that start’s settlement', async () => { + const first = claude.connections[0]! + await sendAndEcho(first, 'Run it in the background.') + for (const [index, captured] of RESUMED_BY_MESSAGE.filter(({ at }) => at <= 4_765).entries()) { + frame(first, { uuid: `launch-${index}`, ...captured.frame }) + } + await eventually(async () => + expect(await savedRows()).toEqual([expect.objectContaining({ state: 'working' })]) + ) + await adapter.closeSession(SESSION) + await eventually(async () => + expect(await savedRows()).toEqual([expect.objectContaining({ state: 'unverifiable' })]) + ) + // The next child reports the helper running again before its start returns, so before the host + // installs it and settles what the first child left. + const opened = claude.connections.push.bind(claude.connections) + vi.spyOn(claude.connections, 'push').mockImplementation((connection) => { + const initialize = connection.initializationResult + connection.initializationResult = async () => { + const initialized = await initialize() + for (const [index, captured] of RESUMED_BY_MESSAGE.filter( + ({ at }) => at >= 16_142 && at <= 16_143 + ).entries()) { + frame(connection, { uuid: `resume-${index}`, ...captured.frame }) + } + return initialized + } + return opened(connection) + }) + const body = hostTestMessage('Carry on.') + await host.send(CALLER, { envelope: envelope('agentSession.send', { body }), body }) + await new Promise((resolve) => setTimeout(resolve, 50)) + expect(claude.connections).toHaveLength(2) + expect(await savedRows()).toEqual([expect.objectContaining({ id: 'agent-1', state: 'working' })]) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts index 3721a7e1d270..2d6117537b98 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts @@ -19,6 +19,7 @@ import { heldStructuredAgentSessionCurrentWork } from './structured-agent-sessio import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' import type { StructuredAgentSessionStatusSubscriber } from './structured-agent-session-status-feed' import { createStructuredAgentSessionHostStatusFeed } from './structured-agent-session-host-status-feed' +import type { StructuredAgentSessionChildWorkProducer } from './structured-agent-session-subagent-rows' import { StructuredAgentSessionTurnCompletionFeed, type StructuredAgentSessionTurnCompletionSubscriber @@ -102,8 +103,11 @@ export class StructuredAgentSessionClientDelivery { publishConversationName = (sessionId: string): void => this.statusFeed.publishConversationName(sessionId) - publishChildWork = (sessionId: string, evidence: AgentChildWorkEvidence[]): void => - this.statusFeed.publishChildWork(sessionId, evidence) + publishChildWork = ( + sessionId: string, + evidence: AgentChildWorkEvidence[], + producer?: StructuredAgentSessionChildWorkProducer + ): void => this.statusFeed.publishChildWork(sessionId, evidence, producer) /** What the feed publishes as `stopping`: only a working session is still being stopped. */ readStopping = (sessionId: string): boolean => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-crash-turn-end.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-crash-turn-end.test.ts index cd9ecc7e7418..0bdc715f292f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-crash-turn-end.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-crash-turn-end.test.ts @@ -32,7 +32,7 @@ import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' import { openAgentSessionJournal } from '../agent-session-journal/journal-store-factory' -import { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { JournalLifecycleBatchAppender } from '../agent-session-journal/journal-lifecycle-batch-appender' import { STALE_SESSION_ROW_PREFIX } from '../../../shared/agent-session-stop-row-identity' import { AgentSessionAcquisitionExitUnprovenError, @@ -514,10 +514,10 @@ describe('a turn a read reached before the reconcile proved its owner dead', () it('stays running when the open cannot write its settle, and a later open settles it', async () => { let now = RELAUNCHED_AT const log = recordingStructuredAgentSessionLogger() - const append = AgentSessionJournal.prototype.appendLifecycleBatch + const append = JournalLifecycleBatchAppender.prototype.appendPlanned let refused = false - vi.spyOn(AgentSessionJournal.prototype, 'appendLifecycleBatch').mockImplementation(function ( - this: AgentSessionJournal, + vi.spyOn(JournalLifecycleBatchAppender.prototype, 'appendPlanned').mockImplementation(function ( + this: JournalLifecycleBatchAppender, input ) { if (!refused && input.settlementId.startsWith(STALE_SESSION_ROW_PREFIX)) { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts index 07510c754e1c..21d59549faa0 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts @@ -4,14 +4,13 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { openAgentSessionJournal } from '../agent-session-journal/journal-store-factory' -import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' import { agentSessionFailureFact, MAX_PROVIDER_DIAGNOSTIC_CHARS, providerDiagnostic } from '../../../shared/agent-session-failure' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' -import { settleStructuredAgentSessionDeadGeneration } from './structured-agent-session-dead-generation-settlement' +import { settleObservedExitForTest } from './structured-agent-session-observed-exit.test-fixture' import { captureUnfinishedStructuredAgentSessionWork, unfinishedStructuredAgentSessionWorkWasInterrupted @@ -93,7 +92,7 @@ describe('dead structured-session generation settlement', () => { await seedUnfinishedWork() await expect( - settleStructuredAgentSessionDeadGeneration({ + settleObservedExitForTest({ journal, sessionId: SESSION, fence: 8, @@ -102,7 +101,7 @@ describe('dead structured-session generation settlement', () => { verdict: { state: 'unverifiable' }, showUnexpectedExitOutcome: false }) - ).resolves.toEqual({ ok: true }) + ).resolves.toMatchObject({ ok: true }) const snapshot = journal.snapshot() expect(snapshot.submissions).toEqual([ @@ -141,9 +140,9 @@ describe('dead structured-session generation settlement', () => { showUnexpectedExitOutcome: true } - await expect(settleStructuredAgentSessionDeadGeneration(input)).resolves.toEqual({ ok: true }) + await expect(settleObservedExitForTest(input)).resolves.toMatchObject({ ok: true }) const settledCursor = journal.cursor() - await expect(settleStructuredAgentSessionDeadGeneration(input)).resolves.toEqual({ ok: true }) + await expect(settleObservedExitForTest(input)).resolves.toMatchObject({ ok: true }) expect(journal.cursor()).toEqual(settledCursor) expect( @@ -158,7 +157,7 @@ describe('dead structured-session generation settlement', () => { it('cuts short a call the proven death interrupted', async () => { await seedUnfinishedWork() - await settleStructuredAgentSessionDeadGeneration({ + await settleObservedExitForTest({ journal, sessionId: SESSION, fence: 7, @@ -193,7 +192,7 @@ describe('dead structured-session generation settlement', () => { verdict: { state: 'interrupted' as const, completedAt: 1_000 }, showUnexpectedExitOutcome: true } - expect(await settleStructuredAgentSessionDeadGeneration(input)).toEqual({ ok: true }) + expect(await settleObservedExitForTest(input)).toMatchObject({ ok: true }) const settled = journal.snapshot() expect(settled.items).toContainEqual(terminal) expect(settled.items.map((item) => item.body)).toContainEqual({ @@ -203,7 +202,7 @@ describe('dead structured-session generation settlement', () => { state: 'failed', endedAs: 'interrupted' }) - expect(await settleStructuredAgentSessionDeadGeneration(input)).toEqual({ ok: true }) + expect(await settleObservedExitForTest(input)).toMatchObject({ ok: true }) expect(journal.snapshot()).toEqual(settled) }) @@ -211,7 +210,7 @@ describe('dead structured-session generation settlement', () => { await seedUnfinishedWork() await expect( - settleStructuredAgentSessionDeadGeneration({ + settleObservedExitForTest({ journal, sessionId: SESSION, fence: 7, @@ -223,7 +222,7 @@ describe('dead structured-session generation settlement', () => { detail: providerDiagnostic('stack frame '.repeat(4_000), 'log') }) }) - ).resolves.toEqual({ ok: true }) + ).resolves.toMatchObject({ ok: true }) const statuses = journal .snapshot() @@ -238,7 +237,7 @@ describe('dead structured-session generation settlement', () => { it("words Orca's own fault as Orca's, never as the provider stopping", async () => { await seedUnfinishedWork() - await settleStructuredAgentSessionDeadGeneration({ + await settleObservedExitForTest({ journal, sessionId: SESSION, fence: 7, @@ -263,51 +262,6 @@ describe('dead structured-session generation settlement', () => { ]) }) - it('retries an already settled expected close without writing through a closed journal gate', async () => { - const settledItem: AgentJournalRenderItem = { - itemId: 'codex:thread-1:turn-1:0', - revision: 2, - sequence: 2, - observedAt: 1_000, - body: { - kind: 'turn', - turnId: 'turn-1', - state: 'interrupted', - completedAt: 1_000 - } - } - const settledSnapshot = journal.snapshot() - const closedJournal: Pick< - AgentSessionJournal, - 'snapshot' | 'submissions' | 'rejectQueuedSubmissions' | 'appendLifecycleBatch' | 'itemFence' - > = { - itemFence: () => undefined, - snapshot: () => ({ - ...settledSnapshot, - items: [settledItem] - }), - submissions: () => [], - rejectQueuedSubmissions: async () => { - throw new Error('journal_closed') - }, - appendLifecycleBatch: async () => { - throw new Error('journal_closed') - } - } - - await expect( - settleStructuredAgentSessionDeadGeneration({ - journal: closedJournal, - sessionId: SESSION, - fence: 7, - settlementId: `expected-close:${SESSION}:7:generation-1`, - pendingSubmissionReason: 'provider_closed_before_acknowledgement', - verdict: { state: 'interrupted', completedAt: 1_000 }, - showUnexpectedExitOutcome: false - }) - ).resolves.toEqual({ ok: true }) - }) - it('settles a live unknown submission even when no unfinished item remains', async () => { await journal.appendSubmission({ clientMessageId: 'client-unknown', @@ -323,7 +277,7 @@ describe('dead structured-session generation settlement', () => { }) await expect( - settleStructuredAgentSessionDeadGeneration({ + settleObservedExitForTest({ journal, sessionId: SESSION, fence: 7, @@ -332,7 +286,7 @@ describe('dead structured-session generation settlement', () => { verdict: { state: 'interrupted', completedAt: 1_000 }, showUnexpectedExitOutcome: false }) - ).resolves.toEqual({ ok: true }) + ).resolves.toMatchObject({ ok: true }) expect(journal.submissions()).toEqual([ expect.objectContaining({ @@ -352,7 +306,7 @@ describe('dead structured-session generation settlement', () => { fence: 7 }) - await settleStructuredAgentSessionDeadGeneration({ + await settleObservedExitForTest({ journal, sessionId: SESSION, fence: 7, @@ -413,7 +367,7 @@ describe('dead structured-session generation settlement', () => { { fence: 7, ...child, turnScope: AGENT_JOURNAL_THREAD_SCOPE } ) - await settleStructuredAgentSessionDeadGeneration({ + await settleObservedExitForTest({ journal, sessionId: SESSION, fence: 8, @@ -497,7 +451,7 @@ describe('whether a dead generation interrupted anything', () => { await seedIdlePendingApproval() await expect( - settleStructuredAgentSessionDeadGeneration({ + settleObservedExitForTest({ journal, sessionId: SESSION, fence: 7, @@ -510,7 +464,7 @@ describe('whether a dead generation interrupted anything', () => { 1_000 ) }) - ).resolves.toEqual({ ok: true }) + ).resolves.toMatchObject({ ok: true }) const snapshot = journal.snapshot() expect(snapshot.items.some((item) => item.body.kind === 'status')).toBe(false) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts index 0a32aa14d6fd..41890413e13c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts @@ -3,7 +3,6 @@ import { MAX_PROVIDER_DIAGNOSTIC_CHARS, type SubmissionRejectionFact } from '../../../shared/agent-session-failure' -import { STALE_SESSION_ROW_PREFIX } from '../../../shared/agent-session-stop-row-identity' import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission' import { AGENT_JOURNAL_THREAD_SCOPE, @@ -20,7 +19,8 @@ import { runningCallEnd, terminalAgentJournalBody } from '../agent-session-journal/journal-terminal-settlement' -import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import type { ResolveDispatchInput } from '../agent-session-journal/journal-store-contracts' +import { lostLiveWorkJournalBody } from '../agent-session-journal/journal-subagent-liveness' import { agentSessionFailureWords, type AgentSessionFailureWordsContext @@ -30,48 +30,29 @@ import { hasStructuredAgentSessionStartFailureRow, structuredAgentSessionStartFailureRow } from './structured-agent-session-start-failure-row' -import type { AgentSessionDeathEvidence } from '../../../shared/agent-session-record' import { - endedByPersonsStop, - provenUnverifiableTurnRevisions, - provenUnverifiedToolCallRevisions, runningTurnLifecycleRevisions, - lastProvenTurnLiveAt, - turnVerdictFromDeathEvidence, - turnVerdictFromReplacedRuntime, watchedExitRevisions, type StructuredAgentSessionTurnVerdict, type StructuredAgentSessionWatchedExit } from './structured-agent-session-stale-turn-verdict' -import { - exitedRootTurnScope, - settledRootTurnScope -} from './structured-agent-session-exit-turn-scope' +import { exitedRootTurnScope } from './structured-agent-session-exit-turn-scope' import { hasUnfinishedStructuredAgentSessionWork, - isInProgressStructuredAgentSessionItem, type DeadGenerationJournal } from './structured-agent-session-unfinished-work' import { codexUnopenedSendResolutions } from './structured-agent-session-unopened-send-withdrawal' -import type { AgentSessionReplacedRuntime } from '../../runtime/agent-session-replaced-runtime' -import { - isUnverifiableTurn, - staleStopRow, - type StaleCut -} from './structured-agent-session-stale-stop-row' + +export { settleStaleStructuredAgentSessionState } from './structured-agent-session-stale-state-settlement' /** Bounds the exit reason the lease keeps as log evidence; a provider diagnostic is held to the * same cap. */ export const MAX_UNEXPECTED_EXIT_REASON_CHARS = MAX_PROVIDER_DIAGNOSTIC_CHARS -/** Whether the settlement was written, and what stopped it when it was not. */ -export type StructuredAgentSessionDeadGenerationSettlement = - | { ok: true } - | { ok: false; error: unknown } - -export async function settleStructuredAgentSessionDeadGeneration(input: { +export type StructuredAgentSessionDeadGenerationInput = { journal: DeadGenerationJournal sessionId: string + /** The fence the rows are written at: the lease's current one once the exit released it. */ fence: number settlementId: string verdict: StructuredAgentSessionTurnVerdict @@ -89,146 +70,83 @@ export async function settleStructuredAgentSessionDeadGeneration(input: { exit?: StructuredAgentSessionWatchedExit /** A person's Stop ended a starting child before what it was handed could run: each is rejected so. */ unrunRejection?: SubmissionRejectionFact -}): Promise { - try { - const hasUnfinishedWork = hasUnfinishedStructuredAgentSessionWork(input.journal, input.exit) - const showUnexpectedExitOutcome = input.showUnexpectedExitOutcome ?? hasUnfinishedWork - if (!showUnexpectedExitOutcome && !hasUnfinishedWork) { - return { ok: true } - } - // A queued message is the delivery loop's to settle: it was never handed to this child. A send - // a child still starting was handed and never echoed did not run, and its root is gone: it is - // rejected, with the child's own diagnostic or as the Stop that ended it. A proven child's - // handed-over sends stay in doubt. - const startupFailure = input.exitedDuringStartup - ? structuredAgentSessionStartFailure({ exit: input.exitFailure }, input.failureTextContext) - : null - const closed = input.unrunRejection - const unrun = - startupFailure ?? (closed && agentSessionFailureWords(closed, { surface: 'rejection' })) - const withdrawn = unrun ? [] : codexUnopenedSendResolutions(input.journal, input.fence) - const withdrawnIds = new Set(withdrawn.map((entry) => entry.clientMessageId)) - const dispatches = [ - ...withdrawn, - ...journalPendingSubmissionResolutions( - (input.journal.submissions?.() ?? input.journal.pendingSubmissions?.() ?? []).filter( - (entry) => !withdrawnIds.has(entry.clientMessageId) - ), - input.fence, - unrun ? { rejection: unrun } : { reason: input.pendingSubmissionReason } - ) - ] - const items = input.journal.snapshot().items - const proven = watchedExitRevisions(items, input.exit, input.journal) - const mutations: JournalLifecycleMutationInput[] = [] - if (showUnexpectedExitOutcome && input.exitedDuringStartup && startupFailure) { - const startKey = input.exitedDuringStartup.generation ?? input.settlementId - // A start a message waited on is the delivery loop's to record, before or after this exit, - // in the words it rejected the message with; this row is for a command, goal or rewind start. - // A row already written stays: rejected is terminal, so its words are not reworded. - const recordedByDeliveryLoop = - input.journal.submissions?.().some(isQueuedAgentJournalSubmission) || - hasStructuredAgentSessionStartFailureRow(items, startKey) - if (!recordedByDeliveryLoop) { - mutations.push(structuredAgentSessionStartFailureRow(startKey, startupFailure)) - } - } else if (showUnexpectedExitOutcome) { - // The turn the exit ended, and an error so no fold ever hides why it stopped. - mutations.push({ - kind: 'item', - identity: { provider: 'orca', clientMessageId: input.settlementId }, - body: { - kind: 'status', - ...agentSessionFailureWords( - input.exitFailure ?? agentSessionFailureFact('providerExited'), - { - ...input.failureTextContext, - surface: 'row' - } - ), - tone: 'error' - }, - turnScope: exitedRootTurnScope(withRevisions(items, proven), input.verdict) - }) - } - const bodies = new Map(items.map((item) => [item.itemId, item.body])) - for (const item of items) { - // Ended as its turn is: a proven death cuts a running call short. An open reasoning row is - // ended too, but is not unfinished work: its running turn already says so. - const end = runningCallEnd(item.turnScope, (id) => bodies.get(id), input.verdict.state) - const body = endedUnseenMessageBody(item.body) ?? terminalAgentJournalBody(item.body, end) - if (body) { - mutations.push({ - kind: 'item', - itemId: item.itemId, - body, - turnScope: item.turnScope ?? AGENT_JOURNAL_THREAD_SCOPE - }) - } - } - mutations.push(...runningTurnLifecycleRevisions(items, input.verdict), ...proven) - await input.journal.appendLifecycleBatch({ - settlementId: `dead-generation:${input.settlementId}`, - fence: input.fence, - recovered: true, - mutations, - dispatches - }) - return { ok: true } - } catch (error) { - // Returned rather than logged: each caller logs it under its own scope. - return { ok: false, error } - } } -/** - * Settles whatever a generation with no child in this process left running: found when a new child - * is acquired, or when a chat is reopened for reading. Derived from the journal, the lease's death - * evidence and the runtimes this one replaced each time, so nothing is owed in between. Either proof - * ends the turn interrupted, and a proof written after an earlier settle revises what that settle - * left `unverifiable`, the turn and the calls it closed alike. Must run before a new child's - * buffered events land, or a live turn would be judged. - */ -export async function settleStaleStructuredAgentSessionState(input: { - journal: AgentSessionJournal - sessionId: string - fence: number - acquisitionGeneration: string | null - deathEvidence: AgentSessionDeathEvidence | null - /** The runtimes this one replaced for this chat: every owner they held lost its pipes with them. */ - replaced?: AgentSessionReplacedRuntime - /** Who the exit row names. */ - failureTextContext?: AgentSessionFailureWordsContext -}): Promise { - const { journal, replaced } = input - const items = journal.snapshot().items - // Each turn is judged by the evidence only if it names that turn's owner. - const deathVerdictFor = (item: AgentJournalRenderItem) => - turnVerdictFromDeathEvidence( - input.deathEvidence, - journal.itemFence(item.itemId), - lastProvenTurnLiveAt(journal, item) - ) - const replacedVerdictFor = (item: AgentJournalRenderItem) => - turnVerdictFromReplacedRuntime( - replaced, - journal.itemFence(item.itemId), - lastProvenTurnLiveAt(journal, item) +/** The exit's rows, read from the journal as it stands. `owns`: which fences are the exited + * generation's (and later); the rest are an earlier generation's, settled by its own rule. */ +export function planStructuredAgentSessionDeadGeneration( + input: StructuredAgentSessionDeadGenerationInput, + owns: (fence: number | undefined) => boolean = () => true +): { mutations: JournalLifecycleMutationInput[]; dispatches: ResolveDispatchInput[] } { + const hasUnfinishedWork = hasUnfinishedStructuredAgentSessionWork(input.journal, input.exit) + const showUnexpectedExitOutcome = input.showUnexpectedExitOutcome ?? hasUnfinishedWork + if (!showUnexpectedExitOutcome && !hasUnfinishedWork) { + return { mutations: [], dispatches: [] } + } + // A queued message is the delivery loop's to settle: it was never handed to this child. A send + // a child still starting was handed and never echoed did not run, and its root is gone: it is + // rejected, with the child's own diagnostic or as the Stop that ended it. A proven child's + // handed-over sends stay in doubt. + const startupFailure = input.exitedDuringStartup + ? structuredAgentSessionStartFailure({ exit: input.exitFailure }, input.failureTextContext) + : null + const closed = input.unrunRejection + const unrun = + startupFailure ?? (closed && agentSessionFailureWords(closed, { surface: 'rejection' })) + const withdrawn = unrun ? [] : codexUnopenedSendResolutions(input.journal, input.fence) + const withdrawnIds = new Set(withdrawn.map((entry) => entry.clientMessageId)) + const dispatches = [ + ...withdrawn, + ...journalPendingSubmissionResolutions( + (input.journal.submissions?.() ?? input.journal.pendingSubmissions?.() ?? []).filter( + (entry) => !withdrawnIds.has(entry.clientMessageId) && owns(entry.fence) + ), + input.fence, + unrun ? { rejection: unrun } : { reason: input.pendingSubmissionReason } ) - const verdictFor = (item: AgentJournalRenderItem) => { - const byDeath = deathVerdictFor(item) - return byDeath.state === 'interrupted' ? byDeath : replacedVerdictFor(item) + ] + const allItems = input.journal.snapshot().items + const items = allItems.filter((item) => owns(input.journal.itemFence(item.itemId))) + const proven = watchedExitRevisions(items, input.exit, input.journal) + const mutations: JournalLifecycleMutationInput[] = [] + if (showUnexpectedExitOutcome && input.exitedDuringStartup && startupFailure) { + const startKey = input.exitedDuringStartup.generation ?? input.settlementId + // A start a message waited on is the delivery loop's to record, before or after this exit, + // in the words it rejected the message with; this row is for a command, goal or rewind start. + // A row already written stays: rejected is terminal, so its words are not reworded. + const recordedByDeliveryLoop = + input.journal.submissions?.().some(isQueuedAgentJournalSubmission) || + hasStructuredAgentSessionStartFailureRow(allItems, startKey) + if (!recordedByDeliveryLoop) { + mutations.push(structuredAgentSessionStartFailureRow(startKey, startupFailure)) + } + } else if (showUnexpectedExitOutcome) { + // The turn the exit ended, and an error so no fold ever hides why it stopped. + mutations.push({ + kind: 'item', + identity: { provider: 'orca', clientMessageId: input.settlementId }, + body: { + kind: 'status', + ...agentSessionFailureWords( + input.exitFailure ?? agentSessionFailureFact('providerExited'), + { + ...input.failureTextContext, + surface: 'row' + } + ), + tone: 'error' + }, + turnScope: exitedRootTurnScope(withRevisions(items, proven), input.verdict) + }) } - // A successful settlement leaves terminal items; a failed transaction leaves the same prefix. - const generation = input.acquisitionGeneration ?? `seq-${journal.cursor().sequence}` - const settlementId = `${STALE_SESSION_ROW_PREFIX}${input.sessionId}:${input.fence}:${generation}` - // Calls an earlier settle closed with no proof, revised once a proof names their owner. - const mutations = provenUnverifiedToolCallRevisions(items, input.deathEvidence, journal, replaced) + const bodies = new Map(allItems.map((item) => [item.itemId, item.body])) for (const item of items) { - // A turn already settled (a person's Stop) ends its calls as it ended; only a turn still running - // leaves them to the evidence. - const end = runningCallEnd(item.turnScope, (id) => journal.itemBody(id), verdictFor(item).state) - const body = endedUnseenMessageBody(item.body) ?? terminalAgentJournalBody(item.body, end) + // Ended as its turn is: a proven death cuts a running call short. An open reasoning row is + // ended too, but is not unfinished work: its running turn already says so. + const end = runningCallEnd(item.turnScope, (id) => bodies.get(id), input.verdict.state) + const terminal = endedUnseenMessageBody(item.body) ?? terminalAgentJournalBody(item.body, end) + // One revision per item: its live subagents and background tasks end with it. + const body = lostLiveWorkJournalBody(terminal ?? item.body) ?? terminal if (body) { mutations.push({ kind: 'item', @@ -238,42 +156,8 @@ export async function settleStaleStructuredAgentSessionState(input: { }) } } - const proven = provenUnverifiableTurnRevisions(items, input.deathEvidence, journal, replaced) - const turnEnds = [ - ...items.flatMap((item) => runningTurnLifecycleRevisions([item], verdictFor(item))), - ...proven - ] - mutations.push(...turnEnds) - const stopRow = staleStopRow(input, items, (item): StaleCut => { - const unverifiable = isUnverifiableTurn(item) - if (!unverifiable && !isInProgressStructuredAgentSessionItem(item)) { - return null - } - const fence = journal.itemFence(item.itemId) - // An earlier settle's `unverifiable` turn is revised by a death naming its owner or by its replaced runtime. - if ( - unverifiable - ? fence !== undefined && fence === input.deathEvidence?.ownerFence - : deathVerdictFor(item).state === 'interrupted' - ) { - return { by: 'death' } - } - return fence !== undefined && replacedVerdictFor(item).state === 'interrupted' - ? { by: 'replaced', fence } - : null - }) - if (stopRow && !endedByPersonsStop(journal, turnEnds)) { - mutations.unshift({ ...stopRow, turnScope: settledRootTurnScope(items, turnEnds) }) - } - if (mutations.length > 0) { - await journal.appendLifecycleBatch({ - settlementId, - fence: input.fence, - recovered: true, - mutations - }) - } - return mutations.length + mutations.push(...runningTurnLifecycleRevisions(items, input.verdict), ...proven) + return { mutations, dispatches } } function withRevisions( 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 c16548ae2afd..1f48f0680584 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 @@ -51,6 +51,9 @@ export type StructuredAgentSessionPublishOptions = Pick< /** An item write states which turn its row belongs to; the write that creates the row decides. */ export type StructuredAgentSessionItemAppendOptions = StructuredAgentSessionAppendOptions & { turnScope: AgentJournalTurnScope + /** A reusable child's row a later run can resume: this child now executes it + * (`JournalItemRow.executionOwner`). */ + executes?: true } export type StructuredAgentSessionLifecycleJournal = Pick< @@ -81,9 +84,6 @@ export type StructuredAgentSessionRevisionResolver = ( export type StructuredAgentSessionRevisionOptions = StructuredAgentSessionItemAppendOptions -/** Compatibility alias for lifecycle callers that already use this resolver. */ -export type StructuredAgentSessionLifecycleIdentityResolver = StructuredAgentSessionIdentityResolver - export type StructuredAgentSessionEventSink = StructuredAgentSessionTransitionSink & { appendItem( identity: AgentJournalItemIdentity, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-exit-delivery.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-exit-delivery.test.ts new file mode 100644 index 000000000000..b02a6e1808a9 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-exit-delivery.test.ts @@ -0,0 +1,140 @@ +// A queued card when the agent ahead of it exits: every frame from the exit on reads the dead +// generation as ended (not Working, none of its prompts answerable), and the card is sent at once, +// whether the exit's release write lands, waits or keeps failing: an acquisition then replaces the +// owner once it is proven dead, and never one that is alive or unproven. + +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import { + EMPTY_STRUCTURED_AGENT_SESSION, + reduceStructuredAgentSession +} from '../../../shared/structured-agent-session-reducer' +import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types' +import { + exitChild, + leaveUnfinishedWork +} from './structured-agent-session-leftover-settlement.test-fixture' +import { + createQueuedMessageTestRig, + eventually, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { HOST_TEST_SESSION as SESSION } from './structured-agent-session-host-test-data' + +let rig: QueuedMessageTestRig | undefined + +afterEach(async () => { + await rig?.dispose() + rig = undefined + vi.restoreAllMocks() +}) + +/** A card queued behind a running turn that raised a prompt. */ +async function queuedBehindPrompt( + probeOwner?: StructuredAgentSessionHostDeps['probeOwner'] +): Promise<{ current: QueuedMessageTestRig; draftId: string }> { + rig = await createQueuedMessageTestRig({ + restartable: true, + ...(probeOwner ? { probeOwner } : {}) + }) + const current = rig + await current.workingSend() + await leaveUnfinishedWork(current, { prompt: true }) + const queued = await current.send('after this', 'queue-if-active').result + if (!queued.ok || !('queued' in queued.value)) { + throw new Error('expected a queued receipt') + } + return { current, draftId: queued.value.queued.messageId } +} + +/** Every release write refused, as storage that keeps failing would. */ +function refuseReleases(current: QueuedMessageTestRig): () => number { + vi.spyOn(console, 'warn').mockImplementation(() => undefined) + const transition = current.store.transitionHandoff.bind(current.store) + let refused = 0 + vi.spyOn(current.store, 'transitionHandoff').mockImplementation((sessionId, change) => + transition(sessionId, (record) => { + const next = change(record) + if (next.lease.claimStatus === 'released') { + refused += 1 + throw new Error('disk full') + } + return next + }) + ) + return () => refused +} + +/** What a client reads after each frame, from the host's own fields. */ +async function watchChat(current: QueuedMessageTestRig) { + let state = EMPTY_STRUCTURED_AGENT_SESSION + const reads: { working: boolean; settled: boolean; prompts: number; cardSent: boolean }[] = [] + await current.host.subscribe({ + id: 'chat', + sessionId: SESSION, + emit: (event) => { + state = reduceStructuredAgentSession(state, { type: 'event', event }) + reads.push({ + working: state.working === true, + settled: state.latestTurn?.turn.state !== 'running', + prompts: state.actionablePromptIds?.length ?? 0, + cardSent: state.submissions.some((entry) => entry.queuedMessageId !== undefined) + }) + } + }) + return reads +} + +const provenDead: StructuredAgentSessionHostDeps['probeOwner'] = async () => ({ + outcome: 'pid-absent' +}) + +describe('an exit whose settlement lands while its release keeps failing', () => { + it('reads the dead generation as ended in every frame, and the card is sent over the proven-dead owner', async () => { + const { current, draftId } = await queuedBehindPrompt(provenDead) + const refused = refuseReleases(current) + const reads = await watchChat(current) + const before = reads.length + + await exitChild(current) + await eventually(async () => expect(await current.handoff(draftId)).toBeDefined()) + await eventually(() => expect(current.dispatch).toHaveBeenCalledTimes(2)) + + expect(refused()).toBeGreaterThan(0) + const untilSent = reads.slice(before).filter((read) => !read.cardSent) + expect(untilSent.length).toBeGreaterThan(0) + // Not Working, and the dead generation's prompt answerable, in no frame before the card's send. + expect(untilSent.filter((read) => read.working)).toEqual([]) + expect(untilSent.filter((read) => read.prompts > 0)).toEqual([]) + // Its turn was settled before the send, and the lease belongs to the generation that took it. + expect(untilSent.at(-1)?.settled).toBe(true) + expect(current.store.getRecord(SESSION)?.lease.claimStatus).toBe('live') + }) +}) + +describe('an owner this host cannot prove dead', () => { + it.each([ + ['alive', { outcome: 'identity-matched', matchedOn: ['process-start-time'] }], + ['unproven', { outcome: 'indeterminate', reason: 'no start time' }] + ] satisfies [string, AgentSessionOwnerProbe][])( + 'is never replaced when %s: no agent starts over it, and the card comes back', + async (_name, probe) => { + const { current, draftId } = await queuedBehindPrompt(async () => probe) + refuseReleases(current) + const owner = current.store.getRecord(SESSION)?.lease + + await exitChild(current) + await eventually(async () => + expect((await current.handoff(draftId))?.dispatchState).toBe('rejected') + ) + + expect(current.dispatch).toHaveBeenCalledTimes(1) + expect(current.store.getRecord(SESSION)?.lease).toMatchObject({ + claimStatus: 'live', + runtimeFence: owner?.runtimeFence, + ownerProcess: owner?.ownerProcess + }) + expect(await current.drafts()).toEqual([{ messageId: draftId, state: 'returned' }]) + } + ) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts index abfcd7076a03..3c2d07866786 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts @@ -197,6 +197,7 @@ export async function stopStructuredAgentSessionAgentUnderSerialize( journal: session.journal, sessionId, fence: child.fence, + rowFence: Math.max(child.fence, record?.lease.runtimeFence ?? child.fence), generation: child.generation ?? 'unknown', turnItemId: quitCuts, trigger: ending.quit, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts index 8faed8b6a6bf..d9961ef266da 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts @@ -67,8 +67,8 @@ export function createStructuredAgentSessionHostStatusFeed(args: { // field initializer, before its constructor parameters are assigned. statusSink: () => args.deps().statusSink, ...(args.onAgentStarted ? { onAgentStarted: args.onAgentStarted } : {}), - onChildWorkChanged: (sessionId) => { - subagentRows.changed(sessionId) + onChildWorkChanged: (sessionId, producer) => { + subagentRows.changed(sessionId, producer) args.onChildWorkChanged?.(sessionId) } }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts index 26bec39b190f..cbbd289bfc22 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts @@ -18,6 +18,7 @@ export function structuredAgentSessionJournalAppendOptions( return { fence, ownerFence: fence, + ...(options.executes ? { executionOwner: fence } : {}), ...(options.observedAt === undefined ? {} : { observedAt: options.observedAt }), turnScope: options.turnScope, ...agentJournalLinkageFields(options) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-late-proof-tool-ends.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-late-proof-tool-ends.test.ts index b9093d850994..3ba3e3043892 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-late-proof-tool-ends.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-late-proof-tool-ends.test.ts @@ -12,10 +12,8 @@ import type { AgentSessionDeathEvidence } from '../../../shared/agent-session-re import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import { createTrackedJournalOpener } from '../agent-session-journal/journal-host-database-test-support' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' -import { - settleStaleStructuredAgentSessionState, - settleStructuredAgentSessionDeadGeneration -} from './structured-agent-session-dead-generation-settlement' +import { settleStaleStructuredAgentSessionState } from './structured-agent-session-dead-generation-settlement' +import { settleObservedExitForTest } from './structured-agent-session-observed-exit.test-fixture' import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' const SESSION = 'session-late-proof' @@ -153,10 +151,11 @@ describe('a proof written after an unverifiable settle', () => { it('corrects a call an unverifiable restart eviction closed', async () => { await seedTurn('turn-1', 7) - await settleStructuredAgentSessionDeadGeneration({ + await settleObservedExitForTest({ journal, sessionId: SESSION, fence: 8, + ownerFence: 7, settlementId: `restart-eviction:${SESSION}:8`, pendingSubmissionReason: 'provider_exited_before_acknowledgement', verdict: { state: 'unverifiable' }, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-leftover-settlement.test-fixture.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-leftover-settlement.test-fixture.ts index 56f71b566d25..ff5f3060c45c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-leftover-settlement.test-fixture.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-leftover-settlement.test-fixture.ts @@ -100,6 +100,24 @@ export async function exitWhileSettlementFails(current: QueuedMessageTestRig) { return { session, database, deadFence: child.fence } } +/** The child exits on its own with nothing refused. */ +export async function exitChild(current: QueuedMessageTestRig): Promise { + const context = current.host.collaboratorsForTests() + const child = context.sessions.get(SESSION)?.child + if (!child?.generation) { + throw new Error('expected a child') + } + await current.host.handleAdapterEvent({ + type: 'ended', + sessionId: SESSION, + cause: 'unexpected-exit', + reason: 'observed exit', + fence: child.fence, + acquisitionGeneration: child.generation + }) + await context.serialize(SESSION, async () => {}) +} + /** A person's ordinary text send from the composer, with queueing on, at the current fence. */ export function sendText(current: QueuedMessageTestRig, text: string) { const fence = current.store.getRecord(SESSION)?.lease.runtimeFence diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-leftover-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-leftover-settlement.ts new file mode 100644 index 000000000000..a17e90ba1fb1 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-leftover-settlement.ts @@ -0,0 +1,116 @@ +// The one settlement of what ended generations left unfinished. An observed exit runs it, with +// what it saw. +// +// Re-derived each time from the journal, the lease and this host's sight of an exit: every turn, +// call, prompt, reasoning row, subagent roster entry and background task whose execution belongs +// to a generation that is not live (`structuredAgentSessionLiveFence`), and every send handed to +// one and never answered. Its turns end `interrupted` only when death evidence names that +// generation or a runtime this one replaced held it, else `unverifiable`, and a later proof revises +// an earlier `unverifiable`. Rows land at the lease's current fence in ONE transaction; a run with +// nothing left plans no row and writes nothing. A failure leaves rows that hold nothing, since the +// projection counts only the live generation's work; the open or the next start settles them. + +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { + planStructuredAgentSessionDeadGeneration, + type StructuredAgentSessionDeadGenerationInput +} from './structured-agent-session-dead-generation-settlement' +import { + planStaleStructuredAgentSessionState, + type StaleStructuredAgentSessionStateJournal +} from './structured-agent-session-stale-state-settlement' +import type { DeadGenerationJournal } from './structured-agent-session-unfinished-work' +import { STALE_SESSION_ROW_PREFIX } from '../../../shared/agent-session-stop-row-identity' +import { structuredAgentSessionFailureWordsContext } from './structured-agent-session-send-preparation' +import { + structuredAgentSessionLiveFence, + type StructuredAgentSessionWorkEvidence +} from './structured-agent-session-current-work' + +/** Without `replacedRuntime`, a store whose runtime replaced none. */ +export type StructuredAgentSessionLeftoverStore = Pick & + Partial> + +/** An observed exit's settlement: what that child's generation left, settled as the exit says. */ +export type StructuredAgentSessionExitSettlement = Omit< + StructuredAgentSessionDeadGenerationInput, + 'journal' | 'sessionId' | 'fence' +> & { ownerFence: number } + +export type StructuredAgentSessionLeftoverSettlementInput = { + store: StructuredAgentSessionLeftoverStore + sessionId: string + journal: DeadGenerationJournal & StaleStructuredAgentSessionStateJournal + exit?: StructuredAgentSessionExitSettlement + /** This host's sight of a child's root exit, which ends its generation even when the release + * write failed (`StructuredAgentSessionHostSession.lastEndedChild`). */ + ended?: StructuredAgentSessionWorkEvidence['ended'] +} + +export type StructuredAgentSessionLeftoverSettlement = + | { ok: true; planned: number } + | { ok: false; error: unknown } + +export async function settleStructuredAgentSessionLeftovers( + input: StructuredAgentSessionLeftoverSettlementInput +): Promise { + try { + const record = input.store.getRecord(input.sessionId) + if (!record) { + return { ok: true, planned: 0 } + } + const fence = record.lease.runtimeFence + const { journal, sessionId } = input + const replaced = input.store.replacedRuntime?.(sessionId) + // Everything below the live generation is ended; with none live, all of it is. + const liveFence = structuredAgentSessionLiveFence({ + record, + replaced, + ...(input.ended ? { ended: input.ended } : {}) + }) + // An exit's own account holds only until a later generation is live: its rows and sends at + // this fence would be that generation's. The general rule settles what the exit left instead. + const exit = liveFence === null ? input.exit : undefined + const below = exit ? exit.ownerFence : (liveFence ?? fence + 1) + const failureTextContext = structuredAgentSessionFailureWordsContext(record) + let planned = 0 + await journal.appendPlannedLifecycleBatch({ + settlementId: exit + ? `dead-generation:${exit.settlementId}` + : `${STALE_SESSION_ROW_PREFIX}${sessionId}:${fence}:seq-${journal.cursor().sequence}`, + fence, + recovered: true, + plan: () => { + const exited = exit + ? planStructuredAgentSessionDeadGeneration( + { ...exit, journal, sessionId, fence }, + (itemFence) => itemFence === undefined || itemFence >= exit.ownerFence + ) + : { mutations: [], dispatches: [] } + const settledByExit = new Set(exited.dispatches.map((entry) => entry.clientMessageId)) + const stale = planStaleStructuredAgentSessionState({ + journal, + sessionId, + fence, + acquisitionGeneration: null, + // Read with the rows: the proof the lease holds now. On an observed exit this settlement + // runs before the release, so the exit's own account (`exit`) judges instead. + deathEvidence: input.store.getRecord(sessionId)?.lease.deathEvidence ?? null, + replaced, + failureTextContext, + below + }) + const mutations = [...exited.mutations, ...stale.mutations] + const dispatches = [ + ...exited.dispatches, + ...stale.dispatches.filter((entry) => !settledByExit.has(entry.clientMessageId)) + ] + planned = mutations.length + dispatches.length + return { mutations, dispatches } + } + }) + return { ok: true, planned } + } catch (error) { + return { ok: false, error } + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-observed-exit.test-fixture.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-observed-exit.test-fixture.ts new file mode 100644 index 000000000000..fafb872993f0 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-observed-exit.test-fixture.ts @@ -0,0 +1,35 @@ +// An observed exit's settlement as the exit writes it (`settleStructuredAgentSessionLeftovers` +// with the exit's own account), for a chat whose lease that exit released. + +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { + agentSessionLeaseFixture, + agentSessionRecordFixture +} from '../../../shared/agent-session-record.test-fixture' +import type { StructuredAgentSessionDeadGenerationInput } from './structured-agent-session-dead-generation-settlement' +import { settleStructuredAgentSessionLeftovers } from './structured-agent-session-leftover-settlement' + +/** `ownerFence`: the exited generation's; by default every fence, as a chat's lone generation. */ +export function settleObservedExitForTest( + input: StructuredAgentSessionDeadGenerationInput & { + journal: AgentSessionJournal + ownerFence?: number + } +) { + const { journal, sessionId, fence, ownerFence = 0, ...exit } = input + const record = agentSessionRecordFixture( + agentSessionLeaseFixture({ + sessionId, + runtimeFence: fence, + claimStatus: 'released', + ownerProcess: null, + reservedSpawnToken: null + }) + ) + return settleStructuredAgentSessionLeftovers({ + store: { getRecord: () => record }, + sessionId, + journal, + exit: { ...exit, ownerFence } + }) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-orca-stop-row.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-orca-stop-row.ts index 1778f7e43977..c4d7333b9eb6 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-orca-stop-row.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-orca-stop-row.ts @@ -80,7 +80,10 @@ function cutByNobody(item: AgentJournalRenderItem | undefined): boolean { export async function recordStructuredAgentSessionShutdownCut(input: { journal: OrcaStopRowJournal sessionId: string + /** The stopped child's fence, which keys the row. */ fence: number + /** The fence the row is written at: the lease's, which the child's end moved past `fence`. */ + rowFence: number generation: string turnItemId: string | null trigger: AgentSessionResumeTrigger @@ -108,7 +111,7 @@ export async function recordStructuredAgentSessionShutdownCut(input: { await input.journal.appendItem( { provider: 'orca', clientMessageId }, orcaStopRowBody(input.failureTextContext, input.trigger), - { fence: input.fence, turnScope: { kind: 'turn', turnItemId } } + { fence: input.rowFence, turnScope: { kind: 'turn', turnItemId } } ) } catch (error) { input.logger.warn('recording why a quit cut a reply failed', { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts index 2f718261d4e0..f1671e1bc9ff 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts @@ -16,7 +16,10 @@ import type { AgentMessageSource } from '../../../shared/agent-session-message-s import { agentSessionMessagePayload } from '../../../shared/structured-agent-session-send-mutation' import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' import { AgentSessionRecoveryCapsule } from '../../runtime/agent-session-recovery-capsule' -import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { + StructuredAgentSessionHost, + type StructuredAgentSessionHostDeps +} from './structured-agent-session-host' import { rotateStructuredAgentSessionHostInstanceForTests } from './structured-agent-session-queued-pause' import { HOST_TEST_NOW as NOW, @@ -57,6 +60,8 @@ export async function createQueuedMessageTestRig( recoveryCapsule?: true /** Where the chat runs, when not the default worktree. */ location?: AgentSessionExecutionLocation + /** How the host probes a recorded owner; absent answers indeterminate. */ + probeOwner?: StructuredAgentSessionHostDeps['probeOwner'] } = {} ) { const root = await mkdtemp(join(tmpdir(), 'orca-queued-messages-')) @@ -74,6 +79,7 @@ export async function createQueuedMessageTestRig( claimKeyId: 'key-1', mintSpawnToken: () => 'spawn-1', now: () => NOW, + ...(options.probeOwner ? { probeOwner: options.probeOwner } : {}), ...(options.idleSweep ? { idleSweep: options.idleSweep } : {}), ...(options.recoveryCapsule ? { recoveryCapsule: new AgentSessionRecoveryCapsule(root) } : {}) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-reasoning-sweep.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-reasoning-sweep.test.ts index 2ea74bd77499..b29e3dab59ad 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-reasoning-sweep.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-reasoning-sweep.test.ts @@ -12,10 +12,8 @@ import { codexProviderHandle } from '../../../shared/agent-session-provider-hand import { openAgentSessionJournal } from '../agent-session-journal/journal-store-factory' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' -import { - settleStaleStructuredAgentSessionState, - settleStructuredAgentSessionDeadGeneration -} from './structured-agent-session-dead-generation-settlement' +import { settleStaleStructuredAgentSessionState } from './structured-agent-session-dead-generation-settlement' +import { settleObservedExitForTest } from './structured-agent-session-observed-exit.test-fixture' import { captureUnfinishedStructuredAgentSessionWork } from './structured-agent-session-unfinished-work' const SESSION = 'session-reasoning-sweep' @@ -72,7 +70,7 @@ describe('an open reasoning row a dead generation left', () => { it('ends, with no claimed time, when the generation is settled at its exit', async () => { await seedOpenReasoning('running') await expect( - settleStructuredAgentSessionDeadGeneration({ + settleObservedExitForTest({ journal, sessionId: SESSION, fence: 8, @@ -81,7 +79,7 @@ describe('an open reasoning row a dead generation left', () => { verdict: { state: 'interrupted', completedAt: 1_500 }, showUnexpectedExitOutcome: false }) - ).resolves.toEqual({ ok: true }) + ).resolves.toMatchObject({ ok: true }) expect(reasoningBody()).toEqual({ kind: 'message', role: 'reasoning', diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-recovered-turn-clock.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-recovered-turn-clock.test.ts index 6dfcb4f898e9..ba5d609b94e6 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-recovered-turn-clock.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-recovered-turn-clock.test.ts @@ -22,10 +22,8 @@ import { selectStructuredAgentSettledTurns } from '../../../shared/structured-ag import { AgentHookServer, _internals } from '../../agent-hooks/server' import { createTrackedJournalOpener } from '../agent-session-journal/journal-host-database-test-support' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' -import { - settleStaleStructuredAgentSessionState, - settleStructuredAgentSessionDeadGeneration -} from './structured-agent-session-dead-generation-settlement' +import { settleStaleStructuredAgentSessionState } from './structured-agent-session-dead-generation-settlement' +import { settleObservedExitForTest } from './structured-agent-session-observed-exit.test-fixture' import type { StructuredAgentSessionTurnVerdict } from './structured-agent-session-stale-turn-verdict' import { StructuredAgentSessionStatusFeed } from './structured-agent-session-status-feed' import { indexedStatusFeedSession } from './structured-agent-session-status-feed-test-session' @@ -128,8 +126,8 @@ async function sessionWithRunningTurn() { function settleDeadGeneration( journal: AgentSessionJournal, verdict: StructuredAgentSessionTurnVerdict -): ReturnType { - return settleStructuredAgentSessionDeadGeneration({ +): ReturnType { + return settleObservedExitForTest({ journal, sessionId: SESSION, fence: 2, @@ -159,7 +157,7 @@ describe('a turn recovery settled after its host went away', () => { async (_label, verdict, outcome, mark) => { const session = await sessionWithRunningTurn() session.recoverAt(RECOVERED) - expect(await settleDeadGeneration(session.journal, verdict)).toEqual({ ok: true }) + expect(await settleDeadGeneration(session.journal, verdict)).toMatchObject({ ok: true }) session.publish() expect(session.summaries.at(-1)).toMatchObject({ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-resume-eligibility.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-resume-eligibility.ts index 93a7962d4a20..2419651b7143 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-resume-eligibility.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-resume-eligibility.ts @@ -12,17 +12,32 @@ import type { AgentSessionRecord } from '../../../shared/agent-session-record' import { randomUUID } from 'node:crypto' import type { AgentSessionAttachParams } from './structured-agent-session-attach' import { attachParamsForRecord } from './structured-agent-session-conversation-open' +import type { StructuredAgentSessionEndedChild } from './structured-agent-session-host-types' +import { isSurfaceReleasableAgentSessionRecord } from '../../runtime/agent-session-surface-release-transition' -export function isResumableStructuredAgentSessionRecord(record: AgentSessionRecord): boolean { - return agentSessionLeaseIsReleased(record.lease) +/** Released, or still held by a child this host saw end at the lease's own fence (`ended`, its + * `lastEndedChild`), whose release write did not land: the acquisition probes that owner and + * replaces it only once proven dead. Nobody else's exit widens this, so an SSH or unverifiable + * owner is refused as before. */ +export function isResumableStructuredAgentSessionRecord( + record: AgentSessionRecord, + ended?: Pick +): boolean { + return ( + agentSessionLeaseIsReleased(record.lease) || + (ended?.rootGone === true && + ended.fence === record.lease.runtimeFence && + isSurfaceReleasableAgentSessionRecord(record)) + ) } /** Attach params for a resume, or null when this record's lease is somebody else's problem. */ export function structuredAgentSessionResumeParams( record: AgentSessionRecord, - clientOperationId: string + clientOperationId: string, + ended?: Pick ): AgentSessionAttachParams | null { - if (!isResumableStructuredAgentSessionRecord(record)) { + if (!isResumableStructuredAgentSessionRecord(record, ended)) { return null } return attachParamsForRecord(record, { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-resumed-child-owner.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-resumed-child-owner.test.ts new file mode 100644 index 000000000000..980ca36ed4e6 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-resumed-child-owner.test.ts @@ -0,0 +1,143 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, expect, it } from 'vitest' +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' +import { claudeProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { isBackgroundTaskBlock, isSubagentBlock } from '../../../shared/native-chat-types' +import { + ClaudeBackgroundTaskIdentityResolver, + writeClaudeBackgroundTaskRow +} from '../../claude/claude-background-task-row-journal' +import { createTrackedJournalOpener } from '../agent-session-journal/journal-host-database-test-support' +import { StructuredAgentSessionCurrentWork } from './structured-agent-session-current-work' +import { createDeferredStructuredAgentSessionEventSink } from './structured-agent-session-event-sink' +import { testEventSinkLogging } from './structured-agent-session-logger-test-support' +import { settleStaleStructuredAgentSessionState } from './structured-agent-session-stale-state-settlement' +import { + StructuredAgentSessionSubagentRows, + type AgentChildWorkRowView +} from './structured-agent-session-subagent-rows' + +// A subagent or Claude background task that run 1 started and run 2 resumed is run 2's work: run 1's +// death must not settle it, though its row was created at run 1's fence. + +const journals = createTrackedJournalOpener() +let root: string + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-resumed-child-owner-')) +}) + +afterEach(async () => { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +const subagent: AgentChildWorkRowView = { + id: 'record-1', + providerId: 'agent-1', + kind: 'agent', + description: 'Audit the build', + state: 'working', + membership: 'live', + firstObservedAt: 100, + observedAt: 100, + stoppable: true, + invocation: { invocationId: 'toolu_1', generation: 1 }, + residency: 'background' +} + +it('keeps a resumed subagent and background task working when the run that created them dies', async () => { + const journal = await journals.open({ + identity: { + sessionId: 'session-1', + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'claude', + providerHandle: claudeProviderHandle('claude-session', null) + }, + now: () => 9_000, + stateDirectory: root + }) + let fence = 1 + const rows = new StructuredAgentSessionSubagentRows({ + journal: () => journal, + fence: () => fence, + currentWork: (_sessionId, open) => new StructuredAgentSessionCurrentWork(open, fence), + readChildWork: () => [subagent], + onWriteFailed: (_sessionId, error) => { + throw error + } + }) + const tasks = new ClaudeBackgroundTaskIdentityResolver() + // One provider process: it reports the subagent and writes the task row through its own sink. + const run = async () => { + const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging()) + deferred.bind({ journal, fence, publish: () => {} }) + rows.changed('session-1', deferred.sink) + writeClaudeBackgroundTaskRow( + deferred.sink, + tasks, + 'task-1', + { + block: { + type: 'background-task', + taskId: 'task-1', + kind: 'command', + label: 'Build', + state: 'working' + }, + lastSerialized: null, + toolUseId: 'tool-1', + terminalNotificationReceived: false, + generation: 1 + }, + () => AGENT_JOURNAL_THREAD_SCOPE + ) + await expect(deferred.drained()).resolves.toEqual({ ok: true }) + await new Promise((resolve) => setTimeout(resolve, 0)) + deferred.close() + } + + const settleRunOne = () => + settleStaleStructuredAgentSessionState({ + journal, + sessionId: 'session-1', + fence: 2, + acquisitionGeneration: null, + deathEvidence: { + kind: 'exit-observed', + detail: 'run 1 exited', + observedAt: 9_000, + ownerFence: 1 + }, + below: 2 + }) + await run() + fence = 2 + await run() + await settleRunOne() + // A rewind rebuilds the epoch from these rows with both provenance facts, as it stores them. + await journal.replaceEpochItems( + 'handle_forked', + 2, + journal.snapshot().items.map((item) => { + const { ownerFence, executionOwner } = journal.itemProvenance(item.itemId) + return { + identity: parseAgentJournalItemKey(item.itemId)!, + body: journal.itemBody(item.itemId)!, + ownerFence: ownerFence ?? 0, + ...(executionOwner === undefined ? {} : { executionOwner }) + } + }) + ) + await settleRunOne() + + const blocks = journal + .snapshot() + .items.flatMap((item) => (item.body.kind === 'message' ? item.body.blocks : [])) + expect(blocks.filter(isSubagentBlock).map((block) => block.state)).toEqual(['working']) + expect(blocks.filter(isBackgroundTaskBlock).map((block) => block.state)).toEqual(['working']) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts index 4c48b155598c..4c42f005e617 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts @@ -170,13 +170,14 @@ export async function rewindStructuredAgentSession( if (!body) { throw new Error('agent_session_rewind:missing-retained-item') } - const ownerFence = ctx.journal.itemFence(itemId) + const { ownerFence, executionOwner } = ctx.journal.itemProvenance(itemId) return { itemId: providerKey(itemId), body: renameRewindTurnOpener(body, providerKey), observedAt, ...(turnScope ? { turnScope } : {}), ...(ownerFence === undefined ? {} : { ownerFence }), + ...(executionOwner === undefined ? {} : { executionOwner }), ...agentJournalLinkageFields(linkage) } }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-settlement-double.test-fixture.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-settlement-double.test-fixture.ts new file mode 100644 index 000000000000..eb6dc1c28be0 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-settlement-double.test-fixture.ts @@ -0,0 +1,56 @@ +// Doubles for unit tests that drive a settlement without a journal database: a planned batch is +// planned at once and handed to an `appendLifecycleBatch` mock as the batch it would write. + +import type { + AgentJournalCursor, + AgentJournalRenderItem, + AgentJournalSubmission +} from '../../../shared/agent-session-journal-types' +import type { StructuredAgentSessionChildExitSession } from './structured-agent-session-child-exit' +import type { + JournalLifecycleBatchInput, + JournalPlannedLifecycleBatchInput +} from '../agent-session-journal/journal-store-contracts' + +const NO_CURSOR: AgentJournalCursor = { epoch: 'epoch-1', sequence: 0 } + +/** `appendPlannedLifecycleBatch` for a double: a plan with no row reaches no `append`, as the + * writer writes none. */ +export function plannedBatchThrough( + append: (input: JournalLifecycleBatchInput) => Promise +): (input: JournalPlannedLifecycleBatchInput) => Promise { + return async (input) => { + const { mutations, dispatches } = input.plan() + if (mutations.length === 0 && dispatches.length === 0) { + return NO_CURSOR + } + const cursor = await append({ + settlementId: input.settlementId, + fence: input.fence, + ...(input.recovered ? { recovered: input.recovered } : {}), + mutations, + dispatches + }) + return cursor + } +} + +/** An exited child's journal for a test without a database: the reads a settlement makes, with a + * planned batch handed to `appendLifecycleBatch` (`plannedBatchThrough`). */ +export function exitJournalDouble(fields: { + items: () => readonly AgentJournalRenderItem[] + appendLifecycleBatch: (input: JournalLifecycleBatchInput) => Promise + itemFence?: (itemId: string) => number | undefined + submissions?: () => readonly Partial[] +}): StructuredAgentSessionChildExitSession['journal'] { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a settlement reads only these members; the rest of the journal is unreachable from it. + return { + cursor: () => NO_CURSOR, + itemBody: () => null, + itemFence: fields.itemFence ?? (() => undefined), + snapshot: () => ({ items: fields.items() }), + submissions: fields.submissions ?? (() => []), + stopMarks: { latest: () => null, personStopDecides: () => false }, + appendPlannedLifecycleBatch: plannedBatchThrough(fields.appendLifecycleBatch) + } as unknown as StructuredAgentSessionChildExitSession['journal'] +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-state-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-state-settlement.ts new file mode 100644 index 000000000000..b0f4f24fb909 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-state-settlement.ts @@ -0,0 +1,181 @@ +// What a generation that can no longer write left unfinished, settled from the journal, the +// lease's death evidence and the runtimes this one replaced each time it is asked. Either proof ends +// a turn interrupted; anything else is `unverifiable`, until a proof naming its owner revises it. Working subagents and background +// tasks become `unverifiable`. Planned at the batch's own turn in the write queue, so an answer or +// a Stop queued ahead of it is what it settles from. + +import { STALE_SESSION_ROW_PREFIX } from '../../../shared/agent-session-stop-row-identity' +import type { AgentSessionReplacedRuntime } from '../../runtime/agent-session-replaced-runtime' +import { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalRenderItem +} from '../../../shared/agent-session-journal-types' +import type { AgentSessionDeathEvidence } from '../../../shared/agent-session-record' +import type { AgentSessionFailureWordsContext } from '../../../shared/agent-session-failure-words' +import { journalPendingSubmissionResolutions } from '../agent-session-journal/journal-pending-submission-recovery' +import { DISPATCH_DOUBT_PROVIDER_EXITED } from '../agent-session-journal/journal-dispatch-doubt-reasons' +import type { JournalLifecycleMutationInput } from '../agent-session-journal/journal-row-builders' +import type { ResolveDispatchInput } from '../agent-session-journal/journal-store-contracts' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { lostLiveWorkJournalBody } from '../agent-session-journal/journal-subagent-liveness' +import { + endedUnseenMessageBody, + runningCallEnd, + terminalAgentJournalBody +} from '../agent-session-journal/journal-terminal-settlement' +import { + endedByPersonsStop, + provenUnverifiableTurnRevisions, + provenUnverifiedToolCallRevisions, + lastProvenTurnLiveAt, + runningTurnLifecycleRevisions, + turnVerdictFromDeathEvidence, + turnVerdictFromReplacedRuntime +} from './structured-agent-session-stale-turn-verdict' +import { settledRootTurnScope } from './structured-agent-session-exit-turn-scope' +import { isUnverifiableTurn, staleStopRow } from './structured-agent-session-stale-stop-row' +import { isInProgressStructuredAgentSessionItem } from './structured-agent-session-unfinished-work' + +/** What the settlement reads and writes of a chat's journal. */ +export type StaleStructuredAgentSessionStateJournal = Pick< + AgentSessionJournal, + | 'snapshot' + | 'itemFence' + | 'itemBody' + | 'submissions' + | 'stopMarks' + | 'lastProviderActivityAt' + | 'cursor' + | 'appendPlannedLifecycleBatch' +> + +export type StaleStructuredAgentSessionStateInput = { + journal: StaleStructuredAgentSessionStateJournal + sessionId: string + /** The fence the settlement's rows are written at: the lease's current one. */ + fence: number + acquisitionGeneration: string | null + deathEvidence: AgentSessionDeathEvidence | null + /** The runtimes this one replaced for this chat: every owner they held lost its pipes with them. */ + replaced?: AgentSessionReplacedRuntime + /** Who the exit row names. */ + failureTextContext?: AgentSessionFailureWordsContext + /** Only what generations below this fence wrote, which can no longer write, is settled; their + * sends handed over and never answered become doubt. Absent: every item, and no send. */ + below?: number +} + +/** Settles it, and answers how many items it revised. */ +export async function settleStaleStructuredAgentSessionState( + input: StaleStructuredAgentSessionStateInput +): Promise { + const { journal } = input + // A successful settlement leaves terminal items; a failed transaction leaves the same prefix. + const generation = input.acquisitionGeneration ?? `seq-${journal.cursor().sequence}` + let planned = 0 + await journal.appendPlannedLifecycleBatch({ + settlementId: `${STALE_SESSION_ROW_PREFIX}${input.sessionId}:${input.fence}:${generation}`, + fence: input.fence, + recovered: true, + plan: () => { + const plan = planStaleStructuredAgentSessionState(input) + planned = plan.mutations.length + return plan + } + }) + return planned +} + +/** The settlement's rows, read from the journal as it stands. */ +export function planStaleStructuredAgentSessionState( + input: StaleStructuredAgentSessionStateInput +): { + mutations: JournalLifecycleMutationInput[] + dispatches: ResolveDispatchInput[] +} { + const { journal, below } = input + const ended = (fence: number | undefined) => + below === undefined || (fence !== undefined && fence < below) + const items = journal.snapshot().items.filter((item) => ended(journal.itemFence(item.itemId))) + const dispatches = + below === undefined + ? [] + : journalPendingSubmissionResolutions( + journal.submissions().filter((entry) => ended(entry.fence)), + input.fence, + // Doubt, never non-delivery: what a gone owner was handed and never answered. + { reason: DISPATCH_DOUBT_PROVIDER_EXITED } + ) + return { mutations: staleMutations(input, items), dispatches } +} + +function staleMutations( + input: StaleStructuredAgentSessionStateInput, + items: readonly AgentJournalRenderItem[] +): JournalLifecycleMutationInput[] { + const { journal, replaced } = input + // Each turn is judged by the evidence only if it names that turn's owner. + const deathVerdictFor = (item: AgentJournalRenderItem) => + turnVerdictFromDeathEvidence( + input.deathEvidence, + journal.itemFence(item.itemId), + lastProvenTurnLiveAt(journal, item) + ) + const replacedVerdictFor = (item: AgentJournalRenderItem) => + turnVerdictFromReplacedRuntime( + replaced, + journal.itemFence(item.itemId), + lastProvenTurnLiveAt(journal, item) + ) + const verdictFor = (item: AgentJournalRenderItem) => { + const byDeath = deathVerdictFor(item) + return byDeath.state === 'interrupted' ? byDeath : replacedVerdictFor(item) + } + // Calls an earlier settle closed with no proof, revised once a proof names their owner. + const mutations = provenUnverifiedToolCallRevisions(items, input.deathEvidence, journal, replaced) + for (const item of items) { + // A turn already settled (a person's Stop) ends its calls as it ended; only a turn still running + // leaves them to the evidence. + const end = runningCallEnd(item.turnScope, (id) => journal.itemBody(id), verdictFor(item).state) + const terminal = endedUnseenMessageBody(item.body) ?? terminalAgentJournalBody(item.body, end) + // One revision per item: a live subagent or background task it holds ends with it. + const body = lostLiveWorkJournalBody(terminal ?? item.body) ?? terminal + if (body) { + mutations.push({ + kind: 'item', + itemId: item.itemId, + body, + turnScope: item.turnScope ?? AGENT_JOURNAL_THREAD_SCOPE + }) + } + } + const proven = provenUnverifiableTurnRevisions(items, input.deathEvidence, journal, replaced) + const turnEnds = [ + ...items.flatMap((item) => runningTurnLifecycleRevisions([item], verdictFor(item))), + ...proven + ] + mutations.push(...turnEnds) + const stopRow = staleStopRow(input, items, (item) => { + const unverifiable = isUnverifiableTurn(item) + if (!unverifiable && !isInProgressStructuredAgentSessionItem(item)) { + return null + } + const fence = journal.itemFence(item.itemId) + // An earlier settle's `unverifiable` turn is revised by a death naming its owner or by its + // replaced runtime. + if ( + unverifiable + ? fence !== undefined && fence === input.deathEvidence?.ownerFence + : deathVerdictFor(item).state === 'interrupted' + ) { + return { by: 'death' } + } + return fence !== undefined && replacedVerdictFor(item).state === 'interrupted' + ? { by: 'replaced', fence } + : null + }) + if (stopRow && !endedByPersonsStop(journal, turnEnds)) { + mutations.unshift({ ...stopRow, turnScope: settledRootTurnScope(items, turnEnds) }) + } + return mutations +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-open.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-open.test.ts index 204da81a2e6c..8c74b5deba85 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-open.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-open.test.ts @@ -43,7 +43,7 @@ async function exitWithUnwrittenSettlement(refusedWrites = 2): Promise { { turnScope: AGENT_JOURNAL_THREAD_SCOPE } ) await rig.host.flushStreamedEvents(SESSION) - const append = vi.spyOn(open.journal, 'appendLifecycleBatch') + const append = vi.spyOn(open.journal, 'appendPlannedLifecycleBatch') for (let refused = 0; refused < refusedWrites; refused += 1) { append.mockRejectedValueOnce(new Error('disk full')) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts index 9efb8202f880..0d3f5620bad7 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts @@ -17,6 +17,7 @@ import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' import { settleStaleStructuredAgentSessionState } from './structured-agent-session-dead-generation-settlement' +import { plannedBatchThrough } from './structured-agent-session-settlement-double.test-fixture' import { runningTurnLifecycleRevisions, turnVerdictFromDeathEvidence, @@ -314,7 +315,8 @@ describe('stale session state on a cold acquire', () => { lastProviderActivityAt: () => undefined, stopMarks: { latest: () => null }, cursor: () => ({ epoch: 'epoch-1', sequence: 8 }), - appendLifecycleBatch + submissions: () => [], + appendPlannedLifecycleBatch: plannedBatchThrough(appendLifecycleBatch) } as unknown as AgentSessionJournal return { journal, appendLifecycleBatch } } @@ -339,6 +341,7 @@ describe('stale session state on a cold acquire', () => { settlementId: 'stale-session:session-1:14:generation-2', fence: 14, recovered: true, + dispatches: [], mutations: [ { kind: 'item', @@ -369,6 +372,7 @@ describe('stale session state on a cold acquire', () => { settlementId: 'stale-session:session-1:14:generation-2', fence: 14, recovered: true, + dispatches: [], mutations: [ { kind: 'item', diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-failure-exit.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-failure-exit.test.ts index b79b0cd3ff5c..cf49ea0b7934 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-failure-exit.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-failure-exit.test.ts @@ -11,25 +11,29 @@ import { type StructuredAgentSessionChildExitSession } from './structured-agent-session-child-exit' import { recordingStructuredAgentSessionLogger } from './structured-agent-session-logger-test-support' +import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' +import { exitJournalDouble } from './structured-agent-session-settlement-double.test-fixture' const SESSION = 'session-1' const GENERATION = 'generation-1' const REASON = 'claude stream-json exited (code 1): session limit reached' const STARTUP_TEXT = 'Claude stopped before it finished starting. Send your message to try again.' -function startedSession(): StructuredAgentSessionChildExitSession & { - journal: { appendLifecycleBatch: ReturnType } +function startedSession( + submissions?: () => readonly Partial[] +): StructuredAgentSessionChildExitSession & { + appendLifecycleBatch: ReturnType } { + const appendLifecycleBatch = vi.fn(async () => ({ epoch: 'epoch-1', sequence: 1 })) return { child: { generation: GENERATION, fence: 7, phase: 'ready' }, - journal: { - cursor: () => ({ epoch: 'epoch-1', sequence: 0 }), - itemBody: () => null, - itemFence: () => undefined, - // Nothing ran: the start failed before any response or acknowledged prompt. - snapshot: () => ({ items: [] }), - appendLifecycleBatch: vi.fn(async () => ({ epoch: 'epoch-1', sequence: 1 })) - } + appendLifecycleBatch, + // Nothing ran: the start failed before any response or acknowledged prompt. + journal: exitJournalDouble({ + items: () => [], + appendLifecycleBatch, + ...(submissions ? { submissions } : {}) + }) } } @@ -84,7 +88,7 @@ describe('a provider that ends before it finished starting', () => { startupUnproven: true }) - expect(session.journal.appendLifecycleBatch).toHaveBeenCalledWith( + expect(session.appendLifecycleBatch).toHaveBeenCalledWith( expect.objectContaining({ mutations: [ expect.objectContaining({ @@ -108,7 +112,7 @@ describe('a provider that ends before it finished starting', () => { await settleStructuredAgentSessionChildExit(contextFor(session), ended) expect(session.child).toBeNull() - expect(session.journal.appendLifecycleBatch).not.toHaveBeenCalled() + expect(session.appendLifecycleBatch).not.toHaveBeenCalled() }) it("reads a start that failed off the host's own phase when the provider omits the flag", async () => { @@ -122,7 +126,7 @@ describe('a provider that ends before it finished starting', () => { failure: { kind: 'providerExited', detail: { text: REASON, audience: 'log' } } }) - expect(session.journal.appendLifecycleBatch).toHaveBeenCalledWith( + expect(session.appendLifecycleBatch).toHaveBeenCalledWith( expect.objectContaining({ mutations: [ expect.objectContaining({ @@ -145,27 +149,26 @@ describe('a provider that ends before it finished starting', () => { }) it('names /compact as the next step when the start that failed was carrying it', async () => { - const base = startedSession() + const base = startedSession(() => [ + { clientMessageId: 'compact-1', dispatchState: 'pending' as const } + ]) const session = { ...base, child: { generation: GENERATION, fence: 7, phase: 'starting' as const }, - journal: { - ...base.journal, - submissions: () => [{ clientMessageId: 'compact-1', dispatchState: 'pending' as const }], - itemBody: () => structuredAgentSessionCompactBody() - } + journal: { ...base.journal, itemBody: () => structuredAgentSessionCompactBody() } } await settleStructuredAgentSessionChildExit(contextFor(session), ended) const text = 'Claude stopped before it finished starting. Run /compact again.' - expect(session.journal.appendLifecycleBatch).toHaveBeenCalledWith( + expect(session.appendLifecycleBatch).toHaveBeenCalledWith( expect.objectContaining({ dispatches: [ expect.objectContaining({ clientMessageId: 'compact-1', state: 'rejected', reason: text, + // At the exited child's own fence, before its release moves it. fence: 7, recovered: true }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts index 108d43f1b646..9d6586787547 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts @@ -8,6 +8,7 @@ import type { StructuredAgentSessionProviderChild } from './structured-agent-session-host-types' import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' +import type { StructuredAgentSessionChildWorkProducer } from './structured-agent-session-subagent-rows' import type { StructuredAgentSessionStatusObserverOptions } from './structured-agent-session-status-observation' import type { StructuredAgentSessionStatusSink } from './structured-agent-session-status-ownership' @@ -38,8 +39,12 @@ export type StructuredAgentSessionStatusFeedDeps = { /** Resolved on every call: the host builds this feed in a field initializer, before its own * deps are assigned. The sink holds the session's child records; the summary reads them there. */ statusSink?: () => StructuredAgentSessionStatusSink | undefined - /** The session's child records changed, so every other reader of them republishes. */ - onChildWorkChanged?: (sessionId: string) => void + /** The session's child records changed, so every other reader of them republishes; `producer` + * is the write path of the child whose report changed them. */ + onChildWorkChanged?: ( + sessionId: string, + producer?: StructuredAgentSessionChildWorkProducer + ) => void /** The session's agent proved a start: its row's phase became `ready`. */ onAgentStarted?: (sessionId: string) => void } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts index f100d9263296..301b1f0ce039 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts @@ -30,6 +30,7 @@ import { type StructuredAgentSessionJournalProjection } from './structured-agent-session-status-journal-projection' import { structuredStatusSummariesEqual } from './structured-agent-session-status-summary-equality' +import type { StructuredAgentSessionChildWorkProducer } from './structured-agent-session-subagent-rows' import { StructuredAgentSessionStatusOwnership } from './structured-agent-session-status-ownership' export type { StructuredAgentSessionStatusSink } from './structured-agent-session-status-ownership' @@ -284,9 +285,13 @@ export class StructuredAgentSessionStatusFeed { /** Child-work evidence for a session this feed publishes; a failing sink costs nothing else. * The summary then re-reads the records, which is also what re-folds the parent's row. */ - publishChildWork(sessionId: string, evidence: AgentChildWorkEvidence[]): void { + publishChildWork( + sessionId: string, + evidence: AgentChildWorkEvidence[], + producer?: StructuredAgentSessionChildWorkProducer + ): void { const session = this.deps.sessions.get(sessionId) - if (!session || !this.admitChildWork(sessionId, session, evidence)) { + if (!session || !this.admitChildWork(sessionId, session, evidence, producer)) { return } this.publish(sessionId) @@ -314,7 +319,8 @@ export class StructuredAgentSessionStatusFeed { private admitChildWork( sessionId: string, session: StatusFeedSession, - evidence: AgentChildWorkEvidence[] + evidence: AgentChildWorkEvidence[], + producer?: StructuredAgentSessionChildWorkProducer ): boolean { try { this.ownership.publishChildWork(sessionId, evidence, session.params.provider) @@ -323,7 +329,7 @@ export class StructuredAgentSessionStatusFeed { return false } try { - this.deps.onChildWorkChanged?.(sessionId) + this.deps.onChildWorkChanged?.(sessionId, producer) } catch (error) { this.logFailure('child-work-observer', 'child work observer failed', sessionId, error) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-rows.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-rows.ts index d5c599132145..063de98acd44 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-rows.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-rows.ts @@ -7,6 +7,10 @@ // tried again by the next change after its batch settles, so nothing here holds an obligation of // its own. What a lost write leaves saying `working` the journal's reopen pass settles as // unverifiable. +// +// A row a child's report changes is written through that child's own sink, so it lands after the +// settlement its attach writes first, and names that child as the run executing the subagent +// (`JournalItemRow.executionOwner`): a subagent a later run resumes is that run's work. import { AGENT_JOURNAL_THREAD_SCOPE, @@ -26,15 +30,32 @@ import { subagentJournalItemIdentity } from '../agent-session-journal/journal-subagent-item' import type { StructuredAgentSessionCurrentWork } from './structured-agent-session-current-work' +import type { + StructuredAgentSessionEventSink, + StructuredAgentSessionSinkAdmission, + StructuredAgentSessionSinkBarrier +} from './structured-agent-session-event-sink' const UNNAMED = 'subagent' +const LANDED: StructuredAgentSessionSinkBarrier = { ok: true } +const LANDED_ADMISSION: StructuredAgentSessionSinkAdmission = { accepted: true } const THREAD_GROUP = 'thread' /** A child's view, with the host-only facts its saved row also keeps. */ export type AgentChildWorkRowView = AgentChildWorkView & Pick -type Row = { block: NativeChatSubagentBlock; turnScope: AgentJournalTurnScope } +/** The child-bound write path of the child whose report changed the records. */ +export type StructuredAgentSessionChildWorkProducer = Pick< + StructuredAgentSessionEventSink, + 'appendItem' | 'tryAppendItem' | 'written' +> + +type Row = { + block: NativeChatSubagentBlock + turnScope: AgentJournalTurnScope + producer: StructuredAgentSessionChildWorkProducer | undefined +} export type StructuredAgentSessionSubagentRowsDeps = { journal: (sessionId: string) => AgentSessionJournal | undefined @@ -53,17 +74,21 @@ export class StructuredAgentSessionSubagentRows { private readonly queued = new Map>() /** The batch each session's writer has in flight: written, but not yet readable by key. */ private readonly writing = new Map>() + /** The running subagents each child has written once: the rows it executes. */ + private readonly claimed = new WeakMap>() constructor(private readonly deps: StructuredAgentSessionSubagentRowsDeps) {} - /** The session's child records changed: queue each subagent whose row says otherwise. */ - changed(sessionId: string): void { + /** The session's child records changed, by `producer`'s report when a child made one: queue each + * subagent whose row says otherwise, and each running one that child has not written yet. */ + changed(sessionId: string, producer?: StructuredAgentSessionChildWorkProducer): void { const journal = this.deps.journal(sessionId) const children = this.deps.readChildWork(sessionId) if (!journal || !children?.some((child) => child.kind === 'agent')) { return } const queued = this.queued.get(sessionId) ?? new Map() + const claimed = producer && (this.claimed.get(producer) ?? new Set()) const providerIds = new Map(children.map((child) => [child.id, child.providerId])) // The strip's rule: a child that finished while work it started runs is still running. const running = new Set(structuredRunningChildWork(children).map((child) => child.id)) @@ -85,10 +110,16 @@ export class StructuredAgentSessionSubagentRows { const parentId = child.parentChildWorkId === undefined ? undefined : providerIds.get(child.parentChildWorkId) const block = subagentRowBlock(child, id, parentId, saved, turnScope, running.has(child.id)) - if (!saved || rowText(saved) !== rowText(block)) { - queued.set(id, { block, turnScope }) + // A running subagent another child now reports is that child's, even when its row reads the same. + const moved = running.has(child.id) && claimed !== undefined && !claimed.has(id) + if (!saved || moved || rowText(saved) !== rowText(block)) { + queued.set(id, { block, turnScope, producer: producer ?? pending?.producer }) + claimed?.add(id) } } + if (producer && claimed) { + this.claimed.set(producer, claimed) + } if (queued.size > 0) { this.queued.set(sessionId, queued) void this.write(sessionId) @@ -109,14 +140,19 @@ export class StructuredAgentSessionSubagentRows { this.writing.set(sessionId, rows) // The journal serializes its writes; these land in the order they were queued. await Promise.all( - [...rows].map(([id, row]) => - journal - .appendItem(subagentJournalItemIdentity(id), subagentJournalBody(row.block), { + [...rows].map(([id, row]) => { + const identity = subagentJournalItemIdentity(id) + const body = subagentJournalBody(row.block) + if (row.producer) { + return this.writeThrough(sessionId, id, row, row.producer) + } + return journal + .appendItem(identity, body, { fence: this.deps.fence(sessionId), turnScope: row.turnScope }) .catch((error: unknown) => this.deps.onWriteFailed(sessionId, error)) - ) + }) ) } } finally { @@ -124,6 +160,39 @@ export class StructuredAgentSessionSubagentRows { } } + /** Through the reporting child's sink, in its order and at its fence. A write it refuses under + * backpressure is queued again once what it holds has landed; any failure frees the claim. */ + private async writeThrough( + sessionId: string, + id: string, + row: Row, + producer: StructuredAgentSessionChildWorkProducer + ): Promise { + const identity = subagentJournalItemIdentity(id) + const body = subagentJournalBody(row.block) + const options = { turnScope: row.turnScope, executes: true as const } + const admission = producer.tryAppendItem + ? producer.tryAppendItem(identity, body, options) + : (producer.appendItem(identity, body, options), LANDED_ADMISSION) + const landed = await (producer.written?.() ?? Promise.resolve(LANDED)) + if (admission.accepted && landed.ok) { + return + } + this.claimed.get(producer)?.delete(id) + if (!landed.ok) { + this.deps.onWriteFailed(sessionId, landed.error) + } else if (!admission.accepted && admission.reason === 'backpressure') { + const queued = this.queued.get(sessionId) ?? new Map() + // A newer report of the same subagent supersedes this one. + if (!queued.has(id)) { + queued.set(id, row) + } + this.queued.set(sessionId, queued) + } else if (!admission.accepted) { + this.deps.onWriteFailed(sessionId, new Error(`the sink refused the row: ${admission.reason}`)) + } + } + private take(sessionId: string): Map | undefined { const rows = this.queued.get(sessionId) this.queued.delete(sessionId) 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 1fb10c9d8f28..0891fa6f1705 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 @@ -14,6 +14,11 @@ import { type StructuredAgentSessionChildExitSession } from './structured-agent-session-child-exit' import { recordingStructuredAgentSessionLogger } from './structured-agent-session-logger-test-support' +import { + exitJournalDouble, + plannedBatchThrough +} from './structured-agent-session-settlement-double.test-fixture' +import type { JournalLifecycleBatchInput } from '../agent-session-journal/journal-store-contracts' const exitOutcome = (agent: string): string => `${agent} stopped while this response was in progress. You can continue in this conversation.` @@ -101,7 +106,8 @@ describe('provider-exit settlement', () => { itemFence: () => 7, lastProviderActivityAt: () => undefined, stopMarks: { latest: () => null, personStopDecides: () => false }, - appendLifecycleBatch + submissions: () => [], + appendPlannedLifecycleBatch: plannedBatchThrough(appendLifecycleBatch) } } as unknown as StructuredAgentSessionHostSession @@ -173,7 +179,8 @@ describe('provider-exit settlement', () => { itemBody: () => null, itemFence: () => undefined, snapshot: () => ({ items }), - appendLifecycleBatch + submissions: () => [], + appendPlannedLifecycleBatch: plannedBatchThrough(appendLifecycleBatch) } } as unknown as StructuredAgentSessionHostSession const store = { @@ -279,19 +286,13 @@ describe('provider-exit settlement', () => { ...(initialState === 'running' ? {} : { completedAt: 40 }) }) ] - const appendLifecycleBatch = vi.fn(async (_input: { mutations: readonly unknown[] }) => ({ + const appendLifecycleBatch = vi.fn(async (_input: JournalLifecycleBatchInput) => ({ epoch: 'epoch-1', sequence: 3 })) const session: StructuredAgentSessionChildExitSession = { child: { generation: GENERATION, fence: 7, phase: 'ready' }, - journal: { - cursor: () => ({ epoch: 'epoch-1', sequence: 0 }), - itemBody: () => null, - itemFence: () => undefined, - snapshot: () => ({ items }), - appendLifecycleBatch - } + journal: exitJournalDouble({ items: () => items, appendLifecycleBatch }) } const { store } = mutableStore() @@ -340,16 +341,15 @@ 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' }, - journal: { - cursor: () => ({ epoch: 'epoch-1', sequence: 0 }), - itemBody: () => null, - itemFence: () => undefined, - snapshot: () => ({ items: [] }), - appendLifecycleBatch: vi.fn(async () => ({ epoch: 'epoch-1', sequence: 1 })), - submissions: () => [{ clientMessageId: 'client-1', dispatchState: 'pending' }] - } + journal: exitJournalDouble({ + items: () => [], + appendLifecycleBatch, + // Handed to the child that exited, at its fence. + submissions: () => [{ clientMessageId: 'client-1', dispatchState: 'pending', fence: 7 }] + }) } const { store } = mutableStore() @@ -371,8 +371,11 @@ describe('provider-exit settlement', () => { acquisitionGeneration: GENERATION }) - expect(session.journal.appendLifecycleBatch).toHaveBeenCalledWith( + // Written at the exited child's own fence, before its release moves it, as is every row its + // settlement writes. + expect(appendLifecycleBatch).toHaveBeenCalledWith( expect.objectContaining({ + fence: 7, dispatches: [ { clientMessageId: 'client-1', @@ -399,17 +402,12 @@ describe('provider-exit settlement', () => { it('releases without offering a restart while terminal settlement is failing', async () => { const session: StructuredAgentSessionChildExitSession = { child: { generation: GENERATION, fence: 7, phase: 'ready' }, - journal: { - cursor: () => ({ epoch: 'epoch-1', sequence: 0 }), - itemBody: () => null, - itemFence: () => undefined, - snapshot: () => ({ - items: [lifecycleItem('turn-failing', 1, { state: 'running', startedAt: 1 })] - }), + journal: exitJournalDouble({ + items: () => [lifecycleItem('turn-failing', 1, { state: 'running', startedAt: 1 })], appendLifecycleBatch: vi.fn(async () => { throw new Error('journal still unavailable') }) - } + }) } const log = recordingStructuredAgentSessionLogger() const generationEnded = vi.fn() @@ -434,6 +432,8 @@ describe('provider-exit settlement', () => { await settleStructuredAgentSessionChildExit(context, event) expect(session.child).toBeNull() + // The death first, before anything is written; then the exit's own account, for the retry to + // write once the journal takes it. expect(generationEnded).toHaveBeenCalledTimes(2) expect(generationEnded).toHaveBeenNthCalledWith(1, SESSION) expect(generationEnded).toHaveBeenLastCalledWith(SESSION, { restate: true }) 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 ed1e642da39e..e0a98e7d7716 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 @@ -11,7 +11,7 @@ import { import type { UnopenedSendJournal } from './structured-agent-session-unopened-send-withdrawal' export type DeadGenerationJournal = UnopenedSendJournal & { - appendLifecycleBatch: AgentSessionJournal['appendLifecycleBatch'] + appendPlannedLifecycleBatch: AgentSessionJournal['appendPlannedLifecycleBatch'] pendingSubmissions?: AgentSessionJournal['pendingSubmissions'] itemFence: AgentSessionJournal['itemFence'] } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.ts index e1022722955f..deeb87fb23e9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.ts @@ -24,7 +24,7 @@ export type UnopenedSendJournal = { | 'handoverRecorded' | 'handedOverAt' | 'acceptedSequence' - > & { reason?: string | null })[] + > & { reason?: string | null; fence?: number })[] resolveDispatch?: AgentSessionJournal['resolveDispatch'] } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-wedged-profile-migration.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-wedged-profile-migration.test.ts index 6727b0144c03..32a0b6c2c2b2 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-wedged-profile-migration.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-wedged-profile-migration.test.ts @@ -38,7 +38,8 @@ import { } from '../../runtime/agent-session-record-store-test-harness' import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' import { openAgentSessionJournal } from '../agent-session-journal/journal-store-factory' -import { AgentSessionJournal } from '../agent-session-journal/journal-store' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { JournalLifecycleBatchAppender } from '../agent-session-journal/journal-lifecycle-batch-appender' import * as recoveryResolution from './structured-agent-session-recovery-resolution' import { StructuredAgentSessionHost } from './structured-agent-session-host' import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types' @@ -464,7 +465,7 @@ describe('already-wedged profiles become usable on load', () => { }) // The read restore's settlement fails, and nothing retries it. const failing = vi - .spyOn(AgentSessionJournal.prototype, 'appendLifecycleBatch') + .spyOn(JournalLifecycleBatchAppender.prototype, 'appendPlanned') .mockRejectedValue(new Error('journal unavailable')) await host.restoreReadableSessions() failing.mockRestore() diff --git a/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts b/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts index 53f2b20cc52a..f1b5a3466fb0 100644 --- a/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts +++ b/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts @@ -89,6 +89,7 @@ export function retainedRowReplacement(row: RetainedRow): AgentJournalProducerLi observedAt: number turnScope?: AgentJournalTurnScope ownerFence: number + executionOwner?: number } { const identity = parseAgentJournalItemKey(row.itemId) if (!identity) { @@ -109,6 +110,7 @@ export function retainedRowReplacement(row: RetainedRow): AgentJournalProducerLi // The old fold's provenance, so a rewind never makes a dead generation's work current. A record // from before rows carried it states none: 0, which is never current. ownerFence: row.ownerFence ?? 0, + ...(row.executionOwner === undefined ? {} : { executionOwner: row.executionOwner }), ...(agentId === undefined ? {} : { agentId }), ...(parentAgentId === undefined ? {} : { parentAgentId }), ...(providerParentRef === undefined ? {} : { providerParentRef }), diff --git a/src/main/runtime/structured-agent-runtime-registrations.ts b/src/main/runtime/structured-agent-runtime-registrations.ts index 3532428938c8..d6bda7b14ace 100644 --- a/src/main/runtime/structured-agent-runtime-registrations.ts +++ b/src/main/runtime/structured-agent-runtime-registrations.ts @@ -58,8 +58,8 @@ function createCodexAdapter(context: StructuredAgentAdapterContext): StructuredA ...(deps.openCodexConnection ? { openConnection: deps.openCodexConnection } : {}), ...(deps.readProcessStartTime ? { readProcessStartTime: deps.readProcessStartTime } : {}), modelCatalog: agentModelCatalogStore, - onChildWorkEvidence: (sessionId, evidence) => - host()?.publishChildWorkEvidence(sessionId, evidence), + onChildWorkEvidence: (sessionId, evidence, producer) => + host()?.publishChildWorkEvidence(sessionId, evidence, producer), onDispatchSettledLate: followUps.onDispatchSettledLate, onPrimaryThreadStoppedRunning: followUps.releaseUnansweredDispatches, logger: deps.logger, @@ -97,8 +97,8 @@ function createClaudeAdapter( resolveClaudeInheritedEnv: context.environment.resolveBaseEnvironment, onLifecycleEvent: context.deliverLifecycle, logger: deps.logger, - onChildWorkEvidence: (sessionId, evidence) => - host()?.publishChildWorkEvidence(sessionId, evidence), + onChildWorkEvidence: (sessionId, evidence, producer) => + host()?.publishChildWorkEvidence(sessionId, evidence, producer), onDispatchSettledLate: followUps.onDispatchSettledLate, onSessionIdle: followUps.releaseUnansweredDispatches, ...(deps.openClaudeConnection ? { openClaudeConnection: deps.openClaudeConnection } : {}), @@ -127,8 +127,8 @@ function createAcpAdapter( readLaunch: composeStructuredLaunchRead(definition, sources, part), resolveLaunchArgs: () => deps.resolveLaunchArgs(spec.agent), connect: (launch, options) => createAcpAgentConnection(launch, options), - onChildWorkEvidence: (sessionId, evidence) => - context.host()?.publishChildWorkEvidence(sessionId, evidence), + onChildWorkEvidence: (sessionId, evidence, producer) => + context.host()?.publishChildWorkEvidence(sessionId, evidence, producer), ...(deps.readProcessStartTime ? { readProcessStartTime: deps.readProcessStartTime } : {}), onDispatchSettledLate: followUps.onDispatchSettledLate, logger: deps.logger, diff --git a/src/shared/agent-session-rewind.ts b/src/shared/agent-session-rewind.ts index 36e30f9b79a4..1cce2ec76f3f 100644 --- a/src/shared/agent-session-rewind.ts +++ b/src/shared/agent-session-rewind.ts @@ -58,6 +58,9 @@ export const AgentSessionRewindRecordSchema = z.object({ // The row's execution provenance, carried into the rebuilt epoch. Absent on records written // before rows carried it. ownerFence: z.number().int().nonnegative().optional(), + // The run executing a reusable child's row, kept apart from `ownerFence`. Absent on records + // written before rows carried it. + executionOwner: z.number().int().nonnegative().optional(), ...AgentJournalProducerLinkageFields }) )