diff --git a/README.md b/README.md index 916215500..7d03989f4 100644 --- a/README.md +++ b/README.md @@ -96,9 +96,11 @@ See [readme-dev.md](readme-dev.md) for local client configuration, binary packag ### Subagent sessions -Subagent sessions follow the draft [ACP subagent RFD](https://github.com/agentclientprotocol/agent-client-protocol/pull/1992) and are enabled only after bilateral capability negotiation during `initialize`. Without native negotiation, the subagent lifecycle stays an ordinary ACP tool call. +Codex subagents are reported as ACP child sessions that follow the ACP [Subagent Sessions RFD](https://agentclientprotocol.com/rfds/subagents). A v1 client opts in with `clientCapabilities.subagents: {}` during `initialize`. A v2 client gets them without negotiation. The adapter announces each child with `subagent_update`, reports delegated tasks as `session_message`, and references children from collaboration tool calls with `session` content. A client can cancel a child's current work with `session/cancel`. -See [docs/subagent-sessions.md](docs/subagent-sessions.md) for the negotiation, lifecycle events, `session/load` reconstruction, and legacy fallback details. +JetBrains AIR clients keep an older private contract instead. Without either contract, the subagent lifecycle stays an ordinary ACP tool call. + +See [docs/subagent-sessions.md](docs/subagent-sessions.md) for both contracts, `session/load` replay, and the fallback. ### Background terminal tasks diff --git a/docs/air-extensions.md b/docs/air-extensions.md index b3e8d3327..d0b771d3d 100644 --- a/docs/air-extensions.md +++ b/docs/air-extensions.md @@ -1005,15 +1005,14 @@ Recovery never removes the transcript record. ## Native subagent sessions -The adapter implements the draft [ACP subagent RFD](https://github.com/agentclientprotocol/agent-client-protocol/pull/1992). -[Subagent sessions](subagent-sessions.md) describes the lifecycle. +AIR keeps the private subagent contract that predates the ACP Subagent Sessions RFD. +[Subagent sessions](subagent-sessions.md#air-legacy-contract) describes its lifecycle and the standard contract of other clients. This section covers only the AIR bridge. -- The canonical client field is `clientCapabilities.subagents: {}`. -- Released ACP SDKs can strip that draft field. - AIR can instead declare `nativeSubagentSessions` in `_meta.jetbrains.air.capabilities`. -- Either signal enables native subagent sessions. New clients must prefer the canonical field. -- The agent always advertises `agentCapabilities.sessionCapabilities.subagents`. It advertises `nativeSubagentSessions` to AIR. +- AIR declares `nativeSubagentSessions` in `_meta.jetbrains.air.capabilities`, or on v1 `clientCapabilities.subagents: {}`. +- Either signal enables native subagent sessions for AIR. +- The agent advertises `agentCapabilities.sessionCapabilities.subagents` and `nativeSubagentSessions` to AIR only. + A client that is not AIR gets neither, because the RFD defines no Agent-side capability. - Without either signal, a Codex subagent stays an ordinary tool call. AIR gets `_meta.jetbrains.air.subagent: true` on a spawn. Another client gets the tool call without `_meta`. diff --git a/docs/subagent-sessions.md b/docs/subagent-sessions.md index f3e576d7e..c1fa71a72 100644 --- a/docs/subagent-sessions.md +++ b/docs/subagent-sessions.md @@ -1,36 +1,79 @@ # Native subagent sessions -This document describes the draft [ACP subagent RFD](https://github.com/agentclientprotocol/agent-client-protocol/pull/1992) as implemented by `codex-acp`. The implementation is in [codex-acp PR #419](https://github.com/agentclientprotocol/codex-acp/pull/419). The Claude reference is [claude-agent-acp PR #1017](https://github.com/agentclientprotocol/claude-agent-acp/pull/1017). +`codex-acp` reports Codex subagents as ACP child sessions through one of two contracts. A connection gets one contract, never both. -## Capability negotiation +- The **standard contract** follows the ACP [Subagent Sessions RFD](https://agentclientprotocol.com/rfds/subagents): `subagent_update`, `session_message`, and `session` tool call content. +- The **AIR legacy contract** is the older private contract for JetBrains AIR clients: `subagent_spawned` and `subagent_state_update`. -Subagents require bilateral capability negotiation during `initialize`. +Without either contract, a subagent stays an ordinary tool call on the root session. See [Fallback](#fallback). -- The canonical client field is `clientCapabilities.subagents: {}`. -- The agent returns `agentCapabilities.sessionCapabilities.subagents: {}`. -- Because released SDKs may strip the draft field, AIR clients can instead advertise `nativeSubagentSessions` in `_meta.jetbrains.air.capabilities`; this adapter advertises that key to AIR in its initialize response. See [AIR extensions](air-extensions.md#native-subagent-sessions). -- New clients and agents must prefer the canonical field. +## Standard contract -## Lifecycle events +### Capability negotiation -- The adapter sends `subagent_spawned` before any child output. -- It uses the child session ID for later messages, thoughts, plans, tools, permissions, and elicitations. -- It sends one `subagent_state_update` on the immediate parent. -- The adapter advertises an empty child capability object. It does not support targeted child cancel or close operations. -- Child permission and elicitation controls remain visible through the root session. +- A v1 Client opts in with `clientCapabilities.subagents: {}`. Without it, the adapter sends none of the updates below. +- A v2 Client needs no capability. Subagent updates are part of the v2 session model. +- There is no Agent-side capability. The `initialize` response to a standard Client does not contain `agentCapabilities.sessionCapabilities.subagents`. + +### Association updates + +- The adapter sends `subagent_update` on the immediate parent before any update, permission request, or elicitation that bears the child's session ID. Ancestors are announced before descendants. +- A child keeps one session ID across assignments, idle periods, and replay. +- `title` comes from the Codex agent path or thread name. +- `description` is the role that Codex assigned to the child (`agentRole`). It is omitted when Codex reports no role. It never holds the task prompt. +- `capabilities.cancel` is advertised only while the child's current turn can be interrupted. `session/cancel` with the child's ID interrupts that work and its observed descendants. Every other Client request for a child session is rejected. +- `state` reports `running`, `requires_action`, `idle` with a `stopReason`, or `unknown` when observability is lost. A v2 Client also gets the same `state_update` on the child session. There is no terminal child state. +- `cwd` is reported as described in [Directories](#directories). + +### Delegated tasks -## Session load +The task a session gives a child is reported as `session_message`, not as a user message and not in `description`. -`session/load` reconstructs the child tree from Codex history. It reports an orphan as `disconnected` when the history does not prove an outcome. Live timeout, `shutdown`, and `notFound` fallbacks currently use `failed`; this differs from the draft rule for an unknown outcome. +- **Incoming view.** A `userMessage` item in the child's Codex thread is reported on the child session, before the child's response. `messageId` is the Codex item ID. `recipientSessionId` is the child. +- **Outgoing view.** A completed `spawnAgent`, `sendInput`, `sendMessage`, or `followupTask` collaboration call with a prompt is reported on the sender's session. `messageId` is `:to:`. A failed or interrupted call reports no message. +- **Linking.** The two views are linked with `senderMessageId` and `recipientMessageId` when the same text was observed for the same recipient on both sides. Whichever view is observed second carries the reference, and a metadata-only `session_message` adds the reference to the first. +- The incoming view names `senderSessionId` only when it is linked. Codex does not record the sender on the child's item, so an unlinked incoming view names only its recipient. Subagents recorded only as `subAgentActivity` items have no prompt on the parent side, so their tasks have no outgoing view and no sender. +- Neither view is derived from the other. A view is reported only when its Codex item was observed. +- Message IDs are the same live and on replay. -## Standard subagent directories +### Collaboration tool calls -On standard subagent connections, directory information is part of `subagent_update` on the immediate parent. This applies to both ACP versions. It does not change AIR's legacy subagent events. +- The `spawnAgent` call is not shown. The announcement and the task message report the delegation. +- Other collaboration calls (`wait`, `sendInput`, `closeAgent`, and so on) stay visible on the sender's session. Their `content` holds one `{ "type": "session", "sessionId": ... }` entry for each receiver that is already announced. `rawInput` still holds the Codex thread IDs. -A verified loaded-child read reports its absolute `cwd`, including a worktree directory. Known child settings changes update that value. Announcements do not reset directory fields while readback is pending. Omission preserves a retained override; `null` would explicitly restore inheritance from the immediate parent, not report an unknown directory. Readback does not block child output, and a superseded read cannot replace newer settings. +### Session load + +`session/load`, and v2 `session/resume` with `replayFrom`, replay each recorded child conversation once, before the response. Replay reports the same announcements, session messages, and session references as the live path. It reports the latest recorded turn outcome as a historical `idle` state. A missing or incomplete child transcript adds one advisory thought on the root session. Replay grants no cancel capability. + +### Directories + +Directory information is part of `subagent_update` on the immediate parent, on both ACP versions. + +A verified loaded-child read reports its absolute `cwd`, including a worktree directory. Known child settings changes update that value. Announcements do not reset directory fields while readback is pending. Omission preserves a retained override. `null` would explicitly restore inheritance from the immediate parent, not report an unknown directory. Readback does not block child output, and a superseded read cannot replace newer settings. The adapter verifies the child's ID, parent, and loaded state before accepting readback. It does not report relative paths or infer directories from provider IDs. Codex does not expose an authoritative ACP additional-directory list for children, so the adapter omits that field. It does not copy parent roots or treat writable sandbox paths as workspace roots. `[]` would mean explicitly no additional directories, not unknown directories. -## Legacy fallback +## AIR legacy contract + +This contract applies only to a Client that declares `_meta.jetbrains.air`. It does not use `subagent_update`, `session_message`, or `session` tool call content. See [AIR extensions](air-extensions.md#native-subagent-sessions). + +### Capability negotiation + +- AIR opts in with `nativeSubagentSessions` in `_meta.jetbrains.air.capabilities`, or on v1 with `clientCapabilities.subagents: {}`. +- The adapter answers an AIR Client with `agentCapabilities.sessionCapabilities.subagents: {}` on v1 and lists `nativeSubagentSessions` in its own `_meta.jetbrains.air.capabilities`. + +### Lifecycle events + +- The adapter sends `subagent_spawned` before any child output. Its `task` field holds the prompt. +- It uses the child session ID for later messages, thoughts, plans, tools, permissions, and elicitations. +- It sends one `subagent_state_update` on the immediate parent, with `completed`, `failed`, `cancelled`, or `disconnected`. +- The adapter advertises an empty child capability object. It does not support targeted child cancel or close operations. +- On v2 both updates are sent as `_subagent_update`. + +### Session load + +`session/load` reconstructs the child tree from Codex history. It reports an orphan as `disconnected` when the history does not prove an outcome. Live timeout, `shutdown`, and `notFound` fallbacks use `failed`. + +## Fallback -Without native negotiation, subagent lifecycle stays an ordinary ACP tool call. Child permission and elicitation requests stay on the root session. +Without either contract, the subagent lifecycle stays an ordinary ACP tool call. Child permission and elicitation requests stay on the root session. diff --git a/package-lock.json b/package-lock.json index 58893d174..f3051ee73 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,7 +9,7 @@ "version": "2.6.0", "license": "Apache-2.0", "dependencies": { - "@agentclientprotocol/sdk": "~2.5.0", + "@agentclientprotocol/sdk": "~2.6.0", "@openai/codex": "^0.166.0", "cross-spawn": "^7.0.6", "diff": "^9.0.0", @@ -32,9 +32,9 @@ } }, "node_modules/@agentclientprotocol/sdk": { - "version": "2.5.0", - "resolved": "https://registry.npmjs.org/@agentclientprotocol/sdk/-/sdk-2.5.0.tgz", - "integrity": "sha512-MFlXeIuwN6nZSK45SAsSnVjbLgQ2r4fnthoF1lQoGJP3PdaPRpOZeCF2Dds8S4cN8yZP5QbxB1bkJccGskmcYw==", + "version": "2.6.0", + "resolved": "https://registry.npmjs.org/@agentclientprotocol/sdk/-/sdk-2.6.0.tgz", + "integrity": "sha512-3FtL6MkAV5F9qxg0FQV8wmBszIm2xuVPoDWOILIFSDQ+Vq3Twdh7Cw6Hk2GOPUbevgrTUpUAqZXQy7Geaj79mw==", "license": "Apache-2.0", "peerDependencies": { "@modelcontextprotocol/server": "^2.4.0", diff --git a/package.json b/package.json index ce7dc5ffb..04f8e6e42 100644 --- a/package.json +++ b/package.json @@ -68,7 +68,7 @@ "vitest": "^5.0.1" }, "dependencies": { - "@agentclientprotocol/sdk": "~2.5.0", + "@agentclientprotocol/sdk": "~2.6.0", "@openai/codex": "^0.166.0", "cross-spawn": "^7.0.6", "diff": "^9.0.0", diff --git a/readme-dev.md b/readme-dev.md index f26698b78..ba61b1fd1 100644 --- a/readme-dev.md +++ b/readme-dev.md @@ -259,7 +259,10 @@ schema alone: Non-AIR v1 clients opt in with `clientCapabilities.subagents: {}`. Non-AIR v2 clients use standard `subagent_update` associations automatically. AIR keeps its existing opt-in, updates, -and generation-based identities; clients do not receive both contracts. +and generation-based identities; clients do not receive both contracts. Only AIR gets the +Agent-side `sessionCapabilities.subagents` in the `initialize` response. +See [docs/subagent-sessions.md](docs/subagent-sessions.md) for delegated task messages and session +references on collaboration tool calls. A standard child keeps one opaque conversation ID across assignments and parent handler changes. Its ancestors are announced before its output, tool updates, or interactions. Permissions and @@ -294,8 +297,8 @@ including nested children, using legacy or paginated history. Announcements and precede the response. Completed snapshots reconcile tool and rendering state before duplicate live payloads are suppressed; live bookkeeping still runs. In-progress commands retain classification without resetting streamed output. Incomplete text has no safe delta cursor; it and other records without a safe -partial representation are omitted rather than replayed over newer content. Agent-owned input is not -rendered as human input. Missing, omitted, or incomplete child history produces a parent advisory; +partial representation are omitted rather than replayed over newer content. Agent-owned input is replayed as +`session_message`, not as human input. Missing, omitted, or incomplete child history produces a parent advisory; an empty transcript alone does not invent a gap. Replay reports the latest recorded child turn's known success, cancellation, or failure as historical diff --git a/src/AcpSessionExtensions.ts b/src/AcpSessionExtensions.ts index d27c3f474..9bb028a55 100644 --- a/src/AcpSessionExtensions.ts +++ b/src/AcpSessionExtensions.ts @@ -4,9 +4,7 @@ import type {StandardSubagentUpdate, SubagentSpawnedUpdate, SubagentStateUpdate} /** Updates the adapter emits, including its existing AIR subagent contract. */ export type AcpSessionUpdate = - | Exclude + | Exclude | SubagentSpawnedUpdate | SubagentStateUpdate | StandardSubagentUpdate @@ -16,7 +14,7 @@ export type AcpSessionNotification = Omit & { update: AcpSessionUpdate; }; -/** The only cast needed until the ACP SDK publishes the extension updates. */ +/** The only cast needed for AIR's private updates, which the ACP SDK does not type. */ export function asSdkSessionNotification( notification: AcpSessionNotification, ): SessionNotification { diff --git a/src/AcpV2SessionUpdate.ts b/src/AcpV2SessionUpdate.ts index 7ea56de5b..1ecf7a9a8 100644 --- a/src/AcpV2SessionUpdate.ts +++ b/src/AcpV2SessionUpdate.ts @@ -63,6 +63,8 @@ export function toV2SessionUpdate(update: AcpSessionUpdate): acpV2.SessionUpdate case "notice": case "compaction_update": case "compaction_summary_chunk": + case "session_message": + case "session_message_chunk": return update; case "config_option_update": return {...update, configOptions: toV2ConfigOptions(update.configOptions)}; @@ -163,6 +165,8 @@ export function toV2ToolCallContent(content: acp.ToolCallContent): acpV2.ToolCal // Same shape on v2; the terminal itself is sent by `toV2SessionUpdates`. case "terminal": return content; + case "session": + return content; case "diff": return {type: "diff", ...toV2Diff(content)}; } diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index e12ce5f6b..dcc15a8a9 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -137,7 +137,6 @@ import { } from "./AcpExtensions"; import {AcpToolCallRenderer} from "./tool-calls/AcpToolCallRenderer"; import {ClientCapabilities} from "./tool-calls/ClientCapabilities"; -import {CollabAgentReporter} from "./tool-calls/reporters/CollabAgentReporter"; import {CommandReporter} from "./tool-calls/reporters/CommandReporter"; import {CompactionReporter} from "./tool-calls/reporters/CompactionReporter"; import {DynamicToolReporter} from "./tool-calls/reporters/DynamicToolReporter"; @@ -220,6 +219,7 @@ import { AIR_META_KEY, AIR_SESSION_FAILURE_KEY, clientSupportsAirCapability, + isAirClient, JETBRAINS_META_KEY, publishAirSessionMetadata, } from "./AirExtension"; @@ -687,7 +687,7 @@ export class CodexAcpServer { id => this.codexAcpClient.appServerClient.runningTurnId(id), models, { read: (id, parent) => this.readSubagentConfiguration(id, parent), enqueuePublication: publish => this.codexAcpClient.enqueueSessionTransition(sessionId, publish), - }) + }, item => this.userMessageContent(item)) : new CodexSubagentEventRouter(sessionId, this.supportsNativeSubagents(), session, release, models); if (router instanceof StandardSubagentEventRouter) this.getSubagentInteractionHandlers(router, sessionId); return router; @@ -830,7 +830,8 @@ export class CodexAcpServer { fork: { }, title: {}, additionalDirectories: {}, - subagents: {}, + // Standard subagent sessions have no Agent-side capability. Only AIR's contract negotiates both sides. + ...(isAirClient(_params.clientCapabilities) ? {subagents: {}} : {}), systemPrompt: {}, }; return { @@ -1985,7 +1986,8 @@ export class CodexAcpServer { } async resumeSessionV2(params: acpV2.ResumeSessionRequest): Promise { - const {replayFrom, systemPrompt, ...rest} = params; + // Initial config option values are not advertised, so they are not applied. + const {replayFrom, systemPrompt, configOptions: _configOptions, ...rest} = params; const request = {...rest, systemPrompt: toV1SystemPrompt(systemPrompt)}; if (replayFrom == null) { await this.resumeSession(request); @@ -2212,7 +2214,8 @@ export class CodexAcpServer { } async newSessionV2(params: acpV2.NewSessionRequest): Promise { - const {systemPrompt, ...rest} = params; + // Initial config option values are not advertised, so they are not applied. + const {systemPrompt, configOptions: _configOptions, ...rest} = params; const request = {...rest, systemPrompt: toV1SystemPrompt(systemPrompt)}; const {sessionId} = await this.newSession(request); return { @@ -3902,6 +3905,7 @@ export class CodexAcpServer { if (!(router instanceof StandardSubagentEventRouter)) return; const session = new ACPSessionConnection(this.connection, sessionId); const replayedEntities = new Set(); + if (threadId === state.sessionId) router.beginHistoryReplay(); const replayChild = async (childThreadId: string, childId: string | null) => { if (childId === null || replayed.has(childThreadId)) return; if (ancestry.has(childThreadId)) { gaps.add(childThreadId); return; } @@ -3909,6 +3913,7 @@ export class CodexAcpServer { try { const history = await this.codexAcpClient.readSubagentHistory(childThreadId); if (!isOpen()) throw new SessionClosedDuringLoadError(); + await router.observeRole(childThreadId, history.thread.agentRole); const parent = router.parentThreadId(childThreadId); if (parent !== undefined) { try { @@ -3952,21 +3957,30 @@ export class CodexAcpServer { continue; } if (item.type === "collabAgentToolCall" && item.tool === "spawnAgent") { - if (item.senderThreadId === threadId) for (const child of item.receiverThreadIds) { - if (!child.trim() || ancestry.has(child)) { gaps.add(child); continue; } - await replayChild(child, await router.announce(child, sessionId, undefined, item.prompt ?? undefined)); + if (item.senderThreadId === threadId) { + const children: Array<[string, string]> = []; + for (const child of item.receiverThreadIds) { + if (!child.trim() || ancestry.has(child)) { gaps.add(child); continue; } + children.push([child, await router.announce(child, sessionId)]); + } + // The delegated task precedes the child's transcript, which holds the incoming view. + await router.reportOutgoingMessages(item); + for (const [child, childId] of children) await replayChild(child, childId); } continue; } + if (item.type === "collabAgentToolCall" && item.senderThreadId === threadId) await router.reportOutgoingMessages(item); if (threadId !== state.sessionId) { - // Agent-owned input is not an ordinary human message. Incomplete text has no safe delta cursor. - if (item.type === "userMessage" || (!completed && !supportsPartialHistory(item))) { + // Incomplete text has no safe delta cursor. + if (item.type !== "userMessage" && !completed && !supportsPartialHistory(item)) { gaps.add(threadId); continue; } - if (!completed) gaps.add(threadId); + if (item.type !== "userMessage" && !completed) gaps.add(threadId); if (!await router.reconcileHistoryItem(threadId, item, completed, turnId)) { gaps.add(threadId); continue; } if (!isOpen()) throw new SessionClosedDuringLoadError(); + // Agent-owned input is a message between sessions, not an ordinary human message. + if (item.type === "userMessage") { await router.reportIncomingMessage(threadId, item); continue; } } for (const update of await this.createHistoryUpdates(item, state)) { await session.replayUpdate(update, replayedEntities); @@ -4120,7 +4134,7 @@ export class CodexAcpServer { case "dynamicToolCall": return [renderer.render(DynamicToolReporter.started(item))]; case "collabAgentToolCall": - return [renderer.render(CollabAgentReporter.started(item))]; + return [renderer.render(sessionState.subagents.legacyCollaborationStarted(item))]; case "webSearch": return [renderer.render(WebSearchReporter.history(item))]; case "imageView": @@ -4141,11 +4155,15 @@ export class CodexAcpServer { } private createUserMessageUpdates(item: ThreadItem & { type: "userMessage" }): UpdateSessionEvent[] { - const updates: UpdateSessionEvent[] = []; // On v2, a message inserted via `session/prompt` is replayed under the id the client // provided then (`clientId`), so it round-trips as the same message on reconnect; v1 has // no such client-minted id and keeps using Codex's own item id. const messageId = this.protocolVersion === 2 ? (item.clientId ?? item.id) : item.id; + return this.userMessageContent(item).map(block => createUserMessageChunk(block, messageId)); + } + + private userMessageContent(item: ThreadItem & { type: "userMessage" }): acp.ContentBlock[] { + const content: acp.ContentBlock[] = []; const contentBlocks = item.content.map(input => this.userInputToContentBlocks(input)); const attachmentUris = new Set(contentBlocks.flatMap((blocks, index) => item.content[index]?.type === "text" @@ -4156,12 +4174,9 @@ export class CodexAcpServer { ? input.path : (input.type === "image" || input.type === "audio") && "url" in input ? input.url : null; if (nativePath !== null && attachmentUris.has(attachmentFileUri(nativePath) ?? "")) continue; - const blocks = contentBlocks[index]!; - for (const block of blocks) { - updates.push(createUserMessageChunk(block, messageId)); - } + content.push(...contentBlocks[index]!); } - return updates; + return content; } private createReasoningUpdates(item: ThreadItem & { type: "reasoning" }): UpdateSessionEvent[] { diff --git a/src/__tests__/CodexACPAgent/data/initialize-v1-routed-response.json b/src/__tests__/CodexACPAgent/data/initialize-v1-routed-response.json index 37ab826b9..afa99af42 100644 --- a/src/__tests__/CodexACPAgent/data/initialize-v1-routed-response.json +++ b/src/__tests__/CodexACPAgent/data/initialize-v1-routed-response.json @@ -23,7 +23,6 @@ "fork": {}, "title": {}, "additionalDirectories": {}, - "subagents": {}, "systemPrompt": {} }, "mcpCapabilities": { diff --git a/src/__tests__/CodexACPAgent/data/live-core-v1-delegation.json b/src/__tests__/CodexACPAgent/data/live-core-v1-delegation.json new file mode 100644 index 000000000..1b74a8dc5 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/live-core-v1-delegation.json @@ -0,0 +1,198 @@ +[ + { + "sessionId": "thread-1", + "update": { + "sessionId": "thread-1:subagent:worker", + "description": "explorer", + "capabilities": {}, + "sessionUpdate": "subagent_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "spawn:to:worker", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "content": [ + { + "text": "Find the bug", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-1", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "senderMessageId": "spawn:to:worker", + "content": [ + { + "text": "Find the bug", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "spawn:to:worker", + "recipientMessageId": "task-1", + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "content": { + "text": "First answer", + "type": "text" + }, + "messageId": "first", + "sessionUpdate": "agent_message_chunk" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "send", + "title": "sendInput", + "kind": "other", + "status": "in_progress", + "content": [ + { + "sessionId": "thread-1:subagent:worker", + "type": "session" + } + ], + "rawInput": { + "prompt": "Also check Windows", + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "inProgress" + }, + "sessionUpdate": "tool_call" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-2", + "recipientSessionId": "thread-1:subagent:worker", + "content": [ + { + "text": "Also check Windows", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "send:to:worker", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "recipientMessageId": "task-2", + "content": [ + { + "text": "Also check Windows", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-2", + "senderSessionId": "thread-1", + "senderMessageId": "send:to:worker", + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "send", + "status": "completed", + "rawInput": { + "prompt": "Also check Windows", + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "completed" + }, + "sessionUpdate": "tool_call_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "wait", + "status": "completed", + "title": "wait", + "content": [ + { + "sessionId": "thread-1:subagent:worker", + "type": "session" + } + ], + "rawInput": { + "prompt": null, + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "completed" + }, + "sessionUpdate": "tool_call_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "failed-send", + "status": "failed", + "title": "sendInput", + "content": [ + { + "sessionId": "thread-1:subagent:worker", + "type": "session" + } + ], + "rawInput": { + "prompt": "Never sent", + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "failed" + }, + "sessionUpdate": "tool_call_update" + } + } +] diff --git a/src/__tests__/CodexACPAgent/data/live-core-v1-identity.json b/src/__tests__/CodexACPAgent/data/live-core-v1-identity.json index 42954476d..261b2b5ff 100644 --- a/src/__tests__/CodexACPAgent/data/live-core-v1-identity.json +++ b/src/__tests__/CodexACPAgent/data/live-core-v1-identity.json @@ -18,6 +18,20 @@ "sessionUpdate": "subagent_update" } }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "item-user", + "recipientSessionId": "thread-1:subagent:worker", + "content": [ + { + "text": "Agent delegation", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, { "sessionId": "thread-1:subagent:worker", "update": { diff --git a/src/__tests__/CodexACPAgent/data/live-core-v2-delegation.json b/src/__tests__/CodexACPAgent/data/live-core-v2-delegation.json new file mode 100644 index 000000000..131735f5f --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/live-core-v2-delegation.json @@ -0,0 +1,198 @@ +[ + { + "sessionId": "thread-1", + "update": { + "sessionId": "thread-1:subagent:worker", + "description": "explorer", + "capabilities": {}, + "sessionUpdate": "subagent_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "spawn:to:worker", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "content": [ + { + "text": "Find the bug", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-1", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "senderMessageId": "spawn:to:worker", + "content": [ + { + "text": "Find the bug", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "spawn:to:worker", + "recipientMessageId": "task-1", + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "first", + "content": { + "text": "First answer", + "type": "text" + }, + "sessionUpdate": "agent_message_chunk" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "send", + "title": "sendInput", + "kind": "other", + "status": "in_progress", + "content": [ + { + "sessionId": "thread-1:subagent:worker", + "type": "session" + } + ], + "rawInput": { + "prompt": "Also check Windows", + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "inProgress" + }, + "sessionUpdate": "tool_call_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-2", + "recipientSessionId": "thread-1:subagent:worker", + "content": [ + { + "text": "Also check Windows", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "send:to:worker", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "recipientMessageId": "task-2", + "content": [ + { + "text": "Also check Windows", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-2", + "senderSessionId": "thread-1", + "senderMessageId": "send:to:worker", + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "send", + "status": "completed", + "rawInput": { + "prompt": "Also check Windows", + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "completed" + }, + "sessionUpdate": "tool_call_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "wait", + "title": "wait", + "status": "completed", + "content": [ + { + "sessionId": "thread-1:subagent:worker", + "type": "session" + } + ], + "rawInput": { + "prompt": null, + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "completed" + }, + "sessionUpdate": "tool_call_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "failed-send", + "title": "sendInput", + "status": "failed", + "content": [ + { + "sessionId": "thread-1:subagent:worker", + "type": "session" + } + ], + "rawInput": { + "prompt": "Never sent", + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "failed" + }, + "sessionUpdate": "tool_call_update" + } + } +] diff --git a/src/__tests__/CodexACPAgent/data/live-core-v2-identity.json b/src/__tests__/CodexACPAgent/data/live-core-v2-identity.json index 3420d1dee..77e508261 100644 --- a/src/__tests__/CodexACPAgent/data/live-core-v2-identity.json +++ b/src/__tests__/CodexACPAgent/data/live-core-v2-identity.json @@ -25,6 +25,20 @@ "sessionUpdate": "subagent_update" } }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "item-user", + "recipientSessionId": "thread-1:subagent:worker", + "content": [ + { + "text": "Agent delegation", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, { "sessionId": "thread-1:subagent:worker", "update": { diff --git a/src/__tests__/CodexACPAgent/data/subagent-replay-v1-delegation.json b/src/__tests__/CodexACPAgent/data/subagent-replay-v1-delegation.json new file mode 100644 index 000000000..fba982e53 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/subagent-replay-v1-delegation.json @@ -0,0 +1,164 @@ +[ + { + "sessionId": "thread-1", + "update": { + "sessionId": "thread-1:subagent:worker", + "capabilities": {}, + "sessionUpdate": "subagent_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "spawn:to:worker", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "content": [ + { + "text": "Find the bug", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-1", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "senderMessageId": "spawn:to:worker", + "content": [ + { + "text": "Find the bug", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "spawn:to:worker", + "recipientMessageId": "task-1", + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "content": { + "text": "First answer", + "type": "text" + }, + "messageId": "first", + "sessionUpdate": "agent_message_chunk" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-2", + "recipientSessionId": "thread-1:subagent:worker", + "content": [ + { + "text": "Also check Windows", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "content": { + "text": "Second answer", + "type": "text" + }, + "messageId": "second", + "sessionUpdate": "agent_message_chunk" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "sendInput:to:worker", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "recipientMessageId": "task-2", + "content": [ + { + "text": "Also check Windows", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-2", + "senderSessionId": "thread-1", + "senderMessageId": "sendInput:to:worker", + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "sendInput", + "title": "sendInput", + "kind": "other", + "status": "completed", + "content": [ + { + "sessionId": "thread-1:subagent:worker", + "type": "session" + } + ], + "rawInput": { + "prompt": "Also check Windows", + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "completed" + }, + "sessionUpdate": "tool_call" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "wait", + "title": "wait", + "kind": "other", + "status": "completed", + "content": [ + { + "sessionId": "thread-1:subagent:worker", + "type": "session" + } + ], + "rawInput": { + "prompt": null, + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "completed" + }, + "sessionUpdate": "tool_call" + } + } +] diff --git a/src/__tests__/CodexACPAgent/data/subagent-replay-v2-delegation.json b/src/__tests__/CodexACPAgent/data/subagent-replay-v2-delegation.json new file mode 100644 index 000000000..b70da2c19 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/subagent-replay-v2-delegation.json @@ -0,0 +1,184 @@ +[ + { + "sessionId": "thread-1", + "update": { + "sessionId": "thread-1:subagent:worker", + "capabilities": {}, + "sessionUpdate": "subagent_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "spawn:to:worker", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "content": [ + { + "text": "Find the bug", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-1", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "senderMessageId": "spawn:to:worker", + "content": [ + { + "text": "Find the bug", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "spawn:to:worker", + "recipientMessageId": "task-1", + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "first", + "content": [], + "_meta": null, + "sessionUpdate": "agent_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "first", + "content": { + "text": "First answer", + "type": "text" + }, + "sessionUpdate": "agent_message_chunk" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-2", + "recipientSessionId": "thread-1:subagent:worker", + "content": [ + { + "text": "Also check Windows", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "second", + "content": [], + "_meta": null, + "sessionUpdate": "agent_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "second", + "content": { + "text": "Second answer", + "type": "text" + }, + "sessionUpdate": "agent_message_chunk" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "sendInput:to:worker", + "senderSessionId": "thread-1", + "recipientSessionId": "thread-1:subagent:worker", + "recipientMessageId": "task-2", + "content": [ + { + "text": "Also check Windows", + "type": "text" + } + ], + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "task-2", + "senderSessionId": "thread-1", + "senderMessageId": "sendInput:to:worker", + "sessionUpdate": "session_message" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "sendInput", + "title": "sendInput", + "kind": "other", + "status": "completed", + "content": [ + { + "sessionId": "thread-1:subagent:worker", + "type": "session" + } + ], + "rawInput": { + "prompt": "Also check Windows", + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "completed" + }, + "_meta": null, + "sessionUpdate": "tool_call_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "toolCallId": "wait", + "title": "wait", + "kind": "other", + "status": "completed", + "content": [ + { + "sessionId": "thread-1:subagent:worker", + "type": "session" + } + ], + "rawInput": { + "prompt": null, + "senderThreadId": "thread-1", + "receiverThreadIds": [ + "worker" + ], + "agentsStates": {}, + "model": null, + "reasoningEffort": null, + "status": "completed" + }, + "_meta": null, + "sessionUpdate": "tool_call_update" + } + } +] diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index b2383aeb9..0db49aa34 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -55,7 +55,6 @@ describe('CodexACPAgent - initialize', () => { fork: {}, title: {}, additionalDirectories: {}, - subagents: {}, systemPrompt: {}, }, mcpCapabilities: { @@ -99,6 +98,14 @@ describe('CodexACPAgent - initialize', () => { }); }); + it.each([ + ["a standard client", {subagents: {}}, undefined], + ["AIR", {subagents: {}, _meta: {jetbrains: {air: {version: 1, capabilities: []}}}}, {}], + ])('answers %s with the subagent session capability that only the AIR contract negotiates', async (_name, clientCapabilities, expected) => { + const result = await agent.initialize({protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities}); + expect((result.agentCapabilities?.sessionCapabilities as {subagents?: unknown}).subagents).toEqual(expected); + }); + it('should advertise gateway auth when the client opts into gateway auth metadata', async () => { const params: acp.InitializeRequest = { protocolVersion: acp.PROTOCOL_VERSION, diff --git a/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts b/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts index 32494f35c..3a5f2b682 100644 --- a/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts +++ b/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts @@ -3,7 +3,7 @@ import type {ServerNotification} from "../../app-server"; import type {ThreadItem} from "../../app-server/v2"; import { agentMessageDelta, commandExecutionItem, connectSession, codexResponse, createThread, createTurn, cwd, - itemStarted, sessionId, turnStarted, turnCompleted, userMessageItem, type PromptSession, + itemCompleted, itemStarted, sessionId, turnStarted, turnCompleted, userMessageItem, type PromptSession, } from "./v2-prompt-harness"; import {expectConformingV2SessionUpdates} from "./v2-session-update-guard"; import {subagentProtocol} from "../../subagents/AcpSubagents"; @@ -17,6 +17,12 @@ const childId = `${sessionId}:subagent:${worker}`; const activityItem = (id = worker, path = "/root/worker"): Extract => ({type: "subAgentActivity", model: null, reasoningEffort: null, id: `activity-${id}`, agentThreadId: id, agentPath: path, kind: "started"}); +const collab = (tool: Extract["tool"], id: string, prompt: string | null, + status: "inProgress" | "completed" | "failed" = "completed"): ThreadItem => + ({type: "collabAgentToolCall", id, tool, status, senderThreadId: sessionId, + receiverThreadIds: tool === "spawnAgent" && status === "inProgress" ? [] : [worker], + prompt, model: null, reasoningEffort: null, agentsStates: {}}); +const task = (id: string, text: string): ThreadItem => ({...userMessageItem(null, text), id}); const onWorker = (event: ServerNotification, id = worker): ServerNotification => ({...event, params: {...event.params, threadId: id}} as ServerNotification); @@ -86,6 +92,38 @@ describe.each([1, 2] as const)("ACP v%s live standard subagents", version => { expect(childUpdates().filter(entry => entry.update.sessionUpdate === "subagent_update" && entry.update.capabilities != null && Object.keys(entry.update.capabilities).length === 0)).toHaveLength(1); expect(childUpdates().some(entry => entry.update.sessionUpdate === "user_message_chunk")).toBe(false); + // No sender was observed, so the incoming view names only its recipient. + expect(childUpdates().filter(entry => entry.update.sessionUpdate === "session_message").map(entry => entry.update)) + .toEqual([{sessionUpdate: "session_message", messageId: "item-user", recipientSessionId: childId, + content: [{type: "text", text: "Agent delegation"}]}]); + }); + + it("reports delegated tasks as linked session messages and references the child from visible collaboration calls", async () => { + await connect(); + h.emit({method: "thread/started", params: {thread: {...createThread(), id: worker, parentThreadId: sessionId, agentRole: "explorer"}}}); + h.emit(itemStarted(collab("spawnAgent", "spawn", "Find the bug", "inProgress"))); + h.emit(itemCompleted(collab("spawnAgent", "spawn", "Find the bug"))); + h.emit(onWorker(turnStarted("child-turn"))); + h.emit(onWorker(itemStarted(task("task-1", "Find the bug"), "child-turn"))); + h.emit(onWorker(itemCompleted(task("task-1", "Find the bug"), "child-turn"))); + h.emit(onWorker(agentMessageDelta("First answer", "child-turn", "first"))); + h.emit(onWorker(turnCompleted("child-turn"))); + // The child can observe a later instruction before the parent's call completes. + h.emit(itemStarted(collab("sendInput", "send", "Also check Windows", "inProgress"))); + h.emit(onWorker(turnStarted("next-turn"))); + h.emit(onWorker(itemStarted(task("task-2", "Also check Windows"), "next-turn"))); + h.emit(itemCompleted(collab("sendInput", "send", "Also check Windows"))); + h.emit(itemCompleted(collab("wait", "wait", null))); + h.emit(itemCompleted(collab("sendInput", "failed-send", "Never sent", "failed"))); + await flush(h); + const kinds = ["session_message", "tool_call", "tool_call_update", "agent_message_chunk"]; + const updates = h.scopedUpdates.filter(entry => kinds.includes(entry.update.sessionUpdate) + || (entry.update.sessionUpdate === "subagent_update" && "description" in entry.update)); + await expect(JSON.stringify(updates, null, 2) + "\n").toMatchFileSnapshot(`data/live-core-v${version}-delegation.json`); + expect(h.scopedUpdates.some(entry => entry.update.sessionUpdate === "user_message_chunk" + || (entry.update as {sessionUpdate: string}).sessionUpdate === "user_message")).toBe(false); + expect(h.scopedUpdates.some(entry => "toolCallId" in entry.update && entry.update.toolCallId === "spawn")).toBe(false); + expect(JSON.stringify(h.scopedUpdates.filter(entry => entry.update.sessionUpdate === "session_message"))).not.toContain("Never sent"); }); it("waits for immediate parentage and announces ancestors before child traffic", async () => { diff --git a/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts b/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts index 858522302..39ab29cb2 100644 --- a/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts +++ b/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts @@ -130,7 +130,30 @@ describe.each([1, 2] as const)("ACP v%s recorded child replay", version => { } expect(text("Nested answer")).toBe(true); expect(childUpdates().some(entry => entry.update.sessionUpdate === "user_message_chunk")).toBe(false); - expect(gap()).toHaveLength(1); + // The spawn record holds no prompt, so no sender view exists to name or link. + expect(childUpdates().filter(entry => entry.update.sessionUpdate === "session_message").map(entry => entry.update)) + .toEqual([{sessionUpdate: "session_message", messageId: "item-user", recipientSessionId: childId, + content: [{type: "text", text: "Agent-owned input"}]}]); + expect(gap()).toEqual([]); + }); + + it("replays delegated tasks as linked session messages before the child's responses", async () => { + await connect(); + const send = (tool: "sendInput" | "wait", prompt: string | null): ThreadItem => ({...spawn(), id: tool, tool, prompt}); + history(h, { + [sessionId]: {items: [{...spawn(), id: "spawn", prompt: "Find the bug"}, send("sendInput", "Also check Windows"), send("wait", null)]}, + [worker]: {parent: sessionId, items: [ + {...userMessageItem(null, "Find the bug"), id: "task-1"}, message("first", "First answer"), + {...userMessageItem(null, "Also check Windows"), id: "task-2"}, message("second", "Second answer"), + ]}, + }); + await load(); + await flush(h); + const kinds = ["session_message", "tool_call", "tool_call_update", "agent_message_chunk", "agent_message"]; + const updates = h.scopedUpdates.filter(entry => kinds.includes(entry.update.sessionUpdate) + || (entry.update.sessionUpdate === "subagent_update" && !("state" in entry.update))); + await expect(JSON.stringify(updates, null, 2) + "\n").toMatchFileSnapshot(`data/subagent-replay-v${version}-delegation.json`); + expect(gap()).toEqual([]); }); it.each([false, true])("enriches an existing pathless parent before replaying a newly recorded descendant (cold: %s)", async cold => { diff --git a/src/__tests__/scenarios/baseline.ts b/src/__tests__/scenarios/baseline.ts index 89299412d..446b8d0c0 100644 --- a/src/__tests__/scenarios/baseline.ts +++ b/src/__tests__/scenarios/baseline.ts @@ -142,6 +142,7 @@ const MCP_COMMAND = { * The feature changes of the compatibility rule that change the messages of the scenarios: * the `/mcp` command has a new description and an input hint. * Initialize also advertises text-only append system prompts and client-set titles. + * It no longer advertises `sessionCapabilities.subagents`, which the subagent RFD does not define for an Agent. * Only a command that equals the baseline `/mcp` command changes. */ export function withFeatureChanges(messages: RecordedMessage[]): RecordedMessage[] { @@ -152,8 +153,9 @@ export function withFeatureChanges(messages: RecordedMessage[]): RecordedMessage const capabilities = params["agentCapabilities"] as Json | undefined; const session = capabilities?.["sessionCapabilities"] as Json | undefined; if (session) { + const {subagents: _subagents, ...kept} = session; return {...message, params: {...params, agentCapabilities: { - ...capabilities, sessionCapabilities: {...session, systemPrompt: {}, title: {}}, + ...capabilities, sessionCapabilities: {...kept, systemPrompt: {}, title: {}}, }}}; } } diff --git a/src/subagents/StandardSubagentEventRouter.ts b/src/subagents/StandardSubagentEventRouter.ts index 93413747f..9a30b8260 100644 --- a/src/subagents/StandardSubagentEventRouter.ts +++ b/src/subagents/StandardSubagentEventRouter.ts @@ -1,3 +1,4 @@ +import type {ContentBlock} from "@agentclientprotocol/sdk/unstable"; import type {ReasoningEffort, ServerNotification} from "../app-server"; import type {StateUpdate} from "@agentclientprotocol/sdk/experimental/v2/unstable"; import type {Thread, ThreadItem, Turn} from "../app-server/v2"; @@ -11,6 +12,7 @@ import {readableServiceErrorMessage} from "../ServiceErrorMessage"; import {toV2IdleState} from "../AcpV2Prompt"; import {normalizeSessionTitle} from "../SessionTitle"; import {CollabAgentReporter} from "../tool-calls/reporters/CollabAgentReporter"; +import type {ToolFacts} from "../tool-calls/ToolFacts"; import {SubagentActivityReporter} from "../tool-calls/reporters/SubagentActivityReporter"; import type {SubagentState} from "./AcpSubagents"; import { @@ -22,6 +24,7 @@ import {PendingNotificationBuffer} from "./PendingNotificationBuffer"; import {SubagentAssociations, type SubagentAssociation} from "./SubagentAssociations"; import type {SubagentEventRouter} from "./SubagentEventRouter"; import {runSubagentInteraction} from "./SubagentInteraction"; +import {outgoingMessageId, sendsMessage, SubagentMessages, userMessageText} from "./SubagentMessages"; import {logger} from "../Logger"; import {isAbsoluteProtocolPath} from "../PathUtils"; @@ -80,6 +83,7 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { private snapshotting = false; private readonly snapshots = new Map(); private interactionObserver?: (event: ServerNotification, sessionId: string) => Promise; + private readonly messages = new SubagentMessages((update, sessionId) => this.session.update(update, sessionId)); constructor( private readonly rootSessionId: string, @@ -95,6 +99,8 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { read(threadId: string, parentThreadId: string): Promise; enqueuePublication(publish: () => Promise): Promise; }, + private readonly userMessageContent: (item: ThreadItem & {type: "userMessage"}) => ContentBlock[] + = item => [{type: "text", text: userMessageText(item)}], ) { this.deferredLive = new PendingNotificationBuffer(rootSessionId); } @@ -133,6 +139,7 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { if (thread.parentThreadId === this.rootSessionId || parent?.live || child?.live || (!child && this.spawnsInFlight.size > 0)) { await this.observeConfiguration(thread.id, {model: thread.model, effort: thread.reasoningEffort}, "threadStarted"); + await this.observeRole(thread.id, thread.agentRole); } return false; } @@ -150,11 +157,12 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { if (!id.trim() || id === item.senderThreadId || id === this.rootSessionId || this.children.has(id)) continue; const pending = this.pendingFor(id); pending.parentThreadId = item.senderThreadId; - if (item.prompt) pending.description = item.prompt; } await this.materializePending(); + await this.reportOutgoingMessages(item); return true; } + if (item.type === "collabAgentToolCall") await this.reportOutgoingMessages(item); if (item.type === "subAgentActivity") { if (isRootAgentPath(item.agentPath)) return true; const child = this.children.get(item.agentThreadId); @@ -254,7 +262,10 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { return true; case "item/started": if (replayed === true) return true; - if (notification.params.item.type === "userMessage") return true; + if (notification.params.item.type === "userMessage") { + await this.reportIncomingMessage(threadId, notification.params.item); + return true; + } if (child.turnId !== null && child.foregroundTurnId === notification.params.turnId) { await child.foreground.startItem(notification.params.item.id); } @@ -270,7 +281,10 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { if (item.type === "commandExecution") child.streams.commands.restore(item); return true; } - if (item.type === "userMessage") return true; + if (item.type === "userMessage") { + await this.reportIncomingMessage(threadId, item); + return true; + } if (item.type === "plan") { const plans = new CodexPlanStream(this.session.forSession(child.sessionId), this.capabilities); const plan = await plans.completed(item.id, item.text); @@ -385,6 +399,44 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { return parent === undefined ? null : this.announce(threadId, parent, path); } + /** The role Codex assigned describes the child. The task it was given is a message, not a description. */ + async observeRole(threadId: string, role: string | null): Promise { + const description = role?.trim(); + if (!description || this.retired) return; + const child = this.children.get(threadId); + if (!child) { this.pendingFor(threadId).description = description; return; } + if (child.description === description) return; + child.description = description; + if (!this.announced.has(threadId)) return; + await this.session.update({sessionUpdate: "subagent_update", sessionId: child.sessionId, description}, child.parentSessionId); + } + + /** The sender's view of a prompt it sent to announced children. A failed or unfinished call sent nothing. */ + async reportOutgoingMessages(item: ThreadItem & {type: "collabAgentToolCall"}): Promise { + if (this.retired || !sendsMessage(item)) return; + const sender = item.senderThreadId === this.rootSessionId ? this.rootSessionId : this.children.get(item.senderThreadId)?.sessionId; + if (sender === undefined) return; + for (const id of new Set(item.receiverThreadIds)) { + const receiver = this.children.get(id); + if (!receiver || receiver.sessionId === sender) continue; + await this.ensureAnnounced(receiver); + if (!this.announced.has(id)) continue; + await this.messages.outgoing(sender, receiver.sessionId, outgoingMessageId(item, id), item.prompt); + } + } + + /** What the child received. The sender is named only when its outgoing view was observed. */ + async reportIncomingMessage(threadId: string, item: ThreadItem & {type: "userMessage"}): Promise { + const child = this.children.get(threadId); + if (!child || this.retired) return; + await this.ensureAnnounced(child); + if (!this.announced.has(threadId)) return; + await this.messages.incoming(child.sessionId, item.id, userMessageText(item), this.userMessageContent(item)); + } + + /** A replay reports the recorded messages again. */ + beginHistoryReplay(): void { this.messages.reset(); } + private async observeConfiguration(threadId: string, observed: ObservedSubagentConfiguration, source: ChildModel["source"]): Promise { const previous = this.configurations.get(threadId); const cwd = observed.cwd !== undefined && isAbsoluteProtocolPath(observed.cwd) ? observed.cwd : null; @@ -679,6 +731,7 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { if (!child || this.retired) return; child.live = true; await this.observeConfiguration(thread.id, {model: thread.model, effort: thread.reasoningEffort}, "threadStarted"); + await this.observeRole(thread.id, thread.agentRole); if (thread.status.type === "idle") { child.turnId = null; child.foregroundTurnId = null; @@ -734,8 +787,17 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { legacyActivityCompleted(item: ThreadItem & {type: "subAgentActivity"}) { return SubagentActivityReporter.activity(item, "completed", this.legacyActivities.delete(item.id) ? "update" : "start"); } - legacyCollaborationStarted(item: ThreadItem & {type: "collabAgentToolCall"}) { return CollabAgentReporter.started(item); } - legacyCollaborationCompleted(item: ThreadItem & {type: "collabAgentToolCall"}) { return CollabAgentReporter.completed(item); } + legacyCollaborationStarted(item: ThreadItem & {type: "collabAgentToolCall"}) { return this.withSessionReferences(item, CollabAgentReporter.started(item)); } + legacyCollaborationCompleted(item: ThreadItem & {type: "collabAgentToolCall"}) { return this.withSessionReferences(item, CollabAgentReporter.completed(item)); } + + /** A visible collaboration call links to the announced children it operates on. */ + private withSessionReferences(item: ThreadItem & {type: "collabAgentToolCall"}, facts: ToolFacts): ToolFacts { + const content = [...new Set(item.receiverThreadIds)].flatMap(id => { + const child = this.children.get(id); + return child && this.announced.has(id) ? [{type: "session" as const, sessionId: child.sessionId}] : []; + }); + return content.length === 0 ? facts : {...facts, standard: {...facts.standard, content}}; + } } function turnEndState(turn: Turn): StateUpdate | undefined { diff --git a/src/subagents/SubagentMessages.ts b/src/subagents/SubagentMessages.ts new file mode 100644 index 000000000..17ef1e333 --- /dev/null +++ b/src/subagents/SubagentMessages.ts @@ -0,0 +1,98 @@ +import type {ContentBlock, SessionMessage} from "@agentclientprotocol/sdk/unstable"; +import type {ThreadItem} from "../app-server/v2"; + +export type SessionMessageUpdate = SessionMessage & {sessionUpdate: "session_message"}; +type CollabItem = ThreadItem & {type: "collabAgentToolCall"}; +type UserMessageItem = ThreadItem & {type: "userMessage"}; +type View = {sessionId: string; messageId: string; text: string}; + +/** Collaboration tools whose prompt is a message to the receivers. */ +export function sendsMessage(item: CollabItem): item is CollabItem & {prompt: string} { + switch (item.tool) { + case "spawnAgent": + case "sendInput": + case "sendMessage": + case "followupTask": + return item.status === "completed" && !!item.prompt; + case "resumeAgent": + case "wait": + case "closeAgent": + case "interruptAgent": + case "listAgents": + return false; + } +} + +/** The outgoing view's ID in the sender's transcript. A tool call can address several receivers. */ +export function outgoingMessageId(item: CollabItem, receiverThreadId: string): string { + return `${item.id}:to:${encodeURIComponent(receiverThreadId)}`; +} + +export function userMessageText(item: UserMessageItem): string { + return item.content.map(input => input.type === "text" ? input.text : "").join(""); +} + +/** + * Reports each view of a message between sessions once, from the side that was observed. + * Views are linked only when the same text was observed for the same recipient on both sides. + */ +export class SubagentMessages { + private readonly reported = new Set(); + private readonly unlinkedOutgoing = new Map(); + private readonly unlinkedIncoming = new Map(); + + constructor(private readonly send: (update: SessionMessageUpdate, sessionId: string) => Promise) {} + + async outgoing(senderSessionId: string, recipientSessionId: string, messageId: string, text: string): Promise { + if (!this.claim(senderSessionId, messageId)) return; + const incoming = take(this.unlinkedIncoming, recipientSessionId, text); + if (!incoming) push(this.unlinkedOutgoing, recipientSessionId, {sessionId: senderSessionId, messageId, text}); + await this.send({sessionUpdate: "session_message", messageId, senderSessionId, recipientSessionId, + ...(incoming ? {recipientMessageId: incoming.messageId} : {}), + content: [{type: "text", text}]}, senderSessionId); + if (incoming) { + await this.send({sessionUpdate: "session_message", messageId: incoming.messageId, + senderSessionId, senderMessageId: messageId}, recipientSessionId); + } + } + + async incoming(recipientSessionId: string, messageId: string, text: string, content: ContentBlock[]): Promise { + if (!this.claim(recipientSessionId, messageId)) return; + const outgoing = take(this.unlinkedOutgoing, recipientSessionId, text); + if (!outgoing) push(this.unlinkedIncoming, recipientSessionId, {sessionId: recipientSessionId, messageId, text}); + await this.send({sessionUpdate: "session_message", messageId, recipientSessionId, + ...(outgoing ? {senderSessionId: outgoing.sessionId, senderMessageId: outgoing.messageId} : {}), + content}, recipientSessionId); + if (outgoing) { + await this.send({sessionUpdate: "session_message", messageId: outgoing.messageId, + recipientMessageId: messageId}, outgoing.sessionId); + } + } + + /** A replay reports every recorded view again. */ + reset(): void { + this.reported.clear(); + this.unlinkedOutgoing.clear(); + this.unlinkedIncoming.clear(); + } + + private claim(sessionId: string, messageId: string): boolean { + const key = `${sessionId}\n${messageId}`; + if (this.reported.has(key)) return false; + this.reported.add(key); + return true; + } +} + +function push(views: Map, recipientSessionId: string, view: View): void { + const waiting = views.get(recipientSessionId) ?? []; + waiting.push(view); + views.set(recipientSessionId, waiting); +} + +function take(views: Map, recipientSessionId: string, text: string): View | null { + const waiting = views.get(recipientSessionId); + const index = waiting?.findIndex(view => view.text === text) ?? -1; + if (!waiting || index < 0) return null; + return waiting.splice(index, 1)[0] ?? null; +}