feat: codex steer

AGENTS.md

@@ -1,5 +1,10 @@
 # Repository Guidelines
 
+## Pull Requests
+
+- Always open pull requests against the `loro-dev` organization repository
+  (`loro-dev/acp-extension-codex`), not a personal fork.
+
 ## Project Structure
 
 - `src/` — ACP server implementation. Entry point: `src/index.ts`.
@@ -28,6 +33,7 @@
 
 - Codex app-server usage: see https://github.com/openai/codex/blob/main/codex-rs/app-server/README.md when touching protocol/transport details, adding or consuming JSON-RPC methods, handling approvals/turn events, or updating generated schema/clients.
 - App-server events: prefer `thread/*`, `turn/*`, and `item/*` event surfaces; avoid the deprecated `codex/event/*` API (planned removal). Keep implementations aligned with generated types in `src/app-server` (including `v2` exports).
+- Steer uses app-server `turn/steer` on the tracked active turn. Correlate `clientUserMessageId` and acknowledge only the matching `item/completed(userMessage)`; never emulate steer with a second `turn/start`.
 - Codex reasoning summaries can echo trailing empty HTML comments from model instructions. Keep
   that provider-specific cleanup in `src/ReasoningText.ts` across live deltas and history replay;
   do not filter assistant text, raw reasoning, or HTML globally in the client renderer.

README.md

@@ -13,6 +13,7 @@ Use [OpenAI Codex](https://github.com/openai/codex) from [Agent Client Protocol]
 - 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.
 - Client-provided MCP servers over command-based stdio config and HTTP transport.
+- Acknowledged steering of an active Codex turn through app-server `turn/steer`.
 - Slash commands: `/status`, `/mcp`, `/skills`, `/review`, `/review-branch`, `/review-commit`, `/compact`, and `/logout`, as well as configured skills.
 
 ## Installation
@@ -44,6 +45,15 @@ The adapter advertises ACP auth methods during initialization. Clients can authe
 - API key via `CODEX_API_KEY` or `OPENAI_API_KEY`.
 - A custom OpenAI-compatible gateway, when the client opts in to the gateway auth capability.
 
+## Steering extension
+
+The initialize response advertises `_meta.codex.steer` with the applied notification
+method and active-turn configuration policy. To steer a running prompt, send another
+`session/prompt` with a unique `_meta.codex.steer.id`. The adapter calls Codex
+`turn/steer` and emits `_codex/steerApplied { sessionId, steerId }` only after Codex
+commits the correlated user-message item. The steer keeps the active turn's model,
+mode, and configuration; slash commands cannot be steered.
+
 ## Runtime options
 
 - `CODEX_API_KEY` - API key used when the API-key auth method is selected. Takes precedence over `OPENAI_API_KEY`.

package.json

@@ -3,7 +3,7 @@
   "publishConfig": {
     "access": "public"
   },
-  "version": "1.1.2",
+  "version": "1.2.0",
   "description": "An ACP-compatible coding agent powered by Codex",
   "main": "dist/index.js",
   "bin": {
@@ -75,4 +75,4 @@
     "vscode-jsonrpc": "^9.0.1",
     "zod": "^4.0.0"
   }
-}
+}

src/AcpExtensions.ts

@@ -10,6 +10,31 @@ export const LEGACY_SET_SESSION_MODEL_METHOD = "session/set_model";
 export const ACP_EXT_SESSION_USAGE_UPDATE_METHOD = "_acp_ext:session_usage_update";
 export const ACP_EXT_SESSION_RATE_LIMITS_METHOD = "_acp_ext:session_rate_limits";
 export const ACP_EXT_CODEX_PROPOSED_PLAN_METHOD = "_acp_ext:codex_proposed_plan";
+export const CODEX_STEER_APPLIED_METHOD = "_codex/steerApplied";
+
+export type CodexSteerCapability = {
+    version: 1;
+    appliedNotification: typeof CODEX_STEER_APPLIED_METHOD;
+    upstreamTurn: "same";
+    configPolicy: "active";
+}
+
+export const CODEX_STEER_CAPABILITY: CodexSteerCapability = {
+    version: 1,
+    appliedNotification: CODEX_STEER_APPLIED_METHOD,
+    upstreamTurn: "same",
+    configPolicy: "active",
+};
+
+export function getCodexSteerId(meta: unknown): string | null {
+    if (typeof meta !== "object" || meta === null) return null;
+    const codex = (meta as Record<string, unknown>)["codex"];
+    if (typeof codex !== "object" || codex === null) return null;
+    const steer = (codex as Record<string, unknown>)["steer"];
+    if (typeof steer !== "object" || steer === null) return null;
+    const id = (steer as Record<string, unknown>)["id"];
+    return typeof id === "string" && id.length > 0 ? id : null;
+}
 
 export type LegacySessionModel = {
     modelId: string;

src/CodexAcpClient.ts

@@ -603,6 +603,20 @@ export class CodexAcpClient {
         return await this.codexClient.runTurn(params, onTurnStarted);
     }
 
+    async sendSteer(
+        request: acp.PromptRequest,
+        expectedTurnId: string,
+        steerId: string,
+    ): Promise<string> {
+        const response = await this.codexClient.turnSteer({
+            threadId: request.sessionId,
+            input: buildPromptItems(request.prompt),
+            expectedTurnId,
+            clientUserMessageId: steerId,
+        });
+        return response.turnId;
+    }
+
     resolveTurnInterrupted(params: { threadId: string, turnId: string }): void {
         this.codexClient.resolveTurnInterrupted(params.threadId, params.turnId);
     }

src/CodexAcpServer.ts

@@ -7,7 +7,7 @@ import {type CodexAuthRequest, getCodexAuthMethods, isCodexAuthRequest} from "./
 import {CodexAcpClient, type SessionMetadata, type SessionMetadataWithThread} from "./CodexAcpClient";
 import type {McpStartupResult} from "./CodexAppServerClient";
 import {ACPSessionConnection, type AcpClientConnection, type UpdateSessionEvent} from "./ACPSessionConnection";
-import type {CollaborationMode, InputModality, ReasoningEffort} from "./app-server";
+import type {CollaborationMode, InputModality, ReasoningEffort, ServerNotification} from "./app-server";
 import type {
     Account,
     Model,
@@ -42,6 +42,9 @@ import {
     type LegacySessionModelState,
     type LegacySetSessionModelRequest,
     type LegacySetSessionModelResponse,
+    CODEX_STEER_CAPABILITY,
+    CODEX_STEER_APPLIED_METHOD,
+    getCodexSteerId,
     isExtMethodRequest,
     LEGACY_SET_SESSION_MODEL_METHOD,
 } from "./AcpExtensions";
@@ -135,11 +138,17 @@ interface ActivePrompt {
     cancelSignal: Promise<null>;
     signal: AbortSignal;
     currentTurn: { threadId: string, turnId: string } | null;
+    outcome: Promise<acp.PromptResponse> | null;
     requestCancel: () => void;
     requestClose: () => void;
     complete: () => void;
 }
 
+interface PendingSteer {
+    activePrompt: ActivePrompt;
+    turnId: string;
+}
+
 export class CodexAcpServer {
     private static readonly MODEL_NAME_TOKEN_OVERRIDES: Record<string, string> = {
         gpt: "GPT",
@@ -162,6 +171,7 @@ export class CodexAcpServer {
     private readonly pendingMcpStartupSessions: Map<string, PendingMcpStartupSession>;
     private readonly pendingTurnStarts: Map<string, PendingTurnStart>;
     private readonly activePrompts: Map<string, ActivePrompt>;
+    private readonly pendingSteers: Map<string, Map<string, PendingSteer>>;
     private readonly closingSessions: Map<string, number>;
     private readonly sessionGenerations: Map<string, number>;
     private readonly sessionOpenGenerations: Map<string, number>;
@@ -177,6 +187,7 @@ export class CodexAcpServer {
         this.pendingMcpStartupSessions = new Map();
         this.pendingTurnStarts = new Map();
         this.activePrompts = new Map();
+        this.pendingSteers = new Map();
         this.closingSessions = new Map();
         this.sessionGenerations = new Map();
         this.sessionOpenGenerations = new Map();
@@ -233,7 +244,12 @@ export class CodexAcpServer {
                     acp: false,
                     http: true,
                     sse: false
-                }
+                },
+                _meta: {
+                    codex: {
+                        steer: CODEX_STEER_CAPABILITY,
+                    },
+                },
             },
             authMethods: getCodexAuthMethods(_params.clientCapabilities),
         };
@@ -1262,7 +1278,6 @@ export class CodexAcpServer {
             resolveCancelSignal = resolve;
         });
         const abortController = new AbortController();
-
         let completed = false;
         let closeRequested = false;
         const activePrompt: ActivePrompt = {
@@ -1271,6 +1286,7 @@ export class CodexAcpServer {
             cancelSignal,
             signal: abortController.signal,
             currentTurn: null,
+            outcome: null,
             requestCancel: () => {
                 if (abortController.signal.aborted) {
                     return;
@@ -1294,6 +1310,7 @@ export class CodexAcpServer {
                 if (this.activePrompts.get(sessionId) === activePrompt) {
                     this.activePrompts.delete(sessionId);
                 }
+                this.clearPendingSteers(sessionId, activePrompt);
                 resolveCompletion();
             },
         };
@@ -1302,6 +1319,80 @@ export class CodexAcpServer {
         return activePrompt;
     }
 
+    private clearPendingSteers(sessionId: string, activePrompt: ActivePrompt): void {
+        const pending = this.pendingSteers.get(sessionId);
+        if (!pending) return;
+        for (const [steerId, steer] of pending) {
+            if (steer.activePrompt === activePrompt) pending.delete(steerId);
+        }
+        if (pending.size === 0) this.pendingSteers.delete(sessionId);
+    }
+
+    private async handleSteerAppliedNotification(
+        sessionId: string,
+        event: ServerNotification,
+        activePrompt: ActivePrompt,
+    ): Promise<void> {
+        if (event.method !== "item/completed" || event.params.item.type !== "userMessage") return;
+        const steerId = event.params.item.clientId;
+        if (!steerId) return;
+        const pending = this.pendingSteers.get(sessionId);
+        if (!pending) return;
+        const steer = pending.get(steerId);
+        if (!steer || steer.activePrompt !== activePrompt || steer.turnId !== event.params.turnId) return;
+        pending.delete(steerId);
+        if (pending.size === 0) this.pendingSteers.delete(sessionId);
+        await this.connection.notify(CODEX_STEER_APPLIED_METHOD, {sessionId, steerId});
+    }
+
+    private async steerPrompt(params: acp.PromptRequest): Promise<acp.PromptResponse> {
+        const steerId = getCodexSteerId(params._meta);
+        if (!steerId) throw RequestError.invalidRequest("Missing Codex steer id");
+        const firstText = params.prompt[0]?.type === "text" ? params.prompt[0].text : "";
+        if (firstText.startsWith("/")) {
+            throw RequestError.invalidRequest("Slash commands cannot steer an active Codex turn");
+        }
+
+        let activePrompt = this.activePrompts.get(params.sessionId);
+        if (!activePrompt) throw RequestError.invalidRequest("No active Codex turn to steer");
+        if (!activePrompt.currentTurn) {
+            await this.pendingTurnStarts.get(params.sessionId)?.promise;
+            activePrompt = this.activePrompts.get(params.sessionId);
+        }
+        const turn = activePrompt?.currentTurn;
+        if (!activePrompt || !turn || activePrompt.signal.aborted) {
+            throw RequestError.invalidRequest("No active Codex turn to steer");
+        }
+
+        const pending = this.pendingSteers.get(params.sessionId) ?? new Map<string, PendingSteer>();
+        if (pending.has(steerId)) {
+            throw RequestError.invalidRequest(`Duplicate Codex steer id: ${steerId}`);
+        }
+        pending.set(steerId, {activePrompt, turnId: turn.turnId});
+        this.pendingSteers.set(params.sessionId, pending);
+        try {
+            const steeredTurnId = await this.runWithProcessCheck(
+                () => this.codexAcpClient.sendSteer(params, turn.turnId, steerId),
+            );
+            if (steeredTurnId !== turn.turnId) {
+                throw RequestError.internalError(
+                    undefined,
+                    `Codex steered unexpected turn ${steeredTurnId}; expected ${turn.turnId}`,
+                );
+            }
+        } catch (error) {
+            if (pending.get(steerId)?.activePrompt === activePrompt) {
+                pending.delete(steerId);
+                if (pending.size === 0) this.pendingSteers.delete(params.sessionId);
+            }
+            throw error;
+        }
+        if (!activePrompt.outcome) {
+            throw RequestError.internalError(undefined, "Active Codex prompt has no tracked outcome");
+        }
+        return await activePrompt.outcome;
+    }
+
     private cancelBeforeTurnStarted(activePrompt: ActivePrompt): Promise<null> {
         return activePrompt.cancelSignal.then(() => {
             if (activePrompt.currentTurn === null) {
@@ -1462,6 +1553,23 @@ export class CodexAcpServer {
     }
 
     async prompt(params: acp.PromptRequest, signal?: AbortSignal): Promise<acp.PromptResponse> {
+        if (getCodexSteerId(params._meta)) {
+            return await this.steerPrompt(params);
+        }
+        if (this.activePrompts.has(params.sessionId)) {
+            throw RequestError.invalidRequest(
+                "A Codex prompt is already active; use the advertised steer extension",
+            );
+        }
+        const outcome = this.runPrompt(params, signal);
+        const activePrompt = this.activePrompts.get(params.sessionId);
+        if (activePrompt) {
+            activePrompt.outcome = outcome;
+        }
+        return await outcome;
+    }
+
+    private async runPrompt(params: acp.PromptRequest, signal?: AbortSignal): Promise<acp.PromptResponse> {
         logger.log("Prompt received", {
             sessionId: params.sessionId,
             prompt: params.prompt,
@@ -1470,14 +1578,8 @@ export class CodexAcpServer {
         sessionState.currentTurnId = null;
         sessionState.lastTokenUsage = null;
         const activePrompt = this.trackActivePrompt(params.sessionId);
-        let pendingTurnStart: PendingTurnStart | null = null;
-        const ensurePendingTurnStart = (): PendingTurnStart => {
-            if (pendingTurnStart === null) {
-                pendingTurnStart = this.createPendingTurnStart();
-                this.pendingTurnStarts.set(params.sessionId, pendingTurnStart);
-            }
-            return pendingTurnStart;
-        };
+        const pendingTurnStart = this.createPendingTurnStart();
+        this.pendingTurnStarts.set(params.sessionId, pendingTurnStart);
         const disposePromptRequestCancellation = this.observePromptRequestCancellation(signal, sessionState, activePrompt);
 
         try {
@@ -1491,6 +1593,7 @@ export class CodexAcpServer {
             );
             await this.codexAcpClient.subscribeToSessionEvents(params.sessionId,
                 async (event) => {
+                    await this.handleSteerAppliedNotification(params.sessionId, event, activePrompt);
                     await elicitationHandler.handleNotification(event);
                     return eventHandler.handleNotification(event);
                 },
@@ -1502,9 +1605,6 @@ export class CodexAcpServer {
             }
 
             const commandPromise = this.availableCommands.tryHandleCommand(params.prompt, sessionState, {
-                onTurnStartPending: () => {
-                    ensurePendingTurnStart();
-                },
                 onTurnStarted: (turnId, threadId) => {
                     const turn = {threadId, turnId};
                     activePrompt.currentTurn = turn;
@@ -1513,7 +1613,7 @@ export class CodexAcpServer {
                         return;
                     }
                     sessionState.currentTurnId = turnId;
-                    pendingTurnStart?.resolve(turnId);
+                    pendingTurnStart.resolve(turnId);
                 },
             });
             void commandPromise.catch((err) => {
@@ -1573,7 +1673,6 @@ export class CodexAcpServer {
                 sessionState.currentModelSupportsFast,
             );
             const collaborationMode = this.createPlanModeCollaborationMode(sessionState, modelId);
-            ensurePendingTurnStart();
             const sendPromptPromise = this.runWithProcessCheck(
                 () => this.codexAcpClient.sendPrompt(
                     params,
@@ -1592,7 +1691,7 @@ export class CodexAcpServer {
                             return;
                         }
                         sessionState.currentTurnId = turnId;
-                        pendingTurnStart?.resolve(turnId);
+                        pendingTurnStart.resolve(turnId);
                     },
                     () => this.promptShouldStop(params.sessionId, activePrompt),
                 ));