From 07124e7b1d15fb6be6179082a4af01d69a0f1500 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 20 Aug 2026 13:53:26 +0400 Subject: [PATCH 1/5] feat: add native ACP subagent sessions --- src/ACPSessionConnection.ts | 15 +- src/CodexAcpClient.ts | 99 ++-- src/CodexAcpServer.ts | 38 +- src/CodexApprovalHandler.ts | 10 +- src/CodexElicitationHandler.ts | 17 +- src/CodexEventHandler.ts | 161 ++++++- .../CodexACPAgent/collab-agent-events.test.ts | 424 ++++++++++++++++++ src/acp-subagents.ts | 66 +++ 8 files changed, 770 insertions(+), 60 deletions(-) create mode 100644 src/acp-subagents.ts diff --git a/src/ACPSessionConnection.ts b/src/ACPSessionConnection.ts index 286630ac..84b24294 100644 --- a/src/ACPSessionConnection.ts +++ b/src/ACPSessionConnection.ts @@ -1,5 +1,8 @@ import * as acp from "@agentclientprotocol/sdk"; -import type {SessionNotification} from "@agentclientprotocol/sdk"; +import { + type AcpSessionUpdate, + asSdkSessionNotification, +} from "./acp-subagents"; export type AcpClientConnection = Pick; @@ -12,12 +15,12 @@ export class ACPSessionConnection { this.sessionId = sessionId; } - async update(update: UpdateSessionEvent) { - await this.connection.notify(acp.methods.client.session.update, { - sessionId: this.sessionId, + async update(update: UpdateSessionEvent, sessionId: string = this.sessionId) { + await this.connection.notify(acp.methods.client.session.update, asSdkSessionNotification({ + sessionId, update: update - }); + })); } } -export type UpdateSessionEvent = SessionNotification["update"]; +export type UpdateSessionEvent = AcpSessionUpdate; diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 4e7f2aa4..c429b7fa 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -525,6 +525,10 @@ export class CodexAcpClient { await this.codexClient.threadUnsubscribe({threadId: sessionId}); } finally { this.codexClient.clearThreadHandlers(sessionId); + for (const childSessionId of this.subagentSubscriptions.get(sessionId) ?? []) { + this.codexClient.clearThreadHandlers(childSessionId); + } + this.subagentSubscriptions.delete(sessionId); } } @@ -763,35 +767,76 @@ export class CodexAcpClient { sessionId: string, eventHandler: (result: ServerNotification) => void | Promise, approvalHandler: ApprovalHandler, - elicitationHandler: ElicitationHandler + elicitationHandler: ElicitationHandler, + supportsSubagents: boolean, ) { - this.codexClient.onServerNotification(sessionId, (event) => { + const dispatch = (event: ServerNotification) => { this.enqueueSessionNotification(sessionId, () => eventHandler(event)); + }; + const registerInteractiveHandlers = (targetSessionId: string): void => { + this.codexClient.onApprovalRequest(targetSessionId, { + handleCommandExecution: async (params) => { + await this.waitForSessionNotifications(sessionId); + return await approvalHandler.handleCommandExecution(params); + }, + handleFileChange: async (params) => { + await this.waitForSessionNotifications(sessionId); + return await approvalHandler.handleFileChange(params); + }, + handlePermissionsRequest: async (params) => { + await this.waitForSessionNotifications(sessionId); + return await approvalHandler.handlePermissionsRequest(params); + }, + }); + this.codexClient.onElicitationRequest(targetSessionId, { + handleElicitation: async (params) => { + await this.waitForSessionNotifications(sessionId); + return await elicitationHandler.handleElicitation(params); + }, + handleUserInput: async (params) => { + await this.waitForSessionNotifications(sessionId); + return await elicitationHandler.handleUserInput(params); + }, + }); + }; + const subscribeDiscoveredChildren = (event: ServerNotification): void => { + if (!supportsSubagents + || (event.method !== "item/started" && event.method !== "item/completed") + || event.params.item.type !== "collabAgentToolCall" + || event.params.item.tool !== "spawnAgent") { + return; + } + let children = this.subagentSubscriptions.get(sessionId); + if (!children) { + children = new Set(); + this.subagentSubscriptions.set(sessionId, children); + } + for (const childSessionId of event.params.item.receiverThreadIds) { + if (childSessionId.trim() === "") continue; + if (childSessionId === sessionId || childSessionId === event.params.threadId) continue; + if (children.has(childSessionId)) continue; + children.add(childSessionId); + this.codexClient.onServerNotification(childSessionId, (childEvent) => { + const eventThreadId = (childEvent.params as {threadId?: unknown}).threadId; + if (eventThreadId !== childSessionId) { + // Notifications without a thread id are broadcast by + // CodexAppServerClient. The root handler owns those; + // processing them here would duplicate them once per child. + return; + } + subscribeDiscoveredChildren(childEvent); + dispatch(childEvent); + }); + registerInteractiveHandlers(childSessionId); + } + }; + this.codexClient.onServerNotification(sessionId, (event) => { + // Register synchronously before queueing the spawn update. App-server + // may emit the first child event immediately after the root event. + subscribeDiscoveredChildren(event); + dispatch(event); }); - this.codexClient.onApprovalRequest(sessionId, { - handleCommandExecution: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await approvalHandler.handleCommandExecution(params); - }, - handleFileChange: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await approvalHandler.handleFileChange(params); - }, - handlePermissionsRequest: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await approvalHandler.handlePermissionsRequest(params); - }, - }); - this.codexClient.onElicitationRequest(sessionId, { - handleElicitation: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await elicitationHandler.handleElicitation(params); - }, - handleUserInput: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await elicitationHandler.handleUserInput(params); - }, - }); + registerInteractiveHandlers(sessionId); } async waitForSessionNotifications(sessionId: string): Promise { @@ -821,6 +866,8 @@ export class CodexAcpClient { }); } + private readonly subagentSubscriptions = new Map>(); + async sendPrompt( request: acp.PromptRequest, agentMode: AgentMode, diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 6e0c394b..a5ccc3e8 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -94,6 +94,10 @@ import { createUserMessageChunk, } from "./ContentChunks"; import {sameThreadGoalSnapshot, type ThreadGoalSnapshot, toThreadGoalSnapshot,} from "./ThreadGoalSnapshot"; +import { + clientSupportsSubagents, + type SubagentAwareSessionCapabilities, +} from "./acp-subagents"; import {randomUUID} from "node:crypto"; import {once} from "node:events"; import { @@ -310,6 +314,14 @@ export class CodexAcpServer { this.terminalOutputMode = resolveTerminalOutputMode(_params.clientCapabilities); this.booleanConfigOptionsSupported = clientSupportsBooleanConfigOptions(_params.clientCapabilities); await this.runWithProcessCheck(() => this.codexAcpClient.initialize(_params)); + const sessionCapabilities: SubagentAwareSessionCapabilities = { + resume: { }, + list: { }, + close: { }, + delete: { }, + additionalDirectories: {}, + ...(clientSupportsSubagents(_params.clientCapabilities) ? {subagents: {}} : {}), + }; return { protocolVersion: acp.PROTOCOL_VERSION, agentInfo: { @@ -327,13 +339,7 @@ export class CodexAcpServer { embeddedContext: true, image: true }, - sessionCapabilities: { - resume: { }, - list: { }, - close: { }, - delete: { }, - additionalDirectories: {}, - }, + sessionCapabilities, mcpCapabilities: { acp: false, http: true, @@ -2240,6 +2246,7 @@ export class CodexAcpServer { : null; let agentFileChangeReportTurnId: string | null = null; let agentFileChangeReportUnavailableReason: AgentFileChangeReportUnavailableReason = "providerError"; + let promptWasCancelled = false; let recoverableSessionFailure = sessionState.sessionFailure; sessionState.currentTurnId = null; sessionState.lastTokenUsage = null; @@ -2266,6 +2273,7 @@ export class CodexAcpServer { } }; const cancelledPromptResponse = (): acp.PromptResponse => { + promptWasCancelled = true; agentFileChangeReportTurnId = null; agentFileChangeReportUnavailableReason = "cancelled"; return this.cancelledPromptResponse(sessionState); @@ -2278,12 +2286,12 @@ export class CodexAcpServer { clientSupportsPlanUpdates(this.clientCapabilities), clientSupportsTypedSessionFailures(this.clientCapabilities), this.sessionFailureEpoch, + clientSupportsSubagents(this.clientCapabilities), ); eventHandler = promptEventHandler; - const approvalHandler = new CodexApprovalHandler(this.connection, sessionState, activePrompt.signal); + const approvalHandler = new CodexApprovalHandler(this.connection, activePrompt.signal); const elicitationHandler = new CodexElicitationHandler( this.connection, - sessionState, this.clientCapabilities, activePrompt.signal, ); @@ -2304,7 +2312,8 @@ export class CodexAcpServer { } }, approvalHandler, - elicitationHandler); + elicitationHandler, + clientSupportsSubagents(this.clientCapabilities)); if (activePrompt.signal.aborted) { return cancelledPromptResponse(); @@ -2455,6 +2464,8 @@ export class CodexAcpServer { return cancelledPromptResponse(); } + await this.codexAcpClient.waitForSessionNotifications(params.sessionId); + await eventHandler.waitForNativeSubagents(activePrompt.signal); await this.codexAcpClient.waitForSessionNotifications(params.sessionId); await eventHandler.flushPendingErrors(); await eventHandler.handleFailedTurn(turnCompleted.turn); @@ -2548,6 +2559,8 @@ export class CodexAcpServer { return cancelledPromptResponse(); } + await this.codexAcpClient.waitForSessionNotifications(params.sessionId); + await eventHandler.waitForNativeSubagents(activePrompt.signal); await this.codexAcpClient.waitForSessionNotifications(params.sessionId); await eventHandler.flushPendingErrors(); await eventHandler.handleFailedTurn(turnCompleted.turn); @@ -2619,6 +2632,11 @@ export class CodexAcpServer { // The app-server subscription is session-scoped and outlives this prompt. Flip routing before // awaiting disposal so queued late notifications cannot enter prompt-local buffers. promptNotificationsActive = false; + await eventHandler?.finishOutstandingNativeSubagents( + promptWasCancelled || activePrompt.signal.aborted || this.sessionIsClosing(params.sessionId) + ? "cancelled" + : "failed", + ); if (agentFileChangeReportRequest !== null) { await this.publishAgentFileChangeReport( sessionState, diff --git a/src/CodexApprovalHandler.ts b/src/CodexApprovalHandler.ts index 7c2e08a9..98827c12 100644 --- a/src/CodexApprovalHandler.ts +++ b/src/CodexApprovalHandler.ts @@ -1,5 +1,4 @@ import * as acp from "@agentclientprotocol/sdk"; -import type {SessionState} from "./CodexAcpServer"; import type {ApprovalHandler} from "./CodexAppServerClient"; import type { CommandExecutionApprovalDecision, @@ -56,16 +55,13 @@ function permissionOption( export class CodexApprovalHandler implements ApprovalHandler { private readonly connection: AcpClientConnection; - private readonly sessionState: SessionState; private readonly cancellationSignal: AbortSignal | undefined; constructor( connection: AcpClientConnection, - sessionState: SessionState, cancellationSignal?: AbortSignal, ) { this.connection = connection; - this.sessionState = sessionState; this.cancellationSignal = cancellationSignal; } @@ -73,7 +69,7 @@ export class CodexApprovalHandler implements ApprovalHandler { params: CommandExecutionRequestApprovalParams ): Promise { try { - const sessionId = this.sessionState.sessionId; + const sessionId = params.threadId; const acpRequest = this.buildCommandPermissionRequest(sessionId, params); const response = await this.connection.request( acp.methods.client.session.requestPermission, @@ -91,7 +87,7 @@ export class CodexApprovalHandler implements ApprovalHandler { params: FileChangeRequestApprovalParams ): Promise { try { - const sessionId = this.sessionState.sessionId; + const sessionId = params.threadId; const acpRequest = this.buildFileChangePermissionRequest(sessionId, params); const response = await this.connection.request( acp.methods.client.session.requestPermission, @@ -109,7 +105,7 @@ export class CodexApprovalHandler implements ApprovalHandler { params: PermissionsRequestApprovalParams ): Promise { try { - const sessionId = this.sessionState.sessionId; + const sessionId = params.threadId; const acpRequest = this.buildPermissionsRequest(sessionId, params); const response = await this.connection.request( acp.methods.client.session.requestPermission, diff --git a/src/CodexElicitationHandler.ts b/src/CodexElicitationHandler.ts index b0ca0eab..52b1e9c0 100644 --- a/src/CodexElicitationHandler.ts +++ b/src/CodexElicitationHandler.ts @@ -1,5 +1,4 @@ import * as acp from "@agentclientprotocol/sdk"; -import type { SessionState } from "./CodexAcpServer"; import type { ElicitationHandler } from "./CodexAppServerClient"; import type { ServerNotification } from "./app-server"; import type {JsonValue} from "./app-server/serde_json/JsonValue"; @@ -270,7 +269,6 @@ function buildToolApprovalOptions(persistOptions: Set): acp.Permis export class CodexElicitationHandler implements ElicitationHandler { private readonly connection: AcpClientConnection; - private readonly sessionState: SessionState; private readonly clientCapabilities: acp.ClientCapabilities | null; private readonly cancellationSignal: AbortSignal | undefined; // In Rust, the MCP elicitation handler receives ElicitationRequestEvent directly from the MCP @@ -295,12 +293,10 @@ export class CodexElicitationHandler implements ElicitationHandler { constructor( connection: AcpClientConnection, - sessionState: SessionState, clientCapabilities: acp.ClientCapabilities | null = null, cancellationSignal?: AbortSignal ) { this.connection = connection; - this.sessionState = sessionState; this.clientCapabilities = clientCapabilities; this.cancellationSignal = cancellationSignal; } @@ -337,7 +333,7 @@ export class CodexElicitationHandler implements ElicitationHandler { if (params.mode === "url" && result.action === "accept") { this.trackUrlElicitation(params.threadId, params.elicitationId); } - await this.publishAcceptedMcpToolApproval(context, result.action === "accept"); + await this.publishAcceptedMcpToolApproval(params.threadId, context, result.action === "accept"); return result; } @@ -351,7 +347,7 @@ export class CodexElicitationHandler implements ElicitationHandler { const optionId = response.outcome.optionId; if (optionId !== McpApprovalOptionId.Decline) { await this.connection.notify(acp.methods.client.session.update, { - sessionId: this.sessionState.sessionId, + sessionId: params.threadId, update: { sessionUpdate: "tool_call_update", toolCallId: correlatedCallId, status: "in_progress" }, }); } @@ -462,7 +458,7 @@ export class CodexElicitationHandler implements ElicitationHandler { context: McpElicitationContext ): acp.CreateElicitationRequest { const base = { - sessionId: this.sessionState.sessionId, + sessionId: params.threadId, ...(context.correlatedCallId ? { toolCallId: context.correlatedCallId } : {}), message: params.message, _meta: metaRecord(params._meta), @@ -546,7 +542,7 @@ export class CodexElicitationHandler implements ElicitationHandler { const firstQuestion = params.questions[0]; return { - sessionId: this.sessionState.sessionId, + sessionId: params.threadId, toolCallId: params.itemId, mode: "form", message: params.questions.length === 1 && firstQuestion @@ -569,7 +565,7 @@ export class CodexElicitationHandler implements ElicitationHandler { params: McpServerElicitationRequestParams, context: McpElicitationContext ): { request: acp.RequestPermissionRequest; correlatedCallId: string | undefined } { - const sessionId = this.sessionState.sessionId; + const sessionId = params.threadId; const messageContent: acp.ToolCallContent = { type: "content", content: { type: "text", text: params.message }, @@ -715,6 +711,7 @@ export class CodexElicitationHandler implements ElicitationHandler { } private async publishAcceptedMcpToolApproval( + sessionId: string, context: McpElicitationContext, accepted: boolean ): Promise { @@ -722,7 +719,7 @@ export class CodexElicitationHandler implements ElicitationHandler { return; } await this.connection.notify(acp.methods.client.session.update, { - sessionId: this.sessionState.sessionId, + sessionId, update: { sessionUpdate: "tool_call_update", toolCallId: context.correlatedCallId, status: "in_progress" }, }); } diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index 7360b6f9..e07dcb4e 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -72,6 +72,7 @@ import { createAgentTextThoughtChunk, } from "./ContentChunks"; import {sameThreadGoalSnapshot, type ThreadGoalSnapshot, toThreadGoalSnapshot} from "./ThreadGoalSnapshot"; +import type {SubagentState} from "./acp-subagents"; import {logger} from "./Logger"; import {randomUUID} from "node:crypto"; import { @@ -101,6 +102,29 @@ type SessionFailurePolicy = { const MAX_SESSION_FAILURE_TITLE_LENGTH = 240; +function toSubagentTerminalState(status: string): SubagentState | undefined { + switch (status) { + case "completed": + return "completed"; + case "interrupted": + return "cancelled"; + case "errored": + case "shutdown": + case "notFound": + return "failed"; + case "pendingInit": + case "running": + return undefined; + default: + return undefined; + } +} + +function fallbackSubagentName(sessionId: string): string { + const suffix = sessionId.length > 8 ? sessionId.slice(-8) : sessionId; + return `Agent ${suffix}`; +} + const SESSION_FAILURE_POLICY: Record = { transport_lost: { category: "connection", @@ -225,6 +249,13 @@ export class CodexEventHandler { private readonly terminalCommandOutputIds = new Set(); private readonly agentMessagePhases = new Map(); private readonly activeSubAgentActivities = new Set(); + private readonly nativeSubagents = new Map(); + private readonly nativeSubagentWaiters = new Set<() => void>(); constructor( connection: AcpClientConnection, @@ -232,6 +263,7 @@ export class CodexEventHandler { supportsPlanUpdates = false, supportsTypedSessionFailures = false, sessionFailureEpoch: string = randomUUID(), + private readonly supportsSubagents = false, ) { this.sessionState = sessionState; this.supportsPlanUpdates = supportsPlanUpdates; @@ -361,9 +393,136 @@ export class CodexEventHandler { async handleNotification(notification: ServerNotification) { await this.flushPendingErrors(); + if (await this.handleNativeSubagentNotification(notification)) { + return; + } + const notificationThreadId = (notification.params as {threadId?: unknown}).threadId; + if (typeof notificationThreadId === "string" + && this.nativeSubagents.get(notificationThreadId)?.terminalState !== undefined) { + logger.log(`Ignoring update for terminal subagent ${notificationThreadId}`); + return; + } const updateEvent = await this.createUpdateEvent(notification); if (updateEvent) { - await this.session.update(updateEvent); + await this.session.update(updateEvent, this.notificationSessionId(notification)); + } + } + + private notificationSessionId(notification: ServerNotification): string { + const threadId = (notification.params as {threadId?: unknown}).threadId; + return typeof threadId === "string" && this.nativeSubagents.has(threadId) + ? threadId + : this.session.sessionId; + } + + private async handleNativeSubagentNotification(notification: ServerNotification): Promise { + if (!this.supportsSubagents + || notification.method !== "item/started" && notification.method !== "item/completed") { + return false; + } + const item = notification.params.item; + if (item.type === "subAgentActivity") { + const hasNativeRepresentation = this.nativeSubagents.has(item.agentThreadId); + if (hasNativeRepresentation && item.kind === "interrupted") { + await this.finishNativeSubagent(item.agentThreadId, "cancelled"); + } + // Activity for an announced child is redundant with its dedicated + // lifecycle/session. Unknown activity still needs the legacy tool + // representation so the client does not lose provider output. + return hasNativeRepresentation; + } + if (item.type !== "collabAgentToolCall") { + return false; + } + + let hasNativeSpawnRepresentation = false; + if (item.tool === "spawnAgent") { + const parentSessionId = this.nativeSubagents.has(item.senderThreadId) + ? item.senderThreadId + : this.session.sessionId; + for (const childSessionId of item.receiverThreadIds) { + if (childSessionId.trim().length === 0) { + logger.log("Ignoring spawned subagent with an empty thread id"); + continue; + } + if (childSessionId === parentSessionId || childSessionId === this.session.sessionId) { + logger.log(`Ignoring self-referential spawned subagent ${childSessionId}`); + continue; + } + if (this.nativeSubagents.has(childSessionId)) { + hasNativeSpawnRepresentation = true; + continue; + } + const child = { + parentSessionId, + name: fallbackSubagentName(childSessionId), + task: item.prompt?.trim() || "Delegated task", + }; + this.nativeSubagents.set(childSessionId, child); + hasNativeSpawnRepresentation = true; + await this.session.update({ + sessionUpdate: "subagent_spawned", + subagentSessionId: childSessionId, + name: child.name, + task: child.task, + capabilities: {}, + }, parentSessionId); + } + } + + for (const [childSessionId, state] of Object.entries(item.agentsStates)) { + if (!state) continue; + const terminalState = toSubagentTerminalState(state.status); + if (terminalState) { + await this.finishNativeSubagent(childSessionId, terminalState); + } + } + // Only spawn has an equivalent ACP subagent lifecycle representation. + // Keep sendInput/resume/wait/close as ordinary tool calls; suppressing + // them would silently discard provider operations from the transcript. + return item.tool === "spawnAgent" && hasNativeSpawnRepresentation; + } + + private async finishNativeSubagent( + childSessionId: string, + state: SubagentState, + ): Promise { + const child = this.nativeSubagents.get(childSessionId); + if (!child || child.terminalState !== undefined) return; + child.terminalState = state; + await this.session.update({ + sessionUpdate: "subagent_state_update", + subagentSessionId: childSessionId, + state, + }, child.parentSessionId); + for (const waiter of this.nativeSubagentWaiters) waiter(); + this.nativeSubagentWaiters.clear(); + } + + async waitForNativeSubagents(signal: AbortSignal): Promise { + while ([...this.nativeSubagents.values()].some(child => child.terminalState === undefined)) { + if (signal.aborted) return; + await new Promise((resolve) => { + const onAbort = () => { + this.nativeSubagentWaiters.delete(onChange); + resolve(); + }; + const onChange = () => { + signal.removeEventListener("abort", onAbort); + resolve(); + }; + this.nativeSubagentWaiters.add(onChange); + signal.addEventListener("abort", onAbort, {once: true}); + }); + } + } + + async finishOutstandingNativeSubagents(state: SubagentState): Promise { + // Children are registered after their parents. Finish descendants first + // so every lifecycle update is delivered on a still-live parent stream. + const childSessionIds = [...this.nativeSubagents.keys()].reverse(); + for (const childSessionId of childSessionIds) { + await this.finishNativeSubagent(childSessionId, state); } } diff --git a/src/__tests__/CodexACPAgent/collab-agent-events.test.ts b/src/__tests__/CodexACPAgent/collab-agent-events.test.ts index 5ad2c72a..a89662e5 100644 --- a/src/__tests__/CodexACPAgent/collab-agent-events.test.ts +++ b/src/__tests__/CodexACPAgent/collab-agent-events.test.ts @@ -1,5 +1,6 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { ServerNotification } from "../../app-server"; +import type { ClientCapabilities } from "@agentclientprotocol/sdk"; import type { SessionState } from "../../CodexAcpServer"; import { AgentMode } from "../../AgentMode"; import { @@ -110,4 +111,427 @@ describe("CodexEventHandler - collab agent tool call events", () => { "data/subagent-activity-flow.json" ); }); + + it("emits native lifecycle and routes child output after capability negotiation", async () => { + const clientCapabilities = {subagents: {}} as ClientCapabilities & { + subagents: Record; + }; + const initializeResponse = await mockFixture.getCodexAcpAgent().initialize({ + protocolVersion: 1, + clientCapabilities, + }); + expect( + (initializeResponse.agentCapabilities?.sessionCapabilities as {subagents?: unknown}).subagents + ).toEqual({}); + const notifications: ServerNotification[] = [ + { + method: "item/started", + params: { + threadId: sessionId, + turnId: "turn-1", + startedAtMs: 0, + item: { + type: "collabAgentToolCall", + id: "call-spawn-weather", + tool: "spawnAgent", + status: "inProgress", + senderThreadId: "thread-main", + receiverThreadIds: ["thread-paris"], + prompt: "Find the current weather in Paris.", + model: null, + reasoningEffort: null, + agentsStates: { + "thread-paris": {status: "running", message: "Checking weather"}, + }, + }, + }, + }, + { + method: "item/agentMessage/delta", + params: { + threadId: "thread-paris", + turnId: "turn-child", + itemId: "child-message", + delta: "Weather found", + }, + }, + { + method: "item/completed", + params: { + threadId: sessionId, + turnId: "turn-1", + completedAtMs: 0, + item: { + type: "collabAgentToolCall", + id: "call-spawn-weather", + tool: "spawnAgent", + status: "completed", + senderThreadId: "thread-main", + receiverThreadIds: ["thread-paris"], + prompt: "Find the current weather in Paris.", + model: null, + reasoningEffort: null, + agentsStates: { + "thread-paris": {status: "completed", message: null}, + }, + }, + }, + }, + ]; + + await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, notifications); + + const updates = mockFixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0]); + expect(updates).toEqual([ + { + sessionId, + update: { + sessionUpdate: "subagent_spawned", + subagentSessionId: "thread-paris", + name: "Agent ad-paris", + task: "Find the current weather in Paris.", + capabilities: {}, + }, + }, + { + sessionId: "thread-paris", + update: { + sessionUpdate: "agent_message_chunk", + content: {type: "text", text: "Weather found"}, + messageId: "child-message", + }, + }, + { + sessionId, + update: { + sessionUpdate: "subagent_state_update", + subagentSessionId: "thread-paris", + state: "completed", + }, + }, + ]); + + mockFixture.setPermissionResponse({outcome: {outcome: "selected", optionId: "allow_once"}}); + await mockFixture.sendServerRequest("item/commandExecution/requestApproval", { + threadId: "thread-paris", + turnId: "turn-child", + itemId: "child-command", + reason: "Check the weather service", + startedAtMs: 0, + environmentId: null, + proposedExecpolicyAmendment: null, + }); + const permissionRequest = mockFixture.getAcpConnectionEvents([]) + .find(event => event.method === "requestPermission" && event.args[0].toolCall.toolCallId === "child-command"); + expect(permissionRequest?.args[0].sessionId).toBe("thread-paris"); + }); + + it("routes nested agents through their immediate parent sessions", async () => { + await mockFixture.getCodexAcpAgent().initialize({ + protocolVersion: 1, + clientCapabilities: {subagents: {}} as ClientCapabilities, + }); + const collabItem = ( + threadId: string, + senderThreadId: string, + receiverThreadId: string, + id: string, + status: "running" | "completed", + ): ServerNotification => ({ + method: status === "running" ? "item/started" : "item/completed", + params: { + threadId, + turnId: `turn-${threadId}`, + ...(status === "running" ? {startedAtMs: 0} : {completedAtMs: 0}), + item: { + type: "collabAgentToolCall", + id, + tool: "spawnAgent", + status: status === "running" ? "inProgress" : "completed", + senderThreadId, + receiverThreadIds: [receiverThreadId], + prompt: `Task for ${receiverThreadId}`, + model: null, + reasoningEffort: null, + agentsStates: {[receiverThreadId]: {status, message: null}}, + }, + }, + } as ServerNotification); + await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, [ + collabItem(sessionId, sessionId, "child-1", "spawn-1", "running"), + collabItem("child-1", "child-1", "grandchild-1", "spawn-2", "running"), + { + method: "item/agentMessage/delta", + params: { + threadId: "grandchild-1", + turnId: "turn-grandchild", + itemId: "grandchild-message", + delta: "Nested result", + }, + }, + collabItem("child-1", "child-1", "grandchild-1", "spawn-2", "completed"), + collabItem(sessionId, sessionId, "child-1", "spawn-1", "completed"), + ]); + + const updates = mockFixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0]); + expect(updates.map(({sessionId: target, update}) => [target, update.sessionUpdate])).toEqual([ + [sessionId, "subagent_spawned"], + ["child-1", "subagent_spawned"], + ["grandchild-1", "agent_message_chunk"], + ["child-1", "subagent_state_update"], + [sessionId, "subagent_state_update"], + ]); + }); + + it("deduplicates lifecycle, rejects blank IDs, and ignores late child output", async () => { + await mockFixture.getCodexAcpAgent().initialize({ + protocolVersion: 1, + clientCapabilities: {subagents: {}} as ClientCapabilities, + }); + const spawn = (method: "item/started" | "item/completed"): ServerNotification => ({ + method, + params: { + threadId: sessionId, + turnId: "turn-1", + ...(method === "item/started" ? {startedAtMs: 0} : {completedAtMs: 0}), + item: { + type: "collabAgentToolCall", + id: "spawn", + tool: "spawnAgent", + status: method === "item/started" ? "inProgress" : "completed", + senderThreadId: sessionId, + receiverThreadIds: ["", "child-1", "child-1"], + prompt: "Task", + model: null, + reasoningEffort: null, + agentsStates: {"child-1": {status: method === "item/started" ? "running" : "completed", message: null}}, + }, + }, + } as ServerNotification); + await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, [ + spawn("item/started"), + spawn("item/completed"), + spawn("item/completed"), + { + method: "item/agentMessage/delta", + params: { + threadId: "child-1", + turnId: "turn-child", + itemId: "late-message", + delta: "Too late", + }, + }, + ]); + + const updates = mockFixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0].update); + expect(updates).toHaveLength(2); + expect(updates.map(update => update.sessionUpdate)).toEqual([ + "subagent_spawned", + "subagent_state_update", + ]); + }); + + it("keeps unsupported collaboration controls visible in native mode", async () => { + await mockFixture.getCodexAcpAgent().initialize({ + protocolVersion: 1, + clientCapabilities: {subagents: {}} as ClientCapabilities, + }); + const collab = ( + method: "item/started" | "item/completed", + tool: "spawnAgent" | "sendInput", + id: string, + status: "running" | "completed", + ): ServerNotification => ({ + method, + params: { + threadId: sessionId, + turnId: "turn-1", + ...(method === "item/started" ? {startedAtMs: 0} : {completedAtMs: 0}), + item: { + type: "collabAgentToolCall", + id, + tool, + status: method === "item/started" ? "inProgress" : "completed", + senderThreadId: sessionId, + receiverThreadIds: ["child-1"], + prompt: tool === "spawnAgent" ? "Child task" : "Additional direction", + model: null, + reasoningEffort: null, + agentsStates: {"child-1": {status, message: null}}, + }, + }, + } as ServerNotification); + + await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, [ + collab("item/started", "spawnAgent", "spawn", "running"), + collab("item/started", "sendInput", "send-input", "running"), + collab("item/completed", "sendInput", "send-input", "running"), + collab("item/completed", "spawnAgent", "spawn", "completed"), + ]); + + const updates = mockFixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0].update); + expect(updates.map(update => [update.sessionUpdate, update.toolCallId, update.title])).toEqual([ + ["subagent_spawned", undefined, undefined], + ["tool_call", "send-input", "sendInput"], + ["tool_call_update", "send-input", "sendInput"], + ["subagent_state_update", undefined, undefined], + ]); + }); + + it("falls back to tool representation when a native spawn cannot be represented", async () => { + await mockFixture.getCodexAcpAgent().initialize({ + protocolVersion: 1, + clientCapabilities: {subagents: {}} as ClientCapabilities, + }); + await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, [{ + method: "item/completed", + params: { + threadId: sessionId, + turnId: "turn-1", + completedAtMs: 0, + item: { + type: "collabAgentToolCall", + id: "self-spawn", + tool: "spawnAgent", + status: "failed", + senderThreadId: sessionId, + receiverThreadIds: [sessionId], + prompt: "Invalid task", + model: null, + reasoningEffort: null, + agentsStates: {}, + }, + }, + }]); + + const updates = mockFixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0].update); + expect(updates).toHaveLength(1); + expect(updates[0]).toMatchObject({ + sessionUpdate: "tool_call_update", + toolCallId: "self-spawn", + title: "spawnAgent", + status: "failed", + }); + }); + + it("does not duplicate global notifications after subscribing to a child", async () => { + await mockFixture.getCodexAcpAgent().initialize({ + protocolVersion: 1, + clientCapabilities: {subagents: {}} as ClientCapabilities, + }); + await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, [ + { + method: "item/started", + params: { + threadId: sessionId, + turnId: "turn-1", + startedAtMs: 0, + item: { + type: "collabAgentToolCall", + id: "spawn", + tool: "spawnAgent", + status: "inProgress", + senderThreadId: sessionId, + receiverThreadIds: ["child-1"], + prompt: "Child task", + model: null, + reasoningEffort: null, + agentsStates: {"child-1": {status: "running", message: null}}, + }, + }, + }, + {method: "warning", params: {threadId: null, message: "Global warning"}}, + { + method: "item/completed", + params: { + threadId: sessionId, + turnId: "turn-1", + completedAtMs: 0, + item: { + type: "collabAgentToolCall", + id: "spawn", + tool: "spawnAgent", + status: "completed", + senderThreadId: sessionId, + receiverThreadIds: ["child-1"], + prompt: "Child task", + model: null, + reasoningEffort: null, + agentsStates: {"child-1": {status: "completed", message: null}}, + }, + }, + }, + ]); + + const warningUpdates = mockFixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0].update) + .filter(update => update.sessionUpdate === "agent_message_chunk" + && update.content?.text.includes("Global warning")); + expect(warningUpdates).toHaveLength(1); + }); + + it("keeps the parent prompt open until every announced child is terminal", async () => { + await mockFixture.getCodexAcpAgent().initialize({ + protocolVersion: 1, + clientCapabilities: {subagents: {}} as ClientCapabilities, + }); + const appServer = mockFixture.getCodexAppServerClient(); + const turn = {id: "turn-1", items: [], status: "inProgress" as const, error: null}; + const completedTurn = {...turn, status: "completed" as const}; + let completeTurn!: () => void; + const completed = new Promise<{threadId: string; turn: typeof completedTurn}>(resolve => { + completeTurn = () => resolve({threadId: sessionId, turn: completedTurn}); + }); + appServer.turnStart = vi.fn().mockResolvedValue({turn}); + appServer.awaitTurnCompleted = vi.fn().mockReturnValue(completed); + vi.spyOn(mockFixture.getCodexAcpAgent(), "getSessionState").mockReturnValue(sessionState); + + const prompt = mockFixture.getCodexAcpAgent().prompt({ + sessionId, + prompt: [{type: "text", text: "Delegate work"}], + }); + await vi.waitFor(() => expect(appServer.turnStart).toHaveBeenCalled()); + const spawn = (status: "running" | "completed") => mockFixture.sendServerNotification({ + method: status === "running" ? "item/started" : "item/completed", + params: { + threadId: sessionId, + turnId: "turn-1", + ...(status === "running" ? {startedAtMs: 0} : {completedAtMs: 0}), + item: { + type: "collabAgentToolCall", + id: "spawn", + tool: "spawnAgent", + status: status === "running" ? "inProgress" : "completed", + senderThreadId: sessionId, + receiverThreadIds: ["child-1"], + prompt: "Child task", + model: null, + reasoningEffort: null, + agentsStates: {"child-1": {status, message: null}}, + }, + }, + }); + spawn("running"); + await mockFixture.getCodexAcpClient().waitForSessionNotifications(sessionId); + completeTurn(); + + let promptSettled = false; + void prompt.finally(() => { promptSettled = true; }); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(promptSettled).toBe(false); + + spawn("completed"); + await expect(prompt).resolves.toMatchObject({stopReason: "end_turn"}); + }); }); diff --git a/src/acp-subagents.ts b/src/acp-subagents.ts new file mode 100644 index 00000000..94534708 --- /dev/null +++ b/src/acp-subagents.ts @@ -0,0 +1,66 @@ +import type { + ClientCapabilities, + SessionCapabilities, + SessionNotification, +} from "@agentclientprotocol/sdk"; + +/** + * Temporary typed surface for agentclientprotocol/agent-client-protocol#1992. + * + * The wire contract is already defined by the ACP draft, but the published + * TypeScript SDK does not contain it yet. Keep the compatibility boundary in + * this file so it can be replaced by SDK exports without changing lifecycle + * code when the draft ships. + */ +export type SubagentSessionCapabilities = { + cancel?: boolean; + close?: boolean; + _meta?: Record | null; +}; + +export type SubagentSpawnedUpdate = { + sessionUpdate: "subagent_spawned"; + subagentSessionId: string; + name: string; + task: string; + capabilities: SubagentSessionCapabilities; + _meta?: Record | null; +}; + +export type SubagentState = "completed" | "failed" | "cancelled"; + +export type SubagentStateUpdate = { + sessionUpdate: "subagent_state_update"; + subagentSessionId: string; + state: SubagentState; + _meta?: Record | null; +}; + +export type AcpSessionUpdate = + | SessionNotification["update"] + | SubagentSpawnedUpdate + | SubagentStateUpdate; + +export type AcpSessionNotification = Omit & { + update: AcpSessionUpdate; +}; + +export type SubagentAwareSessionCapabilities = SessionCapabilities & { + subagents?: Record; +}; + +export function clientSupportsSubagents( + capabilities?: ClientCapabilities | null, +): boolean { + const subagents = ( + capabilities as (ClientCapabilities & { subagents?: unknown }) | null | undefined + )?.subagents; + return typeof subagents === "object" && subagents !== null && !Array.isArray(subagents); +} + +/** The only cast needed until the TypeScript SDK publishes PR #1992. */ +export function asSdkSessionNotification( + notification: AcpSessionNotification, +): SessionNotification { + return notification as SessionNotification; +} From 9491017d95548f0afc242fa9667d7cb708ae9a3e Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 20 Aug 2026 18:15:28 +0400 Subject: [PATCH 2/5] fix: require bilateral subagent negotiation --- src/CodexAcpClient.ts | 54 ++++++++---- src/CodexEventHandler.ts | 10 ++- .../CodexACPAgent/collab-agent-events.test.ts | 44 ++++++++-- .../data/collab-agent-tool-call-flow.json | 83 ------------------- .../data/subagent-activity-flow.json | 29 ------- 5 files changed, 80 insertions(+), 140 deletions(-) delete mode 100644 src/__tests__/CodexACPAgent/data/collab-agent-tool-call-flow.json delete mode 100644 src/__tests__/CodexACPAgent/data/subagent-activity-flow.json diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index c429b7fa..981ff0f4 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -773,35 +773,48 @@ export class CodexAcpClient { const dispatch = (event: ServerNotification) => { this.enqueueSessionNotification(sessionId, () => eventHandler(event)); }; - const registerInteractiveHandlers = (targetSessionId: string): void => { + const registerInteractiveHandlers = (targetSessionId: string, includeElicitations = true): void => { this.codexClient.onApprovalRequest(targetSessionId, { handleCommandExecution: async (params) => { await this.waitForSessionNotifications(sessionId); - return await approvalHandler.handleCommandExecution(params); + return await approvalHandler.handleCommandExecution( + !supportsSubagents && targetSessionId !== sessionId + ? {...params, threadId: sessionId} + : params + ); }, handleFileChange: async (params) => { await this.waitForSessionNotifications(sessionId); - return await approvalHandler.handleFileChange(params); + return await approvalHandler.handleFileChange( + !supportsSubagents && targetSessionId !== sessionId + ? {...params, threadId: sessionId} + : params + ); }, handlePermissionsRequest: async (params) => { await this.waitForSessionNotifications(sessionId); - return await approvalHandler.handlePermissionsRequest(params); - }, - }); - this.codexClient.onElicitationRequest(targetSessionId, { - handleElicitation: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await elicitationHandler.handleElicitation(params); - }, - handleUserInput: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await elicitationHandler.handleUserInput(params); + return await approvalHandler.handlePermissionsRequest( + !supportsSubagents && targetSessionId !== sessionId + ? {...params, threadId: sessionId} + : params + ); }, }); + if (includeElicitations) { + this.codexClient.onElicitationRequest(targetSessionId, { + handleElicitation: async (params) => { + await this.waitForSessionNotifications(sessionId); + return await elicitationHandler.handleElicitation(params); + }, + handleUserInput: async (params) => { + await this.waitForSessionNotifications(sessionId); + return await elicitationHandler.handleUserInput(params); + }, + }); + } }; const subscribeDiscoveredChildren = (event: ServerNotification): void => { - if (!supportsSubagents - || (event.method !== "item/started" && event.method !== "item/completed") + if ((event.method !== "item/started" && event.method !== "item/completed") || event.params.item.type !== "collabAgentToolCall" || event.params.item.tool !== "spawnAgent") { return; @@ -825,9 +838,14 @@ export class CodexAcpClient { return; } subscribeDiscoveredChildren(childEvent); - dispatch(childEvent); + if (supportsSubagents) { + dispatch(childEvent); + } }); - registerInteractiveHandlers(childSessionId); + // Without native subagent negotiation only permission requests + // cross the hidden child boundary. Other child interaction and + // transcript events remain private to the provider. + registerInteractiveHandlers(childSessionId, supportsSubagents); } }; this.codexClient.onServerNotification(sessionId, (event) => { diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index e07dcb4e..51eedd2c 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -416,11 +416,17 @@ export class CodexEventHandler { } private async handleNativeSubagentNotification(notification: ServerNotification): Promise { - if (!this.supportsSubagents - || notification.method !== "item/started" && notification.method !== "item/completed") { + if (notification.method !== "item/started" && notification.method !== "item/completed") { return false; } const item = notification.params.item; + if (!this.supportsSubagents) { + // Subagents are an all-or-nothing negotiated surface. Legacy collaboration + // tools must not leak a second, tool-shaped representation to clients that + // did not advertise native child sessions. Approval requests use their own + // ACP request path and remain available on the root session. + return item.type === "collabAgentToolCall" || item.type === "subAgentActivity"; + } if (item.type === "subAgentActivity") { const hasNativeRepresentation = this.nativeSubagents.has(item.agentThreadId); if (hasNativeRepresentation && item.kind === "interrupted") { diff --git a/src/__tests__/CodexACPAgent/collab-agent-events.test.ts b/src/__tests__/CodexACPAgent/collab-agent-events.test.ts index a89662e5..9cbb1049 100644 --- a/src/__tests__/CodexACPAgent/collab-agent-events.test.ts +++ b/src/__tests__/CodexACPAgent/collab-agent-events.test.ts @@ -25,7 +25,7 @@ describe("CodexEventHandler - collab agent tool call events", () => { agentMode: AgentMode.DEFAULT_AGENT_MODE, }); - it("maps live collab agent tool calls to ACP tool call updates", async () => { + it("hides collaboration lifecycle when the client lacks subagent capability but keeps permissions", async () => { const notifications: ServerNotification[] = [ { method: "item/started", @@ -77,16 +77,37 @@ describe("CodexEventHandler - collab agent tool call events", () => { }, }, }, + { + method: "item/agentMessage/delta", + params: { + threadId: sessionId, + turnId: "turn-1", + itemId: "parent-message", + delta: "Visible parent output", + }, + }, ]; await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, notifications); - await expect(`${mockFixture.getAcpConnectionDump([])}\n`).toMatchFileSnapshot( - "data/collab-agent-tool-call-flow.json" - ); + expect(JSON.stringify(mockFixture.getAcpConnectionEvents([]))).not.toContain("call-spawn-weather"); + + mockFixture.setPermissionResponse({outcome: {outcome: "selected", optionId: "allow_once"}}); + await mockFixture.sendServerRequest("item/commandExecution/requestApproval", { + threadId: "thread-paris", + turnId: "turn-child", + itemId: "child-command", + reason: "Check the weather service", + startedAtMs: 0, + environmentId: null, + proposedExecpolicyAmendment: null, + }); + const permissionRequest = mockFixture.getAcpConnectionEvents([]) + .find(event => event.method === "requestPermission" && event.args[0].toolCall.toolCallId === "child-command"); + expect(permissionRequest?.args[0].sessionId).toBe(sessionId); }); - it("maps live subagent activity to an ACP tool call", async () => { + it("hides legacy subagent activity when the client lacks subagent capability", async () => { const notifications: ServerNotification[] = [ { method: "item/completed", @@ -103,13 +124,20 @@ describe("CodexEventHandler - collab agent tool call events", () => { }, }, }, + { + method: "item/agentMessage/delta", + params: { + threadId: sessionId, + turnId: "turn-1", + itemId: "parent-message", + delta: "Visible parent output", + }, + }, ]; await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, notifications); - await expect(`${mockFixture.getAcpConnectionDump([])}\n`).toMatchFileSnapshot( - "data/subagent-activity-flow.json" - ); + expect(JSON.stringify(mockFixture.getAcpConnectionEvents([]))).not.toContain("call-spawn-weather"); }); it("emits native lifecycle and routes child output after capability negotiation", async () => { diff --git a/src/__tests__/CodexACPAgent/data/collab-agent-tool-call-flow.json b/src/__tests__/CodexACPAgent/data/collab-agent-tool-call-flow.json deleted file mode 100644 index 4391e5a9..00000000 --- a/src/__tests__/CodexACPAgent/data/collab-agent-tool-call-flow.json +++ /dev/null @@ -1,83 +0,0 @@ -{ - "method": "sessionUpdate", - "args": [ - { - "sessionId": "test-session-id", - "update": { - "sessionUpdate": "tool_call", - "toolCallId": "call-spawn-weather", - "kind": "other", - "title": "spawnAgent", - "status": "in_progress", - "rawInput": { - "prompt": "Find the current weather in Paris.", - "senderThreadId": "thread-main", - "receiverThreadIds": [ - "thread-paris" - ], - "agentsStates": { - "thread-paris": { - "status": "running", - "message": "Checking weather" - } - }, - "model": null, - "reasoningEffort": null, - "status": "inProgress" - }, - "_meta": { - "codex": { - "collaboration": { - "tool": "spawnAgent", - "senderThreadId": "thread-main", - "receiverThreadIds": [ - "thread-paris" - ] - } - } - } - } - } - ] -} -{ - "method": "sessionUpdate", - "args": [ - { - "sessionId": "test-session-id", - "update": { - "sessionUpdate": "tool_call_update", - "toolCallId": "call-spawn-weather", - "title": "spawnAgent", - "status": "completed", - "rawInput": { - "prompt": "Find the current weather in Paris.", - "senderThreadId": "thread-main", - "receiverThreadIds": [ - "thread-paris" - ], - "agentsStates": { - "thread-paris": { - "status": "completed", - "message": null - } - }, - "model": null, - "reasoningEffort": null, - "status": "completed" - }, - "_meta": { - "codex": { - "collaboration": { - "tool": "spawnAgent", - "senderThreadId": "thread-main", - "receiverThreadIds": [ - "thread-paris" - ] - } - } - } - } - } - ] -} diff --git a/src/__tests__/CodexACPAgent/data/subagent-activity-flow.json b/src/__tests__/CodexACPAgent/data/subagent-activity-flow.json deleted file mode 100644 index e1b6749c..00000000 --- a/src/__tests__/CodexACPAgent/data/subagent-activity-flow.json +++ /dev/null @@ -1,29 +0,0 @@ -{ - "method": "sessionUpdate", - "args": [ - { - "sessionId": "test-session-id", - "update": { - "sessionUpdate": "tool_call", - "title": "Start subagent weather_research", - "kind": "other", - "toolCallId": "call-spawn-weather", - "status": "completed", - "rawInput": { - "agentThreadId": "thread-paris", - "agentPath": "/root/weather_research", - "activityKind": "started" - }, - "_meta": { - "codex": { - "subagent": { - "threadId": "thread-paris", - "path": "/root/weather_research", - "activity": "started" - } - } - } - } - } - ] -} From e1224d31bf2a5881bce1578019b32cf029f66d59 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 20 Aug 2026 18:42:16 +0400 Subject: [PATCH 3/5] refactor: isolate subagent session logic --- src/ACPSessionConnection.ts | 2 +- src/CodexAcpClient.ts | 98 +-------- src/CodexAcpServer.ts | 2 +- src/CodexEventHandler.ts | 190 ++--------------- .../AcpSubagents.ts} | 9 +- src/subagents/CodexSubagentEventRouter.ts | 191 ++++++++++++++++++ src/subagents/CodexSubagentSubscriptions.ts | 113 +++++++++++ 7 files changed, 338 insertions(+), 267 deletions(-) rename src/{acp-subagents.ts => subagents/AcpSubagents.ts} (82%) create mode 100644 src/subagents/CodexSubagentEventRouter.ts create mode 100644 src/subagents/CodexSubagentSubscriptions.ts diff --git a/src/ACPSessionConnection.ts b/src/ACPSessionConnection.ts index 84b24294..e29a6514 100644 --- a/src/ACPSessionConnection.ts +++ b/src/ACPSessionConnection.ts @@ -2,7 +2,7 @@ import * as acp from "@agentclientprotocol/sdk"; import { type AcpSessionUpdate, asSdkSessionNotification, -} from "./acp-subagents"; +} from "./subagents/AcpSubagents"; export type AcpClientConnection = Pick; diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 981ff0f4..a4a99749 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -63,6 +63,7 @@ import { createReportedAgentFileChangeReport, createUnavailableAgentFileChangeReport, } from "./AgentFileChangeReport"; +import {CodexSubagentSubscriptions} from "./subagents/CodexSubagentSubscriptions"; /** * Well-known provider id for the client-configurable custom LLM gateway. @@ -108,6 +109,7 @@ export class CodexAcpClient { private pendingLoginCompleted: Promise | null = null; private pendingAccountUpdated: Promise | null = null; private readonly sessionNotificationQueues = new Map>(); + private readonly subagents: CodexSubagentSubscriptions; private skillExtraRoots: string[] = []; private configPath: string | null = null; @@ -117,6 +119,7 @@ export class CodexAcpClient { this.config = codexConfig ?? {}; this.modelProvider = modelProvider ?? null; this.gatewayConfig = null; + this.subagents = new CodexSubagentSubscriptions(codexClient); } private readonly defaultClientInfo: ClientInfo = { @@ -525,10 +528,7 @@ export class CodexAcpClient { await this.codexClient.threadUnsubscribe({threadId: sessionId}); } finally { this.codexClient.clearThreadHandlers(sessionId); - for (const childSessionId of this.subagentSubscriptions.get(sessionId) ?? []) { - this.codexClient.clearThreadHandlers(childSessionId); - } - this.subagentSubscriptions.delete(sessionId); + this.subagents.clear(sessionId); } } @@ -773,88 +773,14 @@ export class CodexAcpClient { const dispatch = (event: ServerNotification) => { this.enqueueSessionNotification(sessionId, () => eventHandler(event)); }; - const registerInteractiveHandlers = (targetSessionId: string, includeElicitations = true): void => { - this.codexClient.onApprovalRequest(targetSessionId, { - handleCommandExecution: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await approvalHandler.handleCommandExecution( - !supportsSubagents && targetSessionId !== sessionId - ? {...params, threadId: sessionId} - : params - ); - }, - handleFileChange: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await approvalHandler.handleFileChange( - !supportsSubagents && targetSessionId !== sessionId - ? {...params, threadId: sessionId} - : params - ); - }, - handlePermissionsRequest: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await approvalHandler.handlePermissionsRequest( - !supportsSubagents && targetSessionId !== sessionId - ? {...params, threadId: sessionId} - : params - ); - }, - }); - if (includeElicitations) { - this.codexClient.onElicitationRequest(targetSessionId, { - handleElicitation: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await elicitationHandler.handleElicitation(params); - }, - handleUserInput: async (params) => { - await this.waitForSessionNotifications(sessionId); - return await elicitationHandler.handleUserInput(params); - }, - }); - } - }; - const subscribeDiscoveredChildren = (event: ServerNotification): void => { - if ((event.method !== "item/started" && event.method !== "item/completed") - || event.params.item.type !== "collabAgentToolCall" - || event.params.item.tool !== "spawnAgent") { - return; - } - let children = this.subagentSubscriptions.get(sessionId); - if (!children) { - children = new Set(); - this.subagentSubscriptions.set(sessionId, children); - } - for (const childSessionId of event.params.item.receiverThreadIds) { - if (childSessionId.trim() === "") continue; - if (childSessionId === sessionId || childSessionId === event.params.threadId) continue; - if (children.has(childSessionId)) continue; - children.add(childSessionId); - this.codexClient.onServerNotification(childSessionId, (childEvent) => { - const eventThreadId = (childEvent.params as {threadId?: unknown}).threadId; - if (eventThreadId !== childSessionId) { - // Notifications without a thread id are broadcast by - // CodexAppServerClient. The root handler owns those; - // processing them here would duplicate them once per child. - return; - } - subscribeDiscoveredChildren(childEvent); - if (supportsSubagents) { - dispatch(childEvent); - } - }); - // Without native subagent negotiation only permission requests - // cross the hidden child boundary. Other child interaction and - // transcript events remain private to the provider. - registerInteractiveHandlers(childSessionId, supportsSubagents); - } - }; - this.codexClient.onServerNotification(sessionId, (event) => { - // Register synchronously before queueing the spawn update. App-server - // may emit the first child event immediately after the root event. - subscribeDiscoveredChildren(event); - dispatch(event); + this.subagents.subscribe({ + rootSessionId: sessionId, + supportsSubagents, + dispatch, + approvalHandler, + elicitationHandler, + waitForRootNotifications: () => this.waitForSessionNotifications(sessionId), }); - registerInteractiveHandlers(sessionId); } async waitForSessionNotifications(sessionId: string): Promise { @@ -884,8 +810,6 @@ export class CodexAcpClient { }); } - private readonly subagentSubscriptions = new Map>(); - async sendPrompt( request: acp.PromptRequest, agentMode: AgentMode, diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index a5ccc3e8..b717fd41 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -97,7 +97,7 @@ import {sameThreadGoalSnapshot, type ThreadGoalSnapshot, toThreadGoalSnapshot,} import { clientSupportsSubagents, type SubagentAwareSessionCapabilities, -} from "./acp-subagents"; +} from "./subagents/AcpSubagents"; import {randomUUID} from "node:crypto"; import {once} from "node:events"; import { diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index 51eedd2c..a9edb80a 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -41,8 +41,6 @@ import type { McpStartupCompleteEvent } from "./app-server/McpStartupCompleteEve import {toTokenCount} from "./TokenCount"; import { commandExecutionUsesTerminalOutput, - createCollabAgentToolCallCompleteUpdate, - createCollabAgentToolCallUpdate, createCommandExecutionUpdate, createContextCompactionCompleteUpdate, createContextCompactionStartUpdate, @@ -59,7 +57,6 @@ import { createFuzzyFileSearchComplete, createFuzzyFileSearchStartOrUpdate, createMcpToolCallUpdate, - createSubAgentActivityUpdate, createWebSearchCompleteUpdate, createWebSearchStartUpdate, fuzzyFileSearchToolCallId, @@ -72,7 +69,6 @@ import { createAgentTextThoughtChunk, } from "./ContentChunks"; import {sameThreadGoalSnapshot, type ThreadGoalSnapshot, toThreadGoalSnapshot} from "./ThreadGoalSnapshot"; -import type {SubagentState} from "./acp-subagents"; import {logger} from "./Logger"; import {randomUUID} from "node:crypto"; import { @@ -82,6 +78,8 @@ import { AIR_SESSION_FAILURE_KEY, JETBRAINS_META_KEY, } from "./AirExtension"; +import {CodexSubagentEventRouter} from "./subagents/CodexSubagentEventRouter"; +import type {SubagentState} from "./subagents/AcpSubagents"; export { stripShellPrefix }; @@ -102,29 +100,6 @@ type SessionFailurePolicy = { const MAX_SESSION_FAILURE_TITLE_LENGTH = 240; -function toSubagentTerminalState(status: string): SubagentState | undefined { - switch (status) { - case "completed": - return "completed"; - case "interrupted": - return "cancelled"; - case "errored": - case "shutdown": - case "notFound": - return "failed"; - case "pendingInit": - case "running": - return undefined; - default: - return undefined; - } -} - -function fallbackSubagentName(sessionId: string): string { - const suffix = sessionId.length > 8 ? sessionId.slice(-8) : sessionId; - return `Agent ${suffix}`; -} - const SESSION_FAILURE_POLICY: Record = { transport_lost: { category: "connection", @@ -248,14 +223,7 @@ export class CodexEventHandler { private readonly terminalCommandIds = new Set(); private readonly terminalCommandOutputIds = new Set(); private readonly agentMessagePhases = new Map(); - private readonly activeSubAgentActivities = new Set(); - private readonly nativeSubagents = new Map(); - private readonly nativeSubagentWaiters = new Set<() => void>(); + private readonly subagents: CodexSubagentEventRouter; constructor( connection: AcpClientConnection, @@ -263,13 +231,19 @@ export class CodexEventHandler { supportsPlanUpdates = false, supportsTypedSessionFailures = false, sessionFailureEpoch: string = randomUUID(), - private readonly supportsSubagents = false, + supportsSubagents = false, ) { this.sessionState = sessionState; this.supportsPlanUpdates = supportsPlanUpdates; this.supportsTypedSessionFailures = supportsTypedSessionFailures; this.sessionFailureEpoch = sessionFailureEpoch; this.session = new ACPSessionConnection(connection, sessionState.sessionId); + this.subagents = new CodexSubagentEventRouter( + sessionState.sessionId, + supportsSubagents, + (update, sessionId) => this.session.update(update, sessionId), + (message) => logger.log(message), + ); if (sessionState.sessionFailure !== undefined) { this.failuresById.set(sessionState.sessionFailure.id, sessionState.sessionFailure); } @@ -393,143 +367,24 @@ export class CodexEventHandler { async handleNotification(notification: ServerNotification) { await this.flushPendingErrors(); - if (await this.handleNativeSubagentNotification(notification)) { + if (await this.subagents.handle(notification)) { return; } - const notificationThreadId = (notification.params as {threadId?: unknown}).threadId; - if (typeof notificationThreadId === "string" - && this.nativeSubagents.get(notificationThreadId)?.terminalState !== undefined) { - logger.log(`Ignoring update for terminal subagent ${notificationThreadId}`); + if (this.subagents.shouldIgnore(notification)) { return; } const updateEvent = await this.createUpdateEvent(notification); if (updateEvent) { - await this.session.update(updateEvent, this.notificationSessionId(notification)); - } - } - - private notificationSessionId(notification: ServerNotification): string { - const threadId = (notification.params as {threadId?: unknown}).threadId; - return typeof threadId === "string" && this.nativeSubagents.has(threadId) - ? threadId - : this.session.sessionId; - } - - private async handleNativeSubagentNotification(notification: ServerNotification): Promise { - if (notification.method !== "item/started" && notification.method !== "item/completed") { - return false; - } - const item = notification.params.item; - if (!this.supportsSubagents) { - // Subagents are an all-or-nothing negotiated surface. Legacy collaboration - // tools must not leak a second, tool-shaped representation to clients that - // did not advertise native child sessions. Approval requests use their own - // ACP request path and remain available on the root session. - return item.type === "collabAgentToolCall" || item.type === "subAgentActivity"; - } - if (item.type === "subAgentActivity") { - const hasNativeRepresentation = this.nativeSubagents.has(item.agentThreadId); - if (hasNativeRepresentation && item.kind === "interrupted") { - await this.finishNativeSubagent(item.agentThreadId, "cancelled"); - } - // Activity for an announced child is redundant with its dedicated - // lifecycle/session. Unknown activity still needs the legacy tool - // representation so the client does not lose provider output. - return hasNativeRepresentation; - } - if (item.type !== "collabAgentToolCall") { - return false; - } - - let hasNativeSpawnRepresentation = false; - if (item.tool === "spawnAgent") { - const parentSessionId = this.nativeSubagents.has(item.senderThreadId) - ? item.senderThreadId - : this.session.sessionId; - for (const childSessionId of item.receiverThreadIds) { - if (childSessionId.trim().length === 0) { - logger.log("Ignoring spawned subagent with an empty thread id"); - continue; - } - if (childSessionId === parentSessionId || childSessionId === this.session.sessionId) { - logger.log(`Ignoring self-referential spawned subagent ${childSessionId}`); - continue; - } - if (this.nativeSubagents.has(childSessionId)) { - hasNativeSpawnRepresentation = true; - continue; - } - const child = { - parentSessionId, - name: fallbackSubagentName(childSessionId), - task: item.prompt?.trim() || "Delegated task", - }; - this.nativeSubagents.set(childSessionId, child); - hasNativeSpawnRepresentation = true; - await this.session.update({ - sessionUpdate: "subagent_spawned", - subagentSessionId: childSessionId, - name: child.name, - task: child.task, - capabilities: {}, - }, parentSessionId); - } - } - - for (const [childSessionId, state] of Object.entries(item.agentsStates)) { - if (!state) continue; - const terminalState = toSubagentTerminalState(state.status); - if (terminalState) { - await this.finishNativeSubagent(childSessionId, terminalState); - } + await this.session.update(updateEvent, this.subagents.notificationSessionId(notification)); } - // Only spawn has an equivalent ACP subagent lifecycle representation. - // Keep sendInput/resume/wait/close as ordinary tool calls; suppressing - // them would silently discard provider operations from the transcript. - return item.tool === "spawnAgent" && hasNativeSpawnRepresentation; - } - - private async finishNativeSubagent( - childSessionId: string, - state: SubagentState, - ): Promise { - const child = this.nativeSubagents.get(childSessionId); - if (!child || child.terminalState !== undefined) return; - child.terminalState = state; - await this.session.update({ - sessionUpdate: "subagent_state_update", - subagentSessionId: childSessionId, - state, - }, child.parentSessionId); - for (const waiter of this.nativeSubagentWaiters) waiter(); - this.nativeSubagentWaiters.clear(); } async waitForNativeSubagents(signal: AbortSignal): Promise { - while ([...this.nativeSubagents.values()].some(child => child.terminalState === undefined)) { - if (signal.aborted) return; - await new Promise((resolve) => { - const onAbort = () => { - this.nativeSubagentWaiters.delete(onChange); - resolve(); - }; - const onChange = () => { - signal.removeEventListener("abort", onAbort); - resolve(); - }; - this.nativeSubagentWaiters.add(onChange); - signal.addEventListener("abort", onAbort, {once: true}); - }); - } + await this.subagents.wait(signal); } async finishOutstandingNativeSubagents(state: SubagentState): Promise { - // Children are registered after their parents. Finish descendants first - // so every lifecycle update is delivered on a still-live parent stream. - const childSessionIds = [...this.nativeSubagents.keys()].reverse(); - for (const childSessionId of childSessionIds) { - await this.finishNativeSubagent(childSessionId, state); - } + await this.subagents.finishOutstanding(state); } async flushPendingPlanUpdates(): Promise { @@ -846,15 +701,14 @@ export class CodexEventHandler { this.activeImageGenerationItems.add(event.item.id); return createImageGenerationStartUpdate(event.item); case "collabAgentToolCall": - return createCollabAgentToolCallUpdate(event.item); + return this.subagents.legacyCollaborationStarted(event.item); case "agentMessage": this.rememberAgentMessagePhase(event.item); return null; case "contextCompaction": return createContextCompactionStartUpdate(event.item); case "subAgentActivity": - this.activeSubAgentActivities.add(event.item.id); - return createSubAgentActivityUpdate(event.item, "in_progress", "tool_call"); + return this.subagents.legacyActivityStarted(event.item); case "sleep": case "userMessage": case "hookPrompt": @@ -903,7 +757,7 @@ export class CodexEventHandler { case "webSearch": return createWebSearchCompleteUpdate(event.item); case "collabAgentToolCall": - return createCollabAgentToolCallCompleteUpdate(event.item); + return this.subagents.legacyCollaborationCompleted(event.item); case "agentMessage": this.rememberAgentMessagePhase(event.item); return null; @@ -916,12 +770,8 @@ export class CodexEventHandler { case "contextCompaction": return createContextCompactionCompleteUpdate(event.item); //ignored types - case "subAgentActivity": { - const sessionUpdate = this.activeSubAgentActivities.delete(event.item.id) - ? "tool_call_update" - : "tool_call"; - return createSubAgentActivityUpdate(event.item, "completed", sessionUpdate); - } + case "subAgentActivity": + return this.subagents.legacyActivityCompleted(event.item); case "sleep": case "userMessage": case "hookPrompt": diff --git a/src/acp-subagents.ts b/src/subagents/AcpSubagents.ts similarity index 82% rename from src/acp-subagents.ts rename to src/subagents/AcpSubagents.ts index 94534708..486e53f7 100644 --- a/src/acp-subagents.ts +++ b/src/subagents/AcpSubagents.ts @@ -4,14 +4,7 @@ import type { SessionNotification, } from "@agentclientprotocol/sdk"; -/** - * Temporary typed surface for agentclientprotocol/agent-client-protocol#1992. - * - * The wire contract is already defined by the ACP draft, but the published - * TypeScript SDK does not contain it yet. Keep the compatibility boundary in - * this file so it can be replaced by SDK exports without changing lifecycle - * code when the draft ships. - */ +/** Temporary typed surface for agentclientprotocol/agent-client-protocol#1992. */ export type SubagentSessionCapabilities = { cancel?: boolean; close?: boolean; diff --git a/src/subagents/CodexSubagentEventRouter.ts b/src/subagents/CodexSubagentEventRouter.ts new file mode 100644 index 00000000..ca520733 --- /dev/null +++ b/src/subagents/CodexSubagentEventRouter.ts @@ -0,0 +1,191 @@ +import type {ServerNotification} from "../app-server"; +import type {ThreadItem} from "../app-server/v2"; +import type {UpdateSessionEvent} from "../ACPSessionConnection"; +import { + createCollabAgentToolCallCompleteUpdate, + createCollabAgentToolCallUpdate, + createSubAgentActivityUpdate, +} from "../CodexToolCallMapper"; +import type {SubagentState} from "./AcpSubagents"; + +type Publisher = (update: UpdateSessionEvent, sessionId?: string) => Promise; +type Log = (message: string) => void; + +type NativeSubagent = { + parentSessionId: string; + name: string; + task: string; + terminalState?: SubagentState; +}; + +/** Owns native lifecycle, child routing, waiting, and legacy activity deduplication. */ +export class CodexSubagentEventRouter { + private readonly children = new Map(); + private readonly waiters = new Set<() => void>(); + private readonly activeLegacyActivities = new Set(); + + constructor( + private readonly rootSessionId: string, + private readonly supported: boolean, + private readonly publish: Publisher, + private readonly log: Log, + ) {} + + async handle(notification: ServerNotification): Promise { + if (notification.method !== "item/started" && notification.method !== "item/completed") { + return false; + } + const item = notification.params.item; + if (!this.supported) { + // Permissions use their own ACP request path. Every transcript or + // lifecycle representation stays hidden without bilateral support. + return item.type === "collabAgentToolCall" || item.type === "subAgentActivity"; + } + if (item.type === "subAgentActivity") { + const hasNativeRepresentation = this.children.has(item.agentThreadId); + if (hasNativeRepresentation && item.kind === "interrupted") { + await this.finish(item.agentThreadId, "cancelled"); + } + return hasNativeRepresentation; + } + if (item.type !== "collabAgentToolCall") return false; + + let representedSpawn = false; + if (item.tool === "spawnAgent") { + const parentSessionId = this.children.has(item.senderThreadId) + ? item.senderThreadId + : this.rootSessionId; + for (const childSessionId of item.receiverThreadIds) { + if (childSessionId.trim().length === 0) { + this.log("Ignoring spawned subagent with an empty thread id"); + continue; + } + if (childSessionId === parentSessionId || childSessionId === this.rootSessionId) { + this.log(`Ignoring self-referential spawned subagent ${childSessionId}`); + continue; + } + if (this.children.has(childSessionId)) { + representedSpawn = true; + continue; + } + const child = { + parentSessionId, + name: fallbackName(childSessionId), + task: item.prompt?.trim() || "Delegated task", + }; + this.children.set(childSessionId, child); + representedSpawn = true; + await this.publish({ + sessionUpdate: "subagent_spawned", + subagentSessionId: childSessionId, + name: child.name, + task: child.task, + capabilities: {}, + }, parentSessionId); + } + } + + for (const [childSessionId, state] of Object.entries(item.agentsStates)) { + const terminalState = state && terminalStateOf(state.status); + if (terminalState) await this.finish(childSessionId, terminalState); + } + // `updated` is intentionally not synthesized: the portable protocol + // currently defines only spawn and terminal lifecycle. + return item.tool === "spawnAgent" && representedSpawn; + } + + shouldIgnore(notification: ServerNotification): boolean { + const threadId = (notification.params as {threadId?: unknown}).threadId; + const ignored = typeof threadId === "string" + && this.children.get(threadId)?.terminalState !== undefined; + if (ignored) this.log(`Ignoring update for terminal subagent ${threadId}`); + return ignored; + } + + notificationSessionId(notification: ServerNotification): string { + const threadId = (notification.params as {threadId?: unknown}).threadId; + return typeof threadId === "string" && this.children.has(threadId) + ? threadId + : this.rootSessionId; + } + + legacyActivityStarted(item: ThreadItem & {type: "subAgentActivity"}): UpdateSessionEvent { + this.activeLegacyActivities.add(item.id); + return createSubAgentActivityUpdate(item, "in_progress", "tool_call"); + } + + legacyCollaborationStarted(item: ThreadItem & {type: "collabAgentToolCall"}): UpdateSessionEvent { + return createCollabAgentToolCallUpdate(item); + } + + legacyCollaborationCompleted(item: ThreadItem & {type: "collabAgentToolCall"}): UpdateSessionEvent { + return createCollabAgentToolCallCompleteUpdate(item); + } + + legacyActivityCompleted(item: ThreadItem & {type: "subAgentActivity"}): UpdateSessionEvent { + const sessionUpdate = this.activeLegacyActivities.delete(item.id) + ? "tool_call_update" + : "tool_call"; + return createSubAgentActivityUpdate(item, "completed", sessionUpdate); + } + + async wait(signal: AbortSignal): Promise { + while ([...this.children.values()].some(child => child.terminalState === undefined)) { + if (signal.aborted) return; + await new Promise((resolve) => { + const onAbort = () => { + this.waiters.delete(onChange); + resolve(); + }; + const onChange = () => { + signal.removeEventListener("abort", onAbort); + resolve(); + }; + this.waiters.add(onChange); + signal.addEventListener("abort", onAbort, {once: true}); + }); + } + } + + async finishOutstanding(state: SubagentState): Promise { + for (const childSessionId of [...this.children.keys()].reverse()) { + await this.finish(childSessionId, state); + } + } + + private async finish(childSessionId: string, state: SubagentState): Promise { + const child = this.children.get(childSessionId); + if (!child || child.terminalState !== undefined) return; + child.terminalState = state; + await this.publish({ + sessionUpdate: "subagent_state_update", + subagentSessionId: childSessionId, + state, + }, child.parentSessionId); + for (const waiter of this.waiters) waiter(); + this.waiters.clear(); + } +} + +function terminalStateOf( + status: "pendingInit" | "running" | "completed" | "errored" | "shutdown" | "notFound" | "interrupted", +): SubagentState | undefined { + switch (status) { + case "completed": + return "completed"; + case "interrupted": + return "cancelled"; + case "errored": + case "shutdown": + case "notFound": + return "failed"; + case "pendingInit": + case "running": + return undefined; + } +} + +function fallbackName(sessionId: string): string { + const suffix = sessionId.length > 8 ? sessionId.slice(-8) : sessionId; + return `Agent ${suffix}`; +} diff --git a/src/subagents/CodexSubagentSubscriptions.ts b/src/subagents/CodexSubagentSubscriptions.ts new file mode 100644 index 00000000..44b6ef62 --- /dev/null +++ b/src/subagents/CodexSubagentSubscriptions.ts @@ -0,0 +1,113 @@ +import type { + ApprovalHandler, + CodexAppServerClient, + ElicitationHandler, +} from "../CodexAppServerClient"; +import type {ServerNotification} from "../app-server"; + +type Subscription = { + rootSessionId: string; + supportsSubagents: boolean; + dispatch(event: ServerNotification): void; + approvalHandler: ApprovalHandler; + elicitationHandler: ElicitationHandler; + waitForRootNotifications(): Promise; +}; + +/** Discovers child threads and keeps their output/interaction boundary negotiated. */ +export class CodexSubagentSubscriptions { + private readonly childrenByRoot = new Map>(); + + constructor(private readonly client: CodexAppServerClient) {} + + subscribe(subscription: Subscription): void { + const registerInteractiveHandlers = ( + targetSessionId: string, + includeElicitations = true, + ): void => { + this.client.onApprovalRequest(targetSessionId, { + handleCommandExecution: async (params) => { + await subscription.waitForRootNotifications(); + return await subscription.approvalHandler.handleCommandExecution( + this.rootPermissionParams(subscription, targetSessionId, params), + ); + }, + handleFileChange: async (params) => { + await subscription.waitForRootNotifications(); + return await subscription.approvalHandler.handleFileChange( + this.rootPermissionParams(subscription, targetSessionId, params), + ); + }, + handlePermissionsRequest: async (params) => { + await subscription.waitForRootNotifications(); + return await subscription.approvalHandler.handlePermissionsRequest( + this.rootPermissionParams(subscription, targetSessionId, params), + ); + }, + }); + if (!includeElicitations) return; + this.client.onElicitationRequest(targetSessionId, { + handleElicitation: async (params) => { + await subscription.waitForRootNotifications(); + return await subscription.elicitationHandler.handleElicitation(params); + }, + handleUserInput: async (params) => { + await subscription.waitForRootNotifications(); + return await subscription.elicitationHandler.handleUserInput(params); + }, + }); + }; + + const discover = (event: ServerNotification): void => { + if ((event.method !== "item/started" && event.method !== "item/completed") + || event.params.item.type !== "collabAgentToolCall" + || event.params.item.tool !== "spawnAgent") { + return; + } + const children = this.childrenByRoot.get(subscription.rootSessionId) ?? new Set(); + this.childrenByRoot.set(subscription.rootSessionId, children); + for (const childSessionId of event.params.item.receiverThreadIds) { + if (childSessionId.trim() === "") continue; + if (childSessionId === subscription.rootSessionId + || childSessionId === event.params.threadId + || children.has(childSessionId)) { + continue; + } + children.add(childSessionId); + this.client.onServerNotification(childSessionId, (childEvent) => { + const eventThreadId = (childEvent.params as {threadId?: unknown}).threadId; + if (eventThreadId !== childSessionId) return; + discover(childEvent); + if (subscription.supportsSubagents) subscription.dispatch(childEvent); + }); + // Hidden children keep only root-attributed permission requests. + registerInteractiveHandlers(childSessionId, subscription.supportsSubagents); + } + }; + + this.client.onServerNotification(subscription.rootSessionId, (event) => { + // Register synchronously: app-server may emit child output directly + // after the spawning collaboration item. + discover(event); + subscription.dispatch(event); + }); + registerInteractiveHandlers(subscription.rootSessionId); + } + + clear(rootSessionId: string): void { + for (const childSessionId of this.childrenByRoot.get(rootSessionId) ?? []) { + this.client.clearThreadHandlers(childSessionId); + } + this.childrenByRoot.delete(rootSessionId); + } + + private rootPermissionParams( + subscription: Subscription, + targetSessionId: string, + params: T, + ): T { + return !subscription.supportsSubagents && targetSessionId !== subscription.rootSessionId + ? {...params, threadId: subscription.rootSessionId} + : params; + } +} From 12fe5952f9d2e04a18c13403bc919a6b7812a4ee Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 20 Aug 2026 20:02:24 +0400 Subject: [PATCH 4/5] fix: harden native subagent cleanup Keep spawn and terminal lifecycle retryable when ACP delivery fails, and ensure prompt cleanup continues when the client disconnects. --- src/CodexAcpServer.ts | 14 +++++++++----- src/subagents/CodexSubagentEventRouter.ts | 6 +++--- 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index b717fd41..9d6cae91 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -2632,11 +2632,15 @@ export class CodexAcpServer { // The app-server subscription is session-scoped and outlives this prompt. Flip routing before // awaiting disposal so queued late notifications cannot enter prompt-local buffers. promptNotificationsActive = false; - await eventHandler?.finishOutstandingNativeSubagents( - promptWasCancelled || activePrompt.signal.aborted || this.sessionIsClosing(params.sessionId) - ? "cancelled" - : "failed", - ); + try { + await eventHandler?.finishOutstandingNativeSubagents( + promptWasCancelled || activePrompt.signal.aborted || this.sessionIsClosing(params.sessionId) + ? "cancelled" + : "failed", + ); + } catch (error) { + logger.error("Failed to publish terminal subagent state during prompt cleanup", error); + } if (agentFileChangeReportRequest !== null) { await this.publishAgentFileChangeReport( sessionState, diff --git a/src/subagents/CodexSubagentEventRouter.ts b/src/subagents/CodexSubagentEventRouter.ts index ca520733..b2f82abc 100644 --- a/src/subagents/CodexSubagentEventRouter.ts +++ b/src/subagents/CodexSubagentEventRouter.ts @@ -73,8 +73,6 @@ export class CodexSubagentEventRouter { name: fallbackName(childSessionId), task: item.prompt?.trim() || "Delegated task", }; - this.children.set(childSessionId, child); - representedSpawn = true; await this.publish({ sessionUpdate: "subagent_spawned", subagentSessionId: childSessionId, @@ -82,6 +80,8 @@ export class CodexSubagentEventRouter { task: child.task, capabilities: {}, }, parentSessionId); + this.children.set(childSessionId, child); + representedSpawn = true; } } @@ -156,12 +156,12 @@ export class CodexSubagentEventRouter { private async finish(childSessionId: string, state: SubagentState): Promise { const child = this.children.get(childSessionId); if (!child || child.terminalState !== undefined) return; - child.terminalState = state; await this.publish({ sessionUpdate: "subagent_state_update", subagentSessionId: childSessionId, state, }, child.parentSessionId); + child.terminalState = state; for (const waiter of this.waiters) waiter(); this.waiters.clear(); } From 8ffcf98dca9ef79ca0c0766545f2358731effcb2 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Fri, 21 Aug 2026 23:54:34 +0400 Subject: [PATCH 5/5] feat: complete native subagent compatibility Negotiate native sessions through AIR metadata while released ACP SDKs strip the draft canonical field. Normalize provider activity into child sessions, preserve nested routing, and exclude the root participant from subagent lifecycle. --- README.md | 11 +- src/AirExtension.ts | 17 ++ src/CodexAcpServer.ts | 32 +-- src/CodexEventHandler.ts | 13 +- .../CodexACPAgent/collab-agent-events.test.ts | 261 +++++++++++++++--- .../CodexACPAgent/initialize.test.ts | 2 +- .../typed-session-failure-wire.test.ts | 25 ++ src/__tests__/acp-test-utils.ts | 15 +- src/subagents/AcpSubagents.ts | 10 +- src/subagents/CodexAgentPath.ts | 15 + src/subagents/CodexSubagentEventRouter.ts | 106 +++++-- src/subagents/CodexSubagentSubscriptions.ts | 155 ++++++----- 12 files changed, 512 insertions(+), 150 deletions(-) create mode 100644 src/subagents/CodexAgentPath.ts diff --git a/README.md b/README.md index 5246590a..6e06f608 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ Use [OpenAI Codex](https://github.com/openai/codex) from [Agent Client Protocol] - Model, reasoning effort, fast mode, approval, and sandbox mode configuration. - Text prompts, embedded context, images, resource links, and additional workspace directories. - Shell command, file change, permission request, MCP tool call, terminal output, reasoning, plan, web search, image generation, image view, token usage, and review events. -- Subagent launches as standard ACP tool calls, with Codex thread identity and activity details in namespaced `_meta.codex.subagent` metadata. +- Native ACP subagent sessions with separate child histories and root-routed permissions. - Session-scoped long-running goals through the provider-neutral [goal extension](docs/goal-extension.md). - Client-provided MCP servers over command-based stdio config and HTTP transport. - Slash commands: `/status`, `/mcp`, `/skills`, `/goal`, `/review`, `/review-branch`, `/review-commit`, `/compact`, and `/logout`, as well as configured skills. @@ -75,6 +75,15 @@ npm run bundle:all See [readme-dev.md](readme-dev.md) for local client configuration, binary packaging, and Codex type regeneration. +### Subagent sessions + +Subagents are exposed only after bilateral capability negotiation. Until the released ACP SDKs +preserve the draft `clientCapabilities.subagents` field, a supporting client may advertise +`nativeSubagentSessions` in `_meta.jetbrains.air.capabilities`; the adapter mirrors the capability +in its initialize response. The canonical field remains supported and takes precedence once it is +available. Without either client signal, subagent lifecycle and child output stay hidden while +child permission requests continue to be handled on the root session. + ## License By contributing, you agree that your contributions will be licensed under the Apache 2.0 License. diff --git a/src/AirExtension.ts b/src/AirExtension.ts index cf97b512..9e39a18d 100644 --- a/src/AirExtension.ts +++ b/src/AirExtension.ts @@ -12,5 +12,22 @@ export const AIR_EXTENSION_VERSION_KEY = "version"; export const AIR_EXTENSION_CAPABILITIES_KEY = "capabilities"; export const AIR_SESSION_FAILURE_KEY = "sessionFailure"; export const AIR_AGENT_FILE_CHANGE_REPORT_KEY = "agentFileChangeReport"; +export const AIR_NATIVE_SUBAGENT_SESSIONS_KEY = "nativeSubagentSessions"; export const AIR_AGENT_FILE_CHANGE_REPORT_REQUEST_KEY = "agentFileChangeReportRequest"; export const AIR_EXTENSION_VERSION = 1; + +export function clientSupportsAirCapability( + capabilities: ClientCapabilities | null | undefined, + capability: string, +): boolean { + const jetbrains = capabilities?._meta?.[JETBRAINS_META_KEY] as Record | undefined; + const air = jetbrains?.[AIR_META_KEY] as Record | undefined; + const version = air?.[AIR_EXTENSION_VERSION_KEY]; + const supported = air?.[AIR_EXTENSION_CAPABILITIES_KEY]; + return typeof version === "number" + && Number.isInteger(version) + && version >= AIR_EXTENSION_VERSION + && Array.isArray(supported) + && supported.includes(capability); +} +import type {ClientCapabilities} from "@agentclientprotocol/sdk"; diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 9d6cae91..9b1a490b 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -98,15 +98,18 @@ import { clientSupportsSubagents, type SubagentAwareSessionCapabilities, } from "./subagents/AcpSubagents"; +import {CodexSubagentEventRouter} from "./subagents/CodexSubagentEventRouter"; import {randomUUID} from "node:crypto"; import {once} from "node:events"; import { AIR_AGENT_FILE_CHANGE_REPORT_KEY, + AIR_NATIVE_SUBAGENT_SESSIONS_KEY, AIR_EXTENSION_CAPABILITIES_KEY, AIR_EXTENSION_VERSION, AIR_EXTENSION_VERSION_KEY, AIR_META_KEY, AIR_SESSION_FAILURE_KEY, + clientSupportsAirCapability, JETBRAINS_META_KEY, } from "./AirExtension"; import { @@ -148,6 +151,7 @@ export interface SessionState { sessionTitle: string | null; sessionTitleSource: "unset" | "fallback" | "explicit" | "unknown"; sessionFailure?: SessionFailure; + subagents: CodexSubagentEventRouter; } export type SessionFailureCategory = @@ -173,21 +177,6 @@ export interface SessionFailure { const CODEX_PROCESS_EXITED_ERROR_CODE = 1001; -function clientSupportsAirCapability( - capabilities: acp.ClientCapabilities | null, - capability: string, -): boolean { - const jetbrains = capabilities?._meta?.[JETBRAINS_META_KEY] as Record | undefined; - const air = jetbrains?.[AIR_META_KEY] as Record | undefined; - const version = air?.[AIR_EXTENSION_VERSION_KEY]; - const supported = air?.[AIR_EXTENSION_CAPABILITIES_KEY]; - return typeof version === "number" - && Number.isInteger(version) - && version >= AIR_EXTENSION_VERSION - && Array.isArray(supported) - && supported.includes(capability); -} - function clientSupportsTypedSessionFailures(capabilities: acp.ClientCapabilities | null): boolean { return clientSupportsAirCapability(capabilities, AIR_SESSION_FAILURE_KEY); } @@ -362,6 +351,7 @@ export class CodexAcpServer { [AIR_EXTENSION_CAPABILITIES_KEY]: [ AIR_SESSION_FAILURE_KEY, AIR_AGENT_FILE_CHANGE_REPORT_KEY, + AIR_NATIVE_SUBAGENT_SESSIONS_KEY, ], }, }, @@ -632,6 +622,11 @@ export class CodexAcpServer { goalRevision: 0, sessionTitle: null, sessionTitleSource: "sessionId" in request ? "unknown" : "unset", + subagents: new CodexSubagentEventRouter( + sessionId, + clientSupportsSubagents(this.clientCapabilities), + new ACPSessionConnection(this.connection, sessionId), + ), }; this.sessions.set(sessionId, sessionState); resumeSubscribed = false; @@ -1630,6 +1625,11 @@ export class CodexAcpServer { goalRevision: 0, sessionTitle: null, sessionTitleSource: "unset", + subagents: new CodexSubagentEventRouter( + sessionId, + clientSupportsSubagents(this.clientCapabilities), + new ACPSessionConnection(this.connection, sessionId), + ), }; this.sessions.set(sessionId, sessionState); subscribed = false; @@ -2286,7 +2286,7 @@ export class CodexAcpServer { clientSupportsPlanUpdates(this.clientCapabilities), clientSupportsTypedSessionFailures(this.clientCapabilities), this.sessionFailureEpoch, - clientSupportsSubagents(this.clientCapabilities), + sessionState.subagents, ); eventHandler = promptEventHandler; const approvalHandler = new CodexApprovalHandler(this.connection, activePrompt.signal); diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index a9edb80a..62499587 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -231,19 +231,18 @@ export class CodexEventHandler { supportsPlanUpdates = false, supportsTypedSessionFailures = false, sessionFailureEpoch: string = randomUUID(), - supportsSubagents = false, + subagents: CodexSubagentEventRouter = new CodexSubagentEventRouter( + sessionState.sessionId, + false, + new ACPSessionConnection(connection, sessionState.sessionId), + ), ) { this.sessionState = sessionState; this.supportsPlanUpdates = supportsPlanUpdates; this.supportsTypedSessionFailures = supportsTypedSessionFailures; this.sessionFailureEpoch = sessionFailureEpoch; this.session = new ACPSessionConnection(connection, sessionState.sessionId); - this.subagents = new CodexSubagentEventRouter( - sessionState.sessionId, - supportsSubagents, - (update, sessionId) => this.session.update(update, sessionId), - (message) => logger.log(message), - ); + this.subagents = subagents; if (sessionState.sessionFailure !== undefined) { this.failuresById.set(sessionState.sessionFailure.id, sessionState.sessionFailure); } diff --git a/src/__tests__/CodexACPAgent/collab-agent-events.test.ts b/src/__tests__/CodexACPAgent/collab-agent-events.test.ts index 9cbb1049..1fe50837 100644 --- a/src/__tests__/CodexACPAgent/collab-agent-events.test.ts +++ b/src/__tests__/CodexACPAgent/collab-agent-events.test.ts @@ -1,8 +1,9 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { ServerNotification } from "../../app-server"; -import type { ClientCapabilities } from "@agentclientprotocol/sdk"; import type { SessionState } from "../../CodexAcpServer"; import { AgentMode } from "../../AgentMode"; +import {ACPSessionConnection} from "../../ACPSessionConnection"; +import {CodexSubagentEventRouter} from "../../subagents/CodexSubagentEventRouter"; import { createCodexMockTestFixture, createTestSessionState, @@ -12,18 +13,37 @@ import { describe("CodexEventHandler - collab agent tool call events", () => { let mockFixture: CodexMockTestFixture; + let sessionState: SessionState; const sessionId = "test-session-id"; beforeEach(() => { mockFixture = createCodexMockTestFixture(); + sessionState = createTestSessionState({ + sessionId, + currentModelId: "model-id[effort]", + agentMode: AgentMode.DEFAULT_AGENT_MODE, + }); vi.clearAllMocks(); }); - const sessionState: SessionState = createTestSessionState({ - sessionId, - currentModelId: "model-id[effort]", - agentMode: AgentMode.DEFAULT_AGENT_MODE, - }); + async function initializeNativeSubagents() { + const response = await mockFixture.getCodexAcpAgent().initialize({ + protocolVersion: 1, + clientCapabilities: { + _meta: { + jetbrains: { + air: {version: 1, capabilities: ["nativeSubagentSessions"]}, + }, + }, + }, + }); + sessionState.subagents = new CodexSubagentEventRouter( + sessionId, + true, + new ACPSessionConnection(mockFixture.getAcpConnection(), sessionId), + ); + return response; + } it("hides collaboration lifecycle when the client lacks subagent capability but keeps permissions", async () => { const notifications: ServerNotification[] = [ @@ -140,14 +160,199 @@ describe("CodexEventHandler - collab agent tool call events", () => { expect(JSON.stringify(mockFixture.getAcpConnectionEvents([]))).not.toContain("call-spawn-weather"); }); + it("promotes subagent activity to native lifecycle when collaboration items are absent", async () => { + await initializeNativeSubagents(); + await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, [ + { + method: "item/started", + params: { + threadId: sessionId, + turnId: "turn-1", + startedAtMs: 0, + item: { + type: "subAgentActivity", + id: "activity-started", + kind: "started", + agentThreadId: "child-1", + agentPath: "/root/air_architecture", + }, + }, + }, + { + method: "item/completed", + params: { + threadId: sessionId, + turnId: "turn-1", + completedAtMs: 0, + item: { + type: "subAgentActivity", + id: "activity-started", + kind: "started", + agentThreadId: "child-1", + agentPath: "/root/air_architecture", + }, + }, + }, + { + method: "item/started", + params: { + threadId: "child-1", + turnId: "turn-child", + startedAtMs: 0, + item: { + type: "subAgentActivity", + id: "nested-started", + kind: "started", + agentThreadId: "grandchild-1", + agentPath: "/root/air_architecture/tests", + }, + }, + }, + { + method: "item/agentMessage/delta", + params: { + threadId: "grandchild-1", + turnId: "turn-grandchild", + itemId: "grandchild-message", + delta: "Nested result", + }, + }, + { + method: "item/completed", + params: { + threadId: "child-1", + turnId: "turn-child", + completedAtMs: 0, + item: { + type: "subAgentActivity", + id: "nested-interrupted", + kind: "interrupted", + agentThreadId: "grandchild-1", + agentPath: "/root/air_architecture/tests", + }, + }, + }, + { + method: "turn/completed", + params: { + threadId: sessionId, + turn: { + id: "turn-1", + items: [], + itemsView: "notLoaded", + status: "completed", + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }, + }, + }, + ]); + + const updates = mockFixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0]); + expect(updates).toEqual([ + { + sessionId, + update: { + sessionUpdate: "subagent_spawned", + subagentSessionId: "child-1", + name: "air_architecture", + task: "Delegated task for air_architecture", + capabilities: {}, + }, + }, + { + sessionId: "child-1", + update: { + sessionUpdate: "subagent_spawned", + subagentSessionId: "grandchild-1", + name: "tests", + task: "Delegated task for tests", + capabilities: {}, + }, + }, + { + sessionId: "grandchild-1", + update: { + sessionUpdate: "agent_message_chunk", + content: {type: "text", text: "Nested result"}, + messageId: "grandchild-message", + }, + }, + { + sessionId: "child-1", + update: { + sessionUpdate: "subagent_state_update", + subagentSessionId: "grandchild-1", + state: "cancelled", + }, + }, + { + sessionId, + update: { + sessionUpdate: "subagent_state_update", + subagentSessionId: "child-1", + state: "completed", + }, + }, + ]); + }); + + it("does not represent the root activity as a subagent", async () => { + await initializeNativeSubagents(); + await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, [ + { + method: "item/started", + params: { + threadId: sessionId, + turnId: "turn-1", + startedAtMs: 0, + item: { + type: "subAgentActivity", + id: "root-activity", + kind: "started", + agentThreadId: "root-activity-thread", + agentPath: "/root", + }, + }, + }, + { + method: "item/completed", + params: { + threadId: sessionId, + turnId: "turn-1", + completedAtMs: 0, + item: { + type: "subAgentActivity", + id: "root-activity", + kind: "started", + agentThreadId: "root-activity-thread", + agentPath: "/root/", + }, + }, + }, + { + method: "item/agentMessage/delta", + params: { + threadId: sessionId, + turnId: "turn-1", + itemId: "parent-message", + delta: "Visible root output", + }, + }, + ]); + + expect(mockFixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0].update)) + .not.toContainEqual(expect.objectContaining({sessionUpdate: "subagent_spawned"})); + }); + it("emits native lifecycle and routes child output after capability negotiation", async () => { - const clientCapabilities = {subagents: {}} as ClientCapabilities & { - subagents: Record; - }; - const initializeResponse = await mockFixture.getCodexAcpAgent().initialize({ - protocolVersion: 1, - clientCapabilities, - }); + const initializeResponse = await initializeNativeSubagents(); expect( (initializeResponse.agentCapabilities?.sessionCapabilities as {subagents?: unknown}).subagents ).toEqual({}); @@ -257,10 +462,7 @@ describe("CodexEventHandler - collab agent tool call events", () => { }); it("routes nested agents through their immediate parent sessions", async () => { - await mockFixture.getCodexAcpAgent().initialize({ - protocolVersion: 1, - clientCapabilities: {subagents: {}} as ClientCapabilities, - }); + await initializeNativeSubagents(); const collabItem = ( threadId: string, senderThreadId: string, @@ -316,10 +518,7 @@ describe("CodexEventHandler - collab agent tool call events", () => { }); it("deduplicates lifecycle, rejects blank IDs, and ignores late child output", async () => { - await mockFixture.getCodexAcpAgent().initialize({ - protocolVersion: 1, - clientCapabilities: {subagents: {}} as ClientCapabilities, - }); + await initializeNativeSubagents(); const spawn = (method: "item/started" | "item/completed"): ServerNotification => ({ method, params: { @@ -366,10 +565,7 @@ describe("CodexEventHandler - collab agent tool call events", () => { }); it("keeps unsupported collaboration controls visible in native mode", async () => { - await mockFixture.getCodexAcpAgent().initialize({ - protocolVersion: 1, - clientCapabilities: {subagents: {}} as ClientCapabilities, - }); + await initializeNativeSubagents(); const collab = ( method: "item/started" | "item/completed", tool: "spawnAgent" | "sendInput", @@ -415,10 +611,7 @@ describe("CodexEventHandler - collab agent tool call events", () => { }); it("falls back to tool representation when a native spawn cannot be represented", async () => { - await mockFixture.getCodexAcpAgent().initialize({ - protocolVersion: 1, - clientCapabilities: {subagents: {}} as ClientCapabilities, - }); + await initializeNativeSubagents(); await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, [{ method: "item/completed", params: { @@ -453,10 +646,7 @@ describe("CodexEventHandler - collab agent tool call events", () => { }); it("does not duplicate global notifications after subscribing to a child", async () => { - await mockFixture.getCodexAcpAgent().initialize({ - protocolVersion: 1, - clientCapabilities: {subagents: {}} as ClientCapabilities, - }); + await initializeNativeSubagents(); await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, [ { method: "item/started", @@ -510,10 +700,7 @@ describe("CodexEventHandler - collab agent tool call events", () => { }); it("keeps the parent prompt open until every announced child is terminal", async () => { - await mockFixture.getCodexAcpAgent().initialize({ - protocolVersion: 1, - clientCapabilities: {subagents: {}} as ClientCapabilities, - }); + await initializeNativeSubagents(); const appServer = mockFixture.getCodexAppServerClient(); const turn = {id: "turn-1", items: [], status: "inProgress" as const, error: null}; const completedTurn = {...turn, status: "completed" as const}; diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index 739c4028..57887361 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -73,7 +73,7 @@ describe('CodexACPAgent - initialize', () => { jetbrains: { air: { version: 1, - capabilities: ["sessionFailure", "agentFileChangeReport"], + capabilities: ["sessionFailure", "agentFileChangeReport", "nativeSubagentSessions"], }, }, }, diff --git a/src/__tests__/CodexACPAgent/typed-session-failure-wire.test.ts b/src/__tests__/CodexACPAgent/typed-session-failure-wire.test.ts index 790c2614..115ed58d 100644 --- a/src/__tests__/CodexACPAgent/typed-session-failure-wire.test.ts +++ b/src/__tests__/CodexACPAgent/typed-session-failure-wire.test.ts @@ -11,6 +11,31 @@ const typedFailureCapabilities: acp.ClientCapabilities = { }; describe("typed session failures over ACP transport", () => { + it("negotiates native subagents through AIR metadata across the SDK boundary", async () => { + const fixture = createWireFixture(); + const response = await fixture.client.initialize({ + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: { + _meta: { + jetbrains: { + air: {version: 1, capabilities: ["nativeSubagentSessions"]}, + }, + }, + }, + }); + + expect((response.agentCapabilities!.sessionCapabilities as {subagents?: unknown}).subagents) + .toEqual({}); + expect(response._meta).toMatchObject({ + jetbrains: { + air: { + version: 1, + capabilities: expect.arrayContaining(["nativeSubagentSessions"]), + }, + }, + }); + }); + it("returns a sanitized process-exit failure in the decoded prompt response", async () => { const fixture = createWireFixture({ exitCode: 1, diff --git a/src/__tests__/acp-test-utils.ts b/src/__tests__/acp-test-utils.ts index 6b9e20c0..5cf73d5d 100644 --- a/src/__tests__/acp-test-utils.ts +++ b/src/__tests__/acp-test-utils.ts @@ -4,7 +4,7 @@ import {CodexAcpClient} from '../CodexAcpClient'; import {CodexAppServerClient, type CodexConnectionEvent} from '../CodexAppServerClient'; import {startCodexConnection} from "../CodexJsonRpcConnection"; import {CodexAcpServer, type SessionState} from "../CodexAcpServer"; -import type {AcpClientConnection} from "../ACPSessionConnection"; +import {ACPSessionConnection, type AcpClientConnection} from "../ACPSessionConnection"; import type {ServerNotification} from "../app-server"; import type {MessageConnection} from "vscode-jsonrpc/node"; import path from "node:path"; @@ -14,6 +14,7 @@ import {AgentMode} from "../AgentMode"; import {DEFAULT_COLLABORATION_MODE} from "../CollaborationModeConfig"; import {expect, vi} from "vitest"; import type {Model, ReasoningEffortOption} from "../app-server/v2"; +import {CodexSubagentEventRouter} from "../subagents/CodexSubagentEventRouter"; export type MethodCallEvent = { method: string; args: any[] }; @@ -69,6 +70,7 @@ export interface TestFixture { getAcpConnectionEvents(ignoredFields: string[]): MethodCallEvent[], getAcpConnectionDump(ignoredFields: string[]): string, clearAcpConnectionDump(): void, + getAcpConnection(): AcpClientConnection, } export interface CodexConnectionDumpOptions { @@ -167,6 +169,9 @@ export function createBaseTestFixture(config: ConnectionConfig): TestFixture { }, clearAcpConnectionDump() { acpConnectionEvents.splice(0, acpConnectionEvents.length); + }, + getAcpConnection(): AcpClientConnection { + return acpConnection; } }; } @@ -379,6 +384,7 @@ function anonymizeValue(value: any, path: string[], fieldsToAnonymize: Set): SessionState { + const sessionId = overrides?.sessionId ?? "session-id"; return { currentTurnId: null, lastTokenUsage: null, @@ -390,7 +396,7 @@ export function createTestSessionState(overrides?: Partial): Sessi authProvider: null, cwd: "/test/cwd", additionalDirectories: [], - sessionId: "session-id", + sessionId, currentModelId: "model-id[effort]", availableModels: [], supportedReasoningEfforts: [], @@ -403,6 +409,11 @@ export function createTestSessionState(overrides?: Partial): Sessi goalRevision: 0, sessionTitle: null, sessionTitleSource: "unknown", + subagents: new CodexSubagentEventRouter( + sessionId, + false, + new ACPSessionConnection({notify: vi.fn(), request: vi.fn()} as AcpClientConnection, sessionId), + ), ...overrides, }; } diff --git a/src/subagents/AcpSubagents.ts b/src/subagents/AcpSubagents.ts index 486e53f7..6803c40d 100644 --- a/src/subagents/AcpSubagents.ts +++ b/src/subagents/AcpSubagents.ts @@ -3,6 +3,10 @@ import type { SessionCapabilities, SessionNotification, } from "@agentclientprotocol/sdk"; +import { + AIR_NATIVE_SUBAGENT_SESSIONS_KEY, + clientSupportsAirCapability, +} from "../AirExtension"; /** Temporary typed surface for agentclientprotocol/agent-client-protocol#1992. */ export type SubagentSessionCapabilities = { @@ -48,7 +52,11 @@ export function clientSupportsSubagents( const subagents = ( capabilities as (ClientCapabilities & { subagents?: unknown }) | null | undefined )?.subagents; - return typeof subagents === "object" && subagents !== null && !Array.isArray(subagents); + if (typeof subagents === "object" && subagents !== null && !Array.isArray(subagents)) { + return true; + } + + return clientSupportsAirCapability(capabilities, AIR_NATIVE_SUBAGENT_SESSIONS_KEY); } /** The only cast needed until the TypeScript SDK publishes PR #1992. */ diff --git a/src/subagents/CodexAgentPath.ts b/src/subagents/CodexAgentPath.ts new file mode 100644 index 00000000..be08e8e5 --- /dev/null +++ b/src/subagents/CodexAgentPath.ts @@ -0,0 +1,15 @@ +export function normalizeAgentPath(path: string): string { + const normalized = path.trim().replace(/\/+$/, ""); + return normalized || "/root"; +} + +export function isRootAgentPath(path: string): boolean { + const normalized = normalizeAgentPath(path); + return normalized === "/root" || normalized === "root"; +} + +export function nameFromAgentPath(path: string, fallback: string): string { + const normalized = normalizeAgentPath(path); + const name = normalized.slice(normalized.lastIndexOf("/") + 1).trim(); + return name || fallback; +} diff --git a/src/subagents/CodexSubagentEventRouter.ts b/src/subagents/CodexSubagentEventRouter.ts index b2f82abc..4bd844e7 100644 --- a/src/subagents/CodexSubagentEventRouter.ts +++ b/src/subagents/CodexSubagentEventRouter.ts @@ -1,25 +1,27 @@ import type {ServerNotification} from "../app-server"; import type {ThreadItem} from "../app-server/v2"; -import type {UpdateSessionEvent} from "../ACPSessionConnection"; +import {ACPSessionConnection, type UpdateSessionEvent} from "../ACPSessionConnection"; +import {logger} from "../Logger"; import { createCollabAgentToolCallCompleteUpdate, createCollabAgentToolCallUpdate, createSubAgentActivityUpdate, } from "../CodexToolCallMapper"; import type {SubagentState} from "./AcpSubagents"; - -type Publisher = (update: UpdateSessionEvent, sessionId?: string) => Promise; -type Log = (message: string) => void; +import {isRootAgentPath, nameFromAgentPath, normalizeAgentPath} from "./CodexAgentPath"; type NativeSubagent = { parentSessionId: string; name: string; task: string; + path?: string; terminalState?: SubagentState; }; /** Owns native lifecycle, child routing, waiting, and legacy activity deduplication. */ export class CodexSubagentEventRouter { + private static readonly DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1000; + private readonly children = new Map(); private readonly waiters = new Set<() => void>(); private readonly activeLegacyActivities = new Set(); @@ -27,11 +29,15 @@ export class CodexSubagentEventRouter { constructor( private readonly rootSessionId: string, private readonly supported: boolean, - private readonly publish: Publisher, - private readonly log: Log, + private readonly session: ACPSessionConnection, ) {} async handle(notification: ServerNotification): Promise { + if (notification.method === "turn/completed") { + const state = terminalStateFromTurn(notification.params.turn.status); + if (state) await this.finishOutstanding(state); + return false; + } if (notification.method !== "item/started" && notification.method !== "item/completed") { return false; } @@ -42,7 +48,28 @@ export class CodexSubagentEventRouter { return item.type === "collabAgentToolCall" || item.type === "subAgentActivity"; } if (item.type === "subAgentActivity") { - const hasNativeRepresentation = this.children.has(item.agentThreadId); + // Codex reports the root participant through the same activity item + // shape as children. It is the parent conversation, not a subagent. + if (isRootAgentPath(item.agentPath)) return true; + let hasNativeRepresentation = this.children.has(item.agentThreadId); + if (!hasNativeRepresentation && item.kind !== "interrupted") { + const name = nameFromAgentPath(item.agentPath, fallbackName(item.agentThreadId)); + const parentSessionId = this.parentSessionIdForPath(item.agentPath); + await this.session.update({ + sessionUpdate: "subagent_spawned", + subagentSessionId: item.agentThreadId, + name, + task: `Delegated task for ${name}`, + capabilities: {}, + }, parentSessionId); + this.children.set(item.agentThreadId, { + parentSessionId, + name, + task: `Delegated task for ${name}`, + path: normalizeAgentPath(item.agentPath), + }); + hasNativeRepresentation = true; + } if (hasNativeRepresentation && item.kind === "interrupted") { await this.finish(item.agentThreadId, "cancelled"); } @@ -57,11 +84,11 @@ export class CodexSubagentEventRouter { : this.rootSessionId; for (const childSessionId of item.receiverThreadIds) { if (childSessionId.trim().length === 0) { - this.log("Ignoring spawned subagent with an empty thread id"); + logger.log("Ignoring spawned subagent with an empty thread id"); continue; } if (childSessionId === parentSessionId || childSessionId === this.rootSessionId) { - this.log(`Ignoring self-referential spawned subagent ${childSessionId}`); + logger.log(`Ignoring self-referential spawned subagent ${childSessionId}`); continue; } if (this.children.has(childSessionId)) { @@ -73,7 +100,7 @@ export class CodexSubagentEventRouter { name: fallbackName(childSessionId), task: item.prompt?.trim() || "Delegated task", }; - await this.publish({ + await this.session.update({ sessionUpdate: "subagent_spawned", subagentSessionId: childSessionId, name: child.name, @@ -98,7 +125,7 @@ export class CodexSubagentEventRouter { const threadId = (notification.params as {threadId?: unknown}).threadId; const ignored = typeof threadId === "string" && this.children.get(threadId)?.terminalState !== undefined; - if (ignored) this.log(`Ignoring update for terminal subagent ${threadId}`); + if (ignored) logger.log(`Ignoring update for terminal subagent ${threadId}`); return ignored; } @@ -129,21 +156,43 @@ export class CodexSubagentEventRouter { return createSubAgentActivityUpdate(item, "completed", sessionUpdate); } - async wait(signal: AbortSignal): Promise { + async wait( + signal: AbortSignal, + timeoutMs = CodexSubagentEventRouter.DEFAULT_WAIT_TIMEOUT_MS, + ): Promise { + const deadline = Date.now() + timeoutMs; while ([...this.children.values()].some(child => child.terminalState === undefined)) { if (signal.aborted) return; - await new Promise((resolve) => { + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0) { + logger.log(`Timed out waiting for subagents in session ${this.rootSessionId}; marking them failed`); + await this.finishOutstanding("failed"); + return; + } + const changed = await new Promise((resolve) => { + const timeout = setTimeout(() => { + this.waiters.delete(onChange); + signal.removeEventListener("abort", onAbort); + resolve(false); + }, remainingMs); const onAbort = () => { + clearTimeout(timeout); this.waiters.delete(onChange); - resolve(); + resolve(true); }; const onChange = () => { + clearTimeout(timeout); signal.removeEventListener("abort", onAbort); - resolve(); + resolve(true); }; this.waiters.add(onChange); signal.addEventListener("abort", onAbort, {once: true}); }); + if (!changed) { + logger.log(`Timed out waiting for subagents in session ${this.rootSessionId}; marking them failed`); + await this.finishOutstanding("failed"); + return; + } } } @@ -156,7 +205,7 @@ export class CodexSubagentEventRouter { private async finish(childSessionId: string, state: SubagentState): Promise { const child = this.children.get(childSessionId); if (!child || child.terminalState !== undefined) return; - await this.publish({ + await this.session.update({ sessionUpdate: "subagent_state_update", subagentSessionId: childSessionId, state, @@ -165,6 +214,16 @@ export class CodexSubagentEventRouter { for (const waiter of this.waiters) waiter(); this.waiters.clear(); } + + private parentSessionIdForPath(path: string): string { + const normalized = normalizeAgentPath(path); + const separator = normalized.lastIndexOf("/"); + if (separator <= 0) return this.rootSessionId; + const parentPath = normalized.slice(0, separator); + return [...this.children.entries()] + .find(([, child]) => child.path === parentPath)?.[0] + ?? this.rootSessionId; + } } function terminalStateOf( @@ -185,6 +244,21 @@ function terminalStateOf( } } +function terminalStateFromTurn( + status: "inProgress" | "completed" | "interrupted" | "failed", +): SubagentState | undefined { + switch (status) { + case "completed": + return "completed"; + case "interrupted": + return "cancelled"; + case "failed": + return "failed"; + case "inProgress": + return undefined; + } +} + function fallbackName(sessionId: string): string { const suffix = sessionId.length > 8 ? sessionId.slice(-8) : sessionId; return `Agent ${suffix}`; diff --git a/src/subagents/CodexSubagentSubscriptions.ts b/src/subagents/CodexSubagentSubscriptions.ts index 44b6ef62..e9d5a5c8 100644 --- a/src/subagents/CodexSubagentSubscriptions.ts +++ b/src/subagents/CodexSubagentSubscriptions.ts @@ -4,6 +4,7 @@ import type { ElicitationHandler, } from "../CodexAppServerClient"; import type {ServerNotification} from "../app-server"; +import {isRootAgentPath} from "./CodexAgentPath"; type Subscription = { rootSessionId: string; @@ -14,91 +15,107 @@ type Subscription = { waitForRootNotifications(): Promise; }; +type SessionSubscription = { + current: Subscription; + children: Set; +}; + /** Discovers child threads and keeps their output/interaction boundary negotiated. */ export class CodexSubagentSubscriptions { - private readonly childrenByRoot = new Map>(); + private readonly sessions = new Map(); constructor(private readonly client: CodexAppServerClient) {} subscribe(subscription: Subscription): void { - const registerInteractiveHandlers = ( - targetSessionId: string, - includeElicitations = true, - ): void => { - this.client.onApprovalRequest(targetSessionId, { - handleCommandExecution: async (params) => { - await subscription.waitForRootNotifications(); - return await subscription.approvalHandler.handleCommandExecution( - this.rootPermissionParams(subscription, targetSessionId, params), - ); - }, - handleFileChange: async (params) => { - await subscription.waitForRootNotifications(); - return await subscription.approvalHandler.handleFileChange( - this.rootPermissionParams(subscription, targetSessionId, params), - ); - }, - handlePermissionsRequest: async (params) => { - await subscription.waitForRootNotifications(); - return await subscription.approvalHandler.handlePermissionsRequest( - this.rootPermissionParams(subscription, targetSessionId, params), - ); - }, - }); - if (!includeElicitations) return; - this.client.onElicitationRequest(targetSessionId, { - handleElicitation: async (params) => { - await subscription.waitForRootNotifications(); - return await subscription.elicitationHandler.handleElicitation(params); - }, - handleUserInput: async (params) => { - await subscription.waitForRootNotifications(); - return await subscription.elicitationHandler.handleUserInput(params); - }, - }); - }; - - const discover = (event: ServerNotification): void => { - if ((event.method !== "item/started" && event.method !== "item/completed") - || event.params.item.type !== "collabAgentToolCall" - || event.params.item.tool !== "spawnAgent") { - return; - } - const children = this.childrenByRoot.get(subscription.rootSessionId) ?? new Set(); - this.childrenByRoot.set(subscription.rootSessionId, children); - for (const childSessionId of event.params.item.receiverThreadIds) { - if (childSessionId.trim() === "") continue; - if (childSessionId === subscription.rootSessionId - || childSessionId === event.params.threadId - || children.has(childSessionId)) { - continue; - } - children.add(childSessionId); - this.client.onServerNotification(childSessionId, (childEvent) => { - const eventThreadId = (childEvent.params as {threadId?: unknown}).threadId; - if (eventThreadId !== childSessionId) return; - discover(childEvent); - if (subscription.supportsSubagents) subscription.dispatch(childEvent); - }); - // Hidden children keep only root-attributed permission requests. - registerInteractiveHandlers(childSessionId, subscription.supportsSubagents); - } - }; + const existing = this.sessions.get(subscription.rootSessionId); + if (existing) { + existing.current = subscription; + return; + } + const session = {current: subscription, children: new Set()}; + this.sessions.set(subscription.rootSessionId, session); this.client.onServerNotification(subscription.rootSessionId, (event) => { // Register synchronously: app-server may emit child output directly // after the spawning collaboration item. - discover(event); - subscription.dispatch(event); + this.discover(session, event); + session.current.dispatch(event); }); - registerInteractiveHandlers(subscription.rootSessionId); + this.registerInteractiveHandlers(session, subscription.rootSessionId); } clear(rootSessionId: string): void { - for (const childSessionId of this.childrenByRoot.get(rootSessionId) ?? []) { + for (const childSessionId of this.sessions.get(rootSessionId)?.children ?? []) { this.client.clearThreadHandlers(childSessionId); } - this.childrenByRoot.delete(rootSessionId); + this.sessions.delete(rootSessionId); + } + + private discover(session: SessionSubscription, event: ServerNotification): void { + if (event.method !== "item/started" && event.method !== "item/completed") { + return; + } + const item = event.params.item; + const childSessionIds = item.type === "collabAgentToolCall" && item.tool === "spawnAgent" + ? item.receiverThreadIds + : item.type === "subAgentActivity" && item.kind !== "interrupted" && !isRootAgentPath(item.agentPath) + ? [item.agentThreadId] + : []; + for (const childSessionId of childSessionIds) { + if (childSessionId.trim() === "") continue; + if (childSessionId === session.current.rootSessionId + || childSessionId === event.params.threadId + || session.children.has(childSessionId)) { + continue; + } + session.children.add(childSessionId); + this.client.onServerNotification(childSessionId, (childEvent) => { + const eventThreadId = (childEvent.params as {threadId?: unknown}).threadId; + if (eventThreadId !== childSessionId) return; + this.discover(session, childEvent); + if (session.current.supportsSubagents) session.current.dispatch(childEvent); + }); + // Hidden children keep only root-attributed permission requests. + this.registerInteractiveHandlers(session, childSessionId); + } + } + + private registerInteractiveHandlers(session: SessionSubscription, targetSessionId: string): void { + this.client.onApprovalRequest(targetSessionId, { + handleCommandExecution: async (params) => { + const current = session.current; + await current.waitForRootNotifications(); + return await current.approvalHandler.handleCommandExecution( + this.rootPermissionParams(current, targetSessionId, params), + ); + }, + handleFileChange: async (params) => { + const current = session.current; + await current.waitForRootNotifications(); + return await current.approvalHandler.handleFileChange( + this.rootPermissionParams(current, targetSessionId, params), + ); + }, + handlePermissionsRequest: async (params) => { + const current = session.current; + await current.waitForRootNotifications(); + return await current.approvalHandler.handlePermissionsRequest( + this.rootPermissionParams(current, targetSessionId, params), + ); + }, + }); + this.client.onElicitationRequest(targetSessionId, { + handleElicitation: async (params) => { + const current = session.current; + await current.waitForRootNotifications(); + return await current.elicitationHandler.handleElicitation(params); + }, + handleUserInput: async (params) => { + const current = session.current; + await current.waitForRootNotifications(); + return await current.elicitationHandler.handleUserInput(params); + }, + }); } private rootPermissionParams(