From d25177395c0bed66716903cf686e783c02d8c07c Mon Sep 17 00:00:00 2001 From: Leon Zhao Date: Sat, 11 Jul 2026 06:51:06 +0000 Subject: [PATCH 1/2] docs: require PRs target the loro-dev org repo Add a Pull Requests rule to AGENTS.md so PRs are opened against loro-dev/acp-extension-codex rather than a personal fork. Model: claude-opus-4-8[1m] Co-Authored-By: Claude Opus 4.8 (1M context) --- AGENTS.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/AGENTS.md b/AGENTS.md index 434c6cd0..c59ddac4 100644 --- a/AGENTS.md +++ b/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`. From 803e42a15b2060e4f530953cb900fc066f1f6ee8 Mon Sep 17 00:00:00 2001 From: Leon Zhao Date: Sun, 12 Jul 2026 15:33:21 +0000 Subject: [PATCH 2/2] feat: add acknowledged Codex steer Model: gpt-5 --- AGENTS.md | 1 + README.md | 10 ++ package-lock.json | 4 +- package.json | 4 +- src/AcpExtensions.ts | 25 ++++ src/CodexAcpClient.ts | 14 ++ src/CodexAcpServer.ts | 133 +++++++++++++++--- src/CodexAppServerClient.ts | 6 + .../CodexACPAgent/initialize.test.ts | 10 ++ src/__tests__/CodexACPAgent/steer.test.ts | 119 ++++++++++++++++ 10 files changed, 305 insertions(+), 21 deletions(-) create mode 100644 src/__tests__/CodexACPAgent/steer.test.ts diff --git a/AGENTS.md b/AGENTS.md index c59ddac4..8b751a27 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -33,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. diff --git a/README.md b/README.md index e4345a35..5e837f89 100644 --- a/README.md +++ b/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`. diff --git a/package-lock.json b/package-lock.json index 36c0ac73..1ed03c97 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "acp-extension-codex", - "version": "1.0.1", + "version": "1.2.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "acp-extension-codex", - "version": "1.0.1", + "version": "1.2.0", "license": "Apache-2.0", "dependencies": { "@agentclientprotocol/sdk": "^1.2.1", diff --git a/package.json b/package.json index 2c25cc17..7bde9244 100644 --- a/package.json +++ b/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" } -} \ No newline at end of file +} diff --git a/src/AcpExtensions.ts b/src/AcpExtensions.ts index 411f5807..598d6c1e 100644 --- a/src/AcpExtensions.ts +++ b/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)["codex"]; + if (typeof codex !== "object" || codex === null) return null; + const steer = (codex as Record)["steer"]; + if (typeof steer !== "object" || steer === null) return null; + const id = (steer as Record)["id"]; + return typeof id === "string" && id.length > 0 ? id : null; +} export type LegacySessionModel = { modelId: string; diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index e884df56..13cdde65 100644 --- a/src/CodexAcpClient.ts +++ b/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 { + 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); } diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index c846294f..2fa15dd6 100644 --- a/src/CodexAcpServer.ts +++ b/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; signal: AbortSignal; currentTurn: { threadId: string, turnId: string } | null; + outcome: Promise | null; requestCancel: () => void; requestClose: () => void; complete: () => void; } +interface PendingSteer { + activePrompt: ActivePrompt; + turnId: string; +} + export class CodexAcpServer { private static readonly MODEL_NAME_TOKEN_OVERRIDES: Record = { gpt: "GPT", @@ -162,6 +171,7 @@ export class CodexAcpServer { private readonly pendingMcpStartupSessions: Map; private readonly pendingTurnStarts: Map; private readonly activePrompts: Map; + private readonly pendingSteers: Map>; private readonly closingSessions: Map; private readonly sessionGenerations: Map; private readonly sessionOpenGenerations: Map; @@ -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 { + 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 { + 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(); + 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 { return activePrompt.cancelSignal.then(() => { if (activePrompt.currentTurn === null) { @@ -1462,6 +1553,23 @@ export class CodexAcpServer { } async prompt(params: acp.PromptRequest, signal?: AbortSignal): Promise { + 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 { 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), )); diff --git a/src/CodexAppServerClient.ts b/src/CodexAppServerClient.ts index d1bc210c..500bd91f 100644 --- a/src/CodexAppServerClient.ts +++ b/src/CodexAppServerClient.ts @@ -64,6 +64,8 @@ import type { PermissionsRequestApprovalResponse, ItemCompletedNotification, } from "./app-server/v2"; +import type {TurnSteerParams} from "./app-server/v2/TurnSteerParams"; +import type {TurnSteerResponse} from "./app-server/v2/TurnSteerResponse"; export interface ApprovalHandler { handleCommandExecution(params: CommandExecutionRequestApprovalParams): Promise; @@ -241,6 +243,10 @@ export class CodexAppServerClient { return await this.sendRequest({ method: "turn/start", params: params }); } + async turnSteer(params: TurnSteerParams): Promise { + return await this.sendRequest({ method: "turn/steer", params }); + } + async runTurn(params: TurnStartParams, onTurnStarted?: (turnId: string) => void): Promise { const capturedCompletions: Array = []; const releaseCapture = this.captureTurnCompletions(params.threadId, (event) => { diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index 0bff8791..85ec1caa 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -58,6 +58,16 @@ describe('CodexACPAgent - initialize', () => { http: true, sse: false, }, + _meta: { + codex: { + steer: { + version: 1, + appliedNotification: "_codex/steerApplied", + upstreamTurn: "same", + configPolicy: "active", + }, + }, + }, }, authMethods: getCodexAuthMethods(), }); diff --git a/src/__tests__/CodexACPAgent/steer.test.ts b/src/__tests__/CodexACPAgent/steer.test.ts new file mode 100644 index 00000000..411d8e68 --- /dev/null +++ b/src/__tests__/CodexACPAgent/steer.test.ts @@ -0,0 +1,119 @@ +import {describe, expect, it, vi} from "vitest"; +import {CODEX_STEER_APPLIED_METHOD} from "../../AcpExtensions"; +import {setupPromptTestSession} from "../acp-test-utils"; +import type {TurnCompletedNotification} from "../../app-server/v2"; + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((innerResolve, innerReject) => { + resolve = innerResolve; + reject = innerReject; + }); + return {promise, resolve, reject}; +} + +describe("CodexACPAgent - steer", () => { + it("steers the active app-server turn and acknowledges its committed user message", async () => { + const {mockFixture, sessionState, turnStartSpy} = setupPromptTestSession(); + const appServer = mockFixture.getCodexAppServerClient(); + const completion = deferred(); + const steerResponse = deferred<{turnId: string}>(); + vi.spyOn(appServer, "awaitTurnCompleted").mockReturnValue(completion.promise); + const turnSteerSpy = vi.spyOn(appServer, "turnSteer").mockReturnValue(steerResponse.promise); + + const originalPrompt = mockFixture.getCodexAcpAgent().prompt({ + sessionId: sessionState.sessionId, + prompt: [{type: "text", text: "start"}], + }); + await vi.waitFor(() => expect(turnStartSpy).toHaveBeenCalledOnce()); + + const steerPrompt = mockFixture.getCodexAcpAgent().prompt({ + sessionId: sessionState.sessionId, + prompt: [{type: "text", text: "change direction"}], + _meta: {codex: {steer: {id: "steer-1"}}}, + }); + await vi.waitFor(() => expect(turnSteerSpy).toHaveBeenCalledWith({ + threadId: sessionState.sessionId, + input: [{type: "text", text: "change direction", text_elements: []}], + expectedTurnId: "turn-id", + clientUserMessageId: "steer-1", + })); + expect(turnStartSpy).toHaveBeenCalledOnce(); + expect(mockFixture.getAcpConnectionEvents([])).not.toContainEqual(expect.objectContaining({ + method: "notify", + args: [CODEX_STEER_APPLIED_METHOD, expect.anything()], + })); + + mockFixture.sendServerNotification({ + method: "item/completed", + params: { + threadId: sessionState.sessionId, + turnId: "turn-id", + completedAtMs: 1, + item: { + type: "userMessage", + id: "item-1", + clientId: "steer-1", + content: [{type: "text", text: "change direction", text_elements: []}], + }, + }, + }); + await mockFixture.getCodexAcpClient().waitForSessionNotifications(sessionState.sessionId); + expect(mockFixture.getAcpConnectionEvents([])).toContainEqual({ + method: "notify", + args: [CODEX_STEER_APPLIED_METHOD, { + sessionId: sessionState.sessionId, + steerId: "steer-1", + }], + }); + + steerResponse.resolve({turnId: "turn-id"}); + completion.resolve({ + threadId: sessionState.sessionId, + turn: { + id: "turn-id", + items: [], + itemsView: "notLoaded", + status: "completed", + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }, + }); + await expect(originalPrompt).resolves.toMatchObject({stopReason: "end_turn"}); + await expect(steerPrompt).resolves.toMatchObject({stopReason: "end_turn"}); + }); + + it("rejects an unmarked concurrent prompt instead of starting an ambiguous turn", async () => { + const {mockFixture, sessionState, turnStartSpy} = setupPromptTestSession(); + const completion = deferred(); + vi.spyOn(mockFixture.getCodexAppServerClient(), "awaitTurnCompleted").mockReturnValue(completion.promise); + const originalPrompt = mockFixture.getCodexAcpAgent().prompt({ + sessionId: sessionState.sessionId, + prompt: [{type: "text", text: "start"}], + }); + await vi.waitFor(() => expect(turnStartSpy).toHaveBeenCalledOnce()); + + await expect(mockFixture.getCodexAcpAgent().prompt({ + sessionId: sessionState.sessionId, + prompt: [{type: "text", text: "ambiguous"}], + })).rejects.toThrow("Invalid request"); + expect(turnStartSpy).toHaveBeenCalledOnce(); + completion.resolve({ + threadId: sessionState.sessionId, + turn: { + id: "turn-id", + items: [], + itemsView: "notLoaded", + status: "completed", + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }, + }); + await originalPrompt; + }); +});