Skip to content

feat: support standard ACP subagent sessions #655

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
benbrandt wants to merge 5 commits into
base: main
Choose a base branch
from
Open

feat: support standard ACP subagent sessions #655

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5f318be
feat: support standard ACP subagent sessions
benbrandt Oct 20, 2026
8119d4a
fix: preserve subagent interactions and replay ordering
benbrandt Oct 21, 2026
5030668
fix: retain child state across restoration and parent turns
benbrandt Oct 21, 2026
37a7429
test: avoid host sandbox dependencies in terminal lifecycle fixtures
benbrandt Oct 21, 2026
d6da913
fix: keep mock server diagnostics out of HTTP responses
benbrandt Oct 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 42 additions & 1 deletion readme-dev.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,6 +183,47 @@ turn. Item completion still reports compaction progress, but does not release th
The v1 prompt response and v2 idle state wait for the turn to finish; v2's early prompt-acceptance
response is unchanged.

#### Standard subagent sessions

Non-AIR v1 clients opt in with `clientCapabilities.subagents: {}`. On v2, standard subagent
sessions need no feature marker. AIR retains its existing private opt-in, fallback, updates, and
generation-based identities; a connection receives one contract, not both. The v1 marker used by
older AIR clients also keeps that existing behavior.

Standard `subagent_update` associations appear in the immediate parent's session before any child
traffic. A child keeps its opaque ACP session ID across delegations, idle work, and replay.
Nested children are not flattened under the root; traffic without sufficient parentage is buffered
until the ancestor chain is known. Forked histories get a separate child identity namespace.

Work state is independent of conversation lifetime. An observed work ending reports `idle` with
its stop reason, not a permanently completed child. Lost observability reports `unknown`, not a
fabricated failure. On v2, the same state also appears on the child's own `state_update` stream.
Child work, plans, permissions, and elicitation can outlive the parent's foreground prompt.
Child item-rendering state is conversation-owned too, so a new parent prompt does not reset
an active child's terminal or streamed item tracking.
Child Markdown plans use completed snapshots rather than interim draft deltas.
Parent cancellation interrupts observed descendant work and withdraws child interactions; late
answers cannot restart finished work. Closing or replacing the runtime invalidates those scopes.

Children initially advertise empty capabilities. Client-initiated prompts, load/resume/fork,
cancel, close/delete, configuration changes, and private session mutations are not enabled.
Responding to child-scoped Agent-to-Client requests is still allowed. Replayed associations and
transcripts do not grant controls or confirm current activity; overlapping completed items are
deduplicated and fresh work snapshots follow the replay response. A copied history alone never
authorizes interrupting the original child runtime.
Restored child routes require provider-confirmed parentage through the verified ancestor chain.
Only already-loaded runtimes are rejoined; inactive owned children are not loaded just for replay,
but their routes can receive later live work. Foreign or unverified fork descendants stay history-only.
Child history uses the same traversal for both Codex spawn formats and reconciles completed
items with live tool tracking before suppressing overlapping notifications.
Load/resume attachment operations for the same standard parent are serialized. Fresh child
traffic follows the actual response write, including when the response is part of a batch;
overlapping attachments publish after the latest setup's response rather than interleaving
with its replay. Close and cancel remain independent, and a close invalidates queued setups.

Dedicated `session_message` reporting and optional child controls remain follow-ups. Ordinary
runtime tool reports remain available for operations such as waiting and follow-up tasks.

#### AIR v2 client contract

These are behaviors the AIR (JetBrains) ACP client relies on that aren't obvious from the wire
Expand All @@ -203,7 +244,7 @@ schema alone:
channel. The same `_meta` v1 attaches to `PromptResponse` is copied onto the v2 idle update.
- Custom session updates are renamed on v2: `subagent_spawned`/`subagent_state_update` become a
single `_subagent_update` (gated on the AIR `nativeSubagentSessions` capability; this contract is
retained until the adapter migrates to the standard subagent model), and
retained independently of standard subagent sessions), and
`async_task_spawned`/`async_task_state_update` become `_async_task_spawned`/`_async_task_state_update`
(gated on the AIR `asyncTasks` capability). Payload shapes are unchanged.
- Foreground state is independent of prompt acceptance. `running` can precede the user-message
Expand Down
13 changes: 13 additions & 0 deletions src/ACPSessionConnection.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -248,6 +248,19 @@ export class ACPSessionConnection {
await this.v2Connection.updateState(this.sessionId, state);
}

/** Standard children carry one state snapshot on their own v2 stream and their parent's roster. */
async updateSubagentState(childSessionId: string, parentSessionId: string, state: acpV2.StateUpdate): Promise<void> {
if (this.v2Connection) {
await this.v2Connection.client.notify(acpV2.methods.client.session.update, {
sessionId: childSessionId, update: {sessionUpdate: "state_update", ...state},
});
}
await this.update({sessionUpdate: "subagent_update", sessionId: childSessionId, state}, parentSessionId);
}

forSession(sessionId: string): ACPSessionConnection {
return new ACPSessionConnection(this.connection, sessionId);
}
/** `startReplayMessage`, which only exists on v2 and so needs no v1 shape. */
async startReplayMessage(kind: ReplayMessageKind, messageId: string): Promise<void> {
if (!this.v2Connection) {
Expand Down
115 changes: 87 additions & 28 deletions src/AcpAgentRouter.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import {z} from "zod";
import type {CodexAcpServer} from "./CodexAcpServer";
import {type AcpClientConnection, AcpV2Connection} from "./ACPSessionConnection";
import packageJson from "../package.json";
import {AcpResponseDelivery} from "./AcpResponseDelivery";
import {
GOAL_CONTROL_METHOD, LEGACY_SET_SESSION_MODEL_METHOD,
SESSION_STEERING_METHOD,
Expand Down Expand Up @@ -65,17 +66,75 @@ const hooksTrustParamsParser = z.object({
*/
export function createAcpAgentRouter(
createAgent: (connection: AcpClientConnection | AcpV2Connection) => CodexAcpServer,
): Pick<acpV2.AgentProtocolRouter, "connect"> {
return {
connect(stream, options) {
const responses = new AcpResponseDelivery();
const connection = createRouter(createAgent, responses).connect(responses.observe(stream), options);
void connection.closed.then(() => responses.dispose());
return connection;
},
};
}

function createRouter(
createAgent: (connection: AcpClientConnection | AcpV2Connection) => CodexAcpServer,
responses: AcpResponseDelivery,
): acpV2.AgentProtocolRouter {
let codexAcpServer: CodexAcpServer | null = null;
const getAgent = (): CodexAcpServer => {
const getAgent = (params?: {sessionId?: unknown}): CodexAcpServer => {
if (!codexAcpServer) {
throw acp.RequestError.internalError("ACP agent is not connected");
}
if (typeof params?.sessionId === "string") codexAcpServer.assertSessionMutationAllowed(params.sessionId);
return codexAcpServer;
};
type SetupState = {tail: Promise<void>; publication: Promise<void>; epoch: number; pending: number};
const setups = new Map<string, SetupState>();
const setup = <T>(sessionId: string, requestId: acpV2.JsonRpcId, operation: () => Promise<T>, signal: AbortSignal): Promise<T> => {
const agent = getAgent({sessionId});
if (!agent.usesStandardSubagents()) return operation();
const assertCurrent = agent.captureSessionSetup(sessionId);
const state = setups.get(sessionId) ?? {tail: Promise.resolve(), publication: Promise.resolve(), epoch: 0, pending: 0};
setups.set(sessionId, state);
state.pending++;
const finish = () => {
if (--state.pending === 0 && setups.get(sessionId) === state) setups.delete(sessionId);
};
let resolve!: (response: T) => void;
let reject!: (error: unknown) => void;
const response = new Promise<T>((done, failed) => { resolve = done; reject = failed; });
const task = state.tail.catch(() => {}).then(async () => {
let registered = false;
try {
// A publication may start while this task is yielding to an earlier resolved promise.
for (;;) {
const publication = state.publication;
await publication;
if (publication === state.publication) break;
}
if (signal.aborted) throw new acp.RequestError(-32800, "Session setup was cancelled");
assertCurrent();
const epoch = ++state.epoch;
responses.afterWrite(requestId, () => {
if (epoch === state.epoch) state.publication = agent.afterSessionSetupResponse(sessionId);
void state.publication.then(finish, finish);
});
registered = true;
resolve(await operation());
} catch (error) {
reject(error);
if (!registered) finish();
}
});
// Do not wait for delivery here: two requests in one batch share a response write.
state.tail = task;
return response;
};
const attachAgent = (agent: CodexAcpServer, signal: AbortSignal) => {
codexAcpServer = agent;
signal.addEventListener("abort", () => {
setups.clear();
agent.sessionIndex.dispose();
if (codexAcpServer === agent) {
codexAcpServer = null;
Expand All @@ -87,32 +146,32 @@ export function createAcpAgentRouter(
.onConnect((connection) => attachAgent(createAgent(connection.client), connection.signal))
.onRequest(acp.methods.agent.initialize, (ctx) => getAgent().initialize(ctx.params))
.onRequest(acp.methods.agent.session.new, (ctx) => getAgent().newSession(ctx.params))
.onRequest(acp.methods.agent.session.load, (ctx) => getAgent().loadSession(ctx.params))
.onRequest(acp.methods.agent.session.fork, (ctx) => getAgent().forkSession(ctx.params))
.onRequest(acp.methods.agent.session.load, (ctx) => setup(ctx.params.sessionId, ctx.requestId, () => getAgent(ctx.params).loadSession(ctx.params), ctx.signal))
.onRequest(acp.methods.agent.session.fork, (ctx) => getAgent(ctx.params).forkSession(ctx.params))
.onRequest(acp.methods.agent.session.list, (ctx) => getAgent().listSessions(ctx.params))
.onRequest(acp.methods.agent.session.delete, (ctx) => getAgent().deleteSession(ctx.params))
.onRequest(acp.methods.agent.session.resume, (ctx) => getAgent().resumeSession(ctx.params))
.onRequest(acp.methods.agent.session.close, (ctx) => getAgent().closeSession(ctx.params))
.onRequest(acp.methods.agent.session.setMode, (ctx) => getAgent().setSessionMode(ctx.params))
.onRequest(acp.methods.agent.session.setConfigOption, (ctx) => getAgent().setSessionConfigOption(ctx.params))
.onRequest(acp.methods.agent.session.delete, (ctx) => getAgent(ctx.params).deleteSession(ctx.params))
.onRequest(acp.methods.agent.session.resume, (ctx) => setup(ctx.params.sessionId, ctx.requestId, () => getAgent(ctx.params).resumeSession(ctx.params), ctx.signal))
.onRequest(acp.methods.agent.session.close, (ctx) => getAgent(ctx.params).closeSession(ctx.params))
.onRequest(acp.methods.agent.session.setMode, (ctx) => getAgent(ctx.params).setSessionMode(ctx.params))
.onRequest(acp.methods.agent.session.setConfigOption, (ctx) => getAgent(ctx.params).setSessionConfigOption(ctx.params))
.onRequest(acp.methods.agent.authenticate, (ctx) => getAgent().authenticate(ctx.params, ctx.requestId))
.onRequest(acp.methods.agent.logout, (ctx) => getAgent().logout(ctx.params))
.onRequest(acp.methods.agent.providers.list, (ctx) => getAgent().listProviders(ctx.params))
.onRequest(acp.methods.agent.providers.set, (ctx) => getAgent().setProvider(ctx.params))
.onRequest(acp.methods.agent.providers.disable, (ctx) => getAgent().disableProvider(ctx.params))
.onRequest(acp.methods.agent.session.prompt, (ctx) => getAgent().prompt(ctx.params, ctx.signal))
.onRequest(acp.methods.agent.session.prompt, (ctx) => getAgent(ctx.params).prompt(ctx.params, ctx.signal))
.onNotification(acp.methods.agent.session.cancel, (ctx) => getAgent().cancel(ctx.params))
.onRequest("authentication/status", emptyExtensionParamsParser, (ctx) => getAgent().extMethod("authentication/status", ctx.params))
.onRequest("authentication/logout", emptyExtensionParamsParser, (ctx) => getAgent().extMethod("authentication/logout", ctx.params))
.onRequest(LEGACY_SET_SESSION_MODEL_METHOD, legacySetSessionModelParamsParser, (ctx) => getAgent().extMethod(LEGACY_SET_SESSION_MODEL_METHOD, ctx.params))
.onRequest(SESSION_STEERING_METHOD, sessionSteerParamsParser, (ctx) => getAgent().extMethod(SESSION_STEERING_METHOD, ctx.params))
.onRequest(ASYNC_TASK_STOP_METHOD, asyncTaskStopParamsParser, (ctx) => getAgent().extMethod(ASYNC_TASK_STOP_METHOD, ctx.params))
.onRequest(LEGACY_SET_SESSION_MODEL_METHOD, legacySetSessionModelParamsParser, (ctx) => getAgent(ctx.params).extMethod(LEGACY_SET_SESSION_MODEL_METHOD, ctx.params))
.onRequest(SESSION_STEERING_METHOD, sessionSteerParamsParser, (ctx) => getAgent(ctx.params).extMethod(SESSION_STEERING_METHOD, ctx.params))
.onRequest(ASYNC_TASK_STOP_METHOD, asyncTaskStopParamsParser, (ctx) => getAgent(ctx.params).extMethod(ASYNC_TASK_STOP_METHOD, ctx.params))
.onRequest(CODEX_HOOKS_LIST_METHOD, hooksListParamsParser, (ctx) => getAgent().listHooks(ctx.params.cwd))
.onRequest(CODEX_HOOKS_TRUST_METHOD, hooksTrustParamsParser, (ctx) => getAgent().trustHooks(ctx.params.cwd, ctx.params.hooks))
.onRequest(GOAL_CONTROL_METHOD, goalControlParamsParser, (ctx) => getAgent().extMethod(GOAL_CONTROL_METHOD, ctx.params))
.onRequest(SESSION_RENAME_METHOD, sessionRenameParamsParser, (ctx) => getAgent().sessionIndex.rename(ctx.params))
.onRequest(SESSION_ARCHIVE_METHOD, sessionArchiveParamsParser, (ctx) => getAgent().sessionIndex.setArchived(ctx.params, true))
.onRequest(SESSION_UNARCHIVE_METHOD, sessionArchiveParamsParser, (ctx) => getAgent().sessionIndex.setArchived(ctx.params, false))
.onRequest(GOAL_CONTROL_METHOD, goalControlParamsParser, (ctx) => getAgent(ctx.params).extMethod(GOAL_CONTROL_METHOD, ctx.params))
.onRequest(SESSION_RENAME_METHOD, sessionRenameParamsParser, (ctx) => getAgent(ctx.params).sessionIndex.rename(ctx.params))
.onRequest(SESSION_ARCHIVE_METHOD, sessionArchiveParamsParser, (ctx) => getAgent(ctx.params).sessionIndex.setArchived(ctx.params, true))
.onRequest(SESSION_UNARCHIVE_METHOD, sessionArchiveParamsParser, (ctx) => getAgent(ctx.params).sessionIndex.setArchived(ctx.params, false))
.onRequest(SESSION_LIST_SUBSCRIBE_METHOD, sessionListSubscriptionParamsParser, (ctx) => getAgent().sessionIndex.subscribeList(ctx.params))
.onRequest(SESSION_LIST_UNSUBSCRIBE_METHOD, sessionListSubscriptionParamsParser, (ctx) => getAgent().sessionIndex.unsubscribeList(ctx.params));

Expand All @@ -122,26 +181,26 @@ export function createAcpAgentRouter(
// No `session/load`: v2 folds it into `session/resume` with `replayFrom`.
.onRequest(acpV2.methods.agent.session.new, (ctx) => getAgent().newSessionV2(ctx.params))
.onRequest(acpV2.methods.agent.session.list, (ctx) => getAgent().listSessions(ctx.params))
.onRequest(acpV2.methods.agent.session.delete, (ctx) => getAgent().deleteSession(ctx.params))
.onRequest(acpV2.methods.agent.session.resume, (ctx) => getAgent().resumeSessionV2(ctx.params))
.onRequest(acpV2.methods.agent.session.fork, (ctx) => getAgent().forkSessionV2(ctx.params))
.onRequest(acpV2.methods.agent.session.close, (ctx) => getAgent().closeSession(ctx.params))
.onRequest(acpV2.methods.agent.session.delete, (ctx) => getAgent(ctx.params).deleteSession(ctx.params))
.onRequest(acpV2.methods.agent.session.resume, (ctx) => setup(ctx.params.sessionId, ctx.requestId, () => getAgent(ctx.params).resumeSessionV2(ctx.params), ctx.signal))
.onRequest(acpV2.methods.agent.session.fork, (ctx) => getAgent(ctx.params).forkSessionV2(ctx.params))
.onRequest(acpV2.methods.agent.session.close, (ctx) => getAgent(ctx.params).closeSession(ctx.params))
// No `session/set_mode`: v2 removed the modes API; modes are config options.
.onRequest(acpV2.methods.agent.session.setConfigOption, (ctx) => getAgent().setSessionConfigOptionV2(ctx.params))
.onRequest(acpV2.methods.agent.session.setConfigOption, (ctx) => getAgent(ctx.params).setSessionConfigOptionV2(ctx.params))
// The v1-only `authentication/status|logout` extensions are not registered on v2.
.onRequest(acpV2.methods.agent.auth.login, (ctx) => getAgent().authenticateV2(ctx.params, ctx.requestId))
.onRequest(acpV2.methods.agent.auth.logout, (ctx) => getAgent().logoutV2(ctx.params))
.onRequest(acpV2.methods.agent.session.prompt, (ctx) => getAgent().promptV2(ctx.params, ctx.signal))
.onRequest(acpV2.methods.agent.session.prompt, (ctx) => getAgent(ctx.params).promptV2(ctx.params, ctx.signal))
.onNotification(acpV2.methods.agent.session.cancel, (ctx) => getAgent().cancel(ctx.params))
.onRequest(acpV2.methods.agent.providers.list, (ctx) => getAgent().listProviders(ctx.params))
.onRequest(acpV2.methods.agent.providers.set, (ctx) => getAgent().setProvider(ctx.params))
.onRequest(acpV2.methods.agent.providers.disable, (ctx) => getAgent().disableProvider(ctx.params))
.onRequest(SESSION_STEERING_METHOD, sessionSteerParamsParser, (ctx) => getAgent().extMethod(SESSION_STEERING_METHOD, ctx.params))
.onRequest(GOAL_CONTROL_METHOD, goalControlParamsParser, (ctx) => getAgent().extMethod(GOAL_CONTROL_METHOD, ctx.params))
.onRequest(ASYNC_TASK_STOP_METHOD, asyncTaskStopParamsParser, (ctx) => getAgent().extMethod(ASYNC_TASK_STOP_METHOD, ctx.params))
.onRequest(SESSION_RENAME_METHOD, sessionRenameParamsParser, (ctx) => getAgent().sessionIndex.rename(ctx.params))
.onRequest(SESSION_ARCHIVE_METHOD, sessionArchiveParamsParser, (ctx) => getAgent().sessionIndex.setArchived(ctx.params, true))
.onRequest(SESSION_UNARCHIVE_METHOD, sessionArchiveParamsParser, (ctx) => getAgent().sessionIndex.setArchived(ctx.params, false))
.onRequest(SESSION_STEERING_METHOD, sessionSteerParamsParser, (ctx) => getAgent(ctx.params).extMethod(SESSION_STEERING_METHOD, ctx.params))
.onRequest(GOAL_CONTROL_METHOD, goalControlParamsParser, (ctx) => getAgent(ctx.params).extMethod(GOAL_CONTROL_METHOD, ctx.params))
.onRequest(ASYNC_TASK_STOP_METHOD, asyncTaskStopParamsParser, (ctx) => getAgent(ctx.params).extMethod(ASYNC_TASK_STOP_METHOD, ctx.params))
.onRequest(SESSION_RENAME_METHOD, sessionRenameParamsParser, (ctx) => getAgent(ctx.params).sessionIndex.rename(ctx.params))
.onRequest(SESSION_ARCHIVE_METHOD, sessionArchiveParamsParser, (ctx) => getAgent(ctx.params).sessionIndex.setArchived(ctx.params, true))
.onRequest(SESSION_UNARCHIVE_METHOD, sessionArchiveParamsParser, (ctx) => getAgent(ctx.params).sessionIndex.setArchived(ctx.params, false))
.onRequest(SESSION_LIST_SUBSCRIBE_METHOD, sessionListSubscriptionParamsParser, (ctx) => getAgent().sessionIndex.subscribeList(ctx.params))
.onRequest(SESSION_LIST_UNSUBSCRIBE_METHOD, sessionListSubscriptionParamsParser, (ctx) => getAgent().sessionIndex.unsubscribeList(ctx.params));

Expand Down
Loading
Loading