diff --git a/config/tsconfig.node.json b/config/tsconfig.node.json index ff4e9a762764..7d4d951d2d62 100644 --- a/config/tsconfig.node.json +++ b/config/tsconfig.node.json @@ -10,6 +10,7 @@ "../src/renderer/src/lib/skill-freshness-display-status.ts", "../src/renderer/src/components/native-chat/native-chat-resolution-receipt.ts", "../src/renderer/src/components/native-chat/structured-agent-question-projection.ts", + "../src/renderer/src/components/native-chat/native-chat-subagent-delegation.ts", "../src/preload/**/*", "../src/shared/**/*", "../src/relay/**/*", diff --git a/src/main/codex/codex-background-task-frames.ts b/src/main/codex/codex-background-task-frames.ts index f01863956b6c..06536a857b8c 100644 --- a/src/main/codex/codex-background-task-frames.ts +++ b/src/main/codex/codex-background-task-frames.ts @@ -1,22 +1,23 @@ import type { NativeChatSubagentState } from '../../shared/native-chat-types' -import { - codexSubagentLabel, - isCodexRootAgentActivity, - readCodexSubagentActivity -} from './codex-subagent-activity' +import { readCodexSubagentAnnouncements } from './codex-subagent-activity' import { codexChildTurnState } from './codex-subagent-executions' import { readRecord } from './codex-item-field-readers' import { readCodexThreadItem } from './codex-structured-item-translation' import { readCodexTurnId } from './codex-structured-thread-facts' +export type CodexAnnouncedChild = { + agentThreadId: string + label: string | null + parentTurnId: string | null | undefined + /** The reporting thread, for a spawn: the agent that spawned the child. */ + spawnerThreadId: string | undefined +} + export type CodexBackgroundTaskFrame = | { - kind: 'subagent' - agentThreadId: string - label: string | null - parentTurnId: string | null | undefined - /** The reporting thread, for a `started` activity: the agent that spawned the child. */ - spawnerThreadId: string | undefined + /** An item naming children, several for a call on more than one helper. */ + kind: 'subagents' + children: CodexAnnouncedChild[] } | { kind: 'turn' @@ -78,23 +79,14 @@ export function readCodexBackgroundTaskFrame( return null } const item = readCodexThreadItem(readRecord(event.params).item) - const activity = item && readCodexSubagentActivity(item) - if ( - !activity || - activity.agentThreadId === primaryThreadId || - isCodexRootAgentActivity(activity) - ) { - return null - } - return { - kind: 'subagent', - agentThreadId: activity.agentThreadId, - label: codexSubagentLabel(activity), - parentTurnId: - activity.kind === 'started' || activity.kind === 'interacted' - ? readCodexTurnId(event.params) - : undefined, - // Only `started` names the spawner: other kinds ride whichever agent acted. - spawnerThreadId: activity.kind === 'started' ? event.threadId : undefined - } + const children = (item ? readCodexSubagentAnnouncements(item, primaryThreadId) : []).map( + (announcement) => ({ + agentThreadId: announcement.agentThreadId, + label: announcement.label, + parentTurnId: announcement.namesParentTurn ? readCodexTurnId(event.params) : undefined, + // Only a spawn names the spawner: other announcements ride whichever agent acted. + spawnerThreadId: announcement.spawned ? event.threadId : undefined + }) + ) + return children.length > 0 ? { kind: 'subagents', children } : null } diff --git a/src/main/codex/codex-background-task-tracker.test.ts b/src/main/codex/codex-background-task-tracker.test.ts index 0a70442a0b37..6c32896140b3 100644 --- a/src/main/codex/codex-background-task-tracker.test.ts +++ b/src/main/codex/codex-background-task-tracker.test.ts @@ -76,10 +76,8 @@ function command( describe('readCodexBackgroundTaskFrame', () => { it('reads activity as child metadata without inferring execution state', () => { expect(readCodexBackgroundTaskFrame(activity('interacted'), PRIMARY)).toEqual({ - kind: 'subagent', - agentThreadId: CHILD, - label: 'count_a', - parentTurnId: PARENT_TURN + kind: 'subagents', + children: [{ agentThreadId: CHILD, label: 'count_a', parentTurnId: PARENT_TURN }] }) }) diff --git a/src/main/codex/codex-background-task-tracker.ts b/src/main/codex/codex-background-task-tracker.ts index 6429a84d4fda..9fd2880dc75e 100644 --- a/src/main/codex/codex-background-task-tracker.ts +++ b/src/main/codex/codex-background-task-tracker.ts @@ -73,13 +73,15 @@ export class CodexBackgroundTaskTracker { : []) ] const frame = readCodexBackgroundTaskFrame(event, this.primaryThreadId) - if (frame?.kind === 'subagent') { - this.executions.register( - frame.agentThreadId, - frame.label, - frame.parentTurnId, - frame.spawnerThreadId - ) + if (frame?.kind === 'subagents') { + for (const child of frame.children) { + this.executions.register( + child.agentThreadId, + child.label, + child.parentTurnId, + child.spawnerThreadId + ) + } } else if (frame?.kind === 'thread-closed') { this.executions.closeThread(frame.threadId) } else if (frame && frame.threadId !== this.primaryThreadId) { diff --git a/src/main/codex/codex-child-work-evidence.ts b/src/main/codex/codex-child-work-evidence.ts index 93bda7a54a1f..af946afeedc0 100644 --- a/src/main/codex/codex-child-work-evidence.ts +++ b/src/main/codex/codex-child-work-evidence.ts @@ -100,15 +100,13 @@ export class CodexChildWorkEvidence { commands: readonly CodexBackgroundCommandChange[] ): void { this.queueCommands(commands) - const threadId = this.childThread(event, frame) - if (threadId === null) { - return - } - const facts = this.factsFor(threadId) - if (facts && event.threadId === threadId) { - this.record(facts, event) + for (const threadId of this.childThreads(event, frame)) { + const facts = this.factsFor(threadId) + if (facts && event.threadId === threadId) { + this.record(facts, event) + } + this.queueChild(threadId) } - this.queueChild(threadId) } /** The provider session is gone, with the commands it ended: no child it still ran can report @@ -125,17 +123,17 @@ export class CodexChildWorkEvidence { return pending.map((edge) => edge(observedAt)) } - private childThread( + private childThreads( event: CodexBackgroundTaskEvent, frame: CodexBackgroundTaskFrame | null - ): string | null { - const threadId = - frame?.kind === 'subagent' - ? frame.agentThreadId + ): string[] { + const threadIds = + frame?.kind === 'subagents' + ? frame.children.map((child) => child.agentThreadId) : frame || CHILD_FRAME_METHODS.has(event.method) - ? event.threadId - : null - return threadId === this.primaryThreadId ? null : threadId + ? [event.threadId] + : [] + return threadIds.filter((threadId) => threadId !== this.primaryThreadId) } /** A command belongs to the child thread that launched it; the session's own agent is no owner. diff --git a/src/main/codex/codex-collab-agent-item-translation.ts b/src/main/codex/codex-collab-agent-item-translation.ts new file mode 100644 index 000000000000..f12d22b7339d --- /dev/null +++ b/src/main/codex/codex-collab-agent-item-translation.ts @@ -0,0 +1,154 @@ +// A Codex `collabAgentToolCall` item → a tool row that names the helper it acted on. +// +// The row is the call the agent made (`spawn_agent`, `wait_agent`, `close_agent`, …); the helper +// itself is the subagent roster's row. A helper is named the way the roster names it, so the two +// rows read as the same child. + +import type { AgentJournalItemBody } from '../../shared/agent-session-journal-types' +import { CODEX_COLLAB_ROW_AGENTS_KEY } from '../../shared/codex-collab-agent-tools' +import { + boundInlineText, + boundToolInput, + DEFAULT_JOURNAL_PAYLOAD_LIMITS +} from '../native-chat/agent-session-journal/journal-payload-bounds' +import { + codexCollabHelperLabel, + codexCollabToolName, + readCodexCollabAgentToolCall, + type CodexCollabAgentToolCall +} from './codex-collab-agent-tool-call' +import { readString } from './codex-item-field-readers' +import { codexItemRunState } from './codex-item-run-state' +import type { CodexThreadItem } from './codex-thread-item-identity' + +/** The roster's name for a helper thread, or null for one it holds no name for. */ +export type CodexHelperName = (threadId: string) => string | null + +/** Who the call acted on. A spawn names its helper by its prompt until the roster holds the + * thread it became; a helper with no name (its spawn was never seen) is named by its thread id. */ +function helperNames(call: CodexCollabAgentToolCall, helperName?: CodexHelperName): string { + if (call.tool === 'spawnAgent') { + const spawned = call.receiverThreadIds[0] + return (spawned && helperName?.(spawned)) || codexCollabHelperLabel(call.prompt) || '' + } + return call.receiverThreadIds.map((threadId) => helperName?.(threadId) ?? threadId).join(', ') +} + +/** Codex's own words for a helper's `CollabAgentStatus`, as its client shows them. */ +const HELPER_STATUS_TEXT = new Map([ + ['pendingInit', 'Pending init'], + ['running', 'Running'], + ['interrupted', 'Interrupted'], + ['completed', 'Completed'], + ['shutdown', 'Shutdown'], + ['notFound', 'Not found'] +]) + +const CALL_STATUS_TEXT = new Map([ + ['completed', 'Completed'], + ['failed', 'Failed'], + ['interrupted', 'Interrupted'] +]) + +type HelperState = CodexCollabAgentToolCall['states'][number] + +/** A helper's status summarised as Codex's own client does: a completed helper with its last + * reply, an errored one with its error. */ +function statusSummary(state: HelperState): string | null { + if (state.status === 'errored') { + return `Error - ${state.message ?? 'Agent errored'}` + } + if (state.status === 'completed' && state.message) { + return `Completed - ${state.message}` + } + return state.status === null ? null : (HELPER_STATUS_TEXT.get(state.status) ?? state.status) +} + +/** A finished call's output, worded as Codex's own client words the call. Every finished call has + * one: a client that pairs results by position (one predating result call ids) would otherwise + * draw each later output in the run under the call before its own. The row label already names + * the helper, so the output does not. */ +function outputText(call: CodexCollabAgentToolCall, helperName?: CodexHelperName): string | null { + if (call.status === null || call.status === 'inProgress') { + return null + } + // The status a spawn or close reports predates what it did (a closed helper reads `running`), + // so these say what the call did. + switch (call.tool) { + case 'spawnAgent': + return call.receiverThreadIds.length > 0 ? 'Spawned' : 'Agent spawn failed' + case 'sendInput': + return 'Sent input' + case 'closeAgent': + return 'Closed' + case 'resumeAgent': + return (call.states[0] && statusSummary(call.states[0])) ?? 'Error - Agent resume failed' + } + const reports = call.states.flatMap((state) => { + // A wait returns what each finished helper said. + const text = + call.tool === 'wait' && state.status === 'completed' && state.message + ? state.message + : statusSummary(state) + return text === null ? [] : [{ threadId: state.threadId, text }] + }) + if (reports.length === 0) { + // A wait's end names only helpers that finished (none when it timed out; v2's never names any). + return call.tool === 'wait' + ? 'Finished waiting' + : (CALL_STATUS_TEXT.get(call.status) ?? call.status) + } + if (reports.length === 1 && call.receiverThreadIds.length === 1) { + return reports[0].text + } + return reports + .map(({ threadId, text }) => `${helperName?.(threadId) ?? threadId}: ${text}`) + .join('\n') +} + +/** The prompt is clipped on its own: bounded with the rest, a long one would clip away the + * helper's name and ids the row is read by. */ +function collabRowInput(fields: Record, prompt: string | null): unknown { + const bounded = boundToolInput(fields, DEFAULT_JOURNAL_PAYLOAD_LIMITS) + if (bounded !== fields) { + return bounded + } + const input = prompt + ? { ...fields, prompt: boundInlineText(prompt, DEFAULT_JOURNAL_PAYLOAD_LIMITS).text } + : fields + return Object.keys(input).length > 0 ? input : null +} + +/** `started` is the call's started item, when the caller still holds it. */ +export function codexCollabAgentToolCallBody( + item: CodexThreadItem, + helperName?: CodexHelperName, + started?: CodexThreadItem +): AgentJournalItemBody | null { + const call = readCodexCollabAgentToolCall(item, started) + if (!call) { + return null + } + const description = helperNames(call, helperName) + const model = readString(item, 'model') + const reasoningEffort = readString(item, 'reasoningEffort') + const fields = { + // `description` is the key the row label reads, so the helper's name leads the row. + ...(description ? { description } : {}), + ...(model ? { model } : {}), + ...(reasoningEffort ? { reasoningEffort } : {}), + ...(call.receiverThreadIds.length > 0 + ? { [CODEX_COLLAB_ROW_AGENTS_KEY]: call.receiverThreadIds } + : {}) + } + const text = outputText(call, helperName) + const output = text === null ? null : boundInlineText(text, DEFAULT_JOURNAL_PAYLOAD_LIMITS) + return { + kind: 'tool-call', + name: codexCollabToolName(call), + callId: call.id, + input: collabRowInput(fields, call.prompt), + state: codexItemRunState(item), + ...(output === null ? {} : { output: output.bounded }) + } +} diff --git a/src/main/codex/codex-collab-agent-tool-call.ts b/src/main/codex/codex-collab-agent-tool-call.ts new file mode 100644 index 000000000000..01809a7fb51d --- /dev/null +++ b/src/main/codex/codex-collab-agent-tool-call.ts @@ -0,0 +1,102 @@ +// Reading Codex's `collabAgentToolCall` items: the calls an agent makes to spawn, message, wait on +// and close its helpers. +// +// Codex's default multi-agent mode reports a helper ONLY this way; it emits no `subAgentActivity`. +// Shapes are from the app-server's generated schema (0.155) and a live default-mode session: +// * `spawnAgent` starts with `receiverThreadIds: []`. The helper's thread id first appears when +// the call ends, beside the `prompt` it was given — even when the call ends `failed`. +// * Every other call names its helpers from its start. A finished call reports a thread id it +// does not know as `notFound`; any other state says the helper exists. +// * Codex core also knows each receiver's nickname and role, but the app-server item does not +// carry them yet; `readCodexSubagentAnnouncements` is where they would be adopted. +// * The item names no nickname or task path, so the prompt is the only text that tells one +// helper from another. +// * A wait that times out ends naming no receiver and no state. +// * `agentsStates` is the caller's last-known snapshot of each receiver. The helper's own turn +// frames own its execution, so nothing here reads it as execution state: only `notFound`, +// which says the receiver is no helper, and the snapshot a wait or resume reports, which its +// row shows as output. A spawn's or close's predates what the call did, so its row does not. + +import { codexCollabToolRowName } from '../../shared/codex-collab-agent-tools' +import { collapsedToolInputPrefix } from '../../shared/native-chat-tool-preview-prefix' +import { readRecord, readString } from './codex-item-field-readers' +import type { CodexThreadItem } from './codex-thread-item-identity' + +export const CODEX_COLLAB_AGENT_TOOL_CALL_ITEM_TYPE = 'collabAgentToolCall' + +/** Row length for a helper named by its prompt: long enough to tell two helpers apart. */ +const MAX_HELPER_LABEL_CHARS = 80 + +export type CodexCollabAgentToolCall = { + id: string + /** The schema's `CollabAgentTool`: `spawnAgent`, `sendInput`, `wait`, `closeAgent`, … */ + tool: string + /** `inProgress`, `completed`, `failed` or `interrupted`. */ + status: string | null + receiverThreadIds: string[] + /** The receivers that are helpers: every one but those the call reports `notFound`. */ + helperThreadIds: string[] + prompt: string | null + /** Each receiver's reported snapshot, in `receiverThreadIds` order: its `CollabAgentStatus` + * (`pendingInit`, `running`, `completed`, `errored`, `shutdown`, …) and its message. */ + states: { threadId: string; status: string | null; message: string | null }[] +} + +function readReceiverThreadIds(item: CodexThreadItem | undefined): string[] { + return Array.isArray(item?.receiverThreadIds) + ? item.receiverThreadIds.filter((id): id is string => typeof id === 'string' && id.length > 0) + : [] +} + +/** `started` is the same call's started item, when it was seen: a call keeps naming the helpers + * it started on even when it ends naming fewer, as a timed-out wait ends naming none. */ +export function readCodexCollabAgentToolCall( + item: CodexThreadItem, + started?: CodexThreadItem +): CodexCollabAgentToolCall | null { + const tool = readString(item, 'tool') + if (item.type !== CODEX_COLLAB_AGENT_TOOL_CALL_ITEM_TYPE || tool === null) { + return null + } + const receiverThreadIds = [ + ...new Set([...readReceiverThreadIds(started), ...readReceiverThreadIds(item)]) + ] + const snapshots = readRecord(item.agentsStates) + const states = receiverThreadIds.flatMap((threadId) => { + if (!Object.hasOwn(snapshots, threadId)) { + return [] + } + const snapshot = readRecord(snapshots[threadId]) + return [ + { threadId, status: readString(snapshot, 'status'), message: readString(snapshot, 'message') } + ] + }) + return { + id: item.id, + tool, + status: readString(item, 'status'), + receiverThreadIds, + helperThreadIds: receiverThreadIds.filter( + (threadId) => readString(readRecord(snapshots[threadId]), 'status') !== 'notFound' + ), + prompt: readString(item, 'prompt'), + states + } +} + +/** A helper's row label: the head of the prompt it was spawned with, on one line. */ +export function codexCollabHelperLabel(prompt: string | null): string | null { + const collapsed = prompt === null ? '' : collapsedToolInputPrefix(prompt) + if (collapsed.length <= MAX_HELPER_LABEL_CHARS) { + return collapsed.length > 0 ? collapsed : null + } + const keep = MAX_HELPER_LABEL_CHARS - 1 + // Never end on half a surrogate pair: the label lands in a durable row. + const last = collapsed.charCodeAt(keep - 1) + const end = last >= 0xd800 && last <= 0xdbff ? keep - 1 : keep + return `${collapsed.slice(0, end)}…` +} + +export function codexCollabToolName(call: CodexCollabAgentToolCall): string { + return codexCollabToolRowName(call.tool) +} diff --git a/src/main/codex/codex-collab-call-delegation.test.ts b/src/main/codex/codex-collab-call-delegation.test.ts new file mode 100644 index 000000000000..96131248ddc6 --- /dev/null +++ b/src/main/codex/codex-collab-call-delegation.test.ts @@ -0,0 +1,114 @@ +// A running chat holds a subagent's section open while the parent's newest row delegates to it. +// In Codex's default multi-agent mode that row is the collab call the host writes as a tool row, +// so the rows the real adapter publishes must read as a delegation to the helper they name. + +import { describe, expect, it } from 'vitest' +import { codexCollabRowAgentIds } from '../../shared/codex-collab-agent-tools' +import { projectNativeChatTranscript } from '../../shared/native-chat-transcript-projection' +import type { NativeChatMessage } from '../../shared/native-chat-types' +import { projectStructuredAgentSessionMessages } from '../../shared/structured-agent-session-message-projection' +import { + nativeChatSubagentDelegation, + type NativeChatSubagentDelegation +} from '../../renderer/src/components/native-chat/native-chat-subagent-delegation' +import { + collab, + HELPER, + HELPER_TURN, + PARENT_TURN, + publishedRows, + shell, + spawn, + turn, + waitStarted, + type Frame +} from './codex-collab-call-frames-fixture' +import { THREAD_ID } from './codex-structured-session-adapter-fixture' + +/** The delegation the parent's newest tool run reads as, the row a running chat's frontier judges. */ +async function newestRunDelegation(frames: Frame[]): Promise { + const { conversation } = projectNativeChatTranscript( + projectStructuredAgentSessionMessages(await publishedRows(frames), [], []) + ) + const runs = conversation.filter((message: NativeChatMessage) => + message.blocks.some((block) => block.type === 'tool-call') + ) + const newest = runs.at(-1) + return newest ? nativeChatSubagentDelegation(newest) : null +} + +const spawned = [ + turn('turn/started', THREAD_ID, PARENT_TURN), + spawn('item/started'), + spawn('item/completed'), + turn('turn/started', HELPER, HELPER_TURN), + shell('item/started', HELPER, HELPER_TURN, 'call-helper-shell', 'CHILD_DONE') +] + +describe("a Codex default-mode collab call row as the parent's delegation", () => { + it('delegates to the helper a finished spawn names', async () => { + expect(await newestRunDelegation(spawned)).toEqual({ kind: 'call', agentId: HELPER }) + }) + + it('delegates to the helper a running wait names', async () => { + expect(await newestRunDelegation([...spawned, waitStarted])).toEqual({ + kind: 'call', + agentId: HELPER + }) + }) + + it('delegates to the helper a close names, as a raw collab row did', async () => { + const close = collab('item/started', { + id: 'call-close', + tool: 'closeAgent', + status: 'inProgress', + receiverThreadIds: [HELPER] + }) + expect(await newestRunDelegation([...spawned, close])).toEqual({ + kind: 'call', + agentId: HELPER + }) + }) + + it('reads a spawn still starting, which names no helper, as ordinary output', async () => { + expect( + await newestRunDelegation([ + turn('turn/started', THREAD_ID, PARENT_TURN), + spawn('item/started') + ]) + ).toBeNull() + }) + + it("reads the parent's next call as superseding the delegation", async () => { + expect( + await newestRunDelegation([ + ...spawned, + shell('item/started', THREAD_ID, PARENT_TURN, 'call-parent-shell', 'PARENT_DONE') + ]) + ).toBeNull() + }) + + it('keeps naming the helper when its prompt is past the journal limit', async () => { + const prompt = `Review lane a. ${'x'.repeat(20_000)}` + const longSpawn = [ + turn('turn/started', THREAD_ID, PARENT_TURN), + collab('item/completed', { + id: 'call-spawn', + tool: 'spawnAgent', + status: 'completed', + receiverThreadIds: [HELPER], + prompt + }) + ] + const row = (await publishedRows(longSpawn)) + .map(({ body }) => body) + .find((body) => body.kind === 'tool-call' && body.name === 'spawn_agent') + const input = row?.kind === 'tool-call' ? row.input : undefined + expect(input).toMatchObject({ + description: expect.stringMatching(/^Review lane a\. x+…$/), + prompt: expect.stringContaining('[Orca: output truncated') + }) + expect(codexCollabRowAgentIds(input)).toEqual([HELPER]) + expect(await newestRunDelegation(longSpawn)).toEqual({ kind: 'call', agentId: HELPER }) + }) +}) diff --git a/src/main/codex/codex-collab-call-frames-fixture.ts b/src/main/codex/codex-collab-call-frames-fixture.ts new file mode 100644 index 000000000000..8926e46dcca9 --- /dev/null +++ b/src/main/codex/codex-collab-call-frames-fixture.ts @@ -0,0 +1,127 @@ +// Default-mode collab frames as a live session sent them, and the journal rows the real adapter +// publishes for them: what a client reads. + +import type { AgentJournalRenderItem } from '../../shared/agent-session-journal-types' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { fakeCodex, identityFor, THREAD_ID } from './codex-structured-session-adapter-fixture' +import { CodexStructuredSessionAdapter } from './codex-structured-session-adapter' + +// Shapes as a live default-mode session sent them (codex-cli 0.157); only the sender is remapped. +export const HELPER = '01a0ea72-80bc-7632-a97f-4a9a0d50d08f' +export const PARENT_TURN = 'parent-turn-1' +export const HELPER_TURN = 'helper-turn-1' +export const PROMPT = 'Run exactly one foreground shell command: `sleep 150; echo LATE`.' + +export type Frame = { method: string; params: Record } + +export const turn = ( + method: 'turn/started' | 'turn/completed', + threadId: string, + id: string +): Frame => ({ + method, + params: { threadId, turn: { id, status: 'completed' } } +}) +export const item = ( + method: 'item/started' | 'item/completed', + threadId: string, + turnId: string, + fields: Record +): Frame => ({ method, params: { threadId, turnId, item: fields } }) +export const collab = ( + method: 'item/started' | 'item/completed', + fields: Record +): Frame => + item(method, THREAD_ID, PARENT_TURN, { + type: 'collabAgentToolCall', + senderThreadId: THREAD_ID, + prompt: null, + model: null, + reasoningEffort: null, + agentsStates: {}, + ...fields + }) +export const shell = ( + method: 'item/started' | 'item/completed', + threadId: string, + turnId: string, + id: string, + output: string +): Frame => + item(method, threadId, turnId, { + type: 'commandExecution', + id, + command: `/bin/zsh -lc 'echo ${output}'`, + cwd: '/work/repo', + status: method === 'item/started' ? 'inProgress' : 'completed', + ...(method === 'item/completed' ? { exitCode: 0, aggregatedOutput: `${output}\n` } : {}) + }) + +export const spawn = (method: 'item/started' | 'item/completed'): Frame => + collab(method, { + id: 'call-spawn', + tool: 'spawnAgent', + status: method === 'item/started' ? 'inProgress' : 'completed', + receiverThreadIds: method === 'item/started' ? [] : [HELPER], + prompt: PROMPT, + model: 'gpt-5.5', + reasoningEffort: 'medium', + ...(method === 'item/completed' + ? { agentsStates: { [HELPER]: { status: 'pendingInit', message: null } } } + : {}) + }) +export const waitStarted = collab('item/started', { + id: 'call-wait', + tool: 'wait', + status: 'inProgress', + receiverThreadIds: [HELPER] +}) + +export async function publishedRows(frames: Frame[]): Promise { + const codex = fakeCodex() + const adapter = new CodexStructuredSessionAdapter({ + resolveLaunch: async () => ({ + command: 'codex', + args: ['app-server'], + cwd: '/work/repo', + codexHome: null, + resumeThreadId: null + }), + openConnection: codex.openConnection, + readProcessStartTime: async () => 1_700_000_000_000, + now: () => 1_700_000_000_500 + }) + // The latest revision of each row, in first-written order, as a client's journal holds it. + const rows = new Map() + const journal: StructuredAgentSessionEventSink = { + appendItem: (identity, body, options) => { + const itemId = JSON.stringify(identity) + const previous = rows.get(itemId) + const sequence = previous?.sequence ?? rows.size + 1 + const { agentId, parentAgentId, producerKind } = options + rows.set(itemId, { + ...previous, + ...(agentId === undefined ? {} : { agentId }), + ...(parentAgentId === undefined ? {} : { parentAgentId }), + ...(producerKind === undefined ? {} : { producerKind }), + itemId, + sequence, + revision: (previous?.revision ?? 0) + 1, + observedAt: sequence, + body + }) + }, + appendTombstone: () => {}, + publish: () => {} + } + await adapter.acquire({ + identity: identityFor('session-1'), + fence: 7, + spawnToken: 'spawn-9', + events: journal + }) + for (const frame of frames) { + codex.connections[0]!.handlers.onNotification?.(frame.method, frame.params) + } + return [...rows.values()] +} diff --git a/src/main/codex/codex-collab-call-rows-on-positional-clients.test.ts b/src/main/codex/codex-collab-call-rows-on-positional-clients.test.ts new file mode 100644 index 000000000000..795716ed553c --- /dev/null +++ b/src/main/codex/codex-collab-call-rows-on-positional-clients.test.ts @@ -0,0 +1,247 @@ +// A client that predates result call ids (an older mobile app, or an older desktop reading a newer +// host) projects the journal itself and pairs each tool result with the oldest unanswered call. +// So every finished collab call row must carry an output, or each later output in its run is drawn +// under the call before its own. + +import { describe, expect, it } from 'vitest' +import type { + AgentJournalItemBody, + AgentJournalRenderItem +} from '../../shared/agent-session-journal-types' +import { pairToolBlocks } from '../../shared/native-chat-tool-fold' +import { pairNativeChatToolResults } from '../../shared/native-chat-tool-pairing' +import { projectNativeChatTranscript } from '../../shared/native-chat-transcript-projection' +import type { NativeChatBlock, NativeChatMessage } from '../../shared/native-chat-types' +import { projectStructuredAgentSessionMessages } from '../../shared/structured-agent-session-message-projection' +import { + collab, + HELPER, + HELPER_TURN, + item, + PARENT_TURN, + PROMPT, + publishedRows, + shell, + spawn, + turn, + waitStarted, + type Frame +} from './codex-collab-call-frames-fixture' +import { THREAD_ID } from './codex-structured-session-adapter-fixture' + +type PairedOutput = [call: string, output: string | undefined] + +/** What a client without result call ids draws: each call with the output its pairing gives it. */ +function positionalRuns(rows: AgentJournalRenderItem[]): { + mobile: PairedOutput[][] + desktop: PairedOutput[][] +} { + const withoutCallIds = (message: NativeChatMessage): NativeChatMessage => ({ + ...message, + blocks: message.blocks.map((block): NativeChatBlock => { + if (block.type !== 'tool-result') { + return block + } + const { callId: _callId, ...legacy } = block + return legacy + }) + }) + const transcript = projectNativeChatTranscript( + projectStructuredAgentSessionMessages(rows, [], []).map(withoutCallIds) + ) + // The conversation's runs, then each helper's section's. + const runs = [ + ...transcript.conversation, + ...[...transcript.subagentRows.values()].flat().map((row) => row.message) + ].filter((message) => message.blocks.some((block) => block.type === 'tool-call')) + return { + mobile: runs.map((message) => + pairToolBlocks(message.blocks).map((pair): PairedOutput => [ + pair.call?.name ?? '(none)', + pair.result?.output.trim() + ]) + ), + desktop: runs.map((message) => { + const pairing = pairNativeChatToolResults(message.blocks) + return message.blocks.flatMap((block): PairedOutput[] => + block.type === 'tool-call' + ? [[block.name, pairing.resultByCall.get(block)?.output.trim()]] + : [] + ) + }) + } +} + +const outputOf = (rows: AgentJournalRenderItem[], name: string): AgentJournalItemBody | undefined => + rows.map((row) => row.body).find((body) => body.kind === 'tool-call' && body.name === name) + +describe('Codex collab call rows on a client that pairs results by position', () => { + it('draws each output under its own call through spawn, shell, wait and close', async () => { + const rows = await publishedRows([ + turn('turn/started', THREAD_ID, PARENT_TURN), + spawn('item/started'), + spawn('item/completed'), + turn('turn/started', HELPER, HELPER_TURN), + shell('item/started', THREAD_ID, PARENT_TURN, 'call-parent-shell', 'PARENT_DONE'), + shell('item/completed', THREAD_ID, PARENT_TURN, 'call-parent-shell', 'PARENT_DONE'), + shell('item/started', HELPER, HELPER_TURN, 'call-helper-shell', 'CHILD_DONE'), + shell('item/completed', HELPER, HELPER_TURN, 'call-helper-shell', 'CHILD_DONE'), + item('item/completed', HELPER, HELPER_TURN, { + type: 'agentMessage', + id: 'msg-child', + text: 'CHILD_REPLY' + }), + turn('turn/completed', HELPER, HELPER_TURN), + waitStarted, + collab('item/completed', { + id: 'call-wait', + tool: 'wait', + status: 'completed', + receiverThreadIds: [HELPER], + agentsStates: { [HELPER]: { status: 'completed', message: 'CHILD_REPLY' } } + }), + collab('item/started', { + id: 'call-close', + tool: 'closeAgent', + status: 'inProgress', + receiverThreadIds: [HELPER] + }), + collab('item/completed', { + id: 'call-close', + tool: 'closeAgent', + status: 'completed', + receiverThreadIds: [HELPER], + agentsStates: { [HELPER]: { status: 'completed', message: 'CHILD_REPLY' } } + }), + shell('item/started', THREAD_ID, PARENT_TURN, 'call-parent-shell-2', 'AFTER_CLOSE'), + shell('item/completed', THREAD_ID, PARENT_TURN, 'call-parent-shell-2', 'AFTER_CLOSE') + ]) + // The parent's calls are one run; the helper's own shell is in the helper's section. + const runs = [ + [ + ['spawn_agent', 'Spawned'], + ['shell', 'PARENT_DONE'], + ['wait_agent', 'CHILD_REPLY'], + ['close_agent', 'Closed'], + ['shell', 'AFTER_CLOSE'] + ], + [['shell', 'CHILD_DONE']] + ] + const { mobile, desktop } = positionalRuns(rows) + expect(mobile).toEqual(runs) + expect(desktop).toEqual(runs) + }) + + it('gives a wait that timed out an output, and keeps naming the helpers it waited on', async () => { + const rows = await publishedRows([ + turn('turn/started', THREAD_ID, PARENT_TURN), + spawn('item/completed'), + turn('turn/started', HELPER, HELPER_TURN), + waitStarted, + // Codex ends a timed-out wait naming no receiver and no state. + collab('item/completed', { + id: 'call-wait', + tool: 'wait', + status: 'completed', + receiverThreadIds: [], + agentsStates: {} + }), + collab('item/started', { + id: 'call-close', + tool: 'closeAgent', + status: 'inProgress', + receiverThreadIds: [HELPER] + }), + collab('item/completed', { + id: 'call-close', + tool: 'closeAgent', + status: 'completed', + receiverThreadIds: [HELPER], + agentsStates: { [HELPER]: { status: 'running', message: null } } + }), + shell('item/started', THREAD_ID, PARENT_TURN, 'call-parent-shell', 'PARENT_DONE'), + shell('item/completed', THREAD_ID, PARENT_TURN, 'call-parent-shell', 'PARENT_DONE') + ]) + expect(outputOf(rows, 'wait_agent')).toMatchObject({ + state: 'completed', + input: { description: PROMPT, agents: [HELPER] } + }) + const runs = [ + [ + ['spawn_agent', 'Spawned'], + ['wait_agent', 'Finished waiting'], + // Its snapshot says `running`: the status from before the close. + ['close_agent', 'Closed'], + ['shell', 'PARENT_DONE'] + ] + ] + const { mobile, desktop } = positionalRuns(rows) + expect(mobile).toEqual(runs) + expect(desktop).toEqual(runs) + }) + + it('gives every other finished call what it did or what it reports', async () => { + const OTHER = '01a0ea72-0000-7000-8000-000000000002' + const finished = (id: string, tool: string, fields: Record): Frame[] => [ + collab('item/started', { id, tool, status: 'inProgress', receiverThreadIds: [HELPER] }), + collab('item/completed', { + id, + tool, + status: 'completed', + receiverThreadIds: [HELPER], + ...fields + }) + ] + const rows = await publishedRows([ + turn('turn/started', THREAD_ID, PARENT_TURN), + spawn('item/completed'), + turn('turn/started', HELPER, HELPER_TURN), + // A spawn that created no thread names nothing. + collab('item/completed', { + id: 'call-spawn-failed', + tool: 'spawnAgent', + status: 'failed', + receiverThreadIds: [], + prompt: 'second helper' + }), + ...finished('call-send', 'sendInput', { + prompt: 'keep going', + agentsStates: { [HELPER]: { status: 'running', message: null } } + }), + ...finished('call-resume', 'resumeAgent', { + agentsStates: { [HELPER]: { status: 'completed', message: 'OLD_REPLY' } } + }), + collab('item/started', { + id: 'call-wait', + tool: 'wait', + status: 'inProgress', + receiverThreadIds: [HELPER, OTHER] + }), + // A wait's end names only the helpers that finished; an errored one's message is its error. + collab('item/completed', { + id: 'call-wait', + tool: 'wait', + status: 'failed', + receiverThreadIds: [OTHER], + agentsStates: { [OTHER]: { status: 'errored', message: 'boom' } } + }), + shell('item/started', THREAD_ID, PARENT_TURN, 'call-parent-shell', 'PARENT_DONE'), + shell('item/completed', THREAD_ID, PARENT_TURN, 'call-parent-shell', 'PARENT_DONE') + ]) + const runs = [ + [ + ['spawn_agent', 'Spawned'], + ['spawn_agent', 'Agent spawn failed'], + ['send_input', 'Sent input'], + ['resume_agent', 'Completed - OLD_REPLY'], + ['wait_agent', `${OTHER}: Error - boom`], + ['shell', 'PARENT_DONE'] + ] + ] + const { mobile, desktop } = positionalRuns(rows) + expect(mobile).toEqual(runs) + expect(desktop).toEqual(runs) + // The wait keeps naming both helpers it waited on. + expect(outputOf(rows, 'wait_agent')).toMatchObject({ input: { agents: [HELPER, OTHER] } }) + }) +}) diff --git a/src/main/codex/codex-default-mode-subagents.test.ts b/src/main/codex/codex-default-mode-subagents.test.ts new file mode 100644 index 000000000000..335f903efbbf --- /dev/null +++ b/src/main/codex/codex-default-mode-subagents.test.ts @@ -0,0 +1,704 @@ +// Codex's default multi-agent mode announces a helper only by the `collabAgentToolCall` items that +// spawn, message, wait on or close it. These frames, through the real adapter, must register that +// helper the same way a `subAgentActivity` does: one child in the strip, the host's records and +// the roster row. + +import { describe, expect, it } from 'vitest' +import type { AgentJournalItemBody } from '../../shared/agent-session-journal-types' +import { createAgentChildWorkAdmission } from '../../shared/agent-status-child-work-admission' +import type { AgentChildWorkRecord } from '../../shared/agent-status-child-work' +import { reconcileAgentChildWorkEvidence } from '../../shared/agent-status-child-work-reconciliation' +import { createAgentStatusStore } from '../../shared/agent-status-store' +import { makeStructuredAgentStatusSubject } from '../../shared/agent-status-subject' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { fakeCodex, identityFor, THREAD_ID } from './codex-structured-session-adapter-fixture' +import { CodexStructuredSessionAdapter } from './codex-structured-session-adapter' +import { CodexBackgroundTaskTracker } from './codex-background-task-tracker' +import { createCodexJournalTranslator } from './codex-structured-journal-translation' +import { CodexSubagentExecutions } from './codex-subagent-executions' +import { readCodexSubagentAnnouncements } from './codex-subagent-activity' +import type { CodexThreadItem } from './codex-thread-item-identity' + +const parent = makeStructuredAgentStatusSubject( + { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'folder' + }, + 'session-1' +) + +// The spawn and wait items below are verbatim from a live default-mode session (codex-cli +// 0.155.0-alpha.9.2, `multi_agent` on, `multi_agent_v2` off); only the sender is the fixture's +// thread. That session sent no `subAgentActivity` at all. +const HELPER = '01a0d114-2b8e-73e2-a69b-8e86df067ae9' +const PARENT_TURN = '01a0d112-6f86-7092-8c87-b8618f259efa' +const HELPER_TURN = 'helper-turn-1' +const PROMPT = + 'Run the shell command exactly: `sleep 45; echo CHILD_DONE`. After it completes, reply with the single word `CHILD_REPLY` and nothing else.' +/** The prompt's head, as the helper's row names it. */ +const LABEL = 'Run the shell command exactly: `sleep 45; echo CHILD_DONE`. After it completes,…' +const COMMAND = "/bin/zsh -lc 'sleep 45; echo CHILD_DONE'" + +type Frame = { method: string; params: Record } + +const turn = ( + method: 'turn/started' | 'turn/completed', + threadId: string, + id: string, + status = 'completed' +): Frame => ({ method, params: { threadId, turn: { id, status } } }) +const item = ( + method: 'item/started' | 'item/completed', + threadId: string, + turnId: string, + fields: Record +): Frame => ({ method, params: { threadId, turnId, item: fields } }) +const collab = ( + method: 'item/started' | 'item/completed', + fields: Record +): Frame => + item(method, THREAD_ID, PARENT_TURN, { + type: 'collabAgentToolCall', + senderThreadId: THREAD_ID, + prompt: null, + model: null, + reasoningEffort: null, + agentsStates: {}, + ...fields + }) + +const spawnStarted = collab('item/started', { + id: 'call_dxoSnQY1cHVswlN8MGFJBxHt', + tool: 'spawnAgent', + status: 'inProgress', + receiverThreadIds: [], + prompt: PROMPT, + model: '', + reasoningEffort: 'medium' +}) +const spawnCompleted = collab('item/completed', { + id: 'call_dxoSnQY1cHVswlN8MGFJBxHt', + tool: 'spawnAgent', + status: 'completed', + receiverThreadIds: [HELPER], + prompt: PROMPT, + model: 'gpt-5.5', + reasoningEffort: 'medium', + agentsStates: { [HELPER]: { status: 'pendingInit', message: null } } +}) +const waitStarted = collab('item/started', { + id: 'call_rMn9MIAPavyhOD6ifjH38A8E', + tool: 'wait', + status: 'inProgress', + receiverThreadIds: [HELPER] +}) +const waitCompleted = collab('item/completed', { + id: 'call_rMn9MIAPavyhOD6ifjH38A8E', + tool: 'wait', + status: 'completed', + receiverThreadIds: [HELPER], + agentsStates: { [HELPER]: { status: 'completed', message: 'CHILD_REPLY' } } +}) +const closeCompleted = collab('item/completed', { + id: 'call-close', + tool: 'closeAgent', + status: 'completed', + receiverThreadIds: [HELPER], + agentsStates: { [HELPER]: { status: 'shutdown', message: null } } +}) +const helperShell = (method: 'item/started' | 'item/completed'): Frame => + item(method, HELPER, HELPER_TURN, { + type: 'commandExecution', + id: 'call_3iUPkSEtwkUYstngZm5CNJ52', + command: COMMAND, + cwd: '/work/repo', + // As the live session reported it; a command is tracked the same whatever its source. + source: 'unifiedExecStartup', + status: method === 'item/started' ? 'inProgress' : 'completed', + ...(method === 'item/completed' ? { exitCode: 0, aggregatedOutput: 'CHILD_DONE\n' } : {}) + }) +const helperReply = item('item/completed', HELPER, HELPER_TURN, { + type: 'agentMessage', + id: 'msg-child', + text: 'CHILD_REPLY' +}) +const activityStarted: Frame = item('item/started', THREAD_ID, PARENT_TURN, { + type: 'subAgentActivity', + id: 'activity-helper', + kind: 'started', + agentThreadId: HELPER, + agentPath: '/root/sleeper' +}) + +async function session() { + const codex = fakeCodex() + const store = createAgentStatusStore({ epoch: 'epoch-1', mode: 'authority' }) + expect(store.applyMutation({ parent: { subject: parent } })).not.toBeNull() + let minted = 0 + const admission = createAgentChildWorkAdmission(store, { + mintChildWorkId: () => `child-${++minted}` + }) + const adapter = new CodexStructuredSessionAdapter({ + resolveLaunch: async () => ({ + command: 'codex', + args: ['app-server'], + cwd: '/work/repo', + codexHome: null, + resumeThreadId: null + }), + openConnection: codex.openConnection, + readProcessStartTime: async () => 1_700_000_000_000, + now: () => 1_700_000_000_500, + onChildWorkEvidence: (_sessionId, evidence) => + reconcileAgentChildWorkEvidence({ store, admission, parent, provider: 'codex', evidence }) + }) + // The latest revision of each journal row, in first-written order. + const rows = new Map() + const journal: StructuredAgentSessionEventSink = { + appendItem: (identity, body) => rows.set(JSON.stringify(identity), body), + appendTombstone: () => {}, + publish: () => {} + } + await adapter.acquire({ + identity: identityFor('session-1'), + fence: 7, + spawnToken: 'spawn-9', + events: journal + }) + const send = (...frames: Frame[]): void => { + for (const frame of frames) { + codex.connections[0]!.handlers.onNotification?.(frame.method, frame.params) + } + } + const agents = (): AgentChildWorkRecord[] => + store.getChildren(parent).filter((record) => record.kind === 'agent') + const commands = (): AgentChildWorkRecord[] => + store.getChildren(parent).filter((record) => record.kind === 'command') + const strip = () => adapter.backgroundTaskState('session-1')?.tasks ?? [] + const toolRow = (name: string) => + [...rows.values()].find((body) => body.kind === 'tool-call' && body.name === name) + const rosterRows = () => + [...rows.values()].flatMap((body) => + body.kind === 'message' + ? body.blocks.flatMap((block) => (block.type === 'subagent-group' ? [block] : [])) + : [] + ) + const rawCollabRows = () => + [...rows.values()].filter( + (body) => body.kind === 'status' && body.providerFrame?.kind === 'item:collabAgentToolCall' + ) + return { send, agents, commands, strip, toolRow, rosterRows, rawCollabRows } +} + +describe('Codex default-mode helpers', () => { + it('registers a helper announced only by its spawn call, in the strip, the records and the roster row', async () => { + const run = await session() + run.send(turn('turn/started', THREAD_ID, PARENT_TURN), spawnStarted) + // The call has not said which thread the helper is yet. + expect(run.agents()).toEqual([]) + expect(run.strip()).toEqual([]) + + run.send(spawnCompleted, turn('turn/started', HELPER, HELPER_TURN), waitStarted) + expect(run.agents()).toEqual([ + expect.objectContaining({ + membership: 'live', + state: 'working', + description: LABEL, + invocation: { invocationId: HELPER_TURN, generation: 1 } + }) + ]) + expect(run.strip()).toEqual([ + { id: `codex-agent:${HELPER}`, kind: 'agent', description: LABEL } + ]) + expect(run.rosterRows()).toEqual([ + expect.objectContaining({ + agents: [expect.objectContaining({ id: HELPER, label: LABEL, state: 'working' })] + }) + ]) + + // While the helper runs, its command is the helper's work, not the session's own. + run.send(helperShell('item/started')) + expect(run.strip()).toEqual([ + { id: `codex-agent:${HELPER}`, kind: 'agent', description: LABEL } + ]) + expect(run.commands()).toEqual([ + expect.objectContaining({ + membership: 'live', + description: COMMAND, + parentChildWorkId: run.agents()[0]?.childWorkId + }) + ]) + expect(run.agents()[0]?.operation).toMatchObject({ toolName: 'Bash', input: COMMAND }) + + run.send(helperShell('item/completed')) + // A finished command leaves no record behind. + expect(run.commands()).toEqual([]) + + run.send(helperReply, turn('turn/completed', HELPER, HELPER_TURN)) + expect(run.agents()).toEqual([ + expect.objectContaining({ + membership: 'settled', + outcome: 'succeeded', + lastMessage: 'CHILD_REPLY' + }) + ]) + expect(run.strip()).toEqual([]) + expect(run.rosterRows().at(-1)?.agents).toEqual([ + expect.objectContaining({ id: HELPER, state: 'completed' }) + ]) + }) + + it('labels a command the helper leaves running with the helper, once its turn is over', async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + spawnCompleted, + turn('turn/started', HELPER, HELPER_TURN), + helperShell('item/started'), + turn('turn/completed', HELPER, HELPER_TURN) + ) + expect(run.strip()).toEqual([ + expect.objectContaining({ kind: 'command', description: `${LABEL} — ${COMMAND}` }) + ]) + }) + + it('renders each collab call as a tool row naming its helper, never as a raw provider row', async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + spawnStarted, + spawnCompleted, + turn('turn/started', HELPER, HELPER_TURN), + waitStarted + ) + expect(run.toolRow('wait_agent')).toMatchObject({ + state: 'running', + input: { description: LABEL, agents: [HELPER] } + }) + run.send(turn('turn/completed', HELPER, HELPER_TURN), waitCompleted) + expect(run.toolRow('spawn_agent')).toMatchObject({ + state: 'completed', + input: { description: LABEL, prompt: PROMPT, model: 'gpt-5.5', agents: [HELPER] } + }) + expect(run.toolRow('wait_agent')).toMatchObject({ + state: 'completed', + input: { description: LABEL }, + output: expect.objectContaining({ head: 'CHILD_REPLY' }) + }) + expect(run.rawCollabRows()).toEqual([]) + }) + + it('keeps naming the helper when the turn ends with its wait still open', async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + spawnCompleted, + turn('turn/started', HELPER, HELPER_TURN), + waitStarted, + turn('turn/completed', THREAD_ID, PARENT_TURN, 'interrupted') + ) + expect(run.toolRow('wait_agent')).toMatchObject({ + state: 'failed', + input: { description: LABEL } + }) + }) + + it('registers the helper whichever order its first turn and its spawn arrive in', async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + turn('turn/started', HELPER, HELPER_TURN), + spawnCompleted + ) + expect(run.agents()).toEqual([ + expect.objectContaining({ membership: 'live', state: 'working', description: LABEL }) + ]) + expect(run.strip()).toHaveLength(1) + expect(run.rosterRows()).toEqual([ + expect.objectContaining({ + agents: [expect.objectContaining({ id: HELPER, label: LABEL, state: 'working' })] + }) + ]) + // The roster row does not stand in for the call's own row. + expect(run.toolRow('spawn_agent')).toMatchObject({ + state: 'completed', + input: { description: LABEL } + }) + }) + + it('registers a helper whose spawn call failed but created its thread, as Codex reports an errored one', async () => { + const run = await session() + // The shape Codex builds for a helper that errored at birth: the call fails, yet names the + // thread it created, and that thread can still run. + const spawnFailedWithThread = collab('item/completed', { + id: 'call-failed-spawn', + tool: 'spawnAgent', + status: 'failed', + receiverThreadIds: [HELPER], + prompt: PROMPT, + model: 'gpt-5.5', + reasoningEffort: 'medium', + agentsStates: { [HELPER]: { status: 'errored', message: null } } + }) + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + spawnFailedWithThread, + turn('turn/started', HELPER, HELPER_TURN), + helperShell('item/started') + ) + expect(run.agents()).toEqual([ + expect.objectContaining({ membership: 'live', state: 'working', description: LABEL }) + ]) + // Its command is the helper's, not a bare command of the session's own. + expect(run.strip()).toEqual([ + { id: `codex-agent:${HELPER}`, kind: 'agent', description: LABEL } + ]) + expect(run.rosterRows()).toEqual([ + expect.objectContaining({ + agents: [expect.objectContaining({ id: HELPER, state: 'working' })] + }) + ]) + }) + + it('registers nothing for a spawn that created no thread', async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + collab('item/completed', { + id: 'call-refused-spawn', + tool: 'spawnAgent', + status: 'failed', + receiverThreadIds: [], + prompt: PROMPT + }) + ) + expect(run.agents()).toEqual([]) + expect(run.strip()).toEqual([]) + expect(run.rosterRows()).toEqual([]) + }) + + it('leaves a helper running when its caller failed to close it', async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + spawnCompleted, + turn('turn/started', HELPER, HELPER_TURN), + collab('item/completed', { + id: 'call-close-failed', + tool: 'closeAgent', + status: 'failed', + receiverThreadIds: [HELPER], + agentsStates: { [HELPER]: { status: 'notFound', message: null } } + }) + ) + expect(run.agents()).toEqual([expect.objectContaining({ membership: 'live' })]) + expect(run.strip()).toHaveLength(1) + }) + + it("ends a closed helper only on its own interrupted turn, never on the caller's close item", async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + spawnCompleted, + turn('turn/started', HELPER, HELPER_TURN), + // What Codex reports for a close that failed on a running helper: the status is the + // helper's, not the close's, so `completed` says nothing about whether it stopped. + collab('item/completed', { + id: 'call-close-errored', + tool: 'closeAgent', + status: 'completed', + receiverThreadIds: [HELPER], + agentsStates: { [HELPER]: { status: 'running', message: null } } + }) + ) + expect(run.agents()).toEqual([expect.objectContaining({ membership: 'live' })]) + expect(run.strip()).toHaveLength(1) + expect(run.rosterRows().at(-1)?.agents).toEqual([ + expect.objectContaining({ id: HELPER, state: 'working' }) + ]) + + // A close that works aborts the helper's turn, which Codex reports on the helper's thread. + run.send(closeCompleted, turn('turn/completed', HELPER, HELPER_TURN, 'interrupted')) + expect(run.agents()).toEqual([ + expect.objectContaining({ membership: 'settled', outcome: 'cancelled' }) + ]) + expect(run.strip()).toEqual([]) + expect(run.rosterRows().at(-1)?.agents).toEqual([ + expect.objectContaining({ id: HELPER, state: 'stopped' }) + ]) + expect(run.toolRow('close_agent')).toMatchObject({ + state: 'completed', + input: { description: LABEL } + }) + }) + + it("keeps a closed helper's running command until Codex reports the killed process's exit", async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + spawnCompleted, + turn('turn/started', HELPER, HELPER_TURN), + helperShell('item/started'), + turn('turn/completed', HELPER, HELPER_TURN), + closeCompleted + ) + // The close shuts the helper's thread down and kills its processes; each still reports its + // exit on the helper's thread, and that exit is what ends the command. + expect(run.commands()).toEqual([expect.objectContaining({ membership: 'live' })]) + run.send(helperShell('item/completed')) + expect(run.commands()).toEqual([]) + expect(run.strip()).toEqual([]) + }) + + it('leaves a finished helper finished when its caller closes it', async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + spawnCompleted, + turn('turn/started', HELPER, HELPER_TURN), + turn('turn/completed', HELPER, HELPER_TURN), + closeCompleted + ) + expect(run.agents()).toEqual([ + expect.objectContaining({ membership: 'settled', outcome: 'succeeded' }) + ]) + }) + + it('keeps one child for a session that announces the helper both ways', async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + activityStarted, + spawnCompleted, + turn('turn/started', HELPER, HELPER_TURN), + { ...activityStarted, method: 'item/completed' } + ) + // The activity landed first, so its task name labels the one child. + expect(run.agents()).toEqual([expect.objectContaining({ description: 'sleeper' })]) + expect(run.strip()).toEqual([ + { id: `codex-agent:${HELPER}`, kind: 'agent', description: 'sleeper' } + ]) + expect(run.rosterRows().at(-1)?.agents).toEqual([ + expect.objectContaining({ id: HELPER, label: 'sleeper' }) + ]) + }) +}) + +describe('a helper whose spawn was never seen', () => { + const sendInput = collab('item/completed', { + id: 'call-send-input', + tool: 'sendInput', + status: 'completed', + receiverThreadIds: [HELPER], + prompt: 'Now run it again.', + agentsStates: { [HELPER]: { status: 'running', message: null } } + }) + + it("registers from any call that names it, so its shell is no longer the session's own", async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + turn('turn/started', HELPER, HELPER_TURN), + helperShell('item/started') + ) + expect(run.strip()).toEqual([expect.objectContaining({ kind: 'command' })]) + + run.send(sendInput) + // No spawn named it, so it reads as any unnamed subagent does. + expect(run.strip()).toEqual([{ id: `codex-agent:${HELPER}`, kind: 'agent' }]) + expect(run.agents()).toEqual([ + expect.objectContaining({ membership: 'live', state: 'working' }) + ]) + expect(run.commands()).toEqual([ + expect.objectContaining({ + membership: 'live', + parentChildWorkId: run.agents()[0]?.childWorkId + }) + ]) + expect(run.rosterRows().at(-1)?.agents).toEqual([ + expect.objectContaining({ id: HELPER, label: 'subagent', state: 'working' }) + ]) + expect(run.toolRow('send_input')).toMatchObject({ input: { description: HELPER } }) + }) + + it('registers each helper a call on several names', async () => { + const second = 'helper-two' + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + turn('turn/started', HELPER, HELPER_TURN), + turn('turn/started', second, 'helper-two-turn'), + collab('item/started', { + id: 'call-wait-both', + tool: 'wait', + status: 'inProgress', + receiverThreadIds: [HELPER, second] + }) + ) + expect(run.strip()).toEqual([ + { id: `codex-agent:${HELPER}`, kind: 'agent' }, + { id: `codex-agent:${second}`, kind: 'agent' } + ]) + expect(run.agents()).toEqual([ + expect.objectContaining({ membership: 'live' }), + expect.objectContaining({ membership: 'live' }) + ]) + expect(run.rosterRows().at(-1)?.agents).toEqual([ + expect.objectContaining({ id: HELPER, label: 'subagent' }), + expect.objectContaining({ id: second, label: 'subagent 2' }) + ]) + }) + + it("puts a helper's run in the turn that sent it work, even after that turn ended", async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + spawnCompleted, + turn('turn/started', HELPER, HELPER_TURN), + turn('turn/completed', HELPER, HELPER_TURN), + turn('turn/completed', THREAD_ID, PARENT_TURN), + turn('turn/started', THREAD_ID, 'parent-turn-2'), + { ...sendInput, params: { ...sendInput.params, turnId: 'parent-turn-2' } }, + turn('turn/completed', THREAD_ID, 'parent-turn-2'), + turn('turn/started', HELPER, 'helper-turn-2') + ) + const groups = new Set(run.rosterRows().map((row) => row.groupId)) + expect(groups.size).toBe(2) + expect(run.rosterRows().at(-1)?.agents).toEqual([ + expect.objectContaining({ id: HELPER, state: 'working' }) + ]) + }) + + it('registers every helper a call names, and none it reports not found', () => { + const [second, missing] = ['helper-two', 'helper-missing'] + const waited: CodexThreadItem = { + type: 'collabAgentToolCall', + id: 'call-wait-many', + tool: 'wait', + status: 'completed', + receiverThreadIds: [HELPER, second, missing], + agentsStates: { + [HELPER]: { status: 'completed', message: 'done' }, + [second]: { status: 'running', message: null }, + [missing]: { status: 'notFound', message: null } + } + } + expect(readCodexSubagentAnnouncements(waited, THREAD_ID)).toEqual([ + { agentThreadId: HELPER, label: null, namesParentTurn: false, spawned: false }, + { agentThreadId: second, label: null, namesParentTurn: false, spawned: false } + ]) + }) +}) + +describe('the roster row follows a helper whose turn ends', () => { + const lastRow = (run: Awaited>) => run.rosterRows().at(-1)?.agents + const running = async () => { + const run = await session() + run.send( + turn('turn/started', THREAD_ID, PARENT_TURN), + activityStarted, + turn('turn/started', HELPER, HELPER_TURN) + ) + expect(lastRow(run)).toEqual([expect.objectContaining({ id: HELPER, state: 'working' })]) + return run + } + + it('settles the row with the strip and the record on the failed completion that follows a fatal error', async () => { + const run = await running() + // Codex sends the fatal error, then the helper's failed `turn/completed` 0-32 ms later. + run.send({ + method: 'error', + params: { + threadId: HELPER, + turnId: HELPER_TURN, + willRetry: false, + error: { message: 'Selected model is at capacity.', codexErrorInfo: 'serverOverloaded' } + } + }) + expect(run.strip()).toEqual([expect.objectContaining({ id: `codex-agent:${HELPER}` })]) + expect(run.agents()).toEqual([expect.objectContaining({ membership: 'live' })]) + expect(lastRow(run)).toEqual([expect.objectContaining({ id: HELPER, state: 'working' })]) + + run.send(turn('turn/completed', HELPER, HELPER_TURN, 'failed')) + expect(run.strip()).toEqual([]) + expect(run.agents()).toEqual([ + expect.objectContaining({ membership: 'settled', outcome: 'failed' }) + ]) + expect(lastRow(run)).toEqual([expect.objectContaining({ id: HELPER, state: 'failed' })]) + }) + + it('settles the row with the strip and the record when its thread closes with no turn/completed', async () => { + const run = await running() + run.send({ method: 'thread/closed', params: { threadId: HELPER } }) + expect(run.strip()).toEqual([]) + expect(run.agents()).toEqual([ + expect.objectContaining({ membership: 'settled', outcome: 'unknown' }) + ]) + expect(lastRow(run)).toEqual([expect.objectContaining({ id: HELPER, state: 'unverifiable' })]) + }) +}) + +describe('a restored thread', () => { + function restored(replayed = [spawnCompleted, waitCompleted]) { + const executions = new CodexSubagentExecutions() + const rows = new Map() + const translator = createCodexJournalTranslator({ + sink: { + appendItem: (identity, body) => rows.set(JSON.stringify(identity), body), + appendTombstone: () => {}, + publish: () => {} + }, + primaryThreadId: () => THREAD_ID, + subagentExecutions: executions, + schedule: (run: () => void) => { + run() + return () => {} + } + }) + const tracker = new CodexBackgroundTaskTracker(THREAD_ID, executions) + const admission = translator.restoreThread(THREAD_ID, { + turns: [{ id: PARENT_TURN, items: replayed.map((frame) => frame.params.item) }] + }) + expect(admission).toEqual({ accepted: true }) + const bodies = () => [...rows.values()] + return { executions, tracker, bodies } + } + + it('names the helper on a replayed call row, as the live row did', () => { + const { bodies } = restored() + expect(bodies()).toContainEqual( + expect.objectContaining({ + kind: 'tool-call', + name: 'wait_agent', + input: expect.objectContaining({ description: LABEL }) + }) + ) + }) + + it('claims no running helper for a name it learned from history', () => { + const { executions, tracker, bodies } = restored() + expect(executions.workingChildren()).toEqual([]) + // A live frame republishes the strip and drains evidence: neither holds the helper. + tracker.observe({ + method: 'turn/started', + threadId: THREAD_ID, + params: { threadId: THREAD_ID, turn: { id: 'next-turn', status: 'inProgress' } } + }) + expect(tracker.state).toBeNull() + expect(tracker.drainChildWorkEvidence(1)).toEqual([]) + expect( + bodies().filter( + (body) => + body.kind === 'message' && body.blocks.some((block) => block.type === 'subagent-group') + ) + ).toEqual([]) + }) + + it('knows a helper whose spawn history compacted away, from a call on it', () => { + const { tracker } = restored([waitCompleted]) + tracker.observe({ ...turn('turn/started', HELPER, HELPER_TURN), threadId: HELPER }) + expect(tracker.state?.tasks).toEqual([{ id: `codex-agent:${HELPER}`, kind: 'agent' }]) + }) +}) diff --git a/src/main/codex/codex-item-run-state.ts b/src/main/codex/codex-item-run-state.ts new file mode 100644 index 000000000000..a3b22eff8aed --- /dev/null +++ b/src/main/codex/codex-item-run-state.ts @@ -0,0 +1,16 @@ +import { readString } from './codex-item-field-readers' +import type { CodexThreadItem } from './codex-thread-item-identity' + +/** Codex reports `inProgress` then a terminal status; a zero exit code is the + * only thing that makes a finished command a success. */ +export function codexItemRunState(item: CodexThreadItem): 'running' | 'completed' | 'failed' { + const status = readString(item, 'status') + if (status === null || status === 'inProgress') { + return 'running' + } + if (status !== 'completed') { + return 'failed' + } + const exitCode = item.exitCode + return typeof exitCode === 'number' && exitCode !== 0 ? 'failed' : 'completed' +} diff --git a/src/main/codex/codex-structured-item-translation.ts b/src/main/codex/codex-structured-item-translation.ts index 34f07eefda7e..25b821d90d9f 100644 --- a/src/main/codex/codex-structured-item-translation.ts +++ b/src/main/codex/codex-structured-item-translation.ts @@ -9,6 +9,11 @@ import { import { unhandledProviderFrameJournalItem } from '../native-chat/agent-session-wire/unhandled-provider-frame' import { codexImageItemBody } from './codex-image-item-translation' import { commandActionFacts } from './codex-command-action-class' +import { + codexCollabAgentToolCallBody, + type CodexHelperName +} from './codex-collab-agent-item-translation' +import { codexItemRunState } from './codex-item-run-state' import { readFirstString, readRecord, @@ -66,20 +71,6 @@ export function codexMessageBlocks(item: CodexThreadItem): NativeChatBlock[] { return blocks } -/** Codex reports `inProgress` then a terminal status; a zero exit code is the - * only thing that makes a finished command a success. */ -function commandState(item: CodexThreadItem): 'running' | 'completed' | 'failed' { - const status = readString(item, 'status') - if (status === null || status === 'inProgress') { - return 'running' - } - if (status !== 'completed') { - return 'failed' - } - const exitCode = item.exitCode - return typeof exitCode === 'number' && exitCode !== 0 ? 'failed' : 'completed' -} - export type CodexJournalItem = { body: AgentJournalItemBody | null handled: boolean @@ -103,7 +94,7 @@ function commandItem(item: CodexThreadItem): CodexJournalItem { { command: item.command ?? null, cwd: item.cwd ?? null, ...parsed?.fields }, DEFAULT_JOURNAL_PAYLOAD_LIMITS ), - state: commandState(item), + state: codexItemRunState(item), ...toolExecutionMetadata(item), ...(bounded === null ? {} : { output: bounded.bounded }) }, @@ -127,7 +118,7 @@ function fileChangeItem(item: CodexThreadItem): CodexJournalItem { name: 'apply_patch', callId: item.id, input: boundToolInput({ changes: item.changes ?? null }, DEFAULT_JOURNAL_PAYLOAD_LIMITS), - state: commandState(item) + state: codexItemRunState(item) }, handled: true } @@ -180,7 +171,7 @@ function mcpToolCallItem(item: CodexThreadItem): CodexJournalItem { callId: item.id, ...(server && tool ? { mcpIdentity: { server, tool } } : {}), input: boundToolInput(mcpToolArguments(item.arguments), DEFAULT_JOURNAL_PAYLOAD_LIMITS), - state: failure === null ? commandState(item) : 'failed', + state: failure === null ? codexItemRunState(item) : 'failed', ...(bounded === null ? {} : { output: bounded.bounded }) }, handled: true @@ -234,9 +225,14 @@ function webSearchItem(item: CodexThreadItem): CodexJournalItem { * Journal body for a Codex item, or null for one with nothing to render. * * Known empty items wait for later deltas. Unknown types become bounded status - * rows so a provider release cannot make new activity invisible. + * rows so a provider release cannot make new activity invisible. `started` is a + * finished item's own started revision, when the caller still holds it. */ -export function codexJournalItem(item: CodexThreadItem): CodexJournalItem { +export function codexJournalItem( + item: CodexThreadItem, + helperName?: CodexHelperName, + started?: CodexThreadItem +): CodexJournalItem { if (item.type === 'userMessage' || item.type === 'agentMessage') { const blocks = codexMessageBlocks(item) return { @@ -262,6 +258,10 @@ export function codexJournalItem(item: CodexThreadItem): CodexJournalItem { if (item.type === 'imageView' || item.type === 'imageGeneration') { return { body: codexImageItemBody(item), handled: true } } + const collab = codexCollabAgentToolCallBody(item, helperName, started) + if (collab) { + return { body: collab, handled: true } + } if (item.type === 'plan') { const text = readTextContent(item, 'text') return { diff --git a/src/main/codex/codex-structured-journal-items.ts b/src/main/codex/codex-structured-journal-items.ts index 4e6ebe8e5e08..ec6ac67df5d4 100644 --- a/src/main/codex/codex-structured-journal-items.ts +++ b/src/main/codex/codex-structured-journal-items.ts @@ -12,6 +12,7 @@ import { type CodexThreadItem } from './codex-structured-item-translation' import { createCodexStructuredItemStreams } from './codex-structured-item-streams' +import type { CodexHelperName } from './codex-collab-agent-item-translation' import { boundStreamItem, codexStructuredItemKey } from './codex-structured-item-stream-bounds' import { codexCommandOutlivesTurn } from './codex-command-lifecycle' import type { @@ -46,7 +47,8 @@ export class CodexJournalItems { 'sink' | 'coalesceMs' | 'maxRetainedBytes' | 'schedule' > & { maxMetadataBytes?: number; attributionFor: CodexRowAttribution }, private readonly activeTurn: (threadId: string) => string | null, - private readonly suppress: (threadId: string, turnId: string) => void + private readonly suppress: (threadId: string, turnId: string) => void, + private readonly helperName?: CodexHelperName ) { this.streams = createCodexStructuredItemStreams({ sink: deps.sink, @@ -96,15 +98,17 @@ export class CodexJournalItems { ) { return { handled: true, admission: { accepted: false, reason: 'failed' } } } - const translated = codexJournalItem(item) + const itemKey = codexStructuredItemKey(event.threadId, item.id) + const started = + event.method === 'item/completed' ? this.activeItems.get(itemKey)?.item : undefined + const translated = codexJournalItem(item, this.helperName, started) const command = readCodexJournalString(item, 'command') if (command) { const boundedCommand = Buffer.from(command, 'utf8') .subarray(0, MAX_CODEX_DETAIL_BYTES) .toString('utf8') - this.details.set(codexStructuredItemKey(event.threadId, item.id), boundedCommand) + this.details.set(itemKey, boundedCommand) } - const itemKey = codexStructuredItemKey(event.threadId, item.id) if (!translated.body) { if (event.method === 'item/completed') { this.streams.forget(event.threadId, item.id) @@ -193,7 +197,8 @@ export class CodexJournalItems { threadId, turnId, identity, - item: retainedItem + item: retainedItem, + ...(this.helperName ? { helperName: this.helperName } : {}) }) } @@ -234,7 +239,7 @@ export class CodexJournalItems { } const evicted = this.activeItems.get(oldest) if (evicted) { - const translated = codexJournalItem(evicted.item).body + const translated = codexJournalItem(evicted.item, this.helperName).body if (translated) { const admission = appendCodexLifecycleItem( this.deps.sink, diff --git a/src/main/codex/codex-structured-journal-settlement.ts b/src/main/codex/codex-structured-journal-settlement.ts index a3f4c7f5599c..131de9824aae 100644 --- a/src/main/codex/codex-structured-journal-settlement.ts +++ b/src/main/codex/codex-structured-journal-settlement.ts @@ -20,6 +20,7 @@ import { type CodexTurnOrdinals } from './codex-structured-item-translation' import type { CodexStructuredItemStreams } from './codex-structured-item-streams' +import type { CodexHelperName } from './codex-collab-agent-item-translation' import type { CodexStructuredSessionEvent } from './codex-structured-session-adapter' import { codexCommandOutlivesTurn } from './codex-command-lifecycle' import { @@ -34,6 +35,8 @@ export type CodexActiveJournalItem = { turnId: string | null identity: AgentJournalItemIdentity item: CodexThreadItem + /** Names the helpers a collab call acted on, so a settled revision keeps naming them. */ + helperName?: CodexHelperName } export type CodexPendingJournalPrompt = { @@ -66,7 +69,7 @@ export function settleCodexJournalSession(input: { const streamed = input.streams.snapshot(active.threadId, active.item.id) const translated = streamed ? codexStreamingJournalItem(active.item, streamed.text) - : codexJournalItem(active.item) + : codexJournalItem(active.item, active.helperName) const body = interruptedBody(translated.body) if (body) { mutations.push(settledRow(input.attributionFor, active, body)) @@ -138,7 +141,7 @@ export function settleCodexJournalTurn(input: { const streamed = input.streams.snapshot(active.threadId, active.item.id) const translated = streamed ? codexStreamingJournalItem(active.item, streamed.text) - : codexJournalItem(active.item) + : codexJournalItem(active.item, active.helperName) const body = interruptedBody(translated.body) if (body) { mutations.push(settledRow(input.attributionFor, active, body)) diff --git a/src/main/codex/codex-structured-journal-translation-restore.ts b/src/main/codex/codex-structured-journal-translation-restore.ts index 5dc68d6f764d..9f7efc15ae4c 100644 --- a/src/main/codex/codex-structured-journal-translation-restore.ts +++ b/src/main/codex/codex-structured-journal-translation-restore.ts @@ -1,5 +1,12 @@ import type { AgentJournalTurnLifecycle } from '../../shared/agent-session-journal-types' -import type { CodexTurnOrdinals } from './codex-structured-item-translation' +import { readCodexThreadItem, type CodexTurnOrdinals } from './codex-structured-item-translation' +import type { CodexJournalCompactions } from './codex-structured-journal-compactions' +import type { CodexJournalItems } from './codex-structured-journal-items' +import { + readCodexNotificationThreadItem, + readCodexSubagentAnnouncements +} from './codex-subagent-activity' +import type { CodexSubagentExecutions } from './codex-subagent-executions' import { readCodexJournalRecord, readCodexJournalString @@ -12,6 +19,33 @@ import { } from './codex-structured-journal-translation-turns' import { readCodexTurnDurationMs, readCodexTurnStatus } from './codex-structured-thread-facts' +type CodexHistoryItemEvent = { threadId: string; method: string; params: unknown } + +/** One replayed item. History never runs the live item router, so a helper the replay announces + * is registered here for its name and membership only: `register` starts no execution, so no + * strip entry, record or roster row claims it runs until a live turn of its own says so. */ +export function restoreCodexHistoryItem( + event: CodexHistoryItemEvent, + input: { + primaryThreadId: string | null + compactions: Pick + items: Pick + executions: Pick + } +): CodexJournalTranslationAdmission { + const item = readCodexNotificationThreadItem(event.params, readCodexThreadItem) + const announcements = item ? readCodexSubagentAnnouncements(item, input.primaryThreadId) : [] + for (const announcement of announcements) { + input.executions.register(announcement.agentThreadId, announcement.label, undefined) + } + const compaction = input.compactions.handle(event) + if (compaction) { + return compaction + } + const translated = input.items.handle(event, 'history') + return translated.handled ? translated.admission : { accepted: false, reason: 'untranslated' } +} + /** Old providers may return the complete thread from resume. Keep that fallback * bounded before admitting any rows to the asynchronous sink. */ export const CODEX_RESTORE_MAX_OPERATIONS = 1_024 @@ -22,11 +56,7 @@ export function restoreCodexJournalThread(input: { thread: Record currentTurnIds: Map> ordinals: CodexTurnOrdinals - handleItem: (event: { - threadId: string - method: string - params: unknown - }) => CodexJournalTranslationAdmission + handleItem: (event: CodexHistoryItemEvent) => CodexJournalTranslationAdmission /** Absent when the caller has no session identity to key lifecycle rows by. */ restoreTurnLifecycle?: ( turnLifecycle: AgentJournalTurnLifecycle diff --git a/src/main/codex/codex-structured-journal-translation-writers.ts b/src/main/codex/codex-structured-journal-translation-writers.ts index b13a7212142d..9e577074c5e4 100644 --- a/src/main/codex/codex-structured-journal-translation-writers.ts +++ b/src/main/codex/codex-structured-journal-translation-writers.ts @@ -39,8 +39,12 @@ export function createCodexJournalTranslatorWriters(deps: CodexJournalTranslator }) const producerDeps = { ...deps, attributionFor } const genericFrames = new CodexJournalGenericFrames(producerDeps, activeTurn) - const items = new CodexJournalItems(producerDeps, activeTurn, (threadId, turnId) => - genericFrames.suppress(threadId, turnId) + const items = new CodexJournalItems( + producerDeps, + activeTurn, + (threadId, turnId) => genericFrames.suppress(threadId, turnId), + // A collab call's row names its helpers the way the roster does. + (threadId) => subagents.executions.label(threadId) ) return { activeTurns, diff --git a/src/main/codex/codex-structured-journal-translation.ts b/src/main/codex/codex-structured-journal-translation.ts index 8e05506509a3..abc094bc3064 100644 --- a/src/main/codex/codex-structured-journal-translation.ts +++ b/src/main/codex/codex-structured-journal-translation.ts @@ -7,7 +7,10 @@ import { type CodexJournalTranslatorDeps } from './codex-structured-journal-contracts' import { settleCodexJournalSession } from './codex-structured-journal-settlement' -import { restoreCodexJournalThread } from './codex-structured-journal-translation-restore' +import { + restoreCodexHistoryItem, + restoreCodexJournalThread +} from './codex-structured-journal-translation-restore' import { CodexJournalTurnBoundaries } from './codex-structured-journal-translation-turn-boundaries' import { createCodexJournalTranslatorWriters } from './codex-structured-journal-translation-writers' import { publishCodexTurnLifecycle } from './codex-structured-journal-translation-turns' @@ -113,16 +116,13 @@ export function createCodexJournalTranslator( thread, currentTurnIds: activeTurns.byThread, ordinals: items.ordinals, - handleItem: (event) => { - const compaction = compactions.handle(event) - if (compaction) { - return compaction - } - const translated = items.handle(event, 'history') - return translated.handled - ? translated.admission - : { accepted: false, reason: 'untranslated' } - }, + handleItem: (event) => + restoreCodexHistoryItem(event, { + primaryThreadId: deps.primaryThreadId?.() ?? null, + compactions, + items, + executions: subagents.executions + }), ...(deps.sessionId !== undefined ? { restoreTurnLifecycle: (turnLifecycle) => diff --git a/src/main/codex/codex-subagent-activity.ts b/src/main/codex/codex-subagent-activity.ts index 9761f8634f8c..0a00297dff8a 100644 --- a/src/main/codex/codex-subagent-activity.ts +++ b/src/main/codex/codex-subagent-activity.ts @@ -6,12 +6,19 @@ // * `agentPath` is a tree path (`/root`, `/root/list_directory`); the trailing // segment is a semantic task name and the only label available. There is no // `thread/started` for a child, so nickname/role/depth do not exist. -// * `agentsStates` on `collabAgentToolCall` arrived empty (`{}`) throughout the -// probe, so nothing here reads it; child turn events own execution state. +// * Codex's DEFAULT multi-agent mode sends no `subAgentActivity` at all; a +// helper appears only in the `collabAgentToolCall` items that spawn, message, +// wait on or close it (read in `codex-collab-agent-tool-call.ts`). Either item +// announces the same child, keyed by its thread id, and child turn events own +// its execution state. // * `thread/tokenUsage/updated` reports a per-thread RUNNING TOTAL, so the // latest frame replaces the previous one — it is never accumulated. -import type { CodexThreadItem } from './codex-structured-item-translation' +import { + codexCollabHelperLabel, + readCodexCollabAgentToolCall +} from './codex-collab-agent-tool-call' +import type { CodexThreadItem } from './codex-thread-item-identity' export const CODEX_SUBAGENT_ITEM_TYPE = 'subAgentActivity' export const CODEX_TOKEN_USAGE_METHOD = 'thread/tokenUsage/updated' @@ -85,6 +92,66 @@ export function codexSubagentLabel(activity: CodexSubagentActivity): string | nu return trailing !== undefined && trailing.length > 0 ? trailing : null } +/** A child the item says exists, from either wire shape Codex announces one with. */ +export type CodexSubagentAnnouncement = { + agentThreadId: string + label: string | null + /** The item's turn is the one the child was spawned or messaged from. */ + namesParentTurn: boolean + /** The item is the spawn itself, so the thread that carried it spawned the child. */ + spawned: boolean +} + +/** Collab calls that hand a helper work from the caller's turn, as an `interacted` activity does. */ +const CODEX_COLLAB_TOOLS_THAT_TASK_A_HELPER: ReadonlySet = new Set([ + 'spawnAgent', + 'sendInput' +]) + +/** The children an item announces: a `subAgentActivity`'s, or every helper a collab call names, + * keyed by thread id so repeats announce one child. Neither the tree root nor the session's own + * thread is a child. Any call counts, since a helper whose spawn was never seen (compacted out of + * history) is known only from the calls on it. A spawn counts whatever its status: Codex reports + * `failed` for a helper that errored at birth, yet names the thread it created, and that thread + * can still run. */ +export function readCodexSubagentAnnouncements( + item: CodexThreadItem, + primaryThreadId: string | null +): CodexSubagentAnnouncement[] { + return readAnnouncements(item).filter(({ agentThreadId }) => agentThreadId !== primaryThreadId) +} + +function readAnnouncements(item: CodexThreadItem): CodexSubagentAnnouncement[] { + const activity = readCodexSubagentActivity(item) + if (activity) { + return isCodexRootAgentActivity(activity) + ? [] + : [ + { + agentThreadId: activity.agentThreadId, + label: codexSubagentLabel(activity), + namesParentTurn: activity.kind === 'started' || activity.kind === 'interacted', + spawned: activity.kind === 'started' + } + ] + } + const call = readCodexCollabAgentToolCall(item) + if (!call) { + return [] + } + const spawned = call.tool === 'spawnAgent' + // Only a spawn's prompt describes its helper; another call's prompt is a later message to it, + // so a helper no spawn named reads as any unnamed subagent does. + const label = spawned ? codexCollabHelperLabel(call.prompt) : null + const namesParentTurn = CODEX_COLLAB_TOOLS_THAT_TASK_A_HELPER.has(call.tool) + return call.helperThreadIds.map((agentThreadId) => ({ + agentThreadId, + label, + namesParentTurn, + spawned + })) +} + export type CodexThreadTokenTotal = { threadId: string; totalTokens: number } /** `{threadId, tokenUsage: {total: {totalTokens}}}`. Older builds put the total diff --git a/src/main/codex/codex-subagent-executions.ts b/src/main/codex/codex-subagent-executions.ts index 87d9327c70c0..0d0fe0c5bf4e 100644 --- a/src/main/codex/codex-subagent-executions.ts +++ b/src/main/codex/codex-subagent-executions.ts @@ -25,10 +25,20 @@ export type CodexExecutionChild = { turnCount: number } +/** Told when a child's current execution changes, whichever frame changed it. */ +export type CodexExecutionListener = (child: Readonly) => void + /** Child turn events own execution; activity items only identify the child. */ export class CodexSubagentExecutions { private readonly children = new Map() private readonly settledTurns = new Map() + private readonly listeners = new Set() + + /** Every reader of a child's execution follows it here, so none needs the frames that end it. */ + onExecutionChanged(listener: CodexExecutionListener): () => void { + this.listeners.add(listener) + return () => this.listeners.delete(listener) + } register( agentThreadId: string, @@ -90,6 +100,7 @@ export class CodexSubagentExecutions { } if (state === 'working' || !child.execution || child.execution.turnId === turnId) { child.execution = execution + this.changed(child) } return { child, execution } } @@ -133,6 +144,7 @@ export class CodexSubagentExecutions { for (const child of this.children.values()) { if (child.execution?.state === 'working') { child.execution = { ...child.execution, state: 'unverifiable' } + this.changed(child) } } } @@ -147,6 +159,12 @@ export class CodexSubagentExecutions { return { children: this.children.size, settledTurns: this.settledTurns.size } } + private changed(child: CodexExecutionChild): void { + for (const listener of this.listeners) { + listener(child) + } + } + private child(agentThreadId: string): CodexExecutionChild | undefined { const existing = this.children.get(agentThreadId) if (existing) { diff --git a/src/main/codex/codex-subagent-roster.ts b/src/main/codex/codex-subagent-roster.ts index db02ac7f4547..042183a6e6f8 100644 --- a/src/main/codex/codex-subagent-roster.ts +++ b/src/main/codex/codex-subagent-roster.ts @@ -1,6 +1,8 @@ // The Codex subagent roster: one journal row per spawn group, revised in place. // -// Activity supplies membership; child turn events supply execution state. +// An announcement supplies membership — a `subAgentActivity` item, or in +// Codex's default multi-agent mode any collab call naming the helper — and +// child turn events supply execution state. // // KNOWN LIMITATION: `groups` is process-local and is never seeded from the // journal, while the row's identity is keyed on the group id alone. So once a @@ -21,9 +23,9 @@ import type { StructuredAgentSessionSinkAdmission } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { - codexSubagentLabel, - isCodexRootAgentActivity, readCodexSubagentActivity, + readCodexSubagentAnnouncements, + type CodexSubagentAnnouncement, readCodexThreadTokenTotal } from './codex-subagent-activity' import { @@ -46,9 +48,9 @@ import { import type { CodexThreadItem } from './codex-structured-item-translation' import { MAX_CODEX_SUBAGENT_GROUPS, - MAX_CODEX_SUBAGENTS_PER_GROUP, - MAX_CODEX_TOKEN_USAGE_THREADS + MAX_CODEX_SUBAGENTS_PER_GROUP } from './codex-structured-journal-limits' +import { CodexThreadTokenTotals } from './codex-thread-token-totals' const ADMITTED: StructuredAgentSessionSinkAdmission = { accepted: true } @@ -64,48 +66,54 @@ export type CodexSubagentRosterDeps = { export class CodexSubagentRoster { private readonly groups = new Map() - /** Latest reported total per thread, kept regardless of roster membership: a - * usage frame can arrive before the child's first activity item, and filtering - * at receipt would lose it permanently. Children are selected at write time; - * the map itself is LRU-capped in `handleTokenUsage`. */ - private readonly tokensByThread = new Map() + /** Every thread's total, members or not; children are selected at write time. */ + private readonly tokensByThread = new CodexThreadTokenTotals() private readonly now: () => number - private readonly executions: CodexSubagentExecutions + /** The one owner of child membership and turn state; the rows of calls on a helper read it too. */ + readonly executions: CodexSubagentExecutions + private readonly unfollow: () => void /** Who produced a row, from what this roster learned about each child thread. */ readonly linkage: CodexSubagentLinkage constructor(private readonly deps: CodexSubagentRosterDeps) { this.now = deps.now ?? (() => Date.now()) this.executions = deps.executions ?? new CodexSubagentExecutions() + // The row follows the executions, so every frame that ends a child's turn — its own + // `turn/completed` (a failed one included) or its thread closing — settles it. + // A refused write clears `lastSerialized`, so the next write of the group retries it. + this.unfollow = this.executions.onExecutionChanged( + (child) => child.execution && this.follow(child, child.execution) + ) this.linkage = new CodexSubagentLinkage({ primaryThreadId: deps.primaryThreadId, executions: this.executions }) } - /** Consume a `subAgentActivity` item. Returns null when the item is not one. */ + /** Consume an item that announces children. Null means the item is not this roster's to render: + * a `subAgentActivity` item renders as the roster row alone, while a collab call keeps its own + * row, so it is claimed only to hand back a refused write. */ handleItem(input: { threadId: string turnId: string | null item: CodexThreadItem }): StructuredAgentSessionSinkAdmission | null { - const activity = readCodexSubagentActivity(input.item) - if (!activity) { - return null - } - // The root node is the parent turn itself, not a child it spawned. - if ( - activity.agentThreadId === this.deps.primaryThreadId() || - isCodexRootAgentActivity(activity) - ) { - return ADMITTED - } + const refused = readCodexSubagentAnnouncements(input.item, this.deps.primaryThreadId()) + .map((announcement) => this.announce(input, announcement)) + .find((admission) => !admission.accepted) + return refused ?? (readCodexSubagentActivity(input.item) !== null ? ADMITTED : null) + } + + private announce( + input: { threadId: string; turnId: string | null }, + announcement: CodexSubagentAnnouncement + ): StructuredAgentSessionSinkAdmission { const child = this.executions.register( - activity.agentThreadId, - codexSubagentLabel(activity), - activity.kind === 'started' || activity.kind === 'interacted' ? input.turnId : undefined, - // Only `started` names the spawner: other kinds ride whichever agent acted. - activity.kind === 'started' ? input.threadId : undefined + announcement.agentThreadId, + announcement.label, + announcement.namesParentTurn ? input.turnId : undefined, + // Only a spawn names the spawner: other announcements ride whichever agent acted. + announcement.spawned ? input.threadId : undefined ) if (!child?.execution) { return ADMITTED @@ -146,12 +154,19 @@ export class CodexSubagentRoster { return ADMITTED } const observed = this.executions.observeTurn(input.threadId, input.turnId, input.state) - if (!observed || !observed.child.registered) { + // Followed already if the execution changed; re-derived (idempotently) for its admission. + return observed ? this.follow(observed.child, observed.execution) : ADMITTED + } + + private follow( + child: Readonly, + execution: CodexChildExecution + ): StructuredAgentSessionSinkAdmission { + if (!child.registered) { return ADMITTED } - const { child, execution } = observed - if (input.state === 'working') { - const parent = this.deps.primaryThreadId() ?? input.threadId + if (execution.state === 'working') { + const parent = this.deps.primaryThreadId() ?? child.agentThreadId const group = this.executionGroup(child.agentThreadId, execution.turnId) ?? this.groupFor(parent, this.deps.activeTurn(parent) ?? child.parentTurnId) @@ -159,7 +174,7 @@ export class CodexSubagentRoster { return this.write(group) } for (const group of this.groups.values()) { - if (group.executionTurns.get(input.threadId) !== input.turnId) { + if (group.executionTurns.get(child.agentThreadId) !== execution.turnId) { continue } this.recordExecution(group, child, execution) @@ -177,19 +192,7 @@ export class CodexSubagentRoster { if (!usage) { return null } - // A running total: the newest frame REPLACES the previous one. Summing - // updates would multiply a single child's usage by its frame count. - // Re-insert so the eviction scan below sees recency: `set` on an existing - // key keeps its original position, which would age out an active thread. - this.tokensByThread.delete(usage.threadId) - this.tokensByThread.set(usage.threadId, usage.totalTokens) - while (this.tokensByThread.size > MAX_CODEX_TOKEN_USAGE_THREADS) { - const oldest = this.tokensByThread.keys().next().value - if (typeof oldest !== 'string') { - break - } - this.tokensByThread.delete(oldest) - } + this.tokensByThread.record(usage.threadId, usage.totalTokens) for (const group of this.groups.values()) { if (!group.entries.has(usage.threadId)) { continue @@ -222,6 +225,7 @@ export class CodexSubagentRoster { } dispose(): void { + this.unfollow() this.groups.clear() this.tokensByThread.clear() } diff --git a/src/main/codex/codex-thread-token-totals.test.ts b/src/main/codex/codex-thread-token-totals.test.ts new file mode 100644 index 000000000000..1dab7dc6f6e7 --- /dev/null +++ b/src/main/codex/codex-thread-token-totals.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from 'vitest' +import { MAX_CODEX_TOKEN_USAGE_THREADS } from './codex-structured-journal-limits' +import { CodexThreadTokenTotals } from './codex-thread-token-totals' + +describe('CodexThreadTokenTotals', () => { + it("replaces a thread's total with its newest report instead of summing reports", () => { + const totals = new CodexThreadTokenTotals() + totals.record('child-1', 100) + totals.record('child-1', 250) + expect(totals.get('child-1')).toBe(250) + }) + + it('evicts by recency of report, so a thread still reporting outlives quieter ones', () => { + const totals = new CodexThreadTokenTotals() + totals.record('child-1', 1) + for (let index = 0; index < MAX_CODEX_TOKEN_USAGE_THREADS - 1; index++) { + totals.record(`other-${index}`, index) + } + totals.record('child-1', 2) + totals.record('newest', 3) + expect(totals.get('child-1')).toBe(2) + expect(totals.get('other-0')).toBeUndefined() + }) +}) diff --git a/src/main/codex/codex-thread-token-totals.ts b/src/main/codex/codex-thread-token-totals.ts new file mode 100644 index 000000000000..23bc6a9d1995 --- /dev/null +++ b/src/main/codex/codex-thread-token-totals.ts @@ -0,0 +1,32 @@ +import { MAX_CODEX_TOKEN_USAGE_THREADS } from './codex-structured-journal-limits' + +/** Latest reported token total per thread, kept regardless of roster membership: a + * usage frame can arrive before the child's first activity item, and filtering at + * receipt would lose it permanently. LRU-capped at MAX_CODEX_TOKEN_USAGE_THREADS. */ +export class CodexThreadTokenTotals { + private readonly totals = new Map() + + /** A running total: the newest frame REPLACES the previous one. Summing updates + * would multiply a single child's usage by its frame count. */ + record(threadId: string, totalTokens: number): void { + // Re-insert so eviction sees recency: `set` on an existing key keeps its + // original position, which would age out an active thread. + this.totals.delete(threadId) + this.totals.set(threadId, totalTokens) + while (this.totals.size > MAX_CODEX_TOKEN_USAGE_THREADS) { + const oldest = this.totals.keys().next().value + if (typeof oldest !== 'string') { + break + } + this.totals.delete(oldest) + } + } + + get(threadId: string): number | undefined { + return this.totals.get(threadId) + } + + clear(): void { + this.totals.clear() + } +} diff --git a/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts b/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts index 22b3b26a453c..b4d2286145b0 100644 --- a/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts +++ b/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts @@ -238,10 +238,10 @@ const CODEX_ITEM_CLASSIFICATIONS: Record = // `restoreThread` replays them straight through `items.handle`, which is where // the classification earns its keep. // - // `collabAgentToolCall` is deliberately NOT suppressed with it. Nothing - // guarantees a session reports subagent work as `subAgentActivity` at all; one - // that only ever emits the collab tool call gets no roster row, and suppressing - // that too would leave its fan-out showing nothing. + // `collabAgentToolCall` is deliberately NOT suppressed with it. Codex's default + // multi-agent mode reports subagent work ONLY as that call, which renders as its + // own tool row and never reaches this catalog; were it ever to, suppressing it + // would leave that session's fan-out showing nothing. [CODEX_SUBAGENT_ITEM_TYPE]: 'status-chrome', // `{id, durationMs}` and nothing else — Codex's own transcript renders it as // nothing at all. Every other item type this build does not model carries text diff --git a/src/renderer/src/components/native-chat/native-chat-subagent-delegation.ts b/src/renderer/src/components/native-chat/native-chat-subagent-delegation.ts index 6d3012c271a1..392e858aaa85 100644 --- a/src/renderer/src/components/native-chat/native-chat-subagent-delegation.ts +++ b/src/renderer/src/components/native-chat/native-chat-subagent-delegation.ts @@ -3,10 +3,15 @@ // A roster row names the agents it announces, in the order they were added. A Codex collab // call (spawn, wait, resume, message, close) names the agents it acts on by thread id, which // is the agent id, and delegates to the first; one naming none (a spawn still starting, a -// wait on any agent) is ordinary output. A Claude spawn call names no agent — only its call id, which is re-minted on resume +// wait on any agent) is ordinary output. The host writes it as a tool call; journals from +// before that hold it as a raw status row. A Claude spawn call names no agent — only its call id, which is re-minted on resume // and never a join key — so it belongs to the roster that announces its agent. A drawn row is // a spawn when its newest part is one. +import { + codexCollabRowAgentIds, + isCodexCollabToolRowName +} from '../../../../shared/codex-collab-agent-tools' import { isSubagentGroupBlock, type NativeChatMessage } from '../../../../shared/native-chat-types' export type NativeChatSubagentDelegation = @@ -49,9 +54,14 @@ function derive(message: NativeChatMessage): NativeChatSubagentDelegation | null } // A tool run folds into the message before it, so its last call is the newest thing in it. const newest = message.blocks.findLast((block) => block.type !== 'tool-result') - return newest?.type === 'tool-call' && CLAUDE_SPAWN_TOOLS.has(newest.name) - ? { kind: 'spawn' } - : null + if (newest?.type !== 'tool-call') { + return null + } + if (isCodexCollabToolRowName(newest.name)) { + const [agentId] = codexCollabRowAgentIds(newest.input) + return agentId === undefined ? null : { kind: 'call', agentId } + } + return CLAUDE_SPAWN_TOOLS.has(newest.name) ? { kind: 'spawn' } : null } function receiverThreadIds(head: string): readonly string[] { diff --git a/src/shared/codex-collab-agent-tools.ts b/src/shared/codex-collab-agent-tools.ts new file mode 100644 index 000000000000..a19de5aff2a2 --- /dev/null +++ b/src/shared/codex-collab-agent-tools.ts @@ -0,0 +1,41 @@ +// The tool rows a Codex `collabAgentToolCall` item becomes: the calls an agent makes to spawn, +// message, wait on and close its helpers. The host names each row; readers recognise it here. + +/** The name the model called the tool by, which is what its row shows. The wire item spells the + * tool in camelCase; the model-facing name is snake_case, so this map is not a spelling fix. */ +const CODEX_COLLAB_TOOL_NAMES: ReadonlyMap = new Map([ + ['spawnAgent', 'spawn_agent'], + ['sendInput', 'send_input'], + ['resumeAgent', 'resume_agent'], + ['wait', 'wait_agent'], + ['closeAgent', 'close_agent'], + ['sendMessage', 'send_message'], + ['followupTask', 'followup_task'], + ['interruptAgent', 'interrupt_agent'], + ['listAgents', 'list_agents'] +]) + +const CODEX_COLLAB_ROW_NAMES: ReadonlySet = new Set(CODEX_COLLAB_TOOL_NAMES.values()) + +/** A tool Codex adds later keeps its wire spelling. */ +export function codexCollabToolRowName(wireTool: string): string { + return CODEX_COLLAB_TOOL_NAMES.get(wireTool) ?? wireTool +} + +export function isCodexCollabToolRowName(name: string): boolean { + return CODEX_COLLAB_ROW_NAMES.has(name) +} + +/** The row input's key for the helper thread ids a call acts on, in the order the call names them. */ +export const CODEX_COLLAB_ROW_AGENTS_KEY = 'agents' + +/** The helper thread ids a collab call row names; none when its input was clipped. */ +export function codexCollabRowAgentIds(input: unknown): string[] { + if (typeof input !== 'object' || input === null || !(CODEX_COLLAB_ROW_AGENTS_KEY in input)) { + return [] + } + const ids = input[CODEX_COLLAB_ROW_AGENTS_KEY] + return Array.isArray(ids) + ? ids.filter((id): id is string => typeof id === 'string' && id.length > 0) + : [] +} diff --git a/src/shared/native-chat-tool-fold.ts b/src/shared/native-chat-tool-fold.ts index 7b5509abe8f5..d48d48120ca6 100644 --- a/src/shared/native-chat-tool-fold.ts +++ b/src/shared/native-chat-tool-fold.ts @@ -152,16 +152,29 @@ export type NativeChatToolPair = { result?: NativeChatToolResultBlock } -/** Pair calls and results by FIFO ordinal because transcript blocks carry no tool ids. */ +/** Where in `unanswered` (oldest first) the call `result` answers is, or -1 for none. */ +function answeredToolCallIndex( + unanswered: readonly T[], + result: NativeChatToolResultBlock, + callIdOf: (entry: T) => string | undefined +): number { + if (result.callId === undefined) { + return unanswered.length > 0 ? 0 : -1 + } + return unanswered.findIndex((entry) => callIdOf(entry) === result.callId) +} + +/** Pair results to calls by `answeredToolCallIndex`. Every reader of a run pairs through this + * (`pairNativeChatToolResults` included), so they all agree on who owns an output. */ export function pairToolBlocks( blocks: readonly NativeChatBlock[], limit = Infinity ): NativeChatToolPair[] { const pairs: NativeChatToolPair[] = [] + /** Slots of retained calls not yet answered, oldest first. */ const callSlots: number[] = [] - let resultOrdinal = 0 for (const block of blocks) { - if (pairs.length >= limit && resultOrdinal >= callSlots.length) { + if (pairs.length >= limit && callSlots.length === 0) { break } if (block.type === 'tool-call') { @@ -174,13 +187,13 @@ export function pairToolBlocks( if (block.type !== 'tool-result') { continue } - const slot = callSlots[resultOrdinal] + const answered = answeredToolCallIndex(callSlots, block, (slot) => pairs[slot]?.call?.callId) + const [slot] = answered === -1 ? [] : callSlots.splice(answered, 1) if (slot === undefined) { if (pairs.length < limit) { pairs.push({ result: block }) } } else { - resultOrdinal += 1 pairs[slot]!.result = block } } diff --git a/src/shared/native-chat-tool-icon.test.ts b/src/shared/native-chat-tool-icon.test.ts index f8cabed950cb..f84522443d2d 100644 --- a/src/shared/native-chat-tool-icon.test.ts +++ b/src/shared/native-chat-tool-icon.test.ts @@ -56,6 +56,13 @@ describe('native chat tool icons', () => { expect(nativeChatToolCategory('web search')).toBe('webSearch') }) + it('counts no Codex helper call as running an agent, since a spawn ends once its helper starts', () => { + expect(nativeChatToolCategory('task')).toBe('subAgentActivity') + expect(nativeChatToolCategory('spawn_agent')).toBeNull() + expect(nativeChatToolCategory('wait_agent')).toBeNull() + expect(nativeChatToolCategory('close_agent')).toBeNull() + }) + it('maps the tool names the Claude lane renders verbatim', () => { expect(nativeChatToolIconName('Read')).toBe('eye') expect(nativeChatToolIconName('Bash')).toBe('square-terminal') diff --git a/src/shared/native-chat-tool-icon.ts b/src/shared/native-chat-tool-icon.ts index 1bdc5736c62f..9da8672374a2 100644 --- a/src/shared/native-chat-tool-icon.ts +++ b/src/shared/native-chat-tool-icon.ts @@ -82,6 +82,8 @@ const CATEGORY_BY_ROW_WORD = new Map([ // Claude's tool names, which its lane renders verbatim. ['grep', 'search'], ['glob', 'search'], + // Claude's Task call lasts as long as its subagent. Codex's `spawn_agent` ends once the helper + // starts, so it is deliberately absent: its roster row, not the call, stands for the helper. ['task', 'subAgentActivity'], ['webfetch', 'webSearch'], ['todowrite', 'todoList'], diff --git a/src/shared/native-chat-tool-pair-limit.test.ts b/src/shared/native-chat-tool-pair-limit.test.ts index f279b70c996f..d54bfc9b89ba 100644 --- a/src/shared/native-chat-tool-pair-limit.test.ts +++ b/src/shared/native-chat-tool-pair-limit.test.ts @@ -77,6 +77,43 @@ describe('tool pair limits', () => { expect(pairToolBlocks(blocks, 2)).toEqual(original(blocks, 2)) }) + it('gives a result that names its call to that call, past one that finished with no output', () => { + const named = (name: string, callId: string): NativeChatBlock => ({ + type: 'tool-call', + name, + input: {}, + callId + }) + const [spawn, wait, shell] = [ + named('spawn_agent', 's'), + named('wait_agent', 'w'), + named('Bash', 'x') + ] + const waited: NativeChatBlock = { type: 'tool-result', output: 'CHILD_REPLY', callId: 'w' } + const ran: NativeChatBlock = { type: 'tool-result', output: 'CHILD_DONE', callId: 'x' } + expect(pairToolBlocks([spawn, wait, waited, shell, ran])).toEqual([ + { call: spawn }, + { call: wait, result: waited }, + { call: shell, result: ran } + ]) + }) + + it('gives no retained call the output of a call past the limit', () => { + const named = (callId: string): NativeChatBlock => ({ + type: 'tool-call', + name: 'Bash', + input: {}, + callId + }) + const [spawn, shell, later] = [named('s'), named('x'), named('y')] + const ran: NativeChatBlock = { type: 'tool-result', output: 'x', callId: 'x' } + const laterRan: NativeChatBlock = { type: 'tool-result', output: 'y', callId: 'y' } + expect(pairToolBlocks([spawn, shell, ran, later, laterRan], 2)).toEqual([ + { call: spawn }, + { call: shell, result: ran } + ]) + }) + it('keeps a leading stray result and then stops at the limit', () => { const blocks = [result, call, result, call, result] expect(pairToolBlocks(blocks, 1)).toEqual(original(blocks, 1)) diff --git a/src/shared/native-chat-tool-pairing.test.ts b/src/shared/native-chat-tool-pairing.test.ts index 3d2b5f37a974..72d7c46d656a 100644 --- a/src/shared/native-chat-tool-pairing.test.ts +++ b/src/shared/native-chat-tool-pairing.test.ts @@ -43,6 +43,32 @@ describe('pairNativeChatToolResults', () => { expect(pairedResults.has(orphan)).toBe(false) }) + it('gives a result that names its call to that call, past one that finished with no output', () => { + const named = (name: string, callId: string): NativeChatToolCallBlock => ({ + ...call(name), + callId + }) + const [spawn, wait, shell] = [named('spawn', 's'), named('wait', 'w'), named('shell', 'x')] + const [waited, ran]: NativeChatToolResultBlock[] = [ + { ...result('CHILD_REPLY'), callId: 'w' }, + { ...result('CHILD_DONE'), callId: 'x' } + ] + const { resultByCall } = pairNativeChatToolResults([spawn, wait, waited, shell, ran]) + + expect(resultByCall.has(spawn)).toBe(false) + expect(resultByCall.get(wait)).toBe(waited) + expect(resultByCall.get(shell)).toBe(ran) + }) + + it('leaves a result that names a call not in the run unpaired, rather than give it to another', () => { + const waiting: NativeChatToolCallBlock = { ...call('spawn'), callId: 's' } + const stray: NativeChatToolResultBlock = { ...result('elsewhere'), callId: 'missing' } + const { resultByCall, pairedResults } = pairNativeChatToolResults([waiting, stray]) + + expect(resultByCall.has(waiting)).toBe(false) + expect(pairedResults.has(stray)).toBe(false) + }) + it('ignores blocks that are neither a call nor a result', () => { const text: NativeChatBlock = { type: 'text', text: 'hi' } const [a, ra] = [call('a'), result('r')] diff --git a/src/shared/native-chat-tool-pairing.ts b/src/shared/native-chat-tool-pairing.ts index f4dd3db9d399..490008927c13 100644 --- a/src/shared/native-chat-tool-pairing.ts +++ b/src/shared/native-chat-tool-pairing.ts @@ -7,16 +7,18 @@ // as the command that produced it. Pairing lets the call own its output, so the // run reads as the work it did. // -// Pairing is positional, the same FIFO rule `dropUnattributableToolResults` -// already uses to decide a result is attributable at all: a result answers the -// oldest call that has not been answered yet. +// A result that names its call answers that call, and no other: one naming a call +// that is not waiting (outside a bounded window, say) stays unpaired. One that names +// none is paired positionally, the same FIFO rule `dropUnattributableToolResults` +// already uses to decide a result is attributable at all: it answers the oldest call +// that has not been answered yet. Position alone misattributes every later result once +// one call finishes with no output, which is why a producer that knows the call names it. -import { - isToolCallBlock, - isToolResultBlock, - type NativeChatBlock, - type NativeChatToolCallBlock, - type NativeChatToolResultBlock +import { pairToolBlocks } from './native-chat-tool-fold' +import type { + NativeChatBlock, + NativeChatToolCallBlock, + NativeChatToolResultBlock } from './native-chat-types' export type NativeChatToolPairing = { @@ -31,30 +33,17 @@ export const NO_NATIVE_CHAT_TOOL_PAIRING: NativeChatToolPairing = { pairedResults: new Set() } +/** The desktop run's view of `pairToolBlocks`, so every reader of a run pairs with one loop. */ export function pairNativeChatToolResults( blocks: readonly NativeChatBlock[] ): NativeChatToolPairing { const resultByCall = new Map() const pairedResults = new Set() - const unanswered: NativeChatToolCallBlock[] = [] - let answered = 0 - for (const block of blocks) { - if (isToolCallBlock(block)) { - unanswered.push(block) - continue + for (const { call, result } of pairToolBlocks(blocks)) { + if (call && result) { + resultByCall.set(call, result) + pairedResults.add(result) } - if (!isToolResultBlock(block)) { - continue - } - // Results carry no call id, so the journal's FIFO order is the only stable - // attribution available when calls are interleaved. - const call = unanswered[answered] - if (call === undefined) { - continue - } - answered += 1 - resultByCall.set(call, block) - pairedResults.add(block) } return { resultByCall, pairedResults } } diff --git a/src/shared/native-chat-types.ts b/src/shared/native-chat-types.ts index a1a8b3a51430..6d9d344f5069 100644 --- a/src/shared/native-chat-types.ts +++ b/src/shared/native-chat-types.ts @@ -96,6 +96,8 @@ export type NativeChatToolResultBlock = { type: 'tool-result' output: string isError?: boolean + /** The call this result answers, when the producer knows it; otherwise pairing is positional. */ + callId?: string /** Present only for edit tools whose result reported resolved hunks. */ editPatch?: NativeChatEditPatch } diff --git a/src/shared/structured-agent-session-projection.test.ts b/src/shared/structured-agent-session-projection.test.ts index 8570b0c3374d..c91bd7102b82 100644 --- a/src/shared/structured-agent-session-projection.test.ts +++ b/src/shared/structured-agent-session-projection.test.ts @@ -124,6 +124,23 @@ describe('structured agent session status projection', () => { }) }) + it('names the call a tool row output answers, so a run pairs it by id', () => { + const projected = projectStructuredItemToNativeChat( + item('wait', 1, { + kind: 'tool-call', + name: 'wait_agent', + callId: 'call-wait', + input: null, + state: 'completed', + output: { head: 'CHILD_REPLY', digest: 'd', byteLength: 11, truncated: false } + }) + ) + expect(projected?.blocks).toEqual([ + expect.objectContaining({ type: 'tool-call', callId: 'call-wait' }), + { type: 'tool-result', output: 'CHILD_REPLY', isError: false, callId: 'call-wait' } + ]) + }) + it('projects running, attention, and completed lifecycle states', () => { const running = item('running', 1, { kind: 'status', diff --git a/src/shared/structured-agent-session-projection.ts b/src/shared/structured-agent-session-projection.ts index 39a3d0a67e4b..0502a32798f0 100644 --- a/src/shared/structured-agent-session-projection.ts +++ b/src/shared/structured-agent-session-projection.ts @@ -89,7 +89,9 @@ function itemBlocks(item: AgentJournalRenderItem): { { type: 'tool-result' as const, output: boundedText(body.output), - isError: body.state === 'failed' + isError: body.state === 'failed', + // The call and its output are one journal row, so the result names its call. + ...(body.callId !== undefined ? { callId: body.callId } : {}) } ] : [])