Native chat: one settlement transaction for everything an ended run left (4/5, rebuild of #26867)

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',

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)

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 }
 }

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<AcpStructuredLaneDeps, 'onChildWorkEvidence' | 'onChildWorkFailure'> {
   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',

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<number | null>
   /** 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 } & (

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 } : {}

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)
     }
   }
 

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<number | null>
   mintLinkId?: () => string

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<unknown> | undefined
     hooks.onEvent = (event) => {
       if (event.type === 'ended') {
-        settled = settleStructuredAgentSessionDeadGeneration({
+        settled = settleObservedExitForTest({
           journal: run.journal,
           sessionId: SESSION,
           fence: FENCE,

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<CodexStructuredSessionAdapterDeps, 'onChildWorkEvidence' | 'now'>
+  deps: Pick<CodexStructuredSessionAdapterDeps, 'onChildWorkEvidence' | 'now'>,
+  producer?: StructuredAgentSessionChildWorkProducer
 ): CodexChildWorkSink {
   return {
-    deliver: (evidence) => deps.onChildWorkEvidence?.(sessionId, evidence),
+    deliver: (evidence) => deps.onChildWorkEvidence?.(sessionId, evidence, producer),
     now: () => deps.now?.() ?? Date.now()
   }
 }

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) {

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,

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?: (

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)
       }

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)

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)
 }

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<JournalReducerState, 'itemFences' | 'itemExecutionOwners'>,
+  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<JournalReducerState, 'itemFences' | 'itemExecutionOwners'>,
+  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(

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<AgentJournalCursor> {
+    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,