diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts index 88dda92a8f..6800a7cef0 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts @@ -2115,4 +2115,468 @@ describe("CodexAppServerAgent", () => { // The richer handler returns a typed { answers } object, not a decision string. expect(response).toEqual({ answers: { q1: { answers: ["A"] } } }); }); + + // --- Plan-mode implementation handoff ------------------------------------ + + async function waitUntil(cond: () => boolean): Promise { + for (let i = 0; i < 50 && !cond(); i++) { + await new Promise((r) => setImmediate(r)); + } + } + + // permissionOutcome may be a value or a function (per-call, e.g. a pending promise). + function makePlanAgent(permissionOutcome: unknown) { + const stub = makeStubRpc({ + "thread/start": { thread: { id: "t" } }, + "turn/start": { turn: { id: "turn_1" } }, + }); + const sessionUpdates: Array<{ update?: Record }> = []; + const extNotifications: Array<{ method: string; params: unknown }> = []; + const permissionRequests: Array<{ + toolCall: Record; + options: Array<{ optionId?: string; kind?: string }>; + }> = []; + const client = { + sessionUpdate: async (n: unknown) => { + sessionUpdates.push(n as { update?: Record }); + }, + requestPermission: async (params: { + toolCall: Record; + options: Array<{ optionId?: string; kind?: string }>; + }) => { + permissionRequests.push(params); + return typeof permissionOutcome === "function" + ? permissionOutcome() + : permissionOutcome; + }, + extNotification: async (method: string, params: unknown) => { + extNotifications.push({ method, params }); + }, + } as unknown as AgentSideConnection; + const agent = new CodexAppServerAgent(client, { + processOptions: { binaryPath: "/x/codex" }, + model: "gpt-5.5", + rpcFactory: stub.factory, + }); + return { + agent, + stub, + sessionUpdates, + extNotifications, + permissionRequests, + }; + } + + // Wrapped in an object: returning the pending promise directly would make + // the caller's await adopt it and block until the whole prompt finishes. + async function startPlanTurn( + agent: CodexAppServerAgent, + stub: ReturnType, + meta?: Record, + ): Promise<{ done: ReturnType }> { + await agent.newSession({ + cwd: "/r", + ...(meta ? { _meta: meta } : {}), + } as unknown as NewSessionRequest); + await agent.setSessionConfigOption({ + configId: "mode", + value: "plan", + sessionId: "t", + } as any); + const done = agent.prompt({ + sessionId: "t", + prompt: [{ type: "text", text: "plan a refactor" }], + } as unknown as PromptRequest); + await waitUntil( + () => stub.requests.filter((r) => r.method === "turn/start").length >= 1, + ); + return { done }; + } + + it("offers the implement handoff from the streamed plan and runs the implementation in auto", async () => { + const { agent, stub, sessionUpdates, permissionRequests } = makePlanAgent({ + outcome: { outcome: "selected", optionId: "auto" }, + }); + const { done } = await startPlanTurn(agent, stub); + + // codex 0.140 can stream the proposal without a completed plan item. + stub.emit("item/plan/delta", { + itemId: "p1", + delta: "# The plan\n\n", + }); + stub.emit("item/plan/delta", { itemId: "p1", delta: "1. do it" }); + stub.emit("item/completed", { + item: { + type: "agentMessage", + id: "a1", + text: "The implementation plan is ready.", + }, + }); + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + + // The accepted handoff starts the implementation turn inside the same prompt(). + await waitUntil( + () => stub.requests.filter((r) => r.method === "turn/start").length >= 2, + ); + stub.emit("turn/completed", { + turn: { id: "turn_2", status: "completed" }, + }); + expect((await done).stopReason).toBe("end_turn"); + + // The approval renders as the plan-approval UI (switch_mode + the plan text). + expect(permissionRequests).toHaveLength(1); + expect(permissionRequests[0].toolCall).toMatchObject({ + toolCallId: "p1:implement", + kind: "switch_mode", + rawInput: { plan: "# The plan\n\n1. do it" }, + }); + expect(permissionRequests[0].options.map((o) => o.optionId)).toContain( + "auto", + ); + + // Mode flipped to auto and the host was told. + expect(sessionUpdates).toContainEqual( + expect.objectContaining({ + update: { sessionUpdate: "current_mode_update", currentModeId: "auto" }, + }), + ); + + // The implementation turn runs with auto's policies and an echoed kickoff message. + const turnStarts = stub.requests.filter((r) => r.method === "turn/start"); + expect(turnStarts[1].params).toMatchObject({ + input: [{ type: "text", text: "Implement the plan." }], + approvalPolicy: "on-request", + sandboxPolicy: sandboxPolicyFor("auto"), + collaborationMode: { mode: "default", settings: { model: "gpt-5.5" } }, + }); + expect(sessionUpdates).toContainEqual({ + sessionId: "t", + update: { + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "Implement the plan." }, + }, + }); + }); + + it("stays in plan mode when the handoff is rejected without feedback", async () => { + const { agent, stub, permissionRequests } = makePlanAgent({ + outcome: { outcome: "selected", optionId: "reject_with_feedback" }, + }); + const { done } = await startPlanTurn(agent, stub); + + stub.emit("item/completed", { + item: { type: "plan", id: "p1", text: "# The plan" }, + }); + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + + expect((await done).stopReason).toBe("end_turn"); + expect(permissionRequests).toHaveLength(1); + // No implementation turn started; the picker stays on plan. + expect(stub.requests.filter((r) => r.method === "turn/start")).toHaveLength( + 1, + ); + }); + + it("feeds handoff feedback into another plan turn", async () => { + const { agent, stub, permissionRequests } = makePlanAgent({ + outcome: { outcome: "selected", optionId: "reject_with_feedback" }, + _meta: { customInput: "cover the migration path too" }, + }); + const { done } = await startPlanTurn(agent, stub); + + stub.emit("item/completed", { + item: { type: "plan", id: "p1", text: "# The plan" }, + }); + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + + // The feedback re-enters planning; the revised turn ends without a new plan. + await waitUntil( + () => stub.requests.filter((r) => r.method === "turn/start").length >= 2, + ); + stub.emit("turn/completed", { + turn: { id: "turn_2", status: "completed" }, + }); + expect((await done).stopReason).toBe("end_turn"); + + expect(permissionRequests).toHaveLength(1); + const turnStarts = stub.requests.filter((r) => r.method === "turn/start"); + expect(turnStarts[1].params).toMatchObject({ + input: [{ type: "text", text: "cover the migration path too" }], + // Still planning: the follow-up keeps plan collaboration + the read-only sandbox. + collaborationMode: { mode: "plan", settings: { model: "gpt-5.5" } }, + sandboxPolicy: { type: "readOnly", networkAccess: true }, + }); + }); + + it("skips the handoff when a plan turn ends without a proposed plan", async () => { + const { agent, stub, permissionRequests } = makePlanAgent({ + outcome: { outcome: "selected", optionId: "auto" }, + }); + const { done } = await startPlanTurn(agent, stub); + + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + expect((await done).stopReason).toBe("end_turn"); + expect(permissionRequests).toHaveLength(0); + }); + + it("skips the handoff outside plan mode even if codex emits a plan item", async () => { + const { agent, stub, permissionRequests } = makePlanAgent({ + outcome: { outcome: "selected", optionId: "auto" }, + }); + await agent.newSession({ cwd: "/r" } as unknown as NewSessionRequest); + const done = agent.prompt({ + sessionId: "t", + prompt: [{ type: "text", text: "go" }], + } as unknown as PromptRequest); + await waitUntil( + () => stub.requests.filter((r) => r.method === "turn/start").length >= 1, + ); + + stub.emit("item/completed", { + item: { type: "plan", id: "p1", text: "# The plan" }, + }); + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + expect((await done).stopReason).toBe("end_turn"); + expect(permissionRequests).toHaveLength(0); + }); + + it("defers the cloud idle signal until the handoff settles", async () => { + let resolvePermission: (v: unknown) => void = () => {}; + const { agent, stub, extNotifications, permissionRequests } = makePlanAgent( + () => + new Promise((resolve) => { + resolvePermission = resolve; + }), + ); + const { done } = await startPlanTurn(agent, stub, { taskRunId: "run_1" }); + + stub.emit("item/completed", { + item: { type: "plan", id: "p1", text: "# The plan" }, + }); + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + await waitUntil(() => permissionRequests.length === 1); + + // The plan turn ended, but the prompt is still busy in the handoff — no idle yet. + expect( + extNotifications.filter((n) => n.method === "_posthog/turn_complete"), + ).toHaveLength(0); + + resolvePermission({ outcome: { outcome: "selected", optionId: "auto" } }); + await waitUntil( + () => stub.requests.filter((r) => r.method === "turn/start").length >= 2, + ); + stub.emit("turn/completed", { + turn: { id: "turn_2", status: "completed" }, + }); + expect((await done).stopReason).toBe("end_turn"); + + // Exactly one idle signal, from the implementation turn's completion. + const idle = extNotifications.filter( + (n) => n.method === "_posthog/turn_complete", + ); + expect(idle).toHaveLength(1); + expect((idle[0].params as { stopReason?: string }).stopReason).toBe( + "end_turn", + ); + }); + + it("cancel settles a pending handoff: prompt returns cancelled and idle emits once", async () => { + const { agent, stub, extNotifications, permissionRequests } = makePlanAgent( + // The approval UI never answers. + () => new Promise(() => {}), + ); + const { done } = await startPlanTurn(agent, stub, { taskRunId: "run_1" }); + + stub.emit("item/completed", { + item: { type: "plan", id: "p1", text: "# The plan" }, + }); + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + await waitUntil(() => permissionRequests.length === 1); + + await agent.cancel({ sessionId: "t" } as never); + expect((await done).stopReason).toBe("cancelled"); + expect(stub.requests.filter((r) => r.method === "turn/start")).toHaveLength( + 1, + ); + const idle = extNotifications.filter( + (n) => n.method === "_posthog/turn_complete", + ); + expect(idle).toHaveLength(1); + expect((idle[0].params as { stopReason?: string }).stopReason).toBe( + "cancelled", + ); + }); + + it("ignores an accept that arrives after cancellation", async () => { + let resolvePermission: (v: unknown) => void = () => {}; + const { agent, stub, sessionUpdates, permissionRequests } = makePlanAgent( + () => + new Promise((resolve) => { + resolvePermission = resolve; + }), + ); + const { done } = await startPlanTurn(agent, stub); + + stub.emit("item/completed", { + item: { type: "plan", id: "p1", text: "# The plan" }, + }); + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + await waitUntil(() => permissionRequests.length === 1); + + await agent.cancel({ sessionId: "t" } as never); + expect((await done).stopReason).toBe("cancelled"); + + // The approval UI answers too late: no implementation turn, no mode switch. + resolvePermission({ outcome: { outcome: "selected", optionId: "auto" } }); + for (let i = 0; i < 5; i++) { + await new Promise((r) => setImmediate(r)); + } + expect(stub.requests.filter((r) => r.method === "turn/start")).toHaveLength( + 1, + ); + expect( + sessionUpdates.filter( + (u) => + u.update?.sessionUpdate === "current_mode_update" && + u.update?.currentModeId === "auto", + ), + ).toHaveLength(0); + }); + + it("closes a pending handoff as cancelled and emits cancelled idle", async () => { + const { agent, stub, extNotifications, permissionRequests } = makePlanAgent( + () => new Promise(() => {}), + ); + const { done } = await startPlanTurn(agent, stub, { taskRunId: "run_1" }); + + stub.emit("item/completed", { + item: { type: "plan", id: "p1", text: "# The plan" }, + }); + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + await waitUntil(() => permissionRequests.length === 1); + + await agent.closeSession(); + expect((await done).stopReason).toBe("cancelled"); + const idle = extNotifications.filter( + (n) => n.method === "_posthog/turn_complete", + ); + expect(idle).toHaveLength(1); + expect((idle[0].params as { stopReason?: string }).stopReason).toBe( + "cancelled", + ); + }); + + it("keeps a newer restrictive mode when a stale handoff is accepted", async () => { + let resolvePermission: (v: unknown) => void = () => {}; + const { agent, stub, sessionUpdates, permissionRequests } = makePlanAgent( + () => + new Promise((resolve) => { + resolvePermission = resolve; + }), + ); + const { done } = await startPlanTurn(agent, stub); + + stub.emit("item/completed", { + item: { type: "plan", id: "p1", text: "# The plan" }, + }); + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + await waitUntil(() => permissionRequests.length === 1); + + await agent.setSessionConfigOption({ + configId: "mode", + value: "read-only", + sessionId: "t", + } as any); + expect((await done).stopReason).toBe("end_turn"); + + resolvePermission({ outcome: { outcome: "selected", optionId: "auto" } }); + for (let i = 0; i < 5; i++) await new Promise((r) => setImmediate(r)); + expect(stub.requests.filter((r) => r.method === "turn/start")).toHaveLength( + 1, + ); + expect(sessionUpdates).toContainEqual( + expect.objectContaining({ + update: { + sessionUpdate: "current_mode_update", + currentModeId: "read-only", + }, + }), + ); + expect( + sessionUpdates.filter( + (u) => + u.update?.sessionUpdate === "current_mode_update" && + u.update?.currentModeId === "auto", + ), + ).toHaveLength(0); + }); + + it("ignores a response selecting an option that was not offered", async () => { + const { agent, stub, sessionUpdates } = makePlanAgent({ + // Never offered by the handoff — must not switch modes or implement. + outcome: { outcome: "selected", optionId: "bypassPermissions" }, + }); + const { done } = await startPlanTurn(agent, stub); + + stub.emit("item/completed", { + item: { type: "plan", id: "p1", text: "# The plan" }, + }); + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + expect((await done).stopReason).toBe("end_turn"); + expect(stub.requests.filter((r) => r.method === "turn/start")).toHaveLength( + 1, + ); + expect( + sessionUpdates.filter( + (u) => + u.update?.sessionUpdate === "current_mode_update" && + u.update?.currentModeId !== "plan", + ), + ).toHaveLength(0); + }); + + it("explains how to recover when the approval prompt fails", async () => { + const { agent, stub, sessionUpdates } = makePlanAgent(() => + Promise.reject(new Error("permission UI unavailable")), + ); + const { done } = await startPlanTurn(agent, stub); + + stub.emit("item/completed", { + item: { type: "plan", id: "p1", text: "# The plan" }, + }); + stub.emit("turn/completed", { + turn: { id: "turn_1", status: "completed" }, + }); + expect((await done).stopReason).toBe("end_turn"); + + const texts = sessionUpdates.map( + (u) => (u.update?.content as { text?: string } | undefined)?.text ?? "", + ); + expect(texts.some((t) => t.includes("Still in Plan mode"))).toBe(true); + expect(stub.requests.filter((r) => r.method === "turn/start")).toHaveLength( + 1, + ); + }); }); diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts index 6761b6e0d7..c16a7ce2e8 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts @@ -12,6 +12,7 @@ import type { NewSessionResponse, PromptRequest, PromptResponse, + RequestPermissionResponse, ResumeSessionRequest, ResumeSessionResponse, SetSessionConfigOptionRequest, @@ -22,6 +23,7 @@ import { mcpToolKey, posthogToolMeta } from "@posthog/shared"; import { POSTHOG_NOTIFICATIONS } from "../../acp-extensions"; import { DEFAULT_CODEX_MODEL } from "../../gateway-models"; import type { ProcessSpawnedCallback } from "../../types"; +import { ALLOW_BYPASS } from "../../utils/common"; import { Logger } from "../../utils/logger"; import { nodeReadableToWebReadable, @@ -45,7 +47,7 @@ import { buildTurnCompleteParams, buildUsageBreakdownParams, } from "./ext-notifications"; -import { toCodexInput } from "./input"; +import { type CodexUserInput, toCodexInput } from "./input"; import { buildLocalToolsServer, type LocalToolsMeta } from "./local-tools-mcp"; import { type AppServerItem, @@ -144,6 +146,12 @@ export class CodexAppServerAgent extends BaseAcpAgent { private additionalDirectories?: string[]; /** The session workspace stays writable when extra roots are applied per turn. */ private workspaceDirectory?: string; + /** The in-flight turn's , streamed or completed (drives the implement handoff). */ + private planProposal?: { itemId: string; text: string }; + /** Idle signal deferred while the plan handoff keeps this prompt busy. */ + private deferredTurnComplete?: { usage: AccumulatedUsage }; + /** Settles the pending plan-approval race on cancel/close/preempting prompt. */ + private planHandoffCancel?: () => void; private readonly mcp = new McpManager(); private readonly turns = new TurnController(); private readonly usage = new UsageTracker(); @@ -457,7 +465,10 @@ export class CodexAppServerAgent extends BaseAcpAgent { const value = (params as { value?: unknown }).value; const { modeChanged } = this.config.setOption(configId, value); // collaborationMode rides the next turn/start, so a mode switch only needs current_mode_update here. - if (modeChanged) this.emitCurrentMode(this.config.mode); + if (modeChanged) { + this.emitCurrentMode(this.config.mode); + if (this.config.mode !== "plan") this.planHandoffCancel?.(); + } this.emitConfigOptions(); return { configOptions: this.config.options }; } @@ -535,6 +546,9 @@ export class CodexAppServerAgent extends BaseAcpAgent { } // Reopen the notification gate (a prior interrupt may have left session.cancelled set). this.session.cancelled = false; + // A new prompt while the plan handoff awaits approval implicitly declines it: + // settle the race so the previous prompt() returns and this one owns the turn. + this.planHandoffCancel?.(); // Prepend _meta.prContext (host PR-follow-up / Slack runs) to the FORWARDED prompt, // else codex cloud follow-ups lose the PR-review context. The echo omits it. const prContext = (params._meta as { prContext?: unknown } | undefined) @@ -585,8 +599,17 @@ export class CodexAppServerAgent extends BaseAcpAgent { throw new Error("prompt() called while a turn is already in progress"); } + const stopReason = await this.runTurn(input); + return { stopReason: await this.maybeOfferPlanImplementation(stopReason) }; + } + + /** Start one codex turn and await its completion. */ + private async runTurn(input: CodexUserInput[]): Promise { this.lastAgentMessage = ""; this.resetUsage(); + this.planProposal = undefined; + // A new turn owns the idle boundary; its own completion emits the signal. + this.deferredTurnComplete = undefined; const { completion, turn } = this.turns.begin(); try { const approvalPolicy = this.config.approvalPolicy(); @@ -612,12 +635,193 @@ export class CodexAppServerAgent extends BaseAcpAgent { // Constrain the final message to the task schema for parseable structured output. ...(this.jsonSchema ? { outputSchema: this.jsonSchema } : {}), }); - return { stopReason: await completion }; + return await completion; } finally { this.turns.finishPrompt(turn); } } + /** + * codex plan mode finalizes with a item and (by design) never + * asks "should I proceed?" — the client owns the handoff. Mirror the Claude + * ExitPlanMode flow: offer to implement; on accept switch the mode and run + * the implementation turn inside the same prompt() call. Plan feedback loops + * back into another plan turn, whose revised plan prompts again. + */ + private async maybeOfferPlanImplementation( + stopReason: StopReason, + ): Promise { + let reason = stopReason; + try { + while ( + reason === "end_turn" && + this.config.mode === "plan" && + this.planProposal && + !this.session.cancelled + ) { + const proposal = this.planProposal; + this.planProposal = undefined; + const outcome = await this.requestPlanImplementation(proposal); + // Re-check after the await: a cancel that raced the response wins, so a + // late accept can never start implementation on a cancelled prompt. + if (this.session.cancelled) { + reason = "cancelled"; + break; + } + // A picker change while approval was open owns the mode. Never let a + // stale approval overwrite it with a broader implementation mode. + if (this.config.mode !== "plan") break; + if (outcome.kind === "implement") { + this.config.setOption("mode", outcome.mode); + this.emitCurrentMode(outcome.mode); + this.emitConfigOptions(); + reason = await this.runFollowUpTurn(IMPLEMENT_PLAN_MESSAGE); + break; + } + if (outcome.kind === "feedback") { + reason = await this.runFollowUpTurn(outcome.feedback); + continue; + } + break; + } + } finally { + await this.flushDeferredTurnComplete(reason); + } + return reason; + } + + /** + * Emit the idle signal the handoff's plan turn deferred, unless a newer turn + * took over the boundary (a follow-up turn clears the deferral in runTurn and + * emits its own completion; a preempting prompt() does the same). + */ + private async flushDeferredTurnComplete(reason: StopReason): Promise { + const deferred = this.deferredTurnComplete; + this.deferredTurnComplete = undefined; + if (!deferred || this.turns.isPending) return; + await this.emitTurnCompleteSignal(reason, deferred.usage); + } + + /** Run an adapter-initiated turn, echoed as a user message like a host prompt. */ + private async runFollowUpTurn(text: string): Promise { + this.broadcastUserInput([{ type: "text", text }]); + return this.runTurn(toCodexInput([{ type: "text", text }])); + } + + /** + * The ExitPlanMode-style approval: a switch_mode tool call (routes the host + * to its plan-approval UI) whose option ids are codex mode ids. Cancel or a + * failed prompt stays in plan mode — never silently start implementing. + */ + private async requestPlanImplementation(proposal: { + itemId: string; + text: string; + }): Promise< + | { kind: "implement"; mode: "auto" | "full-access" } + | { kind: "feedback"; feedback: string } + | { kind: "stay" } + > { + const toolCallId = `${proposal.itemId}:implement`; + const toolCall = { + toolCallId, + title: "Ready to code?", + kind: "switch_mode", + content: [ + { + type: "content" as const, + content: { type: "text" as const, text: proposal.text }, + }, + ], + rawInput: { plan: proposal.text }, + }; + const options = [ + { + optionId: "auto", + name: 'Yes, and use "auto" mode', + kind: "allow_always" as const, + }, + ...(ALLOW_BYPASS + ? [ + { + optionId: "full-access", + name: "Yes, and auto-approve everything", + kind: "allow_always" as const, + }, + ] + : []), + { + optionId: "reject_with_feedback", + name: "No, and tell Codex what to do differently", + kind: "reject_once" as const, + _meta: { customInput: true }, + }, + ]; + // Accept only what was offered: a stale or malformed response must not + // select a mode that was hidden from the approval UI. + const offered = new Set(options.map((o) => o.optionId)); + const permission = this.client + .requestPermission({ + sessionId: this.sessionId, + toolCall, + options, + } as unknown as Parameters[0]) + .then( + (res: RequestPermissionResponse) => ({ failed: false as const, res }), + (err: unknown) => ({ failed: true as const, err }), + ); + // Race against cancellation so cancel/close (or a preempting prompt) settles + // the handoff instead of leaving prompt() pending on UI that may never answer. + const cancelled = new Promise((resolve) => { + this.planHandoffCancel = () => resolve(undefined); + }); + const settled = await Promise.race([permission, cancelled]); + this.planHandoffCancel = undefined; + if (!settled) return { kind: "stay" }; + if (settled.failed) { + this.logger.warn("plan implementation prompt failed; staying in plan", { + error: String(settled.err), + }); + // Without this the user sees nothing and Plan mode just sits there. + this.broadcastAgentText( + 'The plan approval prompt could not be shown. Still in Plan mode — switch the mode to Auto and send "Implement the plan." to proceed.', + ); + return { kind: "stay" }; + } + const response = settled.res; + if (this.session.cancelled || response.outcome.outcome !== "selected") { + return { kind: "stay" }; + } + const optionId = response.outcome.optionId; + if (!offered.has(optionId)) return { kind: "stay" }; + if (optionId === "auto") return { kind: "implement", mode: "auto" }; + // Double-gated: only ever offered under ALLOW_BYPASS, and re-checked here. + if (optionId === "full-access" && ALLOW_BYPASS) { + return { kind: "implement", mode: "full-access" }; + } + if (optionId === "reject_with_feedback") { + const feedback = (response as { _meta?: { customInput?: unknown } })._meta + ?.customInput; + if (typeof feedback === "string" && feedback.trim()) { + return { kind: "feedback", feedback: feedback.trim() }; + } + } + return { kind: "stay" }; + } + + /** Emit a plain agent message (user-facing status the model didn't produce). */ + private broadcastAgentText(text: string): void { + if (!this.sessionId) return; + void this.client + .sessionUpdate({ + sessionId: this.sessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text }, + }, + }) + .catch(() => undefined); + } + /** The mode's sandbox with the session's extra writable roots folded into workspaceWrite. */ private sandboxPolicyForTurn(): CodexSandboxPolicy | undefined { const policy = this.config.sandboxPolicy(); @@ -658,6 +862,9 @@ export class CodexAppServerAgent extends BaseAcpAgent { } protected async interrupt(): Promise { + // Settle a pending plan-approval race first so prompt() returns instead of + // waiting on approval UI that may never answer after a cancel. + this.planHandoffCancel?.(); // Stop the server, then finalize through the shared path so a cancelled turn still emits // the cloud idle signal (finalizeTurn claims idempotently). turn/interrupt requires BOTH // threadId and turnId (else -32600); skip the RPC when no turn started. @@ -676,6 +883,8 @@ export class CodexAppServerAgent extends BaseAcpAgent { async closeSession(): Promise { this.commandOutputs.clear(); this.session.abortController.abort(); + this.session.cancelled = true; + this.planHandoffCancel?.(); this.turns.close("cancelled"); this.session.settingsManager.dispose(); // Close the transport BEFORE kill() destroys the stdio streams (else close() blocks on @@ -741,6 +950,10 @@ export class CodexAppServerAgent extends BaseAcpAgent { if (method === APP_SERVER_NOTIFICATIONS.ITEM_COMPLETED) { this.captureAgentMessage(params); + this.capturePlanProposal(params); + } + if (method === APP_SERVER_NOTIFICATIONS.PLAN_DELTA) { + this.captureStreamedPlanProposal(params); } if (method === APP_SERVER_NOTIFICATIONS.TOKEN_USAGE_UPDATED) { @@ -821,6 +1034,32 @@ export class CodexAppServerAgent extends BaseAcpAgent { } } + /** Remember the turn's completed plan item (codex plan mode's authoritative ). */ + private capturePlanProposal(params: unknown): void { + const item = ( + params as { item?: { type?: string; id?: string; text?: string } } + )?.item; + if (item?.type === "plan" && typeof item.text === "string" && item.text) { + this.planProposal = { itemId: item.id ?? "codex-plan", text: item.text }; + } + } + + /** Accumulate the proposal stream used by codex builds that emit no completed plan item. */ + private captureStreamedPlanProposal(params: unknown): void { + const { itemId, delta } = params as { + itemId?: unknown; + delta?: unknown; + }; + if (typeof delta !== "string" || !delta) return; + const proposalId = + typeof itemId === "string" && itemId + ? itemId + : (this.planProposal?.itemId ?? "codex-plan"); + const previousText = + this.planProposal?.itemId === proposalId ? this.planProposal.text : ""; + this.planProposal = { itemId: proposalId, text: previousText + delta }; + } + /** Compaction started: emit `_posthog/status` so the host sets `isCompacting` (gates steer/queue). */ private emitCompactionStarted(): void { if (!this.sessionId) return; @@ -904,45 +1143,66 @@ export class CodexAppServerAgent extends BaseAcpAgent { ); } } - await this.emitTurnComplete(reason, usage, contextUsed); + if (this.willOfferPlanHandoff(reason)) { + // Defer the canonical idle signal: the handoff (and a possible implementation + // turn) keeps this prompt busy, and the cloud host treats turn_complete as + // idle — emitting it now would flush queued prompts into the handoff. + this.deferredTurnComplete = { usage }; + await this.emitUsageBreakdown(contextUsed); + } else { + await this.emitTurnCompleteSignal(reason, usage); + await this.emitUsageBreakdown(contextUsed); + } pending.resolve(reason); } - /** Emit cloud per-turn notifications: `_posthog/turn_complete` (only with a taskRunId) + the usage breakdown (always). */ - private async emitTurnComplete( + /** Whether maybeOfferPlanImplementation will run for a turn that ended this way. */ + private willOfferPlanHandoff(reason: StopReason): boolean { + return ( + reason === "end_turn" && + this.config.mode === "plan" && + !!this.planProposal && + !this.session.cancelled + ); + } + + /** Emit the cloud idle signal `_posthog/turn_complete` (only with a taskRunId). */ + private async emitTurnCompleteSignal( reason: StopReason, usage: AccumulatedUsage, + ): Promise { + if (!this.sessionId || !this.taskRunId) return; + await this.client + .extNotification( + POSTHOG_NOTIFICATIONS.TURN_COMPLETE, + buildTurnCompleteParams( + this.sessionId, + reason, + usage, + ) as unknown as Record, + ) + .catch((err) => + this.logger.warn("turn_complete extNotification failed", err), + ); + } + + /** Emit the `_posthog/usage_update` context breakdown for the host's token UI. */ + private async emitUsageBreakdown( contextUsed: number | undefined, ): Promise { - if (!this.sessionId) return; - if (this.taskRunId) { - await this.client - .extNotification( - POSTHOG_NOTIFICATIONS.TURN_COMPLETE, - buildTurnCompleteParams( - this.sessionId, - reason, - usage, - ) as unknown as Record, - ) - .catch((err) => - this.logger.warn("turn_complete extNotification failed", err), - ); - } - if (contextUsed !== undefined) { - await this.client - .extNotification( - POSTHOG_NOTIFICATIONS.USAGE_UPDATE, - buildUsageBreakdownParams( - this.sessionId, - this.usage.baselineBreakdown, - contextUsed, - ) as unknown as Record, - ) - .catch((err) => - this.logger.warn("usage breakdown extNotification failed", err), - ); - } + if (!this.sessionId || contextUsed === undefined) return; + await this.client + .extNotification( + POSTHOG_NOTIFICATIONS.USAGE_UPDATE, + buildUsageBreakdownParams( + this.sessionId, + this.usage.baselineBreakdown, + contextUsed, + ) as unknown as Record, + ) + .catch((err) => + this.logger.warn("usage breakdown extNotification failed", err), + ); } private handleServerClosed(): void { @@ -1113,6 +1373,9 @@ export class CodexAppServerAgent extends BaseAcpAgent { // BASELINE_TOKENS from codex-rs protocol.rs — the resident floor we can't attribute per-source. const CODEX_BASELINE_TOKENS = 12000; +// The implementation kickoff message, matching codex's own TUI plan handoff. +const IMPLEMENT_PLAN_MESSAGE = "Implement the plan."; + /** codex `TurnStatus` → ACP `StopReason`: interrupted → cancel, failed → refusal, else end. */ function mapTurnStopReason(status: string | undefined): StopReason { if (status === "interrupted") return "cancelled"; diff --git a/packages/agent/src/adapters/codex-app-server/mapping.test.ts b/packages/agent/src/adapters/codex-app-server/mapping.test.ts index 3c9d2ad8ce..776efab120 100644 --- a/packages/agent/src/adapters/codex-app-server/mapping.test.ts +++ b/packages/agent/src/adapters/codex-app-server/mapping.test.ts @@ -42,6 +42,22 @@ describe("mapAppServerNotification", () => { }); }); + it("streams a plan delta as an ACP agent_message_chunk", () => { + const result = mapAppServerNotification( + "s-1", + APP_SERVER_NOTIFICATIONS.PLAN_DELTA, + { itemId: "p1", delta: "## Plan\n" }, + ); + + expect(result).toEqual({ + sessionId: "s-1", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "## Plan\n" }, + }, + }); + }); + it("returns null when the delta is missing or empty", () => { expect( mapAppServerNotification( @@ -525,6 +541,21 @@ describe("mapHistoryItem", () => { ]); }); + it("replays a persisted plan item as an agent_message_chunk", () => { + expect( + mapHistoryItem("s-1", { type: "plan", id: "p1", text: "# The plan" }), + ).toEqual([ + { + sessionId: "s-1", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "# The plan" }, + }, + }, + ]); + expect(mapHistoryItem("s-1", { type: "plan", id: "p1" })).toEqual([]); + }); + it("replays an agentMessage as an agent_message_chunk", () => { expect( mapHistoryItem("s-1", { type: "agentMessage", id: "a1", text: "done" }), @@ -583,11 +614,8 @@ describe("mapHistoryItem", () => { }); }); - it("does not replay ephemeral reasoning/plan items", () => { + it("does not replay ephemeral reasoning items", () => { expect(mapHistoryItem("s-1", { type: "reasoning", id: "r1" })).toEqual([]); - expect( - mapHistoryItem("s-1", { type: "plan", id: "p1", text: "the plan" }), - ).toEqual([]); }); }); diff --git a/packages/agent/src/adapters/codex-app-server/mapping.ts b/packages/agent/src/adapters/codex-app-server/mapping.ts index fbfa99dee8..af9cd206a7 100644 --- a/packages/agent/src/adapters/codex-app-server/mapping.ts +++ b/packages/agent/src/adapters/codex-app-server/mapping.ts @@ -41,6 +41,18 @@ export function mapAppServerNotification( }, }; } + // Plan-mode proposal streaming as agent prose (codex strips it from agentMessage deltas). + case APP_SERVER_NOTIFICATIONS.PLAN_DELTA: { + const delta = readStringField(params, "delta"); + if (!delta) return null; + return { + sessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: delta }, + }, + }; + } case APP_SERVER_NOTIFICATIONS.TOKEN_USAGE_UPDATED: { // Context indicator: renderer reads `used`/`size`; detailed breakdown comes via `_posthog/usage_update`. const usage = readTokenUsage(params); @@ -232,7 +244,7 @@ function dynamicToolText(items: unknown): string | null { /** * Re-renders a persisted `ThreadItem` as the ACP updates a live stream would have produced, * so a reattaching host shows the full transcript. Tool items collapse to one completed - * `tool_call`; ephemeral items (reasoning, plan) are not replayed. + * `tool_call`; reasoning is not replayed. */ export function mapHistoryItem( sessionId: string, @@ -254,8 +266,20 @@ export function mapHistoryItem( ] : []; case "reasoning": - case "plan": return []; + // Replay the proposed plan as agent prose so a reattached host still shows it. + case "plan": + return item.text + ? [ + { + sessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: item.text }, + }, + }, + ] + : []; default: { const tool = describeTool(item); if (!tool || !item.id) return []; diff --git a/packages/agent/src/adapters/codex-app-server/protocol.ts b/packages/agent/src/adapters/codex-app-server/protocol.ts index bd948e9a34..ac6d3074af 100644 --- a/packages/agent/src/adapters/codex-app-server/protocol.ts +++ b/packages/agent/src/adapters/codex-app-server/protocol.ts @@ -29,6 +29,9 @@ export const APP_SERVER_NOTIFICATIONS = { REASONING_TEXT_DELTA: "item/reasoning/textDelta", // Default reasoning stream for gpt-5 models; raw textDelta is off by default, so without this the host sees no reasoning. REASONING_SUMMARY_TEXT_DELTA: "item/reasoning/summaryTextDelta", + // Plan-mode stream. codex strips the plan from agentMessage deltas, + // so without this the host sees nothing while the plan is written. + PLAN_DELTA: "item/plan/delta", TURN_PLAN_UPDATED: "turn/plan/updated", TURN_COMPLETED: "turn/completed", // Fatal turn error; `willRetry:false` means it won't recover on its own. diff --git a/packages/api-client/src/generated.ts b/packages/api-client/src/generated.ts index 94b8b7470f..4f16fdb00c 100644 --- a/packages/api-client/src/generated.ts +++ b/packages/api-client/src/generated.ts @@ -4277,7 +4277,7 @@ export namespace Schemas { }; export type CodeInviteRedeemRequest = { code: string }; export type CodexRuntimeAdapterEnum = "codex"; - export type CodexTaskRunCreateSchemaInitialPermissionModeEnum = "auto" | "read-only" | "full-access"; + export type CodexTaskRunCreateSchemaInitialPermissionModeEnum = "plan" | "auto" | "read-only" | "full-access"; export type CodexTaskRunCreateSchema = { mode?: (TaskExecutionModeEnum & unknown) | undefined; branch?: (string | null) | undefined; diff --git a/packages/api-client/src/posthog-client.test.ts b/packages/api-client/src/posthog-client.test.ts index 454546842c..baf36c1d26 100644 --- a/packages/api-client/src/posthog-client.test.ts +++ b/packages/api-client/src/posthog-client.test.ts @@ -77,7 +77,7 @@ describe("PostHogAPIClient", () => { ); }); - it("maps plan to read-only for cloud Codex runs", async () => { + it("preserves plan for cloud Codex runs", async () => { const client = new PostHogAPIClient( "http://localhost:8000", async () => "token", @@ -106,7 +106,7 @@ describe("PostHogAPIClient", () => { "/api/projects/{project_id}/tasks/{id}/run/", expect.objectContaining({ body: expect.objectContaining({ - initial_permission_mode: "read-only", + initial_permission_mode: "plan", }), }), ); diff --git a/packages/shared/src/execution-modes.test.ts b/packages/shared/src/execution-modes.test.ts index 6a765b56ad..b6be3890ab 100644 --- a/packages/shared/src/execution-modes.test.ts +++ b/packages/shared/src/execution-modes.test.ts @@ -6,7 +6,7 @@ describe("resolveCloudInitialPermissionMode", () => { ["codex", "auto", "auto"], ["codex", "read-only", "read-only"], ["codex", "full-access", "full-access"], - ["codex", "plan", "read-only"], + ["codex", "plan", "plan"], ["codex", "default", "auto"], ["codex", "acceptEdits", "auto"], ["codex", "bypassPermissions", "full-access"], diff --git a/packages/shared/src/execution-modes.ts b/packages/shared/src/execution-modes.ts index e7d7fde243..9b8c0ebf4e 100644 --- a/packages/shared/src/execution-modes.ts +++ b/packages/shared/src/execution-modes.ts @@ -45,6 +45,7 @@ const CLAUDE_CLOUD_PERMISSION_MODES = [ ] as const; const CODEX_CLOUD_PERMISSION_MODES = [ + "plan", "auto", "read-only", "full-access", @@ -65,19 +66,18 @@ function isCodexCloudPermissionMode( return (CODEX_CLOUD_PERMISSION_MODES as readonly string[]).includes(mode); } -// The cloud API's per-adapter mode enums are narrower than the local presets (no "plan" for codex); degrade to the nearest permission ceiling. +// Translate presets that only exist on the other adapter to the nearest permission ceiling. const CODEX_CLOUD_MODE_FALLBACKS: Record< - Exclude, + Exclude, CodexCloudPermissionMode > = { default: "auto", acceptEdits: "auto", - plan: "read-only", bypassPermissions: "full-access", }; const CLAUDE_CLOUD_MODE_FALLBACKS: Record< - Exclude, + Exclude, ClaudeCloudPermissionMode > = { "read-only": "plan",