From c0add67cca411b98fa4e94ed37a487522204ce4d Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 14 Oct 2026 21:39:12 -0700 Subject: [PATCH] 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. --- .../claude-background-task-row-journal.ts | 2 + .../codex/codex-structured-journal-sink.ts | 4 +- .../journal-item-fold.ts | 8 +- .../journal-item-provenance.ts | 11 +- .../agent-session-journal/journal-reducer.ts | 10 +- .../journal-row-builders.ts | 4 + .../journal-row-schema.ts | 8 +- .../journal-row-unusable-annotations.ts | 8 +- .../journal-store-contracts.ts | 3 + .../agent-session-journal/journal-store.ts | 6 +- .../structured-agent-session-event-sink.ts | 6 +- ...ructured-agent-session-host-status-feed.ts | 2 + ...ed-agent-session-journal-append-options.ts | 1 + ...-agent-session-resumed-child-owner.test.ts | 126 ++++++++++++++++++ ...ctured-agent-session-subagent-rows.test.ts | 1 + .../structured-agent-session-subagent-rows.ts | 30 ++++- 16 files changed, 209 insertions(+), 21 deletions(-) create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-resumed-child-owner.test.ts diff --git a/src/main/claude/claude-background-task-row-journal.ts b/src/main/claude/claude-background-task-row-journal.ts index 1419729c5cf..85e85d07e1d 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/codex/codex-structured-journal-sink.ts b/src/main/codex/codex-structured-journal-sink.ts index 9087449584f..2af85c59b3d 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/native-chat/agent-session-journal/journal-item-fold.ts b/src/main/native-chat/agent-session-journal/journal-item-fold.ts index c45d8975b25..c23d0547a1d 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 afb1df346d1..42df3b4a49c 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,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, + 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( 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 ecb9858b0bc..cbc87dc9e43 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 4b5ce901bfa..88574cd829d 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 91b3eb43a8e..88cde342d37 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 e6d25b21df9..53de64ace50 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 2ddfcdfd21c..c9f2feae47b 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 } 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 48992117087..62c03a93153 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -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) 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 880a7dfa86b..b8b5797b755 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-host-status-feed.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts index 8faed8b6a6b..0de2c90c0ec 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 @@ -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) => { 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 26bec39b190..cbbd289bfc2 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-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 00000000000..09444b8541d --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-resumed-child-owner.test.ts @@ -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']) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-rows.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-rows.test.ts index 42395b2b3be..d345f6b1c41 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-rows.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-rows.test.ts @@ -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 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 d5c59913214..18cfc2317f6 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,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 -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() + 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)) )