From b161aa38b1a5fc35737caad2eba6553826104bb5 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 14 Oct 2026 22:13:19 -0700 Subject: [PATCH] Native chat: a subagent row is written through the child whose report it copies Review of c0add67cca4: 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 c0add67cca4 (the row lands `unverifiable`). --- src/main/acp/acp-structured-acquire.ts | 2 +- src/main/acp/acp-structured-lane.ts | 6 +- .../acp-structured-session-adapter-deps.ts | 8 ++- .../claude-structured-session-adapter.ts | 2 +- .../claude/claude-structured-session-state.ts | 8 ++- .../codex/codex-background-task-tracker.ts | 6 +- .../codex/codex-structured-session-acquire.ts | 2 +- .../codex/codex-structured-session-state.ts | 8 ++- ...-agent-session-claude-subagent-row.test.ts | 45 ++++++++++-- ...tructured-agent-session-client-delivery.ts | 8 ++- ...ructured-agent-session-host-status-feed.ts | 6 +- ...-agent-session-resumed-child-owner.test.ts | 7 +- ...ructured-agent-session-status-feed-deps.ts | 9 ++- .../structured-agent-session-status-feed.ts | 14 ++-- ...ctured-agent-session-subagent-rows.test.ts | 1 - .../structured-agent-session-subagent-rows.ts | 69 +++++++++++-------- .../structured-agent-runtime-registrations.ts | 12 ++-- 17 files changed, 148 insertions(+), 65 deletions(-) diff --git a/src/main/acp/acp-structured-acquire.ts b/src/main/acp/acp-structured-acquire.ts index 98323a272a1..6eabd0618ec 100644 --- a/src/main/acp/acp-structured-acquire.ts +++ b/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) diff --git a/src/main/acp/acp-structured-lane.ts b/src/main/acp/acp-structured-lane.ts index b21c0405cd5..97e8dd17435 100644 --- a/src/main/acp/acp-structured-lane.ts +++ b/src/main/acp/acp-structured-lane.ts @@ -2,6 +2,7 @@ // the shared assembler writes them. Events apply strictly in order; one the sink refuses under // backpressure holds every later one until the sink drains, so nothing is reordered or dropped. +import type { StructuredAgentSessionChildWorkProducer } from '../native-chat/agent-session-wire/structured-agent-session-subagent-rows' import { parseAgentJournalItemKey } from '../../shared/agent-session-journal-item-key' import { createProviderTimelineAssembler, @@ -304,10 +305,11 @@ export class AcpStructuredLane { export function acpLaneChildWorkDelivery( deps: AcpStructuredSessionAdapterDeps, - sessionId: string + sessionId: string, + producer?: StructuredAgentSessionChildWorkProducer ): Pick { return { - onChildWorkEvidence: (evidence) => deps.onChildWorkEvidence?.(sessionId, evidence), + onChildWorkEvidence: (evidence) => deps.onChildWorkEvidence?.(sessionId, evidence, producer), onChildWorkFailure: (error) => deps.logger?.error('publishing ACP child work failed', { scope: 'acp-child-work', diff --git a/src/main/acp/acp-structured-session-adapter-deps.ts b/src/main/acp/acp-structured-session-adapter-deps.ts index ec2802e38f1..0de78c9bf20 100644 --- a/src/main/acp/acp-structured-session-adapter-deps.ts +++ b/src/main/acp/acp-structured-session-adapter-deps.ts @@ -1,3 +1,4 @@ +import type { StructuredAgentSessionChildWorkProducer } from '../native-chat/agent-session-wire/structured-agent-session-subagent-rows' import type { StructuredAgentSessionLifecycleEvent } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' import type { @@ -34,7 +35,12 @@ export type AcpStructuredSessionAdapterDeps = { readProcessStartTime?: (pid: number) => Promise /** Every exit, expected or not: the host ends that child's record. */ onEvent?: (event: StructuredAgentSessionLifecycleEvent) => void - onChildWorkEvidence?: (sessionId: string, evidence: AgentChildWorkEvidence[]) => void + /** `producer`: the reporting child's own sink, which writes what the report changes. */ + onChildWorkEvidence?: ( + sessionId: string, + evidence: AgentChildWorkEvidence[], + producer?: StructuredAgentSessionChildWorkProducer + ) => void /** A send this adapter admitted, settled once the agent answered for it. */ onDispatchSettledLate?: ( settlement: { sessionId: string; clientMessageId: string } & ( diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index bbc1d6fea64..8ac3e3c16a4 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -155,7 +155,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda ): void { const evidence = drainClaudeChildWork(session, message, this.deps.now?.() ?? Date.now()) if (evidence.length > 0) { - this.deps.onChildWorkEvidence?.(sessionId, evidence) + this.deps.onChildWorkEvidence?.(sessionId, evidence, session?.events) } } diff --git a/src/main/claude/claude-structured-session-state.ts b/src/main/claude/claude-structured-session-state.ts index f5d7b67777b..41e7d76c0a7 100644 --- a/src/main/claude/claude-structured-session-state.ts +++ b/src/main/claude/claude-structured-session-state.ts @@ -1,3 +1,4 @@ +import type { StructuredAgentSessionChildWorkProducer } from '../native-chat/agent-session-wire/structured-agent-session-subagent-rows' import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' import type { AgentJournalDispatchRejection } from '../../shared/agent-session-failure-words' import type { SubmissionRejectionFact } from '../../shared/agent-session-failure' @@ -103,7 +104,12 @@ export type ClaudeStructuredSessionAdapterDeps = { /** The CLI reported `session_state_changed idle`: its turn is over. */ onSessionIdle?: (input: { sessionId: string }) => void /** What the session's child work did, delivered after the journal handled the frame. */ - onChildWorkEvidence?: (sessionId: string, evidence: AgentChildWorkEvidence[]) => void + /** `producer`: the reporting child's own sink, which writes what the report changes. */ + onChildWorkEvidence?: ( + sessionId: string, + evidence: AgentChildWorkEvidence[], + producer?: StructuredAgentSessionChildWorkProducer + ) => void openConnection?: typeof openClaudeStreamJsonConnection readProcessStartTime?: (pid: number) => Promise mintLinkId?: () => string diff --git a/src/main/codex/codex-background-task-tracker.ts b/src/main/codex/codex-background-task-tracker.ts index 2ce5cf5a4de..69e0f11e4d5 100644 --- a/src/main/codex/codex-background-task-tracker.ts +++ b/src/main/codex/codex-background-task-tracker.ts @@ -1,3 +1,4 @@ +import type { StructuredAgentSessionChildWorkProducer } from '../native-chat/agent-session-wire/structured-agent-session-subagent-rows' import type { AgentSessionBackgroundTask, AgentSessionBackgroundTaskState @@ -23,10 +24,11 @@ export type CodexChildWorkSink = { export function codexChildWorkSink( sessionId: string, - deps: Pick + deps: Pick, + producer?: StructuredAgentSessionChildWorkProducer ): CodexChildWorkSink { return { - deliver: (evidence) => deps.onChildWorkEvidence?.(sessionId, evidence), + deliver: (evidence) => deps.onChildWorkEvidence?.(sessionId, evidence, producer), now: () => deps.now?.() ?? Date.now() } } diff --git a/src/main/codex/codex-structured-session-acquire.ts b/src/main/codex/codex-structured-session-acquire.ts index 2cbc854e54d..92695c45cee 100644 --- a/src/main/codex/codex-structured-session-acquire.ts +++ b/src/main/codex/codex-structured-session-acquire.ts @@ -224,7 +224,7 @@ export async function acquireCodexStructuredSession(input: { assertCodexConnectionOpen(connection, sessionId) acquisitions.deleteIfCurrent(sessionId, attempt) // Where this session's child work goes: the host's records, after each frame is journaled. - const sink = codexChildWorkSink(sessionId, deps) + const sink = codexChildWorkSink(sessionId, deps, acquireInput.events) const session: CodexSession = { account, connection, diff --git a/src/main/codex/codex-structured-session-state.ts b/src/main/codex/codex-structured-session-state.ts index 0e0560c1ccf..de3f46d53ba 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -1,3 +1,4 @@ +import type { StructuredAgentSessionChildWorkProducer } from '../native-chat/agent-session-wire/structured-agent-session-subagent-rows' import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' import type { AgentSessionAccountKind } from '../../shared/agent-session-availability' import type { @@ -95,7 +96,12 @@ export type CodexStructuredSessionAdapterDeps = { /** Where bookkeeping a close or exit does after the child is gone reports a failure. */ logger?: StructuredAgentSessionLogger /** What the session's child work did, delivered after the journal handled the frame. */ - onChildWorkEvidence?: (sessionId: string, evidence: AgentChildWorkEvidence[]) => void + /** `producer`: the reporting child's own sink, which writes what the report changes. */ + onChildWorkEvidence?: ( + sessionId: string, + evidence: AgentChildWorkEvidence[], + producer?: StructuredAgentSessionChildWorkProducer + ) => void /** A send admitted earlier: its identity once Codex echoes it, or its rejection when the turn * Codex answered it into ended without taking it. */ onDispatchSettledLate?: ( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts index eadb81b9812..e2dd8b3d85d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts @@ -63,8 +63,9 @@ beforeEach(async () => { claudeConfigDir: join(root, 'claude-home'), providerSessionId: PROVIDER_SESSION_ID, resumeLeafUuid: null, - resumesTranscript: false, - continuesChain: false + // A later child resumes the transcript the first one started. + resumesTranscript: (store.getRecord(SESSION)?.providerHandleChain.length ?? 0) > 0, + continuesChain: (store.getRecord(SESSION)?.providerHandleChain.length ?? 0) > 0 }), onEvent: (event) => { const mapped = structuredClaudeLifecycleEvent(event) @@ -75,8 +76,8 @@ beforeEach(async () => { onDispatchSettledLate: (settlement) => void host.settleLateDispatch(settlement), persistHandle: async () => undefined, logger: log.logger, - onChildWorkEvidence: (sessionId, evidence) => - host.publishChildWorkEvidence(sessionId, evidence), + onChildWorkEvidence: (sessionId, evidence, producer) => + host.publishChildWorkEvidence(sessionId, evidence, producer), openConnection: claude.openConnection, readProcessStartTime: async () => 1_700_000_000_000, now: () => clock @@ -298,3 +299,39 @@ it('settles a helper still running when its session closes as unverifiable, clai ) expect((await savedRows())[0]).not.toHaveProperty('outcomeBasis') }) + +it('keeps a helper the next child resumes while it is still starting working, past that start’s settlement', async () => { + const first = claude.connections[0]! + await sendAndEcho(first, 'Run it in the background.') + for (const [index, captured] of RESUMED_BY_MESSAGE.filter(({ at }) => at <= 4_765).entries()) { + frame(first, { uuid: `launch-${index}`, ...captured.frame }) + } + await eventually(async () => + expect(await savedRows()).toEqual([expect.objectContaining({ state: 'working' })]) + ) + await adapter.closeSession(SESSION) + await eventually(async () => + expect(await savedRows()).toEqual([expect.objectContaining({ state: 'unverifiable' })]) + ) + // The next child reports the helper running again before its start returns, so before the host + // installs it and settles what the first child left. + const opened = claude.connections.push.bind(claude.connections) + vi.spyOn(claude.connections, 'push').mockImplementation((connection) => { + const initialize = connection.initializationResult + connection.initializationResult = async () => { + const initialized = await initialize() + for (const [index, captured] of RESUMED_BY_MESSAGE.filter( + ({ at }) => at >= 16_142 && at <= 16_143 + ).entries()) { + frame(connection, { uuid: `resume-${index}`, ...captured.frame }) + } + return initialized + } + return opened(connection) + }) + const body = hostTestMessage('Carry on.') + await host.send(CALLER, { envelope: envelope('agentSession.send', { body }), body }) + await new Promise((resolve) => setTimeout(resolve, 50)) + expect(claude.connections).toHaveLength(2) + expect(await savedRows()).toEqual([expect.objectContaining({ id: 'agent-1', state: 'working' })]) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts index 3721a7e1d27..2d6117537b9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts @@ -19,6 +19,7 @@ import { heldStructuredAgentSessionCurrentWork } from './structured-agent-sessio import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' import type { StructuredAgentSessionStatusSubscriber } from './structured-agent-session-status-feed' import { createStructuredAgentSessionHostStatusFeed } from './structured-agent-session-host-status-feed' +import type { StructuredAgentSessionChildWorkProducer } from './structured-agent-session-subagent-rows' import { StructuredAgentSessionTurnCompletionFeed, type StructuredAgentSessionTurnCompletionSubscriber @@ -102,8 +103,11 @@ export class StructuredAgentSessionClientDelivery { publishConversationName = (sessionId: string): void => this.statusFeed.publishConversationName(sessionId) - publishChildWork = (sessionId: string, evidence: AgentChildWorkEvidence[]): void => - this.statusFeed.publishChildWork(sessionId, evidence) + publishChildWork = ( + sessionId: string, + evidence: AgentChildWorkEvidence[], + producer?: StructuredAgentSessionChildWorkProducer + ): void => this.statusFeed.publishChildWork(sessionId, evidence, producer) /** What the feed publishes as `stopping`: only a working session is still being stopped. */ readStopping = (sessionId: string): boolean => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts index 0de2c90c0ec..d9961ef266d 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,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) } }) 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 index 09444b8541d..a58ba128316 100644 --- 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 @@ -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']) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts index 108d43f1b64..9d658678754 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts @@ -8,6 +8,7 @@ import type { StructuredAgentSessionProviderChild } from './structured-agent-session-host-types' import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' +import type { StructuredAgentSessionChildWorkProducer } from './structured-agent-session-subagent-rows' import type { StructuredAgentSessionStatusObserverOptions } from './structured-agent-session-status-observation' import type { StructuredAgentSessionStatusSink } from './structured-agent-session-status-ownership' @@ -38,8 +39,12 @@ export type StructuredAgentSessionStatusFeedDeps = { /** Resolved on every call: the host builds this feed in a field initializer, before its own * deps are assigned. The sink holds the session's child records; the summary reads them there. */ statusSink?: () => StructuredAgentSessionStatusSink | undefined - /** The session's child records changed, so every other reader of them republishes. */ - onChildWorkChanged?: (sessionId: string) => void + /** The session's child records changed, so every other reader of them republishes; `producer` + * is the write path of the child whose report changed them. */ + onChildWorkChanged?: ( + sessionId: string, + producer?: StructuredAgentSessionChildWorkProducer + ) => void /** The session's agent proved a start: its row's phase became `ready`. */ onAgentStarted?: (sessionId: string) => void } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts index f100d926329..80a9b9b1bf4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts @@ -30,6 +30,7 @@ import { type StructuredAgentSessionJournalProjection } from './structured-agent-session-status-journal-projection' import { structuredStatusSummariesEqual } from './structured-agent-session-status-summary-equality' +import type { StructuredAgentSessionChildWorkProducer } from './structured-agent-session-subagent-rows' import { StructuredAgentSessionStatusOwnership } from './structured-agent-session-status-ownership' export type { StructuredAgentSessionStatusSink } from './structured-agent-session-status-ownership' @@ -284,9 +285,13 @@ export class StructuredAgentSessionStatusFeed { /** Child-work evidence for a session this feed publishes; a failing sink costs nothing else. * The summary then re-reads the records, which is also what re-folds the parent's row. */ - publishChildWork(sessionId: string, evidence: AgentChildWorkEvidence[]): void { + publishChildWork( + sessionId: string, + evidence: AgentChildWorkEvidence[], + producer?: StructuredAgentSessionChildWorkProducer + ): void { const session = this.deps.sessions.get(sessionId) - if (!session || !this.admitChildWork(sessionId, session, evidence)) { + if (!session || !this.admitChildWork(sessionId, session, evidence, producer)) { return } this.publish(sessionId) @@ -314,7 +319,8 @@ export class StructuredAgentSessionStatusFeed { private admitChildWork( sessionId: string, session: StatusFeedSession, - evidence: AgentChildWorkEvidence[] + evidence: AgentChildWorkEvidence[], + producer: StructuredAgentSessionChildWorkProducer | 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) } 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 d345f6b1c41..42395b2b3be 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,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 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 18cfc2317f6..21b1b8c53d0 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 @@ -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 +/** 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>() /** The batch each session's writer has in flight: written, but not yet readable by key. */ private readonly writing = new Map>() + /** The running subagents each child has written once: the rows it executes. */ + private readonly claimed = new WeakMap>() constructor(private readonly deps: StructuredAgentSessionSubagentRowsDeps) {} - /** The session's child records changed: queue each subagent whose row says otherwise. */ - changed(sessionId: string): void { + /** The session's child records changed, by `producer`'s report when a child made one: queue each + * subagent whose row says otherwise, and each running one that child has not written yet. */ + changed(sessionId: string, producer?: StructuredAgentSessionChildWorkProducer): void { const journal = this.deps.journal(sessionId) const children = this.deps.readChildWork(sessionId) if (!journal || !children?.some((child) => child.kind === 'agent')) { return } const queued = this.queued.get(sessionId) ?? new Map() - const executionOwner = this.deps.producer(sessionId) + const claimed = producer && (this.claimed.get(producer) ?? new Set()) const providerIds = new Map(children.map((child) => [child.id, child.providerId])) // The strip's rule: a child that finished while work it started runs is still running. const running = new Set(structuredRunningChildWork(children).map((child) => child.id)) @@ -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 { diff --git a/src/main/runtime/structured-agent-runtime-registrations.ts b/src/main/runtime/structured-agent-runtime-registrations.ts index a062e762094..7dbda0b95cc 100644 --- a/src/main/runtime/structured-agent-runtime-registrations.ts +++ b/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,