feat: add live standard ACP subagent sessions

readme-dev.md

@@ -212,7 +212,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
@@ -242,6 +242,29 @@ schema alone:
 - Other AIR `_meta` keys (goal, message phase, mode kind, permission presentation) go only to AIR
   clients, as on v1. Every v2 permission request still gets its prompt `title` and `description`.
 
+#### Live standard subagent sessions
+
+Non-AIR v1 clients opt in with `clientCapabilities.subagents: {}`. Non-AIR v2 clients use
+standard `subagent_update` associations automatically. AIR keeps its existing opt-in, updates,
+and generation-based identities; clients do not receive both contracts.
+
+A standard child keeps one opaque conversation ID across assignments and parent handler changes.
+Its ancestors are announced before its output, tool updates, or interactions. Permissions and
+elicitation use the child's ID and survive parent replay; actual work endings withdraw obsolete
+requests locally, even if the client returns a late answer. Child tools, terminals, plans, titles,
+and observed model options stay separate from the parent.
+
+Capabilities are empty: direct child prompting, loading, configuration, cancellation, and other
+client mutations are not enabled. Parent cancellation interrupts observed live descendants.
+An observed work ending reports `idle`; loss of observability reports `unknown`, not a fabricated
+failure or a permanently closed conversation.
+
+Replay currently restores recoverable associations, not recursive child transcripts or historical
+work outcomes. A parent advisory makes that gap visible. Rejoining a loaded child requires current
+provider parentage; copied fork history alone cannot acquire the original runtime's subscriptions.
+After replay, current child work/model snapshots follow the actual response write, including whole
+JSON-RPC batches. Parent setup operations serialize without waiting on each other's batch delivery.
+
 #### Models of a `providers/set` gateway
 
 `providers/set` lists the models of the gateway (`GET {baseUrl}/models`, with the headers Codex sends to it)

src/ACPSessionConnection.ts

@@ -248,6 +248,17 @@ export class ACPSessionConnection {
         await this.v2Connection.updateState(this.sessionId, state);
     }
 
+    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) {

src/AcpAgentRouter.ts

@@ -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,
@@ -65,17 +66,73 @@ 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; });
+        state.tail = state.tail.catch(() => {}).then(async () => {
+            let registered = false;
+            try {
+                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();
+            }
+        });
+        // Waiting for delivery here would deadlock two setup requests in the same batch.
+        return response;
+    };
     const attachAgent = (agent: CodexAcpServer, signal: AbortSignal) => {
         codexAcpServer = agent;
         signal.addEventListener("abort", () => {
+            setups.clear();
             agent.sessionIndex.dispose();
             if (codexAcpServer === agent) {
                 codexAcpServer = null;
@@ -87,32 +144,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));
 
@@ -122,26 +179,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));
 

src/AcpResponseDelivery.ts

@@ -0,0 +1,42 @@
+import type {AnyWireMessage, JsonRpcId, WireStream} from "@agentclientprotocol/sdk/experimental/v2/unstable";
+
+/** Observe completed writes, including batches, without changing SDK framing. */
+export class AcpResponseDelivery {
+    private readonly callbacks = new Map<JsonRpcId, () => void>();
+    private disposed = false;
+
+    afterWrite(requestId: JsonRpcId, callback: () => void): void {
+        if (!this.disposed) this.callbacks.set(requestId, callback);
+    }
+
+    observe(stream: WireStream): WireStream {
+        return {
+            readable: stream.readable,
+            writable: new WritableStream<AnyWireMessage>({
+                write: async message => {
+                    const writer = stream.writable.getWriter();
+                    try { await writer.write(message); }
+                    finally { writer.releaseLock(); }
+                    for (const entry of Array.isArray(message) ? message : [message]) {
+                        if ("method" in entry || entry.id === null) continue;
+                        const callback = this.callbacks.get(entry.id);
+                        this.callbacks.delete(entry.id);
+                        callback?.();
+                    }
+                },
+                close: async () => {
+                    const writer = stream.writable.getWriter();
+                    try { await writer.close(); }
+                    finally { this.dispose(); writer.releaseLock(); }
+                },
+                abort: async reason => {
+                    const writer = stream.writable.getWriter();
+                    try { await writer.abort(reason); }
+                    finally { this.dispose(); writer.releaseLock(); }
+                },
+            }),
+        };
+    }
+
+    dispose(): void { this.disposed = true; this.callbacks.clear(); }
+}