Native chat: a resumed subagent or Claude background task belongs to the run executing it

A subagent row that run 1 created and run 2 resumed kept run 1's creation fence, so settling run 1's death marked run 2's live subagent unverifiable. Claude background tasks reuse their rows the same way (a matching parent tool use, or a parentless row). Rows of reusable child execution now also name their execution owner: the fence of the child process whose report the write copies, captured when that report arrives, never the lease that is current when a queued write flushes. - Subagent rows: the host captures the live child's fence when its child-work report changes the records, and rewrites a running subagent another child now reports even when its text is unchanged. - Claude background-task rows: written through that child's own event sink, which stamps its bound fence. - Creation provenance (`ownerFence`) is unchanged. Writes that do not state an owner (settlement, Pi's start usage, ACP session/load replay, ACP tasks) keep it, and a row without one falls back to its creation fence. `journal.itemFence()` now answers whose execution an item is (`executionOwner ?? ownerFence`), so every settlement and liveness read, inside its write transaction, follows the run that resumed the work. Also drops the event sink's compatibility alias for the identity resolver (the Codex sink names the resolver itself), keeping the sink under its line limit.
2026-10-31 00:57:45 +00:00 · 2026-10-14 21:39:12 -07:00
parent 67d715bc37
commit c0add67cca
16 changed files with 209 additions and 21 deletions
@@ -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 } : {}
@@ -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) {
@@ -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)
 }
@@ -1,9 +1,18 @@
 // 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)
+}
+
 /** 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(
@@ -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<string, number>
+  /** The run executing a reusable child's item now (`JournalItemRow.executionOwner`), when one said. */
+  itemExecutionOwners: Map<string, number>
  /** Latest saved output per owner, rebuilt during the existing row fold. */
  providerActivityAt: Map<number, number>
  /** 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)
@@ -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)
@@ -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 & {
@@ -60,9 +60,11 @@ function dropUnusableTurnScope(record: Record<string, unknown>): void {

 /** Provenance this build cannot read, removed like linkage: the item's creating fence stands in. */
 function dropUnusableOwnerFence(record: Record<string, unknown>): 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]
+    }
  }
 }

@@ -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
 }
@@ -68,6 +68,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 } 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'
@@ -247,8 +248,9 @@ 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)

  /** Latest saved output at this fence, excluding client actions and recovery bookkeeping. */
  lastProviderActivityAt = (fence: number) => this.state.providerActivityAt.get(fence)
@@ -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,
@@ -34,6 +34,8 @@ export function createStructuredAgentSessionHostStatusFeed(args: {
  const subagentRows = new StructuredAgentSessionSubagentRows({
    journal: (sessionId) => args.sessions.get(sessionId)?.journal,
    fence: (sessionId) => structuredAgentSessionConversationFence(args.deps().store, sessionId),
+    // Read as the report lands, never when the queued write flushes.
+    producer: (sessionId) => args.sessions.get(sessionId)?.child?.fence,
    currentWork: args.currentWork,
    // The views, with the two host-only facts a saved row also keeps from their records.
    readChildWork: (sessionId) => {
@@ -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)
@@ -0,0 +1,126 @@
+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 { 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,
+    producer: () => 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')
+    writeClaudeBackgroundTaskRow(
+      deferred.sink,
+      tasks,
+      'task-1',
+      {
+        block: {
+          type: 'background-task',
+          taskId: 'task-1',
+          kind: 'command',
+          label: 'Build',
+          state: 'running'
+        },
+        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()
+  }
+
+  await run()
+  fence = 2
+  await run()
+  await settleStaleStructuredAgentSessionState({
+    journal,
+    sessionId: 'session-1',
+    fence: 2,
+    acquisitionGeneration: null,
+    deathEvidence: {
+      kind: 'exit-observed',
+      detail: 'run 1 exited',
+      observedAt: 9_000,
+      ownerFence: 1
+    },
+    below: 2
+  })
+
+  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(['running'])
+})
@@ -60,6 +60,7 @@ async function rig() {
  const rows = new StructuredAgentSessionSubagentRows({
    journal: () => journal,
    fence: () => 1,
+    producer: () => 1,
    currentWork: (_sessionId, open) => new StructuredAgentSessionCurrentWork(open, 1),
    readChildWork: () => children,
    onWriteFailed: failed
@@ -7,6 +7,9 @@
 // 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 also names the child process whose report it copies (`JournalItemRow.executionOwner`),
+// captured when that report arrives, so a subagent a later run resumes is that run's work.

 import {
  AGENT_JOURNAL_THREAD_SCOPE,
@@ -26,6 +29,7 @@ import {
  subagentJournalItemIdentity
 } from '../agent-session-journal/journal-subagent-item'
 import type { StructuredAgentSessionCurrentWork } from './structured-agent-session-current-work'
+import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key'

 const UNNAMED = 'subagent'
 const THREAD_GROUP = 'thread'
@@ -34,11 +38,17 @@ const THREAD_GROUP = 'thread'
 export type AgentChildWorkRowView = AgentChildWorkView &
  Pick<AgentChildWorkInput, 'outcomeBasis' | 'residency'>

-type Row = { block: NativeChatSubagentBlock; turnScope: AgentJournalTurnScope }
+type Row = {
+  block: NativeChatSubagentBlock
+  turnScope: AgentJournalTurnScope
+  executionOwner: number | undefined
+}

 export type StructuredAgentSessionSubagentRowsDeps = {
  journal: (sessionId: string) => AgentSessionJournal | undefined
  fence: (sessionId: string) => number
+  /** The fence of the live child whose report changed the records; absent, a row keeps its owner. */
+  producer: (sessionId: string) => number | undefined
  /** The host's read of what runs, whose turn a first row joins. */
  currentWork: (
    sessionId: string,
@@ -64,6 +74,7 @@ export class StructuredAgentSessionSubagentRows {
      return
    }
    const queued = this.queued.get(sessionId) ?? new Map<string, Row>()
+    const executionOwner = this.deps.producer(sessionId)
    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,8 +96,18 @@ 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 owner = pending
+        ? pending.executionOwner
+        : journal.itemFence(agentJournalItemKey(subagentJournalItemIdentity(id)))
+      const moved =
+        running.has(child.id) && executionOwner !== undefined && owner !== executionOwner
+      if (!saved || moved || rowText(saved) !== rowText(block)) {
+        queued.set(id, {
+          block,
+          turnScope,
+          executionOwner: executionOwner ?? pending?.executionOwner
+        })
      }
    }
    if (queued.size > 0) {
@@ -113,7 +134,8 @@ export class StructuredAgentSessionSubagentRows {
            journal
              .appendItem(subagentJournalItemIdentity(id), subagentJournalBody(row.block), {
                fence: this.deps.fence(sessionId),
-                turnScope: row.turnScope
+                turnScope: row.turnScope,
+                ...(row.executionOwner === undefined ? {} : { executionOwner: row.executionOwner })
              })
              .catch((error: unknown) => this.deps.onWriteFailed(sessionId, error))
          )