Native chat: a subagent row is written through the child whose report it copies

Review of c0add67cca: the subagent row writer read its execution owner from
the host's installed child when a report landed. A child that is still being
acquired reports a resumed subagent before the host installs it, so its row
was written at once under the old owner, then the attach's settlement of what
the previous child left marked it unverifiable, and nothing rewrote it until
the next report.

Each adapter (Claude, Codex, ACP) now hands its child-work report to the host
with that child's own event sink, the write path its rows already take. The
host passes it through the status feed to the subagent row writer, which writes
the rows that report changes through that sink with `executes`. They land in
the child's own order, after the attach's settlement, at the fence the sink is
bound to, and that fence is the row's execution owner. A running subagent a
child has not written yet is written once even when its text is unchanged.
Changes the host makes itself (a Stop's acknowledgement, an end) still write
directly and keep the row's owner. The shared child-work evidence type is
unchanged: the sink travels as a separate argument.

Test: a Claude helper the next child reports again while it is still starting
stays working after that start's settlement, through the shipping adapter and
host. It fails on c0add67cca (the row lands `unverifiable`).

src/main/acp/acp-structured-acquire.ts

@@ -190,7 +190,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-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-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/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-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-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' })])
+})

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

src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts

@@ -34,8 +34,6 @@ 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) => {
@@ -69,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)
     }
   })

src/main/native-chat/agent-session-wire/structured-agent-session-resumed-child-owner.test.ts

@@ -64,7 +64,6 @@ it('keeps a resumed subagent and background task working when the run that creat
   const rows = new StructuredAgentSessionSubagentRows({
     journal: () => journal,
     fence: () => fence,
-    producer: () => fence,
     currentWork: (_sessionId, open) => new StructuredAgentSessionCurrentWork(open, fence),
     readChildWork: () => [subagent],
     onWriteFailed: (_sessionId, error) => {
@@ -76,7 +75,7 @@ it('keeps a resumed subagent and background task working when the run that creat
   const run = async () => {
     const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging())
     deferred.bind({ journal, fence, publish: () => {} })
-    rows.changed('session-1')
+    rows.changed('session-1', deferred.sink)
     writeClaudeBackgroundTaskRow(
       deferred.sink,
       tasks,
@@ -87,7 +86,7 @@ it('keeps a resumed subagent and background task working when the run that creat
           taskId: 'task-1',
           kind: 'command',
           label: 'Build',
-          state: 'running'
+          state: 'working'
         },
         lastSerialized: null,
         toolUseId: 'tool-1',
@@ -122,5 +121,5 @@ it('keeps a resumed subagent and background task working when the run that creat
     .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'])
+  expect(blocks.filter(isBackgroundTaskBlock).map((block) => block.state)).toEqual(['working'])
 })

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
 }

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

src/main/native-chat/agent-session-wire/structured-agent-session-subagent-rows.test.ts

@@ -60,7 +60,6 @@ async function rig() {
   const rows = new StructuredAgentSessionSubagentRows({
     journal: () => journal,
     fence: () => 1,
-    producer: () => 1,
     currentWork: (_sessionId, open) => new StructuredAgentSessionCurrentWork(open, 1),
     readChildWork: () => children,
     onWriteFailed: failed

src/main/native-chat/agent-session-wire/structured-agent-session-subagent-rows.ts

@@ -8,8 +8,9 @@
 // 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.
+// 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,
@@ -29,7 +30,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'
+import type { StructuredAgentSessionEventSink } from './structured-agent-session-event-sink'
 
 const UNNAMED = 'subagent'
 const THREAD_GROUP = 'thread'
@@ -38,17 +39,21 @@ const THREAD_GROUP = 'thread'
 export type AgentChildWorkRowView = AgentChildWorkView &
   Pick<AgentChildWorkInput, 'outcomeBasis' | 'residency'>
 
+/** The child-bound write path of the child whose report changed the records. */
+export type StructuredAgentSessionChildWorkProducer = Pick<
+  StructuredAgentSessionEventSink,
+  'appendItem' | 'written'
+>
+
 type Row = {
   block: NativeChatSubagentBlock
   turnScope: AgentJournalTurnScope
-  executionOwner: number | undefined
+  producer: StructuredAgentSessionChildWorkProducer | 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,
@@ -63,18 +68,21 @@ export class StructuredAgentSessionSubagentRows {
   private readonly queued = new Map<string, Map<string, Row>>()
   /** The batch each session's writer has in flight: written, but not yet readable by key. */
   private readonly writing = new Map<string, Map<string, Row>>()
+  /** The running subagents each child has written once: the rows it executes. */
+  private readonly claimed = new WeakMap<StructuredAgentSessionChildWorkProducer, Set<string>>()
 
   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<string, Row>()
-    const executionOwner = this.deps.producer(sessionId)
+    const claimed = producer && (this.claimed.get(producer) ?? new Set<string>())
     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))
@@ -97,19 +105,15 @@ export class StructuredAgentSessionSubagentRows {
         child.parentChildWorkId === undefined ? undefined : providerIds.get(child.parentChildWorkId)
       const block = subagentRowBlock(child, id, parentId, saved, turnScope, running.has(child.id))
       // 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
+      const moved = running.has(child.id) && claimed !== undefined && !claimed.has(id)
       if (!saved || moved || rowText(saved) !== rowText(block)) {
-        queued.set(id, {
-          block,
-          turnScope,
-          executionOwner: executionOwner ?? pending?.executionOwner
-        })
+        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)
@@ -130,15 +134,24 @@ 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), {
-                fence: this.deps.fence(sessionId),
-                turnScope: row.turnScope,
-                ...(row.executionOwner === undefined ? {} : { executionOwner: row.executionOwner })
-              })
-              .catch((error: unknown) => this.deps.onWriteFailed(sessionId, error))
-          )
+          [...rows].map(([id, row]) => {
+            const identity = subagentJournalItemIdentity(id)
+            const body = subagentJournalBody(row.block)
+            if (!row.producer) {
+              return journal
+                .appendItem(identity, body, {
+                  fence: this.deps.fence(sessionId),
+                  turnScope: row.turnScope
+                })
+                .catch((error: unknown) => this.deps.onWriteFailed(sessionId, error))
+            }
+            row.producer.appendItem(identity, body, { turnScope: row.turnScope, executes: true })
+            return row.producer.written?.().then((landed) => {
+              if (!landed.ok) {
+                this.deps.onWriteFailed(sessionId, landed.error)
+              }
+            })
+          })
         )
       }
     } finally {

src/main/runtime/structured-agent-runtime-registrations.ts

@@ -158,8 +158,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,
@@ -197,8 +197,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 } : {}),
@@ -243,8 +243,8 @@ function acpRegistration(spec: AcpLaunchSpec): StructuredAgentRuntimeRegistratio
         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,