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,