From d9435c67d229f6385f8ffc95c3fae8143dd15853 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 2 Feb 2026 12:58:57 +0700 Subject: [PATCH] refactor: migrate Codex provider from MCP to app-server --- .../src/server/agent/provider-registry.ts | 6 +- .../codex-app-server-agent.e2e.test.ts | 40 + .../providers/codex-app-server-agent.test.ts | 710 +++++++ .../agent/providers/codex-app-server-agent.ts | 1723 +++++++++++++++++ 4 files changed, 2476 insertions(+), 3 deletions(-) create mode 100644 packages/server/src/server/agent/providers/codex-app-server-agent.e2e.test.ts create mode 100644 packages/server/src/server/agent/providers/codex-app-server-agent.test.ts create mode 100644 packages/server/src/server/agent/providers/codex-app-server-agent.ts diff --git a/packages/server/src/server/agent/provider-registry.ts b/packages/server/src/server/agent/provider-registry.ts index 733e51ba2..37327f2a3 100644 --- a/packages/server/src/server/agent/provider-registry.ts +++ b/packages/server/src/server/agent/provider-registry.ts @@ -7,7 +7,7 @@ import type { import type { Logger } from "pino"; import { ClaudeAgentClient } from "./providers/claude-agent.js"; -import { CodexMcpAgentClient } from "./providers/codex-mcp-agent.js"; +import { CodexAppServerAgentClient } from "./providers/codex-app-server-agent.js"; import { OpenCodeAgentClient, OpenCodeServerManager } from "./providers/opencode-agent.js"; import { @@ -32,7 +32,7 @@ export interface ProviderDefinition extends AgentProviderDefinition { export function buildProviderRegistry(logger: Logger): Record { const claudeClient = new ClaudeAgentClient({ logger }); - const codexClient = new CodexMcpAgentClient(logger); + const codexClient = new CodexAppServerAgentClient(logger); const opencodeClient = new OpenCodeAgentClient(logger); return { @@ -43,7 +43,7 @@ export function buildProviderRegistry(logger: Logger): Record d.id === "codex")!, - createClient: (logger: Logger) => new CodexMcpAgentClient(logger), + createClient: (logger: Logger) => new CodexAppServerAgentClient(logger), fetchModels: (options) => codexClient.listModels(options), }, opencode: { diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.e2e.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.e2e.test.ts new file mode 100644 index 000000000..b5330af11 --- /dev/null +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.e2e.test.ts @@ -0,0 +1,40 @@ +import { describe, expect, test } from "vitest"; +import { execFileSync } from "node:child_process"; +import os from "node:os"; +import path from "node:path"; +import { mkdtempSync } from "node:fs"; + +import { CodexAppServerAgentClient } from "./codex-app-server-agent.js"; +import { createTestLogger } from "../../../test-utils/test-logger.js"; + +const CODEX_TEST_MODEL = "gpt-5.1-codex-mini"; +const CODEX_TEST_REASONING_EFFORT = "low"; + +function isCodexInstalled(): boolean { + try { + const out = execFileSync("which", ["codex"], { encoding: "utf8" }).trim(); + return out.length > 0; + } catch { + return false; + } +} + +describe("Codex app-server provider (e2e)", () => { + test.runIf(isCodexInstalled())("lists models and runs a simple prompt", async () => { + const client = new CodexAppServerAgentClient(createTestLogger()); + const models = await client.listModels(); + expect(models.some((m) => m.id.includes("gpt-5.1-codex"))).toBe(true); + + const session = await client.createSession({ + provider: "codex", + cwd: mkdtempSync(path.join(os.tmpdir(), "codex-app-server-e2e-")), + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + const result = await session.run("Say hello in one sentence."); + expect(result.finalText.length).toBeGreaterThan(0); + await session.close(); + }, 30000); +}); diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts new file mode 100644 index 000000000..a2250e42f --- /dev/null +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -0,0 +1,710 @@ +import { describe, expect, test } from "vitest"; +import { execFileSync } from "node:child_process"; +import { mkdtempSync, readFileSync, rmSync, writeFileSync, mkdirSync, existsSync } from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { CodexAppServerAgentClient } from "./codex-app-server-agent.js"; +import { createTestLogger } from "../../../test-utils/test-logger.js"; +import type { + AgentPermissionRequest, + AgentPromptContentBlock, + AgentTimelineItem, +} from "../agent-sdk-types.js"; + +const CODEX_TEST_MODEL = "gpt-5.1-codex-mini"; +const CODEX_TEST_REASONING_EFFORT = "low"; + +function isCodexInstalled(): boolean { + try { + const out = execFileSync("which", ["codex"], { encoding: "utf8" }).trim(); + return out.length > 0; + } catch { + return false; + } +} + +function tmpCwd(prefix = "codex-app-server-e2e-"): string { + return mkdtempSync(path.join(os.tmpdir(), prefix)); +} + +function useTempCodexSessionDir(): () => void { + const codexSessionDir = tmpCwd("codex-sessions-"); + const prevSessionDir = process.env.CODEX_SESSION_DIR; + process.env.CODEX_SESSION_DIR = codexSessionDir; + return () => { + if (prevSessionDir === undefined) { + delete process.env.CODEX_SESSION_DIR; + } else { + process.env.CODEX_SESSION_DIR = prevSessionDir; + } + rmSync(codexSessionDir, { recursive: true, force: true }); + }; +} + +function hasShellCommand(item: AgentTimelineItem, commandFragment: string): boolean { + if (item.type !== "tool_call" || item.name !== "shell") return false; + const input = item.input as { command?: string } | undefined; + const command = input?.command ?? ""; + return command.includes(commandFragment); +} + +function hasApplyPatchFile(item: AgentTimelineItem, fileName: string): boolean { + if (item.type !== "tool_call" || item.name !== "apply_patch") return false; + const input = item.input as { files?: Array<{ path?: string }> } | undefined; + const output = item.output as + | { files?: Array<{ path?: string; patch?: string }>; diff?: string } + | undefined; + const inInput = (input?.files ?? []).some((file) => file?.path === fileName); + const inOutput = (output?.files ?? []).some((file) => file?.path === fileName); + const inDiff = typeof output?.diff === "string" && output.diff.includes(fileName); + return inInput || inOutput || inDiff; +} + +describe("Codex app-server provider (integration)", () => { + const logger = createTestLogger(); + + test.runIf(isCodexInstalled())("listModels returns live Codex models", async () => { + const client = new CodexAppServerAgentClient(logger); + const models = await client.listModels(); + expect(models.some((model) => model.id.includes("gpt-5.1-codex"))).toBe(true); + }, 30000); + + test.runIf(isCodexInstalled())("getRuntimeInfo reflects model + mode", async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-runtime-"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + await session.run("Reply with OK."); + const info = await session.getRuntimeInfo(); + await session.close(); + + expect(info.model).toBe(CODEX_TEST_MODEL); + expect(info.modeId).toBe("auto"); + expect(info.sessionId?.length).toBeGreaterThan(0); + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 30000); + + test.runIf(isCodexInstalled())("listCommands includes custom prompts and executeCommand runs them", async () => { + const cleanup = useTempCodexSessionDir(); + const codexHome = process.env.CODEX_HOME ?? path.join(os.homedir(), ".codex"); + const promptsDir = path.join(codexHome, "prompts"); + const promptPath = path.join(promptsDir, "test.md"); + const cwd = tmpCwd("codex-cmd-"); + + mkdirSync(promptsDir, { recursive: true }); + writeFileSync( + promptPath, + "---\ndescription: Test Prompt\n---\nReply with exactly: OK-$ARGUMENTS.", + "utf8" + ); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + const commands = await session.listCommands?.(); + expect(commands?.some((cmd) => cmd.name === "prompts:test")).toBe(true); + + const result = await session.executeCommand?.("prompts:test", "world"); + await session.close(); + + expect(result?.text.trim()).toContain("OK-world"); + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + rmSync(promptPath, { force: true }); + } + }, 60000); + + test.runIf(isCodexInstalled())("command approval flow requests permission and runs command", async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-cmd-approval-"); + const filePath = path.join(cwd, "permission.txt"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + approvalPolicy: "on-request", + paseoPromptInstructions: + "You must use the shell tool for command execution tasks. Do not answer without running the command.", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + let sawPermission = false; + let captured: AgentPermissionRequest | null = null; + let sawPermissionResolved = false; + const timelineItems: AgentTimelineItem[] = []; + + const events = session.stream( + [ + "You must use your shell tool to run the exact command", + "`printf \"ok\" > permission.txt`.", + "If you need approval before running it, request approval first.", + "After approval, run it and reply DONE.", + ].join(" ") + ); + + let failure: string | null = null; + for await (const event of events) { + if (event.type === "permission_requested" && event.request.name === "CodexBash") { + sawPermission = true; + captured = event.request; + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if ( + event.type === "permission_resolved" && + captured && + event.requestId === captured.id && + event.resolution.behavior === "allow" + ) { + sawPermissionResolved = true; + } + if (event.type === "timeline" && event.item.type === "tool_call") { + timelineItems.push(event.item); + } + if (event.type === "turn_failed") { + failure = event.error; + break; + } + if (event.type === "turn_completed") { + break; + } + } + + await session.close(); + + if (failure) { + throw new Error(failure); + } + if (captured) { + expect(sawPermissionResolved).toBe(true); + } + expect(sawPermission || timelineItems.length > 0).toBe(true); + expect( + timelineItems.some((item) => item.type === "tool_call" && item.name === "shell") + ).toBe(true); + expect(existsSync(filePath)).toBe(true); + expect(readFileSync(filePath, "utf8")).toContain("ok"); + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 60000); + + test.runIf(isCodexInstalled())( + "streams responses and maps shell + file change tool calls into timeline items", + async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-stream-"); + const shellFile = path.join(cwd, "shell.txt"); + const patchFile = path.join(cwd, "patch.txt"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "full-access", + approvalPolicy: "on-request", + paseoPromptInstructions: + "You must use shell for commands and apply_patch for file edits. Do not skip tool usage.", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + let sawAssistantMessage = false; + let sawShellTool = false; + let sawPatchTool = false; + let sawShellCompleted = false; + let sawPatchCompleted = false; + const timelineItems: AgentTimelineItem[] = []; + + const shellEvents = session.stream( + "Run the exact shell command `printf \"ok\" > shell.txt`. After it completes, reply SHELL_DONE." + ); + let failure: string | null = null; + for await (const event of shellEvents) { + if (event.type === "permission_requested") { + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if (event.type === "timeline") { + timelineItems.push(event.item); + if (event.item.type === "assistant_message") { + sawAssistantMessage = true; + } + if (hasShellCommand(event.item, "printf")) { + sawShellTool = true; + if (event.item.status === "completed") { + sawShellCompleted = true; + } + } + } + if (event.type === "turn_failed") { + failure = event.error; + break; + } + if (event.type === "turn_completed") { + break; + } + } + + if (failure) { + throw new Error(failure); + } + + const patch = [ + "*** Begin Patch", + "*** Add File: patch.txt", + "+patched", + "*** End Patch", + ].join("\\n"); + const patchEvents = session.stream( + [ + "Use the apply_patch tool and nothing else.", + "Do not use the shell tool for file changes.", + "Do not respond with any message until the apply_patch tool completes.", + "Apply the following patch exactly:", + patch, + "After it completes, reply PATCH_DONE.", + ].join("\\n") + ); + + for await (const event of patchEvents) { + if (event.type === "permission_requested") { + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if (event.type === "timeline") { + timelineItems.push(event.item); + if (event.item.type === "assistant_message") { + sawAssistantMessage = true; + } + if (hasApplyPatchFile(event.item, "patch.txt")) { + sawPatchTool = true; + if (event.item.status === "completed") { + sawPatchCompleted = true; + } + } + } + if (event.type === "turn_failed") { + failure = event.error; + break; + } + if (event.type === "turn_completed") { + break; + } + } + + await session.close(); + + if (failure) { + throw new Error(failure); + } + + expect(sawAssistantMessage).toBe(true); + expect(sawShellTool).toBe(true); + expect(sawPatchTool).toBe(true); + expect(sawShellCompleted || existsSync(shellFile)).toBe(true); + expect(sawPatchCompleted || existsSync(patchFile)).toBe(true); + expect(readFileSync(shellFile, "utf8")).toContain("ok"); + expect(readFileSync(patchFile, "utf8").trim()).toBe("patched"); + + const shellItem = timelineItems.find((item) => hasShellCommand(item, "printf")); + const patchItem = timelineItems.find((item) => hasApplyPatchFile(item, "patch.txt")); + expect(shellItem?.type).toBe("tool_call"); + expect(patchItem?.type).toBe("tool_call"); + if (shellItem?.type === "tool_call") { + expect(shellItem.name).toBe("shell"); + expect(shellItem.input).toBeTruthy(); + } + if (patchItem?.type === "tool_call") { + expect(patchItem.name).toBe("apply_patch"); + expect(patchItem.input || patchItem.output).toBeTruthy(); + } + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, + 120000 + ); + + test.runIf(isCodexInstalled())( + "interrupts long-running commands and emits a canceled turn", + async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-interrupt-"); + + let session: Awaited> | null = null; + let followupSession: Awaited> | null = + null; + let interruptAt: number | null = null; + let stoppedAt: number | null = null; + + try { + const client = new CodexAppServerAgentClient(logger); + session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + approvalPolicy: "on-request", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + const stream = session.stream( + "Run the exact shell command `sleep 60` using your shell tool and do not respond until it finishes." + ); + + let sawSleepCommand = false; + for await (const event of stream) { + if (event.type === "permission_requested") { + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if ( + event.type === "timeline" && + event.item.type === "tool_call" && + event.item.name === "shell" && + hasShellCommand(event.item, "sleep 60") + ) { + sawSleepCommand = true; + if (!interruptAt) { + interruptAt = Date.now(); + await session.interrupt(); + } + } + if (event.type === "turn_canceled") { + stoppedAt = Date.now(); + break; + } + if (event.type === "turn_completed" || event.type === "turn_failed") { + stoppedAt = Date.now(); + break; + } + } + + if (!interruptAt) { + throw new Error("Did not issue interrupt for long-running command"); + } + if (!stoppedAt) { + stoppedAt = Date.now(); + } + const latencyMs = stoppedAt - interruptAt; + expect(sawSleepCommand).toBe(true); + expect(latencyMs).toBeGreaterThanOrEqual(0); + expect(latencyMs).toBeLessThan(10_000); + + await session.close(); + session = null; + + followupSession = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + const followup = await followupSession.run("Reply OK and stop."); + expect(followup.finalText.toLowerCase()).toContain("ok"); + } finally { + await session?.close(); + await followupSession?.close(); + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, + 120000 + ); + + test.runIf(isCodexInstalled())( + "persists session metadata and resumes with history", + async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-resume-"); + const token = `ALPHA-${Date.now()}`; + + let session: Awaited> | null = null; + let resumed: Awaited> | null = null; + + try { + const client = new CodexAppServerAgentClient(logger); + session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + const first = await session.run(`Remember the word ${token} and reply ACK.`); + expect(first.finalText.toLowerCase()).toContain("ack"); + + const handle = session.describePersistence(); + expect(handle?.sessionId).toBeTruthy(); + expect(handle?.metadata?.threadId).toBe(handle?.sessionId); + + await session.close(); + session = null; + + resumed = await client.resumeSession(handle!); + const history: AgentTimelineItem[] = []; + for await (const event of resumed.streamHistory()) { + if (event.type === "timeline") { + history.push(event.item); + } + } + + expect( + history.some( + (item) => item.type === "assistant_message" || item.type === "user_message" + ) + ).toBe(true); + + const response = await resumed.run( + `Respond with the exact token ${token} and stop.` + ); + expect(response.finalText).toContain(token); + + const resumedHandle = resumed.describePersistence(); + expect(resumedHandle?.sessionId).toBe(handle?.sessionId); + expect(resumedHandle?.metadata?.threadId).toBe(handle?.sessionId); + } finally { + await session?.close(); + await resumed?.close(); + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, + 180000 + ); + + test.runIf(isCodexInstalled())( + "emits plan items and resolves collaboration mode mapping", + async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-plan-"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "read-only", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + const result = await Promise.race([ + session.run("Provide a concise 2-step plan and do not execute. Reply PLAN_DONE."), + new Promise((_, reject) => + setTimeout(() => reject(new Error("Plan run timed out")), 60000) + ), + ]); + + const info = await session.getRuntimeInfo(); + const sawCollaborationMode = Boolean(info.extra?.collaborationMode); + await session.close(); + + const sawTodo = result.timeline.some( + (item) => item.type === "todo" && item.items.length > 0 + ); + expect(sawTodo).toBe(true); + expect(sawCollaborationMode).toBe(true); + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, + 120000 + ); + + test.runIf(isCodexInstalled())("file change approval flow requests permission and applies change", async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-file-approval-"); + const targetPath = path.join(cwd, "approval-test.txt"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "full-access", + approvalPolicy: "on-request", + paseoPromptInstructions: + "You must use the apply_patch tool for file edits. Do not use the shell tool for file changes.", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + let sawPermission = false; + let captured: AgentPermissionRequest | null = null; + let sawPermissionResolved = false; + const timelineItems: AgentTimelineItem[] = []; + + const patch = [ + "*** Begin Patch", + "*** Add File: approval-test.txt", + "+ok", + "*** End Patch", + ].join("\\n"); + const events = session.stream( + [ + "Use the apply_patch tool and nothing else.", + "If you need approval before writing files, request approval first.", + "Do not respond with any message until the apply_patch tool completes.", + "Apply the following patch exactly:", + patch, + "After approval, reply FILE_DONE.", + ].join("\\n") + ); + + let failure: string | null = null; + for await (const event of events) { + if (event.type === "permission_requested" && event.request.name === "CodexFileChange") { + sawPermission = true; + captured = event.request; + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if ( + event.type === "permission_resolved" && + captured && + event.requestId === captured.id && + event.resolution.behavior === "allow" + ) { + sawPermissionResolved = true; + } + if (event.type === "timeline" && event.item.type === "tool_call") { + timelineItems.push(event.item); + } + if (event.type === "turn_failed") { + failure = event.error; + break; + } + if (event.type === "turn_completed") { + break; + } + } + + await session.close(); + + if (failure) { + throw new Error(failure); + } + if (captured) { + expect(sawPermissionResolved).toBe(true); + } + expect(sawPermission || timelineItems.length > 0).toBe(true); + expect( + timelineItems.some((item) => item.type === "tool_call" && item.name === "apply_patch") + ).toBe(true); + expect(readFileSync(targetPath, "utf8").trim()).toBe("ok"); + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 60000); + + test.runIf(isCodexInstalled())("tool approval flow requests user input for app tools", async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-tool-approval-"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + await session.connect(); + const rawClient = (session as any).client as { request: (method: string, params?: any) => Promise } | null; + if (!rawClient) { + throw new Error("Codex app-server client unavailable for app/list"); + } + const appsResult = await rawClient.request("app/list", { cursor: null, limit: 10 }); + const apps = Array.isArray(appsResult?.data) ? appsResult.data : []; + const app = apps.find((entry: any) => entry?.isAccessible) ?? apps[0]; + if (!app) { + await session.close(); + return; + } + + const input = [ + { type: "text", text: `$${app.id} Perform a minimal action and wait for approval if required. Reply with TOOL_DONE.` }, + { type: "mention", name: app.name ?? app.id, path: `app://${app.id}` }, + ] as unknown as AgentPromptContentBlock[]; + + let sawPermission = false; + let captured: AgentPermissionRequest | null = null; + let sawPermissionResolved = false; + let failure: string | null = null; + const timelineItems: AgentTimelineItem[] = []; + + for await (const event of session.stream(input)) { + if (event.type === "permission_requested" && event.request.name === "CodexTool") { + sawPermission = true; + captured = event.request; + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if ( + event.type === "permission_resolved" && + captured && + event.requestId === captured.id && + event.resolution.behavior === "allow" + ) { + sawPermissionResolved = true; + } + if (event.type === "timeline" && event.item.type === "tool_call") { + timelineItems.push(event.item); + } + if (event.type === "turn_failed") { + failure = event.error; + break; + } + if (event.type === "turn_completed") { + break; + } + } + + await session.close(); + + if (failure) { + throw new Error(failure); + } + if (captured) { + expect(sawPermissionResolved).toBe(true); + } + expect(sawPermission || timelineItems.length > 0).toBe(true); + if (captured) { + expect(Array.isArray(captured?.metadata?.questions)).toBe(true); + } + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 90000); +}); diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts new file mode 100644 index 000000000..8e1fed0f7 --- /dev/null +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -0,0 +1,1723 @@ +import type { + AgentCapabilityFlags, + AgentClient, + AgentCommandResult, + AgentMode, + AgentModelDefinition, + McpServerConfig, + AgentPermissionRequest, + AgentPermissionResponse, + AgentPromptContentBlock, + AgentPromptInput, + AgentRunOptions, + AgentRunResult, + AgentRuntimeInfo, + AgentSession, + AgentSessionConfig, + AgentSlashCommand, + AgentStreamEvent, + AgentTimelineItem, + AgentUsage, + ListModelsOptions, + ListPersistedAgentsOptions, + PersistedAgentDescriptor, + ToolCallTimelineItem, +} from "../agent-sdk-types.js"; +import type { Logger } from "pino"; + +import { execSync, spawn } from "node:child_process"; +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import fs from "node:fs/promises"; +import { Dirent } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import readline from "node:readline"; + +import { injectLeadingPaseoInstructionTag } from "../paseo-instructions-tag.js"; + +const DEFAULT_TIMEOUT_MS = 14 * 24 * 60 * 60 * 1000; +const CODEX_PROVIDER = "codex" as const; + +const CODEX_APP_SERVER_CAPABILITIES: AgentCapabilityFlags = { + supportsStreaming: true, + supportsSessionPersistence: true, + supportsDynamicModes: false, + supportsMcpServers: true, + supportsReasoningStream: true, + supportsToolInvocations: true, +}; + +const CODEX_MODES: AgentMode[] = [ + { + id: "read-only", + label: "Read Only", + description: + "Read files and answer questions. Manual approval required for edits, commands, or network ops.", + }, + { + id: "auto", + label: "Auto", + description: + "Edit files and run commands but still request approval before escalating scope.", + }, + { + id: "full-access", + label: "Full Access", + description: "Edit files, run commands, and access the network without additional prompts.", + }, +]; + +const DEFAULT_CODEX_MODE_ID = "auto"; + +const MODE_PRESETS: Record< + string, + { approvalPolicy: string; sandbox: string; networkAccess?: boolean } +> = { + "read-only": { + approvalPolicy: "on-request", + sandbox: "read-only", + }, + auto: { + approvalPolicy: "on-request", + sandbox: "workspace-write", + }, + "full-access": { + approvalPolicy: "never", + sandbox: "danger-full-access", + networkAccess: true, + }, +}; + +function validateCodexMode(modeId: string): void { + if (!(modeId in MODE_PRESETS)) { + const validModes = Object.keys(MODE_PRESETS).join(", "); + throw new Error( + `Invalid Codex mode "${modeId}". Valid modes are: ${validModes}` + ); + } +} + +function resolveCodexBinary(): string { + try { + const codexPath = execSync("which codex", { encoding: "utf8" }).trim(); + if (codexPath) { + return codexPath; + } + } catch { + // Fall through to error + } + throw new Error( + "Codex CLI not found. Please install codex globally: npm install -g @openai/codex" + ); +} + +function resolveCodexHomeDir(): string { + return process.env.CODEX_HOME ?? path.join(os.homedir(), ".codex"); +} + +function tokenizeCommandArgs(args: string): string[] { + const tokens: string[] = []; + let current = ""; + let quote: "'" | '"' | null = null; + for (let i = 0; i < args.length; i += 1) { + const ch = args[i]!; + if (quote) { + if (ch === quote) { + quote = null; + continue; + } + if (ch === "\\" && i + 1 < args.length) { + const next = args[i + 1]!; + if (next === quote || next === "\\" || next === "n" || next === "t") { + i += 1; + current += next === "n" ? "\n" : next === "t" ? "\t" : next; + continue; + } + } + current += ch; + continue; + } + + if (ch === "'" || ch === '"') { + quote = ch; + continue; + } + + if (/\s/.test(ch)) { + if (current) { + tokens.push(current); + current = ""; + } + continue; + } + current += ch; + } + if (current) { + tokens.push(current); + } + return tokens; +} + +function parseFrontMatter(markdown: string): { + frontMatter: Record; + body: string; +} { + const lines = markdown.split("\n"); + if (lines[0]?.trim() !== "---") { + return { frontMatter: {}, body: markdown }; + } + let end = -1; + for (let i = 1; i < lines.length; i += 1) { + if (lines[i]?.trim() === "---") { + end = i; + break; + } + } + if (end === -1) { + return { frontMatter: {}, body: markdown }; + } + const metaLines = lines.slice(1, end); + const body = lines.slice(end + 1).join("\n"); + const frontMatter: Record = {}; + for (const line of metaLines) { + const trimmed = line.trim(); + if (!trimmed || trimmed.startsWith("#")) { + continue; + } + const idx = trimmed.indexOf(":"); + if (idx <= 0) { + continue; + } + const key = trimmed.slice(0, idx).trim(); + let value = trimmed.slice(idx + 1).trim(); + value = value.replace(/^['"]/, "").replace(/['"]$/, ""); + if (key && value) { + frontMatter[key] = value; + } + } + return { frontMatter, body }; +} + +async function listCodexCustomPrompts(): Promise { + const codexHome = resolveCodexHomeDir(); + const promptsDir = path.join(codexHome, "prompts"); + let entries: Dirent[]; + try { + entries = await fs.readdir(promptsDir, { withFileTypes: true }); + } catch { + return []; + } + + const commands: AgentSlashCommand[] = []; + for (const entry of entries) { + if (!entry.isFile()) { + continue; + } + if (!entry.name.endsWith(".md")) { + continue; + } + const name = entry.name.slice(0, -".md".length); + if (!name) { + continue; + } + const fullPath = path.join(promptsDir, entry.name); + let content: string; + try { + content = await fs.readFile(fullPath, "utf8"); + } catch { + continue; + } + const parsed = parseFrontMatter(content); + const description = parsed.frontMatter["description"] ?? "Custom prompt"; + const argumentHint = + parsed.frontMatter["argument-hint"] ?? + parsed.frontMatter["argument_hint"] ?? + ""; + commands.push({ + name: `prompts:${name}`, + description, + argumentHint, + }); + } + return commands.sort((a, b) => a.name.localeCompare(b.name)); +} + +async function listCodexSkills(cwd: string): Promise { + const candidates: string[] = []; + candidates.push(path.join(cwd, ".codex", "skills")); + + const repoRoot = (() => { + try { + const output = execSync("git rev-parse --show-toplevel", { + cwd, + encoding: "utf8", + stdio: ["ignore", "pipe", "ignore"], + }); + const trimmed = output.trim(); + return trimmed ? trimmed : null; + } catch { + return null; + } + })(); + if (repoRoot) { + candidates.push(path.join(path.dirname(cwd), ".codex", "skills")); + candidates.push(path.join(repoRoot, ".codex", "skills")); + } + + candidates.push(path.join(resolveCodexHomeDir(), "skills")); + + const commandsByName = new Map(); + + for (const dir of candidates) { + let entries: Dirent[]; + try { + entries = await fs.readdir(dir, { withFileTypes: true }); + } catch { + continue; + } + + for (const entry of entries) { + if (!entry.isDirectory() && !entry.isSymbolicLink()) { + continue; + } + const skillDir = path.join(dir, entry.name); + const skillPath = path.join(skillDir, "SKILL.md"); + let content: string; + try { + content = await fs.readFile(skillPath, "utf8"); + } catch { + continue; + } + const { frontMatter } = parseFrontMatter(content); + const name = frontMatter["name"]; + const description = frontMatter["description"]; + if (!name || !description) { + continue; + } + if (!commandsByName.has(name)) { + commandsByName.set(name, { + name, + description, + argumentHint: "", + }); + } + } + } + + return Array.from(commandsByName.values()).sort((a, b) => + a.name.localeCompare(b.name) + ); +} + +function escapeRegExp(value: string): string { + return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); +} + +function expandCodexCustomPrompt(template: string, args: string | undefined): string { + const trimmedArgs = args ? args.trim() : ""; + const tokens = trimmedArgs ? tokenizeCommandArgs(trimmedArgs) : []; + const named: Record = {}; + const positional: string[] = []; + + for (const token of tokens) { + const idx = token.indexOf("="); + if (idx > 0) { + const key = token.slice(0, idx); + const value = token.slice(idx + 1); + if (key) { + named[key] = value; + continue; + } + } + positional.push(token); + } + + const dollarPlaceholder = "__CODEX_DOLLAR_PLACEHOLDER__"; + let out = template.split("$$").join(dollarPlaceholder); + + out = out.split("$ARGUMENTS").join(trimmedArgs); + + for (let i = 1; i <= 9; i += 1) { + const value = positional[i - 1] ?? ""; + out = out.split(`$${i}`).join(value); + } + + const namedKeys = Object.keys(named).sort((a, b) => b.length - a.length); + for (const key of namedKeys) { + const value = named[key] ?? ""; + const re = new RegExp(`\\$${escapeRegExp(key)}\\b`, "g"); + out = out.replace(re, value); + } + + out = out.split(dollarPlaceholder).join("$"); + return out; +} + +interface CodexMcpServerConfig { + url?: string; + http_headers?: Record; + command?: string; + args?: string[]; + env?: Record; + tool_timeout_sec?: number; +} + +function toCodexMcpConfig(config: McpServerConfig): CodexMcpServerConfig { + switch (config.type) { + case "stdio": + return { + command: config.command, + args: config.args, + env: config.env, + }; + case "http": + return { + url: config.url, + http_headers: config.headers, + }; + case "sse": + return { + url: config.url, + http_headers: config.headers, + }; + } +} + +class Pushable implements AsyncIterable { + private queue: T[] = []; + private resolvers: ((value: IteratorResult) => void)[] = []; + private closed = false; + + push(item: T) { + if (this.closed) { + return; + } + if (this.resolvers.length > 0) { + const resolve = this.resolvers.shift()!; + resolve({ value: item, done: false }); + } else { + this.queue.push(item); + } + } + + end() { + this.closed = true; + while (this.resolvers.length > 0) { + const resolve = this.resolvers.shift()!; + resolve({ value: undefined, done: true }); + } + } + + [Symbol.asyncIterator](): AsyncIterator { + return { + next: (): Promise> => { + if (this.queue.length > 0) { + const value = this.queue.shift()!; + return Promise.resolve({ value, done: false }); + } + if (this.closed) { + return Promise.resolve({ value: undefined, done: true }); + } + return new Promise>((resolve) => { + this.resolvers.push(resolve); + }); + }, + }; + } +} + +type JsonRpcRequest = { + id: number; + method: string; + params?: unknown; +}; + +type JsonRpcResponse = { + id: number; + result?: unknown; + error?: { code?: number; message: string }; +}; + +type JsonRpcNotification = { + method: string; + params?: unknown; +}; + +type JsonRpcMessage = JsonRpcRequest | JsonRpcResponse | JsonRpcNotification; + +type PendingRequest = { + resolve: (value: unknown) => void; + reject: (error: Error) => void; + timer: ReturnType; +}; + +type RequestHandler = (params: unknown) => Promise | unknown; + +type NotificationHandler = (method: string, params: unknown) => void; + +class CodexAppServerClient { + private readonly rl: readline.Interface; + private readonly pending = new Map(); + private readonly requestHandlers = new Map(); + private notificationHandler: NotificationHandler | null = null; + private nextId = 1; + private disposed = false; + private stderrBuffer = ""; + + constructor( + private readonly child: ChildProcessWithoutNullStreams, + private readonly logger: Logger + ) { + this.rl = readline.createInterface({ input: child.stdout }); + this.rl.on("line", (line) => this.handleLine(line)); + + child.stderr.on("data", (chunk) => { + this.stderrBuffer += chunk.toString(); + if (this.stderrBuffer.length > 8192) { + this.stderrBuffer = this.stderrBuffer.slice(-8192); + } + }); + + child.on("exit", (code, signal) => { + const message = + code === 0 && !signal + ? "Codex app-server exited" + : `Codex app-server exited with code ${code ?? "null"} and signal ${signal ?? "null"}`; + const error = new Error(`${message}\n${this.stderrBuffer}`.trim()); + for (const pending of this.pending.values()) { + clearTimeout(pending.timer); + pending.reject(error); + } + this.pending.clear(); + this.disposed = true; + }); + } + + setNotificationHandler(handler: NotificationHandler): void { + this.notificationHandler = handler; + } + + setRequestHandler(method: string, handler: RequestHandler): void { + this.requestHandlers.set(method, handler); + } + + request(method: string, params?: unknown, timeoutMs = DEFAULT_TIMEOUT_MS): Promise { + if (this.disposed) { + return Promise.reject(new Error("Codex app-server client is closed")); + } + const id = this.nextId++; + const payload: JsonRpcRequest = { id, method, params }; + const serialized = JSON.stringify(payload); + this.child.stdin.write(`${serialized}\n`); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + this.pending.delete(id); + reject(new Error(`Codex app-server request timed out for ${method}`)); + }, timeoutMs); + this.pending.set(id, { resolve, reject, timer }); + }); + } + + notify(method: string, params?: unknown): void { + if (this.disposed) { + return; + } + const payload: JsonRpcNotification = { method, params }; + this.child.stdin.write(`${JSON.stringify(payload)}\n`); + } + + async dispose(): Promise { + if (this.disposed) return; + this.disposed = true; + this.rl.close(); + try { + this.child.kill(); + } catch { + // ignore + } + } + + private async handleLine(line: string): Promise { + if (!line.trim()) return; + let msg: JsonRpcMessage; + try { + msg = JSON.parse(line); + } catch (error) { + this.logger.warn({ error, line }, "Failed to parse Codex app-server JSON"); + return; + } + + if (typeof (msg as JsonRpcResponse).id === "number") { + const id = (msg as JsonRpcResponse).id; + if ((msg as JsonRpcResponse).result !== undefined || (msg as JsonRpcResponse).error) { + const pending = this.pending.get(id); + if (!pending) return; + clearTimeout(pending.timer); + this.pending.delete(id); + if ((msg as JsonRpcResponse).error) { + pending.reject(new Error((msg as JsonRpcResponse).error?.message ?? "Unknown error")); + } else { + pending.resolve((msg as JsonRpcResponse).result); + } + return; + } + + // Server-initiated request + if (typeof (msg as JsonRpcRequest).method === "string") { + const request = msg as JsonRpcRequest; + const handler = this.requestHandlers.get(request.method); + try { + const result = handler ? await handler(request.params) : {}; + const response: JsonRpcResponse = { id: request.id, result }; + this.child.stdin.write(`${JSON.stringify(response)}\n`); + } catch (error) { + const response: JsonRpcResponse = { + id: request.id, + error: { message: error instanceof Error ? error.message : String(error) }, + }; + this.child.stdin.write(`${JSON.stringify(response)}\n`); + } + return; + } + } + + if (typeof (msg as JsonRpcNotification).method === "string") { + const notification = msg as JsonRpcNotification; + this.notificationHandler?.(notification.method, notification.params); + } + } +} + +function toAgentUsage(tokenUsage: unknown): AgentUsage | undefined { + if (!tokenUsage || typeof tokenUsage !== "object") return undefined; + const usage = tokenUsage as { last?: { inputTokens?: number; cachedInputTokens?: number; outputTokens?: number } }; + return { + inputTokens: usage.last?.inputTokens, + cachedInputTokens: usage.last?.cachedInputTokens, + outputTokens: usage.last?.outputTokens, + }; +} + +function createToolCallTimelineItem( + data: Omit +): AgentTimelineItem { + return { type: "tool_call", ...data }; +} + +function extractUserText(content: unknown): string | null { + if (!Array.isArray(content)) return null; + const parts: string[] = []; + for (const item of content) { + if (item && typeof item === "object") { + const obj = item as { type?: string; text?: string }; + if (obj.type === "text" && typeof obj.text === "string") { + parts.push(obj.text); + } + } + } + return parts.length > 0 ? parts.join("\n") : null; +} + +function extractContentText(content: unknown): string | null { + if (!Array.isArray(content)) return null; + const parts: string[] = []; + for (const item of content) { + if (item && typeof item === "object") { + const obj = item as { text?: string }; + if (typeof obj.text === "string") { + parts.push(obj.text); + } + } + } + return parts.length > 0 ? parts.join("\n") : null; +} + +function parsePlanTextToTodoItems(text: string): { text: string; completed: boolean }[] { + const lines = text + .split("\n") + .map((line) => line.trim()) + .filter((line) => line.length > 0); + if (lines.length === 0) { + return [{ text, completed: false }]; + } + return lines.map((line) => ({ + text: line.replace(/^[-*]\s+/, ""), + completed: false, + })); +} + +function planStepsToTodoItems(steps: Array<{ step: string; status: string }>): { + text: string; + completed: boolean; +}[] { + return steps.map((entry) => ({ + text: entry.step, + completed: entry.status === "completed", + })); +} + +function threadItemToTimeline( + item: any, + options?: { includeUserMessage?: boolean } +): AgentTimelineItem | null { + if (!item || typeof item !== "object") return null; + const includeUserMessage = options?.includeUserMessage ?? true; + switch (item.type) { + case "userMessage": { + if (!includeUserMessage) { + return null; + } + const text = extractUserText(item.content) ?? ""; + return { type: "user_message", text }; + } + case "agentMessage": { + return { type: "assistant_message", text: item.text ?? "" }; + } + case "plan": { + const text = item.text ?? ""; + const items = parsePlanTextToTodoItems(text); + return { type: "todo", items }; + } + case "reasoning": { + const summary = Array.isArray(item.summary) ? item.summary.join("\n") : ""; + const content = Array.isArray(item.content) ? item.content.join("\n") : ""; + const text = summary || content; + return text ? { type: "reasoning", text } : null; + } + case "commandExecution": { + const output = { + type: "command", + command: item.command, + output: item.aggregatedOutput ?? "", + exitCode: item.exitCode ?? undefined, + }; + return createToolCallTimelineItem({ + name: "shell", + status: item.status, + callId: item.id, + input: { command: item.command, cwd: item.cwd }, + output, + }); + } + case "fileChange": { + const files = Array.isArray(item.changes) + ? item.changes.map((change: any) => ({ + path: change.path, + kind: change.kind, + })) + : []; + const outputFiles = Array.isArray(item.changes) + ? item.changes.map((change: any) => ({ + path: change.path, + patch: change.diff, + kind: change.kind, + })) + : []; + return createToolCallTimelineItem({ + name: "apply_patch", + status: item.status, + callId: item.id, + input: { files }, + output: { files: outputFiles }, + }); + } + case "mcpToolCall": { + if (item.tool === "read_file") { + const pathValue = item.arguments?.path ?? item.arguments?.file_path ?? null; + const content = extractContentText(item.result?.content) ?? ""; + return createToolCallTimelineItem({ + name: "read_file", + status: item.status, + callId: item.id, + input: pathValue ? { path: pathValue } : item.arguments, + output: pathValue + ? { type: "read_file", path: pathValue, content } + : item.result ?? undefined, + error: item.error ?? undefined, + }); + } + return createToolCallTimelineItem({ + name: `${item.server}.${item.tool}`, + status: item.status, + callId: item.id, + input: item.arguments, + output: item.result ?? undefined, + error: item.error ?? undefined, + }); + } + case "webSearch": { + return createToolCallTimelineItem({ + name: "web_search", + status: "completed", + callId: item.id, + input: { query: item.query }, + output: item.action ?? undefined, + }); + } + default: + return null; + } +} + +function toSandboxPolicy(type: string, networkAccess?: boolean): Record { + switch (type) { + case "read-only": + return { type: "readOnly" }; + case "workspace-write": + return { type: "workspaceWrite", networkAccess: networkAccess ?? false }; + case "danger-full-access": + return { type: "dangerFullAccess" }; + default: + return { type: "workspaceWrite", networkAccess: networkAccess ?? false }; + } +} + +class CodexAppServerAgentSession implements AgentSession { + readonly provider = CODEX_PROVIDER; + readonly capabilities = CODEX_APP_SERVER_CAPABILITIES; + + private readonly logger: Logger; + private readonly config: AgentSessionConfig; + private currentMode: string; + private currentThreadId: string | null = null; + private currentTurnId: string | null = null; + private client: CodexAppServerClient | null = null; + private eventQueue: Pushable | null = null; + private cachedRuntimeInfo: AgentRuntimeInfo | null = null; + private historyPending = false; + private persistedHistory: AgentTimelineItem[] = []; + private pendingPermissions = new Map(); + private pendingPermissionHandlers = new Map< + string, + { + resolve: (value: unknown) => void; + kind: "command" | "file" | "tool"; + questions?: Array<{ id: string; options?: Array<{ label?: string; value?: string }> }>; + } + >(); + private resolvedPermissionRequests = new Set(); + private pendingAgentMessages = new Map(); + private pendingReasoning = new Map(); + private latestUsage: AgentUsage | undefined; + private connected = false; + private paseoInstructionsInjected = false; + private collaborationModes: Array<{ + name: string; + mode?: string | null; + model?: string | null; + reasoning_effort?: string | null; + developer_instructions?: string | null; + }> = []; + private resolvedCollaborationMode: { mode: string; settings: Record; name: string } | null = null; + private cachedSkills: Array<{ name: string; description: string; path: string }> = []; + + constructor( + config: AgentSessionConfig, + private readonly resumeHandle: { sessionId: string; metadata?: Record } | null, + logger: Logger + ) { + this.logger = logger.child({ module: "agent", provider: CODEX_PROVIDER }); + if (config.modeId === undefined) { + throw new Error("Codex agent requires modeId to be specified"); + } + validateCodexMode(config.modeId); + this.currentMode = config.modeId; + this.config = config; + + if (this.resumeHandle?.sessionId) { + this.currentThreadId = this.resumeHandle.sessionId; + this.historyPending = true; + this.paseoInstructionsInjected = true; + } + } + + get id(): string | null { + return this.currentThreadId; + } + + async connect(): Promise { + if (this.connected) return; + const binaryPath = resolveCodexBinary(); + const child = spawn(binaryPath, ["app-server"], { + stdio: ["pipe", "pipe", "pipe"], + env: process.env, + }); + this.client = new CodexAppServerClient(child, this.logger); + this.client.setNotificationHandler((method, params) => this.handleNotification(method, params)); + this.registerRequestHandlers(); + + await this.client.request("initialize", { + clientInfo: { + name: "paseo", + title: "Paseo", + version: "0.0.0", + }, + }); + this.client.notify("initialized", {}); + + await this.loadCollaborationModes(); + await this.loadSkills(); + + if (this.currentThreadId) { + await this.loadPersistedHistory(); + await this.ensureThreadLoaded(); + } + + this.connected = true; + } + + private async loadCollaborationModes(): Promise { + if (!this.client) return; + try { + const response = (await this.client.request("collaborationMode/list", {})) as { + data?: Array; + }; + const data = Array.isArray(response?.data) ? response.data : []; + this.collaborationModes = data.map((entry) => ({ + name: String(entry.name ?? ""), + mode: entry.mode ?? null, + model: entry.model ?? null, + reasoning_effort: entry.reasoning_effort ?? null, + developer_instructions: entry.developer_instructions ?? null, + })); + } catch (error) { + this.logger.debug({ error }, "Failed to load collaboration modes"); + this.collaborationModes = []; + } + this.resolvedCollaborationMode = this.resolveCollaborationMode(this.currentMode); + } + + private async loadSkills(): Promise { + if (!this.client) return; + try { + const response = (await this.client.request("skills/list", { + cwd: [this.config.cwd], + })) as { data?: Array }; + const entries = Array.isArray(response?.data) ? response.data : []; + const skills: Array<{ name: string; description: string; path: string }> = []; + for (const entry of entries) { + const list = Array.isArray(entry.skills) ? entry.skills : []; + for (const skill of list) { + if (!skill?.name || !skill?.path) continue; + skills.push({ + name: skill.name, + description: skill.description ?? skill.shortDescription ?? "Skill", + path: skill.path, + }); + } + } + this.cachedSkills = skills; + } catch (error) { + this.logger.debug({ error }, "Failed to load skills list"); + this.cachedSkills = []; + } + } + + private resolveCollaborationMode(modeId: string): { mode: string; settings: Record; name: string } | null { + if (this.collaborationModes.length === 0) return null; + const normalized = modeId.toLowerCase(); + const findByName = (predicate: (name: string) => boolean) => + this.collaborationModes.find((entry) => predicate(entry.name.toLowerCase())); + let match = + normalized === "read-only" + ? findByName((name) => name.includes("read") || name.includes("plan")) + : normalized === "full-access" + ? findByName((name) => name.includes("full") || name.includes("exec")) + : findByName((name) => name.includes("auto") || name.includes("code")); + if (!match) { + match = this.collaborationModes[0] ?? null; + } + if (!match) return null; + const settings: Record = {}; + if (match.model) settings.model = match.model; + if (match.reasoning_effort) settings.reasoning_effort = match.reasoning_effort; + if (match.developer_instructions) settings.developer_instructions = match.developer_instructions; + if (this.config.model) settings.model = this.config.model; + if (this.config.reasoningEffort) settings.reasoning_effort = this.config.reasoningEffort; + return { mode: match.mode ?? "code", settings, name: match.name }; + } + + private registerRequestHandlers(): void { + if (!this.client) return; + + this.client.setRequestHandler("item/commandExecution/requestApproval", (params) => + this.handleCommandApprovalRequest(params) + ); + this.client.setRequestHandler("item/fileChange/requestApproval", (params) => + this.handleFileChangeApprovalRequest(params) + ); + this.client.setRequestHandler("tool/requestUserInput", (params) => + this.handleToolApprovalRequest(params) + ); + } + + private async loadPersistedHistory(): Promise { + if (!this.client || !this.currentThreadId) return; + try { + const response = (await this.client.request("thread/read", { + threadId: this.currentThreadId, + includeTurns: true, + })) as { thread?: { turns?: Array<{ items?: any[] }> } }; + const thread = response?.thread; + if (!thread || !Array.isArray(thread.turns)) return; + const timeline: AgentTimelineItem[] = []; + for (const turn of thread.turns) { + const items = Array.isArray(turn.items) ? turn.items : []; + for (const item of items) { + const timelineItem = threadItemToTimeline(item); + if (timelineItem) { + timeline.push(timelineItem); + } + } + } + if (timeline.length > 0) { + this.persistedHistory = timeline; + this.historyPending = true; + } + } catch (error) { + this.logger.warn({ error }, "Failed to load Codex thread history"); + } + } + + private async ensureThreadLoaded(): Promise { + if (!this.client || !this.currentThreadId) return; + try { + const loaded = (await this.client.request("thread/loaded/list", {})) as { data?: string[] }; + const ids = Array.isArray(loaded?.data) ? loaded.data : []; + if (ids.includes(this.currentThreadId)) { + return; + } + await this.client.request("thread/resume", { threadId: this.currentThreadId }); + } catch (error) { + this.logger.warn({ error }, "Failed to resume Codex thread, starting new thread"); + this.currentThreadId = null; + await this.ensureThread(); + } + } + + async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise { + const events = this.stream(prompt, options); + const timeline: AgentTimelineItem[] = []; + let finalText = ""; + let usage: AgentUsage | undefined; + + for await (const event of events) { + if (event.type === "timeline") { + timeline.push(event.item); + if (event.item.type === "assistant_message") { + finalText = event.item.text; + } + } else if (event.type === "turn_completed") { + usage = event.usage; + } else if (event.type === "turn_failed") { + throw new Error(event.error); + } + } + + const info = await this.getRuntimeInfo(); + return { + sessionId: info.sessionId ?? "", + finalText, + usage, + timeline, + }; + } + + async *stream( + prompt: AgentPromptInput, + options?: AgentRunOptions + ): AsyncGenerator { + await this.connect(); + if (!this.client) return; + + const queue = new Pushable(); + this.eventQueue = queue; + + try { + if (this.currentThreadId) { + await this.ensureThreadLoaded(); + } else { + await this.ensureThread(); + } + const input = this.buildUserInput(prompt); + const preset = MODE_PRESETS[this.currentMode] ?? MODE_PRESETS[DEFAULT_CODEX_MODE_ID]; + const approvalPolicy = this.config.approvalPolicy ?? preset.approvalPolicy; + const sandboxPolicyType = this.config.sandboxMode ?? preset.sandbox; + + const params: Record = { + threadId: this.currentThreadId, + input, + approvalPolicy, + sandboxPolicy: toSandboxPolicy( + sandboxPolicyType, + typeof this.config.networkAccess === "boolean" + ? this.config.networkAccess + : preset.networkAccess + ), + }; + + if (this.config.model) { + params.model = this.config.model; + } + if (this.config.reasoningEffort) { + params.effort = this.config.reasoningEffort; + } + if (this.resolvedCollaborationMode) { + params.collaborationMode = { + mode: this.resolvedCollaborationMode.mode, + settings: this.resolvedCollaborationMode.settings, + }; + } + if (this.config.cwd) { + params.cwd = this.config.cwd; + } + if (options?.outputSchema) { + params.outputSchema = options.outputSchema; + } + + await this.client.request("turn/start", params); + + for await (const event of queue) { + yield event; + if ( + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled" + ) { + break; + } + } + } finally { + if (this.eventQueue === queue) { + this.eventQueue = null; + } + } + } + + async *streamHistory(): AsyncGenerator { + if (!this.historyPending || this.persistedHistory.length === 0) { + return; + } + const history = this.persistedHistory; + this.persistedHistory = []; + this.historyPending = false; + for (const item of history) { + yield { type: "timeline", provider: CODEX_PROVIDER, item }; + } + } + + async getRuntimeInfo(): Promise { + if (this.cachedRuntimeInfo) return { ...this.cachedRuntimeInfo }; + const info: AgentRuntimeInfo = { + provider: CODEX_PROVIDER, + sessionId: this.currentThreadId, + model: this.config.model ?? null, + modeId: this.currentMode ?? null, + extra: this.resolvedCollaborationMode + ? { collaborationMode: this.resolvedCollaborationMode.name } + : undefined, + }; + this.cachedRuntimeInfo = info; + return { ...info }; + } + + async getAvailableModes(): Promise { + return CODEX_MODES; + } + + async getCurrentMode(): Promise { + return this.currentMode ?? null; + } + + async setMode(modeId: string): Promise { + validateCodexMode(modeId); + this.currentMode = modeId; + this.resolvedCollaborationMode = this.resolveCollaborationMode(modeId); + this.cachedRuntimeInfo = null; + } + + getPendingPermissions(): AgentPermissionRequest[] { + return Array.from(this.pendingPermissions.values()); + } + + async respondToPermission( + requestId: string, + response: AgentPermissionResponse + ): Promise { + const pending = this.pendingPermissionHandlers.get(requestId); + if (!pending) { + throw new Error(`No pending Codex app-server permission request with id '${requestId}'`); + } + this.pendingPermissionHandlers.delete(requestId); + this.pendingPermissions.delete(requestId); + this.resolvedPermissionRequests.add(requestId); + + this.emitEvent({ + type: "permission_resolved", + provider: CODEX_PROVIDER, + requestId, + resolution: response, + }); + + if (pending.kind === "command") { + const decision = response.behavior === "allow" + ? "accept" + : response.interrupt + ? "cancel" + : "decline"; + pending.resolve({ decision }); + return; + } + + if (pending.kind === "file") { + const decision = response.behavior === "allow" + ? "accept" + : response.interrupt + ? "cancel" + : "decline"; + pending.resolve({ decision }); + return; + } + + // tool/requestUserInput + const answers: Record = {}; + const questions = pending.questions ?? []; + const decision = response.behavior === "allow" ? "accept" : response.interrupt ? "cancel" : "decline"; + for (const question of questions) { + let picked = decision; + const options = question.options ?? []; + if (options.length > 0) { + const byLabel = options.find((opt) => (opt.label ?? "").toLowerCase().includes(decision)); + const byValue = options.find((opt) => (opt.value ?? "").toLowerCase().includes(decision)); + const option = byLabel ?? byValue ?? options[0]!; + picked = option.value ?? option.label ?? decision; + } + answers[question.id] = { answers: [picked] }; + } + if (questions.length === 0) { + answers["default"] = { answers: [decision] }; + } + pending.resolve({ answers }); + } + + describePersistence(): { provider: typeof CODEX_PROVIDER; sessionId: string; nativeHandle: string; metadata: Record } | null { + if (!this.currentThreadId) return null; + return { + provider: CODEX_PROVIDER, + sessionId: this.currentThreadId, + nativeHandle: this.currentThreadId, + metadata: { + provider: CODEX_PROVIDER, + cwd: this.config.cwd, + title: this.config.title ?? null, + threadId: this.currentThreadId, + modeId: this.currentMode, + model: this.config.model ?? null, + reasoningEffort: this.config.reasoningEffort ?? null, + }, + }; + } + + async interrupt(): Promise { + if (!this.client || !this.currentThreadId || !this.currentTurnId) return; + try { + await this.client.request("turn/interrupt", { + threadId: this.currentThreadId, + turnId: this.currentTurnId, + }); + } catch (error) { + this.logger.warn({ error }, "Failed to interrupt Codex turn"); + } + } + + async close(): Promise { + for (const pending of this.pendingPermissionHandlers.values()) { + pending.resolve({ decision: "cancel" }); + } + this.pendingPermissionHandlers.clear(); + this.pendingPermissions.clear(); + this.resolvedPermissionRequests.clear(); + this.eventQueue?.end(); + this.eventQueue = null; + if (this.client) { + await this.client.dispose(); + } + this.client = null; + this.connected = false; + this.currentThreadId = null; + this.currentTurnId = null; + } + + async listCommands(): Promise { + const prompts = await listCodexCustomPrompts(); + if (!this.connected) { + await this.connect(); + } else { + await this.loadSkills(); + } + const appServerSkills = this.cachedSkills.map((skill) => ({ + name: skill.name, + description: skill.description, + argumentHint: "", + })); + const fallbackSkills = appServerSkills.length === 0 ? await listCodexSkills(this.config.cwd) : []; + return [...appServerSkills, ...fallbackSkills, ...prompts].sort((a, b) => + a.name.localeCompare(b.name) + ); + } + + async executeCommand(commandName: string, args?: string): Promise { + if (commandName.startsWith("prompts:")) { + const promptName = commandName.slice("prompts:".length); + const codexHome = resolveCodexHomeDir(); + const promptPath = path.join(codexHome, "prompts", `${promptName}.md`); + const raw = await fs.readFile(promptPath, "utf8"); + const parsed = parseFrontMatter(raw); + const expanded = expandCodexCustomPrompt(parsed.body, args); + const result = await this.run(expanded); + return { text: result.finalText, timeline: result.timeline, usage: result.usage }; + } + + if (!this.connected) { + await this.connect(); + } else { + await this.loadSkills(); + } + const skill = this.cachedSkills.find((entry) => entry.name === commandName); + if (skill) { + const input = [ + { type: "skill", name: skill.name, path: skill.path }, + ] as unknown as AgentPromptContentBlock[]; + if (args && args.trim().length > 0) { + input.push({ type: "text", text: args.trim() }); + } else { + input.push({ type: "text", text: `$${skill.name}` }); + } + const result = await this.run(input); + return { text: result.finalText, timeline: result.timeline, usage: result.usage }; + } + + const skillPrompt = args ? `$${commandName} ${args}` : `$${commandName}`; + const result = await this.run(skillPrompt); + return { text: result.finalText, timeline: result.timeline, usage: result.usage }; + } + + private async ensureThread(): Promise { + if (!this.client) return; + if (this.currentThreadId) return; + const preset = MODE_PRESETS[this.currentMode] ?? MODE_PRESETS[DEFAULT_CODEX_MODE_ID]; + const approvalPolicy = this.config.approvalPolicy ?? preset.approvalPolicy; + const sandbox = this.config.sandboxMode ?? preset.sandbox; + const innerConfig: Record = {}; + if (this.config.mcpServers) { + const mcpServers: Record = {}; + for (const [name, serverConfig] of Object.entries(this.config.mcpServers)) { + mcpServers[name] = toCodexMcpConfig(serverConfig); + } + innerConfig.mcp_servers = mcpServers; + } + if (this.config.extra?.codex) { + Object.assign(innerConfig, this.config.extra.codex); + } + const response = (await this.client.request("thread/start", { + model: this.config.model ?? null, + cwd: this.config.cwd ?? null, + approvalPolicy, + sandbox, + ...(Object.keys(innerConfig).length > 0 ? { config: innerConfig } : {}), + })) as { thread?: { id?: string } }; + const threadId = response?.thread?.id; + if (!threadId) { + throw new Error("Codex app-server did not return thread id"); + } + this.currentThreadId = threadId; + } + + private buildUserInput(prompt: AgentPromptInput): unknown[] { + if (typeof prompt === "string") { + const text = this.paseoInstructionsInjected + ? prompt + : injectLeadingPaseoInstructionTag(prompt, this.config.paseoPromptInstructions); + this.paseoInstructionsInjected = true; + return [{ type: "text", text }]; + } + const blocks = prompt as AgentPromptContentBlock[]; + if (this.paseoInstructionsInjected) { + return blocks; + } + this.paseoInstructionsInjected = true; + if (blocks.length === 0) { + return blocks; + } + const first = blocks[0]; + if (first && typeof first === "object" && (first as { type?: string }).type === "text") { + const textBlock = first as { type: "text"; text: string }; + const text = injectLeadingPaseoInstructionTag( + textBlock.text ?? "", + this.config.paseoPromptInstructions + ); + return [{ ...textBlock, text }, ...blocks.slice(1)]; + } + const injected = injectLeadingPaseoInstructionTag( + "", + this.config.paseoPromptInstructions + ); + if (injected.trim().length === 0) { + return blocks; + } + return [{ type: "text", text: injected }, ...blocks]; + } + + private emitEvent(event: AgentStreamEvent): void { + if (event.type === "timeline") { + if (event.item.type === "assistant_message") { + this.pendingAgentMessages.clear(); + } + } + this.eventQueue?.push(event); + } + + private handleNotification(method: string, params: unknown): void { + this.emitEvent({ type: "provider_event", provider: CODEX_PROVIDER, raw: { method, params } }); + + switch (method) { + case "thread/started": { + const threadId = (params as any)?.thread?.id; + if (threadId) { + this.currentThreadId = threadId; + this.emitEvent({ type: "thread_started", provider: CODEX_PROVIDER, sessionId: threadId }); + } + break; + } + case "turn/started": { + const turnId = (params as any)?.turn?.id; + if (turnId) { + this.currentTurnId = turnId; + } + this.emitEvent({ type: "turn_started", provider: CODEX_PROVIDER }); + break; + } + case "turn/completed": { + const turn = (params as any)?.turn; + const status = turn?.status; + if (status === "failed") { + const message = turn?.error?.message ?? "Codex turn failed"; + this.emitEvent({ type: "turn_failed", provider: CODEX_PROVIDER, error: message }); + } else if (status === "interrupted") { + this.emitEvent({ type: "turn_canceled", provider: CODEX_PROVIDER, reason: "interrupted" }); + } else { + this.emitEvent({ type: "turn_completed", provider: CODEX_PROVIDER, usage: this.latestUsage }); + } + this.eventQueue?.end(); + break; + } + case "turn/plan/updated": { + const steps = (params as any)?.plan; + if (Array.isArray(steps)) { + const items = planStepsToTodoItems( + steps.map((entry: any) => ({ + step: entry.step ?? "", + status: entry.status ?? "pending", + })) + ); + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: { type: "todo", items }, + }); + } + break; + } + case "turn/diff/updated": { + const diff = (params as any)?.diff; + if (typeof diff === "string" && diff.trim().length > 0) { + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: createToolCallTimelineItem({ + name: "apply_patch", + status: "running", + output: { diff }, + }), + }); + } + break; + } + case "thread/tokenUsage/updated": { + const tokenUsage = (params as any)?.tokenUsage; + this.latestUsage = toAgentUsage(tokenUsage); + break; + } + case "item/agentMessage/delta": { + const itemId = (params as any)?.itemId; + const delta = (params as any)?.delta; + if (itemId && typeof delta === "string") { + const prev = this.pendingAgentMessages.get(itemId) ?? ""; + this.pendingAgentMessages.set(itemId, prev + delta); + } + break; + } + case "item/reasoning/summaryTextDelta": { + const itemId = (params as any)?.itemId; + const delta = (params as any)?.delta; + if (itemId && typeof delta === "string") { + const prev = this.pendingReasoning.get(itemId) ?? []; + prev.push(delta); + this.pendingReasoning.set(itemId, prev); + } + break; + } + case "item/completed": { + const item = (params as any)?.item; + const timelineItem = threadItemToTimeline(item, { includeUserMessage: false }); + if (timelineItem) { + if (timelineItem.type === "assistant_message" && item?.id) { + const buffered = this.pendingAgentMessages.get(item.id); + if (buffered && buffered.length > 0) { + timelineItem.text = buffered; + } + } + if (timelineItem.type === "reasoning" && item?.id) { + const buffered = this.pendingReasoning.get(item.id); + if (buffered && buffered.length > 0) { + timelineItem.text = buffered.join(""); + } + } + this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); + } + break; + } + case "item/started": { + const item = (params as any)?.item; + const timelineItem = threadItemToTimeline(item, { includeUserMessage: false }); + if (timelineItem && timelineItem.type === "tool_call") { + this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); + } + break; + } + default: + break; + } + } + + private handleCommandApprovalRequest(params: unknown): Promise { + const parsed = params as { + itemId: string; + threadId: string; + turnId: string; + command?: string | null; + cwd?: string | null; + reason?: string | null; + }; + const requestId = `permission-${parsed.itemId}`; + const title = parsed.command ? `Run command: ${parsed.command}` : "Run command"; + const request: AgentPermissionRequest = { + id: requestId, + provider: CODEX_PROVIDER, + name: "CodexBash", + kind: "tool", + title, + description: parsed.reason ?? undefined, + input: { + command: parsed.command ?? undefined, + cwd: parsed.cwd ?? undefined, + }, + metadata: { + itemId: parsed.itemId, + threadId: parsed.threadId, + turnId: parsed.turnId, + }, + }; + this.pendingPermissions.set(requestId, request); + this.emitEvent({ type: "permission_requested", provider: CODEX_PROVIDER, request }); + return new Promise((resolve) => { + this.pendingPermissionHandlers.set(requestId, { resolve, kind: "command" }); + }); + } + + private handleFileChangeApprovalRequest(params: unknown): Promise { + const parsed = params as { itemId: string; threadId: string; turnId: string; reason?: string | null }; + const requestId = `permission-${parsed.itemId}`; + const request: AgentPermissionRequest = { + id: requestId, + provider: CODEX_PROVIDER, + name: "CodexFileChange", + kind: "tool", + title: "Apply file changes", + description: parsed.reason ?? undefined, + metadata: { + itemId: parsed.itemId, + threadId: parsed.threadId, + turnId: parsed.turnId, + }, + }; + this.pendingPermissions.set(requestId, request); + this.emitEvent({ type: "permission_requested", provider: CODEX_PROVIDER, request }); + return new Promise((resolve) => { + this.pendingPermissionHandlers.set(requestId, { resolve, kind: "file" }); + }); + } + + private handleToolApprovalRequest(params: unknown): Promise { + const parsed = params as { itemId: string; threadId: string; turnId: string; questions: any[] }; + const requestId = `permission-${parsed.itemId}`; + const request: AgentPermissionRequest = { + id: requestId, + provider: CODEX_PROVIDER, + name: "CodexTool", + kind: "tool", + title: "Tool action requires approval", + description: undefined, + metadata: { + itemId: parsed.itemId, + threadId: parsed.threadId, + turnId: parsed.turnId, + questions: parsed.questions, + }, + }; + this.pendingPermissions.set(requestId, request); + this.emitEvent({ type: "permission_requested", provider: CODEX_PROVIDER, request }); + return new Promise((resolve) => { + this.pendingPermissionHandlers.set(requestId, { + resolve, + kind: "tool", + questions: Array.isArray(parsed.questions) ? parsed.questions : [], + }); + }); + } +} + +export class CodexAppServerAgentClient implements AgentClient { + readonly provider = CODEX_PROVIDER; + readonly capabilities = CODEX_APP_SERVER_CAPABILITIES; + + constructor(private readonly logger: Logger) {} + + async createSession(config: AgentSessionConfig): Promise { + const sessionConfig: AgentSessionConfig = { ...config, provider: CODEX_PROVIDER }; + const session = new CodexAppServerAgentSession(sessionConfig, null, this.logger); + await session.connect(); + return session; + } + + async resumeSession(handle: { sessionId: string; metadata?: Record }, overrides?: Partial): Promise { + const storedConfig = (handle.metadata ?? {}) as AgentSessionConfig; + const merged: AgentSessionConfig = { + ...storedConfig, + ...overrides, + provider: CODEX_PROVIDER, + cwd: overrides?.cwd ?? storedConfig.cwd ?? process.cwd(), + }; + const session = new CodexAppServerAgentSession(merged, handle, this.logger); + await session.connect(); + return session; + } + + async listPersistedAgents( + options?: ListPersistedAgentsOptions + ): Promise { + const binaryPath = resolveCodexBinary(); + const child = spawn(binaryPath, ["app-server"], { + stdio: ["pipe", "pipe", "pipe"], + }); + const client = new CodexAppServerClient(child, this.logger); + + try { + await client.request("initialize", { + clientInfo: { name: "paseo", title: "Paseo", version: "0.0.0" }, + }); + client.notify("initialized", {}); + + const limit = options?.limit ?? 20; + const response = (await client.request("thread/list", { limit })) as { + data?: Array; + }; + const threads = Array.isArray(response?.data) ? response.data : []; + const descriptors: PersistedAgentDescriptor[] = []; + + for (const thread of threads.slice(0, limit)) { + const threadId = thread.id; + const cwd = thread.cwd ?? process.cwd(); + const title = thread.preview ?? null; + let timeline: AgentTimelineItem[] = []; + try { + const read = (await client.request("thread/read", { + threadId, + includeTurns: true, + })) as { thread?: { turns?: Array<{ items?: any[] }> } }; + const turns = read.thread?.turns ?? []; + const items: AgentTimelineItem[] = []; + for (const turn of turns) { + for (const item of turn.items ?? []) { + const timelineItem = threadItemToTimeline(item); + if (timelineItem) items.push(timelineItem); + } + } + timeline = items; + } catch { + timeline = []; + } + + descriptors.push({ + provider: CODEX_PROVIDER, + sessionId: threadId, + cwd, + title, + lastActivityAt: new Date((thread.updatedAt ?? thread.createdAt ?? 0) * 1000), + persistence: { + provider: CODEX_PROVIDER, + sessionId: threadId, + nativeHandle: threadId, + metadata: { + provider: CODEX_PROVIDER, + cwd, + title, + threadId, + }, + }, + timeline, + }); + } + + return descriptors; + } finally { + await client.dispose(); + } + } + + async listModels(_options?: ListModelsOptions): Promise { + const binaryPath = resolveCodexBinary(); + const child = spawn(binaryPath, ["app-server"], { + stdio: ["pipe", "pipe", "pipe"], + }); + const client = new CodexAppServerClient(child, this.logger); + + try { + await client.request("initialize", { + clientInfo: { + name: "paseo", + title: "Paseo", + version: "0.0.0", + }, + }); + client.notify("initialized", {}); + + const response = (await client.request("model/list", {})) as { data?: Array }; + const models = Array.isArray(response?.data) ? response.data : []; + return models.map((model) => ({ + provider: CODEX_PROVIDER, + id: model.id, + label: model.displayName, + description: model.description, + isDefault: model.isDefault, + metadata: { + model: model.model, + defaultReasoningEffort: model.defaultReasoningEffort, + supportedReasoningEfforts: model.supportedReasoningEfforts, + }, + })); + } finally { + await client.dispose(); + } + } +}