From c40de3bd49a782209525474baab124b5e86b4189 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 25 Dec 2025 15:42:22 +0700 Subject: [PATCH] Remove deprecated Codex SDK provider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Delete codex-agent.ts/test.ts/unit.test.ts and related files. Codex MCP is now the only Codex provider, registered for both "codex" and "codex-mcp" provider IDs in bootstrap.ts. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- .../agent/providers/codex-agent.test.ts | 1080 -------- .../src/server/agent/providers/codex-agent.ts | 2247 ----------------- .../agent/providers/codex-agent.unit.test.ts | 28 - packages/server/test-results-after.txt | 16 - packages/server/test-sdk-spawn.ts | 34 - plan.md | 19 + test-sdk-spawn.ts | 34 - 7 files changed, 19 insertions(+), 3439 deletions(-) delete mode 100644 packages/server/src/server/agent/providers/codex-agent.test.ts delete mode 100644 packages/server/src/server/agent/providers/codex-agent.ts delete mode 100644 packages/server/src/server/agent/providers/codex-agent.unit.test.ts delete mode 100644 packages/server/test-results-after.txt delete mode 100644 packages/server/test-sdk-spawn.ts delete mode 100644 test-sdk-spawn.ts diff --git a/packages/server/src/server/agent/providers/codex-agent.test.ts b/packages/server/src/server/agent/providers/codex-agent.test.ts deleted file mode 100644 index d0447055f..000000000 --- a/packages/server/src/server/agent/providers/codex-agent.test.ts +++ /dev/null @@ -1,1080 +0,0 @@ -import { describe, expect, test } from "vitest"; -import { createServer as createHTTPServer } from "http"; -import { randomUUID } from "node:crypto"; -import { mkdtempSync, rmSync, promises as fs } from "node:fs"; -import type { Dirent } from "node:fs"; -import os from "node:os"; -import path from "node:path"; -import express from "express"; -import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; -import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; - -import { CodexAgentClient, isSyntheticRolloutUserMessage } from "./codex-agent.js"; -import { - hydrateStreamState, - isAgentToolCallItem, - type AgentToolCallItem, - type StreamItem, -} from "../../../../../app/src/types/stream.js"; -import { - serializeAgentStreamEvent, - type AgentStreamEventPayload, -} from "../../messages.js"; -import type { - AgentProvider, - AgentSessionConfig, - AgentStreamEvent, - AgentPermissionRequest, - AgentPersistenceHandle, - AgentTimelineItem, -} from "../agent-sdk-types.js"; -import { AgentManager } from "../agent-manager.js"; -import { AgentRegistry } from "../agent-registry.js"; -import { createAgentMcpServer } from "../mcp-server.js"; - -function tmpCwd(): string { - return mkdtempSync(path.join(os.tmpdir(), "codex-agent-e2e-")); -} - -function useTempCodexSessionDir(): () => void { - const prevSessionDir = process.env.CODEX_SESSION_DIR; - const prevHome = process.env.CODEX_HOME; - return () => { - if (prevSessionDir === undefined) { - delete process.env.CODEX_SESSION_DIR; - } else { - process.env.CODEX_SESSION_DIR = prevSessionDir; - } - if (prevHome === undefined) { - delete process.env.CODEX_HOME; - } else { - process.env.CODEX_HOME = prevHome; - } - }; -} - -function log(message: string): void { - console.info(`[CodexAgentTest] ${message}`); -} - -type ToolCallItem = Extract; - -type AgentMcpServerHandle = { - url: string; - close: () => Promise; -}; - -async function startAgentMcpServer(): Promise { - const app = express(); - app.use(express.json()); - const httpServer = createHTTPServer(app); - - const registryDir = await fs.mkdtemp(path.join(os.tmpdir(), "agent-mcp-registry-")); - const registryPath = path.join(registryDir, "agents.json"); - const agentRegistry = new AgentRegistry(registryPath); - const agentManager = new AgentManager({ - clients: {}, - registry: agentRegistry, - }); - - let allowedHosts: string[] | undefined; - const agentMcpTransports = new Map(); - - const createAgentMcpTransport = async (callerAgentId?: string) => { - const agentMcpServer = await createAgentMcpServer({ - agentManager, - agentRegistry, - callerAgentId, - }); - - const transport = new StreamableHTTPServerTransport({ - sessionIdGenerator: () => randomUUID(), - onsessioninitialized: (sessionId) => { - agentMcpTransports.set(sessionId, transport); - log(`Agent MCP session initialized: ${sessionId}`); - }, - onsessionclosed: (sessionId) => { - agentMcpTransports.delete(sessionId); - log(`Agent MCP session closed: ${sessionId}`); - }, - enableDnsRebindingProtection: true, - ...(allowedHosts ? { allowedHosts } : {}), - }); - - transport.onclose = () => { - if (transport.sessionId) { - agentMcpTransports.delete(transport.sessionId); - } - }; - transport.onerror = (error) => { - console.error("[Agent MCP] Transport error:", error); - }; - - await agentMcpServer.connect(transport); - return transport; - }; - - const handleAgentMcpRequest: express.RequestHandler = async (req, res) => { - try { - const sessionId = req.header("mcp-session-id"); - let transport = sessionId ? agentMcpTransports.get(sessionId) : undefined; - - if (!transport) { - if (req.method !== "POST") { - res.status(400).json({ - jsonrpc: "2.0", - error: { code: -32000, message: "Missing or invalid MCP session" }, - id: null, - }); - return; - } - if (!isInitializeRequest(req.body)) { - res.status(400).json({ - jsonrpc: "2.0", - error: { code: -32000, message: "Initialization request expected" }, - id: null, - }); - return; - } - log("Agent MCP initialize request received"); - const callerAgentIdRaw = req.query.callerAgentId; - const callerAgentId = - typeof callerAgentIdRaw === "string" - ? callerAgentIdRaw - : Array.isArray(callerAgentIdRaw) - ? callerAgentIdRaw[0] - : undefined; - transport = await createAgentMcpTransport(callerAgentId); - } - - await transport.handleRequest(req as any, res as any, req.body); - } catch (error) { - console.error("[Agent MCP] Failed to handle request:", error); - if (!res.headersSent) { - res.status(500).json({ - jsonrpc: "2.0", - error: { code: -32603, message: "Internal MCP server error" }, - id: null, - }); - } - } - }; - - app.post("/mcp/agents", handleAgentMcpRequest); - app.get("/mcp/agents", handleAgentMcpRequest); - app.delete("/mcp/agents", handleAgentMcpRequest); - - const port = await new Promise((resolve) => { - httpServer.listen(0, () => { - const address = httpServer.address(); - resolve(typeof address === "object" && address ? address.port : 0); - }); - }); - - allowedHosts = [`127.0.0.1:${port}`, `localhost:${port}`]; - const url = `http://127.0.0.1:${port}/mcp/agents`; - - return { - url, - close: async () => { - await new Promise((resolve) => httpServer.close(() => resolve())); - rmSync(registryDir, { recursive: true, force: true }); - }, - }; -} - -function commandTextFromInput(input: unknown): string | null { - if (!input || typeof input !== "object") { - return null; - } - const command = (input as { command?: unknown }).command; - if (typeof command === "string" && command.length > 0) { - return command; - } - if (Array.isArray(command)) { - const tokens = command.filter((value): value is string => typeof value === "string"); - if (tokens.length > 0) { - return tokens.join(" "); - } - } - return null; -} - -function isSleepCommandToolCall(item: ToolCallItem): boolean { - const display = typeof item.displayName === "string" ? item.displayName.toLowerCase() : ""; - if (display.includes("sleep 60")) { - return true; - } - const inputText = commandTextFromInput(item.input)?.toLowerCase() ?? ""; - return inputText.includes("sleep 60"); -} - -describe("CodexAgentClient (SDK integration)", () => { - test( - "responds with text", - async () => { - const cwd = tmpCwd(); - const restoreSessionDir = useTempCodexSessionDir(); - const client = new CodexAgentClient(); - const config: AgentSessionConfig = { provider: "codex", cwd, modeId: "full-access" }; - let session: Awaited> | null = null; - try { - session = await client.createSession(config); - log("Running single-turn acknowledgment test"); - - const result = await session.run("Reply with the single word ACK and then stop."); - - expect(result.finalText.toLowerCase()).toContain("ack"); - } finally { - await session?.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - }, - 120_000 - ); - - test( - "emits tool or command events when writing a file", - async () => { - const cwd = tmpCwd(); - const restoreSessionDir = useTempCodexSessionDir(); - const client = new CodexAgentClient(); - const config: AgentSessionConfig = { provider: "codex", cwd }; - let session: Awaited> | null = null; - try { - session = await client.createSession(config); - log("Streaming file creation activity"); - - const events = session.stream( - "Create a file named tool-test.txt in the current directory with the contents 'hello world', then stop." - ); - - let sawActivity = false; - - for await (const event of events) { - if ( - event.type === "timeline" && - event.provider === "codex" && - event.item.type === "tool_call" - ) { - sawActivity = true; - } - if ( - event.type === "provider_event" && - event.provider === "codex" && - event.raw?.type && - event.raw.type.startsWith("item") && - ["file_change", "command_execution", "mcp_tool_call"].includes((event.raw as any).item?.type) - ) { - sawActivity = true; - } - - if (event.type === "turn_completed" || event.type === "turn_failed") { - break; - } - } - - expect(sawActivity).toBe(true); - } finally { - await session?.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - }, - 180_000 - ); - - test( - "emits Codex tool calls that hydrate into specific UI entries", - async () => { - const cwd = tmpCwd(); - const restoreSessionDir = useTempCodexSessionDir(); - const client = new CodexAgentClient(); - const config: AgentSessionConfig = { provider: "codex", cwd }; - type HydrationEntry = { event: AgentStreamEventPayload; timestamp: Date }; - let session: Awaited> | null = null; - try { - session = await client.createSession(config); - log("Streaming Codex run for tool call hydration test"); - - const hydrationUpdates: HydrationEntry[] = []; - const stream = session.stream( - [ - "1. Run the command `pwd` using your shell tool and report the output.", - "2. Next, use your apply_patch editing tool (not the shell) to create a file named codex-stream.log containing exactly the single line 'ok'.", - "3. After the patch succeeds, stop.", - ].join("\n") - ); - - for await (const event of stream) { - if (event.type === "timeline" || event.type === "provider_event") { - hydrationUpdates.push({ - event: serializeAgentStreamEvent(event), - timestamp: new Date(), - }); - } - - if (event.type === "turn_completed" || event.type === "turn_failed") { - break; - } - } - - const streamItems = hydrateStreamState(hydrationUpdates); - const commandEntry = streamItems.find( - (item): item is AgentToolCallItem => - isAgentToolCallItem(item) && - item.payload.data.server === "command" && - (item.payload.data.displayName?.includes("pwd") ?? false) - ); - const fileEntry = streamItems.find((item): item is AgentToolCallItem => { - if (!isAgentToolCallItem(item)) { - return false; - } - if (item.payload.data.server !== "file_change") { - return false; - } - const result = item.payload.data.result as { files?: Array<{ path?: string }> } | undefined; - if (!Array.isArray(result?.files)) { - return false; - } - return result.files.some((file) => typeof file?.path === "string" && file.path.includes("codex-stream.log")); - }); - - expect(commandEntry).toBeDefined(); - expect(commandEntry?.payload.data.status).toBe("completed"); - expect(commandEntry?.payload.data.result).toMatchObject({ - exitCode: 0, - }); - - expect(fileEntry).toBeDefined(); - expect(fileEntry?.payload.data.status).toBe("completed"); - } finally { - await session?.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - }, - 180_000 - ); - - test( - "emits distinct non-empty call ids for sequential Codex tool calls", - async () => { - const cwd = tmpCwd(); - const restoreSessionDir = useTempCodexSessionDir(); - const client = new CodexAgentClient(); - const config: AgentSessionConfig = { provider: "codex", cwd, modeId: "full-access" }; - let session: Awaited> | null = null; - try { - session = await client.createSession(config); - log("Streaming Codex run for tool call id uniqueness test"); - - const prompt = [ - "1. Run the command `pwd` using your shell tool and wait for it to finish.", - "2. Use apply_patch (not the shell) to create a file codex-call-id.txt containing only the text 'ok'.", - "3. After the patch completes, run `cat codex-call-id.txt` with the shell tool to read the file back.", - "4. Respond DONE and stop.", - ].join("\n"); - - const stream = session.stream(prompt); - const observedToolCalls: ToolCallItem[] = []; - - for await (const event of stream) { - if ( - event.type === "timeline" && - event.provider === "codex" && - event.item.type === "tool_call" && - event.item.server !== "permission" - ) { - observedToolCalls.push(event.item); - } - - if (event.type === "turn_completed" || event.type === "turn_failed") { - break; - } - } - - expect(observedToolCalls.length).toBeGreaterThanOrEqual(2); - - for (const toolCall of observedToolCalls) { - expect(typeof toolCall.callId).toBe("string"); - expect((toolCall.callId ?? "").trim().length).toBeGreaterThan(0); - } - - const uniqueIds = new Set(observedToolCalls.map((item) => item.callId)); - expect(uniqueIds.size).toBeGreaterThanOrEqual(2); - - const callIdCounts = new Map(); - const callIdStatuses = new Map>(); - for (const toolCall of observedToolCalls) { - const callId = toolCall.callId!; - callIdCounts.set(callId, (callIdCounts.get(callId) ?? 0) + 1); - if (!callIdStatuses.has(callId)) { - callIdStatuses.set(callId, new Set()); - } - if (typeof toolCall.status === "string") { - callIdStatuses.get(callId)!.add(toolCall.status); - } - } - - const commandCallGroups = Array.from(callIdCounts.entries()) - .filter(([callId]) => observedToolCalls.some((call) => call.callId === callId && call.server === "command")); - const multiEventCommandIds = commandCallGroups.filter(([, count]) => count >= 2).map(([callId]) => callId); - expect(multiEventCommandIds.length).toBeGreaterThan(0); - - const hasCompletedCommandLifecycle = multiEventCommandIds.some((callId) => { - const statuses = callIdStatuses.get(callId); - return statuses?.has("completed") && (statuses?.size ?? 0) >= 2; - }); - expect(hasCompletedCommandLifecycle).toBe(true); - } finally { - await session?.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - }, - 180_000 - ); - - test( - "hydrates persisted shell_command tool calls with completed status", - async () => { - const restoreSessionDir = useTempCodexSessionDir(); - const client = new CodexAgentClient(); - const cwd = tmpCwd(); - let session: Awaited> | null = null; - let resumed: Awaited> | null = null; - try { - session = await client.createSession({ provider: "codex", cwd, modeId: "full-access" }); - log("Recording Codex command activity for persistence hydration test"); - - const prompt = [ - "Run the command `pwd` using your shell tool to print the working directory.", - "After the command finishes, respond with DONE and stop.", - ].join("\n"); - - let sawCommand = false; - const stream = session.stream(prompt); - for await (const event of stream) { - if ( - event.type === "timeline" && - event.provider === "codex" && - event.item.type === "tool_call" && - event.item.server === "command" - ) { - sawCommand = true; - } - if (event.type === "turn_completed" || event.type === "turn_failed") { - break; - } - } - expect(sawCommand).toBe(true); - - const handle = session.describePersistence(); - expect(handle).toBeTruthy(); - - await session.close(); - session = null; - - const rolloutPath = handle ? await resolveRolloutPathFromHandle(handle) : null; - expect(rolloutPath).toBeTruthy(); - await prependSyntheticRolloutEntries(rolloutPath!, [ - buildSyntheticInstructionEntry(handle!.sessionId ?? handle!.nativeHandle ?? "synthetic-session"), - buildSyntheticEnvironmentEntry(), - ]); - - resumed = await client.resumeSession(handle!); - const hydrationUpdates: StreamHydrationUpdate[] = []; - for await (const event of resumed.streamHistory()) { - recordTimelineUpdate(hydrationUpdates, event); - } - - const hydratedState = hydrateStreamState(hydrationUpdates); - const commandEntry = hydratedState.find( - (item): item is AgentToolCallItem => - isAgentToolCallItem(item) && item.payload.data.server === "command" - ); - expect(commandEntry).toBeTruthy(); - if (commandEntry) { - expect(commandEntry.payload.data.status).toBe("completed"); - expect(commandEntry.payload.data.result).toMatchObject({ - metadata: { exit_code: 0 }, - }); - } - } finally { - await session?.close(); - await resumed?.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - }, - 180_000 - ); - - test( - "supports multiple turns within the same session", - async () => { - const cwd = tmpCwd(); - const restoreSessionDir = useTempCodexSessionDir(); - const client = new CodexAgentClient(); - const config: AgentSessionConfig = { provider: "codex", cwd, modeId: "full-access" }; - let session: Awaited> | null = null; - try { - session = await client.createSession(config); - log("Running multi-turn continuity test"); - - const first = await session.run("Reply with the exact text ACK-ONE and then stop."); - expect(first.finalText.toLowerCase()).toContain("ack-one"); - - const handleAfterFirst = session.describePersistence(); - expect(handleAfterFirst?.sessionId).toBeTruthy(); - - const second = await session.run("Now reply with another acknowledgment and then stop."); - expect(second.finalText).not.toHaveLength(0); - expect(second.finalText).not.toBe(first.finalText); - - const handleAfterSecond = session.describePersistence(); - expect(handleAfterSecond?.sessionId).toBe(handleAfterFirst?.sessionId); - } finally { - await session?.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - }, - 150_000 - ); - - test( - "can change modes mid-session", - async () => { - const cwd = tmpCwd(); - const restoreSessionDir = useTempCodexSessionDir(); - const client = new CodexAgentClient(); - const config: AgentSessionConfig = { provider: "codex", cwd, modeId: "auto" }; - let session: Awaited> | null = null; - try { - session = await client.createSession(config); - log("Testing mode transitions inside a single session"); - - expect(await session.getCurrentMode()).toBe("auto"); - - const first = await session.run("Reply with ACK-MODE-AUTO and then stop."); - expect(first.finalText.toLowerCase()).toContain("ack-mode-auto"); - - await session.setMode("full-access"); - expect(await session.getCurrentMode()).toBe("full-access"); - - const second = await session.run("Reply with ACK-MODE-FULL and then stop."); - expect(second.finalText.toLowerCase()).toContain("ack-mode-full"); - } finally { - await session?.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - }, - 180_000 - ); - - test( - "resumes a session using a persistence handle", - async () => { - const cwd = tmpCwd(); - const restoreSessionDir = useTempCodexSessionDir(); - const client = new CodexAgentClient(); - const config: AgentSessionConfig = { provider: "codex", cwd }; - let session: Awaited> | null = null; - let resumed: Awaited> | null = null; - try { - session = await client.createSession(config); - log("Recording initial turn before persistence"); - - await session.run("Remember the word ALPHA and confirm with ACK."); - const handle = session.describePersistence(); - expect(handle).not.toBeNull(); - expect(handle?.sessionId).toBeTruthy(); - - await session.close(); - session = null; - - log("Resuming session from persistence handle"); - resumed = await client.resumeSession(handle!); - const replayedHistory: AgentStreamEvent[] = []; - for await (const event of resumed.streamHistory()) { - replayedHistory.push(event); - } - log( - `Replayed ${replayedHistory.length} history events: ${JSON.stringify( - replayedHistory.map((e) => ({ type: e.type, timelineType: e.type === "timeline" ? e.item.type : undefined })), - null, - 2 - )}` - ); - expect( - replayedHistory.some( - (event) => - event.type === "timeline" && event.provider === "codex" && event.item.type === "assistant_message" - ) - ).toBe(true); - - const response = await resumed.run("Respond with ACK-RESUMED and then stop."); - expect(response.finalText.toLowerCase()).toContain("ack-resumed"); - - const resumedHandle = resumed.describePersistence(); - expect(resumedHandle?.sessionId).toBe(handle?.sessionId); - } finally { - await session?.close(); - await resumed?.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - }, - 180_000 - ); - - test( - "interrupts a long-running shell command before it completes", - async () => { - const cwd = tmpCwd(); - const restoreSessionDir = useTempCodexSessionDir(); - const client = new CodexAgentClient(); - const config: AgentSessionConfig = { provider: "codex", cwd, modeId: "full-access" }; - let session: Awaited> | null = null; - let runStartedAt: number | null = null; - let durationMs = 0; - let sawSleepCommand = false; - let interruptIssued = false; - - try { - session = await client.createSession(config); - log("Launching sleep command interrupt test"); - - const prompt = [ - "Run the exact shell command `sleep 60` using your shell tool.", - "Do not run any additional commands or send a response until that command finishes.", - ].join(" "); - - runStartedAt = Date.now(); - const stream = session.stream(prompt); - - for await (const event of stream) { - if (event.type === "permission_requested" && session) { - await session.respondToPermission(event.request.id, { behavior: "allow" }); - } - - if ( - event.type === "timeline" && - event.provider === "codex" && - event.item.type === "tool_call" && - event.item.server === "command" && - isSleepCommandToolCall(event.item) - ) { - sawSleepCommand = true; - if (!interruptIssued) { - interruptIssued = true; - await session.interrupt(); - } - } - - if (event.type === "turn_completed" || event.type === "turn_failed") { - break; - } - } - - if (runStartedAt === null) { - throw new Error("Codex run never started"); - } - durationMs = Date.now() - runStartedAt; - } finally { - if (durationMs === 0 && runStartedAt !== null) { - durationMs = Date.now() - runStartedAt; - } - await session?.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - - expect(sawSleepCommand).toBe(true); - expect(interruptIssued).toBe(true); - expect(durationMs).toBeGreaterThan(0); - expect(durationMs).toBeLessThan(60_000); - }, - 90_000 - ); - - // Codex CLI currently doesn't emit approval request events even when approvalPolicy is set to on-request, - // so we keep this test skipped until upstream support lands. - test.skip( - "emits permission requests and resolves them when approvals are handled (awaiting Codex support)", - async () => { - const cwd = tmpCwd(); - const restoreSessionDir = useTempCodexSessionDir(); - const client = new CodexAgentClient(); - const config: AgentSessionConfig = { - provider: "codex", - cwd, - modeId: "full-access", - extra: { codex: { approvalPolicy: "on-request" } }, - }; - let session: Awaited> | null = null; - try { - session = await client.createSession(config); - log("Processing Codex permission request stream"); - - let captured: AgentPermissionRequest | null = null; - const streamed = session.stream( - "Request approval to run the command `pwd`, then run it after approval and stop." - ); - for await (const event of streamed) { - if (event.type === "permission_requested" && !captured) { - captured = event.request; - expect(session.getPendingPermissions()).toHaveLength(1); - await session.respondToPermission(captured.id, { behavior: "allow" }); - } - if (event.type === "turn_completed" || event.type === "turn_failed") { - break; - } - } - - expect(captured).not.toBeNull(); - expect(session.getPendingPermissions()).toHaveLength(0); - - const replay: AgentStreamEvent[] = []; - for await (const event of session.streamHistory()) { - replay.push(event); - } - - expect( - replay.some( - (event) => - event.type === "permission_resolved" && - event.requestId === captured!.id && - event.resolution.behavior === "allow" - ) - ).toBe(true); - expect( - replay.some( - (event) => - event.type === "timeline" && - event.provider === "codex" && - event.item.type === "tool_call" && - event.item.server === "permission" && - event.item.status === "granted" - ) - ).toBe(true); - } finally { - await session?.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - }, - 180_000 - ); - - test( - "hydrates user messages from persisted history", - async () => { - const cwd = tmpCwd(); - const restoreSessionDir = useTempCodexSessionDir(); - const client = new CodexAgentClient(); - const config: AgentSessionConfig = { provider: "codex", cwd }; - let session: Awaited> | null = null; - let resumed: Awaited> | null = null; - - const promptMarker = `CODEX_USER_${Date.now().toString(36)}`; - const prompt = `Reply only with ${promptMarker} and stop.`; - const liveTimelineUpdates: StreamHydrationUpdate[] = [ - buildUserMessageUpdate("codex", prompt, "msg-codex-hydrated-user"), - ]; - - try { - session = await client.createSession(config); - const events = session.stream(prompt); - for await (const event of events) { - recordTimelineUpdate(liveTimelineUpdates, event); - if (event.type === "turn_completed" || event.type === "turn_failed") { - break; - } - } - - const handle = session.describePersistence(); - expect(handle).toBeTruthy(); - - await session.close(); - session = null; - - resumed = await client.resumeSession(handle!); - const hydrationUpdates: StreamHydrationUpdate[] = []; - for await (const event of resumed.streamHistory()) { - recordTimelineUpdate(hydrationUpdates, event); - } - - const liveState = hydrateStreamState(liveTimelineUpdates); - const hydratedState = hydrateStreamState(hydrationUpdates); - - expect(stateIncludesUserMessage(liveState, promptMarker)).toBe(true); - expect(stateIncludesUserMessage(hydratedState, promptMarker)).toBe(true); - expect(stateIncludesSyntheticMessage(hydratedState)).toBe(false); - - const persisted = await client.listPersistedAgents({ limit: 200 }); - const persistedEntry = persisted.find((entry) => entry.sessionId === handle!.sessionId); - expect(persistedEntry).toBeTruthy(); - const normalizedTitle = (persistedEntry?.title ?? "").toLowerCase(); - expect(normalizedTitle.startsWith("# agents.md instructions for")).toBe(false); - expect(normalizedTitle.startsWith("")).toBe(false); - } finally { - await session?.close(); - await resumed?.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - }, - 180_000 - ); - - test( - "sees agent-control MCP tools (list_agents) via codex", - async () => { - const cwd = tmpCwd(); - const restoreSessionDir = useTempCodexSessionDir(); - const mcpServer = await startAgentMcpServer(); - const client = new CodexAgentClient(); - const config: AgentSessionConfig = { - provider: "codex", - cwd, - modeId: "full-access", - agentControlMcp: { url: mcpServer.url }, - }; - let session: Awaited> | null = null; - try { - session = await client.createSession(config); - log(`Agent MCP URL: ${mcpServer.url}`); - - const prompt = [ - "Use the MCP tool agent_control.list_agents and report the agent IDs you receive.", - "If the tool is unavailable, say exactly: MCP tool list_agents unavailable.", - "Then stop.", - ].join("\n"); - - const toolCalls: ToolCallItem[] = []; - let finalText = ""; - const eventLog: string[] = []; - - for await (const event of session.stream(prompt)) { - eventLog.push(event.type); - if (event.type === "timeline" && event.provider === "codex") { - if (event.item.type === "tool_call") { - toolCalls.push(event.item); - } else if (event.item.type === "assistant_message") { - finalText += `${event.item.text}\n`; - } - } - - if (event.type === "turn_failed") { - console.info("[CodexAgentTest] Turn failed:", event.error); - } - - if (event.type === "turn_completed" || event.type === "turn_failed") { - break; - } - } - - const listAgentsCall = toolCalls.find((call) => call.tool === "list_agents"); - if (!listAgentsCall) { - console.info( - "[CodexAgentTest] MCP tool calls:", - toolCalls.map((call) => `${call.server}.${call.tool}`) - ); - console.info("[CodexAgentTest] Final text:", finalText.trim()); - console.info("[CodexAgentTest] Event types:", eventLog); - } - - expect(listAgentsCall).toBeDefined(); - } finally { - await session?.close(); - await mcpServer.close(); - rmSync(cwd, { recursive: true, force: true }); - restoreSessionDir(); - } - }, - 240_000 - ); -}); - -describe("isSyntheticRolloutUserMessage", () => { - test("flags AGENTS instruction payloads", () => { - const text = `# AGENTS.md instructions for /Users/test - - -Stay safe. -`; - expect(isSyntheticRolloutUserMessage(text)).toBe(true); - }); - - test("flags environment context payloads", () => { - const text = ` - /Users/test/project - never -`; - expect(isSyntheticRolloutUserMessage(text)).toBe(true); - }); - - test("allows real user prompts", () => { - expect(isSyntheticRolloutUserMessage("investigate flaky tests")).toBe(false); - }); -}); - -type StreamHydrationUpdate = { - event: Extract; - timestamp: Date; -}; - -function recordTimelineUpdate(target: StreamHydrationUpdate[], event: AgentStreamEvent) { - if (event.type !== "timeline") { - return; - } - target.push({ - event: event as StreamHydrationUpdate["event"], - timestamp: new Date(), - }); -} - -function buildUserMessageUpdate( - provider: AgentProvider, - text: string, - messageId: string -): StreamHydrationUpdate { - return { - event: { - type: "timeline", - provider, - item: { - type: "user_message", - text, - messageId, - }, - }, - timestamp: new Date(), - }; -} - -function stateIncludesUserMessage(state: StreamItem[], marker: string): boolean { - return state.some( - (item) => item.kind === "user_message" && item.text.toLowerCase().includes(marker.toLowerCase()) - ); -} - -function stateIncludesSyntheticMessage(state: StreamItem[]): boolean { - return state.some((item) => { - if (item.kind !== "user_message") { - return false; - } - const normalized = item.text.trim().toLowerCase(); - return ( - normalized.startsWith("# agents.md instructions for") || - normalized.startsWith("") - ); - }); -} - -async function prependSyntheticRolloutEntries(filePath: string, entries: Array>): Promise { - const existing = await fs.readFile(filePath, "utf8"); - const synthetic = entries.map((entry) => JSON.stringify(entry)).join("\n"); - const nextContent = synthetic.length ? `${synthetic}\n${existing}` : existing; - await fs.writeFile( - filePath, - nextContent.endsWith("\n") ? nextContent : `${nextContent}\n`, - "utf8" - ); -} - -async function resolveRolloutPathFromHandle(handle: AgentPersistenceHandle): Promise { - const pathFromMetadata = typeof handle.metadata?.codexRolloutPath === "string" ? handle.metadata.codexRolloutPath : null; - if (pathFromMetadata && (await fileExists(pathFromMetadata))) { - return pathFromMetadata; - } - - const sessionDir = typeof handle.metadata?.codexSessionDir === "string" ? handle.metadata.codexSessionDir : null; - const sessionId = handle.sessionId ?? (typeof handle.nativeHandle === "string" ? handle.nativeHandle : null); - if (!sessionDir || !sessionId) { - return null; - } - return findRolloutInDir(sessionDir, sessionId, 5); -} - -async function fileExists(filePath: string): Promise { - try { - await fs.access(filePath); - return true; - } catch { - return false; - } -} - -async function findRolloutInDir(root: string, sessionId: string, maxDepth: number): Promise { - const stack: Array<{ dir: string; depth: number }> = [{ dir: root, depth: 0 }]; - while (stack.length > 0) { - const { dir, depth } = stack.pop()!; - let entries: Dirent[]; - try { - entries = await fs.readdir(dir, { withFileTypes: true }); - } catch { - continue; - } - - for (const entry of entries) { - const entryPath = path.join(dir, entry.name); - if (entry.isFile()) { - if ( - entry.name.startsWith("rollout-") && - entry.name.includes(sessionId) && - (entry.name.endsWith(".json") || entry.name.endsWith(".jsonl")) - ) { - return entryPath; - } - } else if (entry.isDirectory() && depth < maxDepth) { - stack.push({ dir: entryPath, depth: depth + 1 }); - } - } - } - return null; -} - -function buildSyntheticInstructionEntry(sessionId: string): Record { - const text = [ - `# AGENTS.md instructions for ${sessionId}`, - "", - "", - "Always obey the operator.", - "", - ].join("\n"); - return buildSyntheticResponseItem(text); -} - -function buildSyntheticEnvironmentEntry(): Record { - const text = [ - "", - " /tmp/workspace", - " never", - " danger-full-access", - "", - ].join("\n"); - return buildSyntheticResponseItem(text); -} - -function buildSyntheticResponseItem(text: string): Record { - return { - timestamp: new Date().toISOString(), - type: "response_item", - payload: { - type: "message", - role: "user", - content: [ - { - type: "input_text", - text, - }, - ], - }, - }; -} diff --git a/packages/server/src/server/agent/providers/codex-agent.ts b/packages/server/src/server/agent/providers/codex-agent.ts deleted file mode 100644 index 305e17fe1..000000000 --- a/packages/server/src/server/agent/providers/codex-agent.ts +++ /dev/null @@ -1,2247 +0,0 @@ -import { randomUUID } from "node:crypto"; -import { promises as fs, constants as fsConstants } from "node:fs"; -import type { Dirent } from "node:fs"; -import os from "node:os"; -import path from "node:path"; -import { execSync } from "node:child_process"; - -import { - Codex, - type ApprovalMode, - type CodexOptions, - type Input as CodexInput, - type ModelReasoningEffort, - type SandboxMode, - type Thread, - type ThreadEvent, - type ThreadItem, - type ThreadOptions, - type TurnOptions, - type Usage as CodexUsage, -} from "@openai/codex-sdk"; - -import type { - AgentCapabilityFlags, - AgentClient, - AgentMode, - AgentPermissionRequest, - AgentPermissionResponse, - AgentPersistenceHandle, - AgentPromptInput, - AgentRunOptions, - AgentRunResult, - AgentSession, - AgentSessionConfig, - AgentStreamEvent, - AgentTimelineItem, - AgentUsage, - AgentRuntimeInfo, - ListPersistedAgentsOptions, - PersistedAgentDescriptor, -} from "../agent-sdk-types.js"; -import { getOrchestratorModeInstructions } from "../orchestrator-instructions.js"; - -type CodexAgentConfig = AgentSessionConfig & { provider: "codex" }; - -type CodexExtraConfig = { - developerInstructions?: string; -}; - -type CodexWrapperInfo = { - path: string; - dir: string; -}; - -type CodexSessionBootstrap = { - codex: Codex; - wrapper?: CodexWrapperInfo; -}; - -const CODEX_CAPABILITIES: AgentCapabilityFlags = { - supportsStreaming: true, - supportsSessionPersistence: true, - supportsDynamicModes: false, - supportsMcpServers: false, - supportsReasoningStream: true, - supportsToolInvocations: true, -}; - -const CODEX_MODES: AgentMode[] = [ - { - id: "read-only", - label: "Read Only", - description: - "Codex can read files and answer questions. Manual approval required for edits, commands, or network ops.", - }, - { - id: "auto", - label: "Auto", - description: - "Codex can edit files and run commands inside the workspace but still asks before escalating scope.", - }, - { - id: "full-access", - label: "Full Access", - description: - "Codex can edit files, run commands, and access the network without additional prompts.", - }, -]; - -const VALID_CODEX_MODES = new Set(CODEX_MODES.map((mode) => mode.id)); - -const DEFAULT_CODEX_MODE_ID = "auto"; - -const MODE_PRESETS: Record< - string, - Partial & { - networkAccessEnabled?: boolean; - webSearchEnabled?: boolean; - } -> = { - "read-only": { - sandboxMode: "read-only", - approvalPolicy: "untrusted", - networkAccessEnabled: false, - webSearchEnabled: false, - }, - auto: { - sandboxMode: "workspace-write", - approvalPolicy: "on-request", - networkAccessEnabled: false, - webSearchEnabled: false, - }, - "full-access": { - sandboxMode: "danger-full-access", - approvalPolicy: "never", - networkAccessEnabled: true, - webSearchEnabled: true, - }, -}; - -type CodexOptionsOverrides = Partial & { - skipGitRepoCheck?: boolean; -}; - -const MAX_ROLLOUT_SEARCH_DEPTH = 5; -const PERSISTED_TIMELINE_LIMIT = 20; -const SHELL_FUNCTION_NAMES = new Set(["shell", "shell_command"]); -type ToolCallTimelineItem = Extract; -type ThreadItemWithOptionalCallId = ThreadItem & { call_id?: string }; - -type ExecPermissionRequestPayload = { - call_id?: string; - command?: string | string[]; - cwd?: string; - reason?: string; - risk?: { description?: string } & Record; - parsed_cmd?: string[]; - [key: string]: unknown; -}; - -type PatchPermissionRequestPayload = { - call_id?: string; - changes?: Record; - grant_root?: string; - reason?: string; - [key: string]: unknown; -}; - -function createToolCallTimelineItem( - data: Omit -): AgentTimelineItem { - return { type: "tool_call", ...data }; -} - -function normalizeCallId(value?: string | null): string | undefined { - if (typeof value !== "string") { - return undefined; - } - const trimmed = value.trim(); - return trimmed.length ? trimmed : undefined; -} - -function extractThreadItemCallId(item: ThreadItem): string | undefined { - const withCallId = item as ThreadItemWithOptionalCallId; - return ( - normalizeCallId(withCallId.call_id) ?? - normalizeCallId((item as { id?: string }).id) - ); -} - -function buildCommandDisplayName(command?: unknown): string { - if (typeof command === "string") { - const trimmed = command.trim(); - if (trimmed.length) { - return trimmed; - } - } - return "Command"; -} - -function buildFileChangeSummary( - files: { path: string; kind: string }[] -): string { - if (!files.length) { - return "File change"; - } - if (files.length === 1) { - return `${files[0].kind ?? "edit"}: ${files[0].path}`; - } - return `${files.length} file changes`; -} - -function coerceSandboxMode(value?: string): SandboxMode | undefined { - if (!value) return undefined; - if ( - value === "read-only" || - value === "workspace-write" || - value === "danger-full-access" - ) { - return value; - } - return undefined; -} - -function coerceApprovalMode(value?: string): ApprovalMode | undefined { - if (!value) return undefined; - if ( - value === "never" || - value === "on-request" || - value === "on-failure" || - value === "untrusted" - ) { - return value; - } - return undefined; -} - -function coerceReasoningEffort( - value?: string -): ModelReasoningEffort | undefined { - if (!value) return undefined; - if ( - value === "minimal" || - value === "low" || - value === "medium" || - value === "high" - ) { - return value; - } - return undefined; -} - -function normalizeExtraOptions( - extra?: Record -): CodexOptionsOverrides | undefined { - if (!extra) return undefined; - const allowedKeys = new Set([ - "model", - "sandboxMode", - "skipGitRepoCheck", - "modelReasoningEffort", - "networkAccessEnabled", - "webSearchEnabled", - "approvalPolicy", - ]); - - const normalized: CodexOptionsOverrides = {}; - for (const [key, value] of Object.entries(extra)) { - if (!allowedKeys.has(key)) continue; - (normalized as Record)[key] = value; - } - return normalized; -} - -function detectSystemCodexPath(): string | undefined { - try { - const codexPath = execSync("which codex", { - encoding: "utf8", - stdio: ["pipe", "pipe", "ignore"], - }).trim(); - if (codexPath && codexPath.length > 0) { - return codexPath; - } - } catch { - // which command failed - codex not in PATH - } - return undefined; -} - -function resolveAgentControlMcpUrl(config: CodexAgentConfig): string | null { - const baseUrl = config.agentControlMcp?.url; - if (!baseUrl || typeof baseUrl !== "string") { - return null; - } - return config.parentAgentId - ? appendCallerAgentId(baseUrl, config.parentAgentId) - : baseUrl; -} - -function resolveAgentControlBearerToken( - config: CodexAgentConfig -): string | null { - const header = config.agentControlMcp?.headers?.Authorization; - if (!header || typeof header !== "string") { - return null; - } - const trimmed = header.trim(); - if (trimmed.toLowerCase().startsWith("bearer ")) { - const token = trimmed.slice("bearer ".length).trim(); - return token.length > 0 ? token : null; - } - if (trimmed.toLowerCase().startsWith("basic ")) { - const token = trimmed.slice("basic ".length).trim(); - return token.length > 0 ? token : null; - } - return null; -} - -async function createCodexWrapperScript( - realCodexPath: string, - mcpUrl: string, - mcpBearerToken: string | null, - developerInstructions: string -): Promise { - const templatePath = await resolveCodexWrapperTemplatePath(); - const template = await fs.readFile(templatePath, "utf8"); - const dir = await fs.mkdtemp(path.join(os.tmpdir(), "codex-wrapper-")); - const wrapperPath = path.join(dir, "codex-wrapper"); - const content = template - .replace("__REAL_CODEX__", escapeBashValue(realCodexPath)) - .replace("__MCP_URL__", escapeBashValue(mcpUrl)) - .replace( - "__MCP_BEARER__", - escapeBashValue(mcpBearerToken ?? "") - ) - .replace( - "__DEV_INSTRUCTIONS__", - escapeBashTomlValue(encodeTomlBasicString(developerInstructions)) - ); - await fs.writeFile(wrapperPath, content, { mode: 0o700 }); - await fs.chmod(wrapperPath, 0o700); - return { path: wrapperPath, dir }; -} - -async function resolveCodexWrapperTemplatePath(): Promise { - let dir = process.cwd(); - for (let i = 0; i < 8; i += 1) { - const candidate = path.join(dir, "scripts", "codex-mcp-wrapper-template.sh"); - if (await fileExists(candidate)) { - return candidate; - } - const parent = path.dirname(dir); - if (parent === dir) { - break; - } - dir = parent; - } - throw new Error( - "Unable to locate scripts/codex-mcp-wrapper-template.sh for Codex MCP wrapper" - ); -} - -function appendCallerAgentId(url: string, agentId: string): string { - try { - const parsed = new URL(url); - parsed.searchParams.set("callerAgentId", agentId); - return parsed.toString(); - } catch { - const separator = url.includes("?") ? "&" : "?"; - return `${url}${separator}callerAgentId=${encodeURIComponent(agentId)}`; - } -} - -function resolveDeveloperInstructions(config: CodexAgentConfig): string | null { - const extras = config.extra?.codex as CodexExtraConfig | undefined; - const override = - extras && typeof extras === "object" - ? extras.developerInstructions - : undefined; - if (typeof override === "string" && override.trim().length > 0) { - return override.trim(); - } - const defaultInstructions = getOrchestratorModeInstructions().trim(); - return defaultInstructions.length > 0 ? defaultInstructions : null; -} - -function encodeTomlBasicString(value: string): string { - return value - .replace(/\\/g, "\\\\") - .replace(/"/g, '\\"') - .replace(/\r?\n/g, "\\n"); -} - -function escapeBashValue(value: string): string { - return value - .replace(/"/g, '\\"') - .replace(/\$/g, "\\$") - .replace(/`/g, "\\`"); -} - -function escapeBashTomlValue(value: string): string { - return value.replace(/\$/g, "\\$").replace(/`/g, "\\`"); -} - -export class CodexAgentClient implements AgentClient { - readonly provider = "codex" as const; - readonly capabilities = CODEX_CAPABILITIES; - private readonly baseOptions: CodexOptions; - private readonly baseCodexPath: string | null; - - constructor(options?: CodexOptions) { - const codexOptions = { ...options }; - - // If no explicit codexPathOverride, try to use system-installed codex from PATH - if (!codexOptions.codexPathOverride) { - const systemCodexPath = detectSystemCodexPath(); - if (systemCodexPath) { - codexOptions.codexPathOverride = systemCodexPath; - console.log(`[Codex] Using system binary: ${systemCodexPath}`); - } else { - console.log( - "[Codex] Using embedded binary (no system codex found in PATH)" - ); - } - } - - this.baseOptions = codexOptions; - this.baseCodexPath = codexOptions.codexPathOverride ?? null; - } - - async createSession(config: AgentSessionConfig): Promise { - const codexConfig = this.assertConfig(config); - const bootstrap = await this.createCodexForSession(codexConfig); - return CodexAgentSession.create( - bootstrap.codex, - codexConfig, - undefined, - bootstrap.wrapper - ); - } - - async resumeSession( - handle: AgentPersistenceHandle, - overrides?: Partial - ): Promise { - const metadata = (handle.metadata ?? {}) as Partial; - const merged = { ...metadata, ...overrides } as Partial; - if (!merged.cwd) { - throw new Error( - "Codex resume requires the original working directory in metadata" - ); - } - const mergedConfig = { ...merged, provider: "codex" } as AgentSessionConfig; - const codexConfig = this.assertConfig(mergedConfig); - const bootstrap = await this.createCodexForSession(codexConfig); - return CodexAgentSession.create( - bootstrap.codex, - codexConfig, - handle, - bootstrap.wrapper - ); - } - - async listPersistedAgents( - options?: ListPersistedAgentsOptions - ): Promise { - const sessionRoot = resolveCodexSessionRoot(); - if (!sessionRoot) { - return []; - } - if (!(await fileExists(sessionRoot))) { - return []; - } - const limit = options?.limit ?? 20; - const candidates = await collectRecentRolloutFiles(sessionRoot, limit * 3); - const descriptors: PersistedAgentDescriptor[] = []; - for (const candidate of candidates) { - const descriptor = await parseRolloutDescriptor( - candidate.path, - candidate.mtime, - sessionRoot - ); - if (descriptor) { - descriptors.push(descriptor); - } - if (descriptors.length >= limit) { - break; - } - } - return descriptors; - } - - private assertConfig(config: AgentSessionConfig): CodexAgentConfig { - if (config.provider !== "codex") { - throw new Error( - `CodexAgentClient received config for provider '${config.provider}'` - ); - } - return config as CodexAgentConfig; - } - - private async createCodexForSession( - config: CodexAgentConfig - ): Promise { - const mcpUrl = resolveAgentControlMcpUrl(config); - const mcpBearerToken = resolveAgentControlBearerToken(config); - const developerInstructions = resolveDeveloperInstructions(config); - if ((!mcpUrl && !developerInstructions) || !this.baseCodexPath) { - if ((mcpUrl || developerInstructions) && !this.baseCodexPath) { - console.warn("[Codex] MCP wrapper skipped (no codex binary path available)"); - } - return { codex: new Codex(this.baseOptions) }; - } - - const wrapper = await createCodexWrapperScript( - this.baseCodexPath, - mcpUrl ?? "", - mcpBearerToken, - developerInstructions ?? "" - ); - const env = Object.fromEntries( - Object.entries({ - ...process.env, - ...(this.baseOptions.env ?? {}), - }).filter((entry): entry is [string, string] => entry[1] !== undefined) - ); - - return { - codex: new Codex({ - ...this.baseOptions, - codexPathOverride: wrapper.path, - env, - }), - wrapper, - }; - } -} - -class CodexAgentSession implements AgentSession { - static async create( - codex: Codex, - config: CodexAgentConfig, - handle?: AgentPersistenceHandle, - wrapper?: CodexWrapperInfo - ): Promise { - const session = new CodexAgentSession(codex, config, handle, wrapper); - if (handle) { - await session.loadReplayHistory(handle); - } - return session; - } - - readonly provider = "codex" as const; - readonly capabilities = CODEX_CAPABILITIES; - - private readonly codex: Codex; - private config: CodexAgentConfig; - private threadOptions: ThreadOptions; - private thread: Thread | null; - private threadId: string | null; - private persistence: AgentPersistenceHandle | null; - private pendingLocalId: string; - private currentMode: string; - private availableModes: AgentMode[] = CODEX_MODES; - private readonly codexSessionDir: string | null; - private rolloutPath: string | null; - private readonly wrapper: CodexWrapperInfo | null; - private historyEvents: AgentStreamEvent[] = []; - private pendingPermissions = new Map(); - private cancelCurrentTurn: (() => void) | null = null; - private cachedRuntimeInfo: AgentRuntimeInfo | null = null; - - constructor( - codex: Codex, - config: CodexAgentConfig, - handle?: AgentPersistenceHandle, - wrapper?: CodexWrapperInfo - ) { - this.codex = codex; - this.config = { ...config }; - this.wrapper = wrapper ?? null; - - // Validate mode if provided - if (config.modeId && !VALID_CODEX_MODES.has(config.modeId)) { - const validModesList = Array.from(VALID_CODEX_MODES).join(", "); - throw new Error( - `Invalid mode '${config.modeId}' for Codex provider. Valid modes: ${validModesList}` - ); - } - - this.currentMode = this.config.modeId ?? DEFAULT_CODEX_MODE_ID; - this.threadOptions = this.buildThreadOptions(this.currentMode); - this.codexSessionDir = this.resolveCodexSessionDir(handle); - this.rolloutPath = this.readRolloutPath(handle); - this.threadId = handle?.sessionId ?? handle?.nativeHandle ?? null; - this.pendingLocalId = this.threadId ?? `codex-${randomUUID()}`; - this.persistence = handle ?? null; - this.cachedRuntimeInfo = null; - if (handle && !this.threadId) { - throw new Error("Codex resume requires a thread id"); - } - this.thread = handle - ? this.codex.resumeThread(this.threadId!, this.threadOptions) - : this.codex.startThread(this.threadOptions); - } - - get id(): string | null { - return this.threadId; - } - - async getRuntimeInfo(): Promise { - if (!this.cachedRuntimeInfo) { - await this.refreshRuntimeInfoFromRollout(); - } - if (this.cachedRuntimeInfo) { - return { ...this.cachedRuntimeInfo }; - } - const modelFromRollout = await this.readRuntimeModelFromRollout(); - const info: AgentRuntimeInfo = { - provider: "codex", - sessionId: this.threadId ?? this.pendingLocalId ?? null, - model: - modelFromRollout ?? - this.threadOptions.model ?? - null, - modeId: this.currentMode ?? null, - }; - this.cachedRuntimeInfo = info; - return { ...info }; - } - - 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); - } - } - - // Update runtime info cache after the turn completes - await this.refreshRuntimeInfoFromRollout(); - - return { - sessionId: this.threadId ?? this.pendingLocalId, - finalText, - usage, - timeline, - }; - } - - async *stream( - prompt: AgentPromptInput, - options?: AgentRunOptions - ): AsyncGenerator { - const thread = this.thread; - if (!thread) { - throw new Error("Codex session is closed"); - } - - const replayEvents = this.drainHistoryEvents(); - for (const event of replayEvents) { - yield event; - } - - const input = this.toCodexInput(prompt); - const turnOptions = this.buildTurnOptions(options); - const { events } = await thread.runStreamed(input, turnOptions); - - let finishedNaturally = false; - let cancelIssued = false; - const requestCancel = () => { - if (cancelIssued) { - return; - } - cancelIssued = true; - if (typeof events.return !== "function") { - return; - } - try { - const cancellation = events.return({ - type: "turn.completed", - } as ThreadEvent); - void cancellation.catch((error) => { - console.warn( - "[CodexAgentSession] Failed to stop Codex stream:", - error - ); - }); - } catch (error) { - console.warn("[CodexAgentSession] Failed to stop Codex stream:", error); - } - }; - - this.cancelCurrentTurn = requestCancel; - - try { - for await (const rawEvent of events) { - yield* this.translateEvent(rawEvent); - if ( - rawEvent.type === "turn.completed" || - rawEvent.type === "turn.failed" || - rawEvent.type === "error" - ) { - finishedNaturally = true; - break; - } - } - } finally { - if (!finishedNaturally) { - requestCancel(); - } - if (this.cancelCurrentTurn === requestCancel) { - this.cancelCurrentTurn = null; - } - await this.refreshRuntimeInfoFromRollout(); - } - } - - async interrupt(): Promise { - this.cancelCurrentTurn?.(); - } - - async *streamHistory(): AsyncGenerator { - const replayEvents = this.drainHistoryEvents(); - for (const event of replayEvents) { - yield event; - } - } - - async getAvailableModes(): Promise { - return this.availableModes; - } - - async getCurrentMode(): Promise { - return this.currentMode ?? null; - } - - async setMode(modeId: string): Promise { - if (modeId === this.currentMode) { - return; - } - - // Validate mode - if (!VALID_CODEX_MODES.has(modeId)) { - const validModesList = Array.from(VALID_CODEX_MODES).join(", "); - throw new Error( - `Invalid mode '${modeId}' for Codex provider. Valid modes: ${validModesList}` - ); - } - - this.currentMode = modeId; - this.config.modeId = modeId; - this.threadOptions = this.buildThreadOptions(modeId); - - if (this.threadId) { - this.thread = this.codex.resumeThread(this.threadId, this.threadOptions); - } else { - this.pendingLocalId = `codex-${randomUUID()}`; - this.thread = this.codex.startThread(this.threadOptions); - } - } - - getPendingPermissions(): AgentPermissionRequest[] { - return Array.from(this.pendingPermissions.values()); - } - - async respondToPermission( - requestId: string, - response: AgentPermissionResponse - ): Promise { - const request = this.pendingPermissions.get(requestId); - if (!request) { - throw new Error( - `No pending Codex permission request with id '${requestId}'` - ); - } - - this.pendingPermissions.delete(requestId); - - const status = response.behavior === "allow" ? "granted" : "denied"; - this.enqueueHistoryEvent({ - type: "timeline", - provider: "codex", - item: createToolCallTimelineItem({ - server: "permission", - tool: request.name, - status, - callId: request.id, - displayName: request.title ?? request.name, - kind: "permission", - input: request.input, - }), - }); - - this.enqueueHistoryEvent({ - type: "permission_resolved", - provider: "codex", - requestId, - resolution: response, - }); - } - - describePersistence(): AgentPersistenceHandle | null { - if (this.persistence) { - return this.persistence; - } - if (!this.threadId) { - return null; - } - const { model: _ignoredModel, ...restConfig } = this.config; - this.persistence = { - provider: "codex", - sessionId: this.threadId, - nativeHandle: this.threadId, - metadata: { - ...restConfig, - codexSessionDir: this.codexSessionDir ?? undefined, - codexRolloutPath: this.rolloutPath ?? undefined, - }, - }; - return this.persistence; - } - - async close(): Promise { - this.thread = null; - this.pendingPermissions.clear(); - if (this.wrapper) { - try { - await fs.rm(this.wrapper.path, { force: true }); - await fs.rm(this.wrapper.dir, { recursive: true, force: true }); - } catch { - // ignore cleanup failures - } - } - } - - // eslint-disable-next-line @typescript-eslint/no-unused-vars - setManagedAgentId(_agentId: string): void { - // Codex agents don't currently use MCP for agent-control - // This is a no-op but satisfies the AgentSession interface - } - - private drainHistoryEvents(): AgentStreamEvent[] { - if (!this.historyEvents.length) { - return []; - } - const events = [...this.historyEvents]; - this.historyEvents = []; - return events; - } - - private enqueueHistoryEvent(event: AgentStreamEvent): void { - this.historyEvents.push(event); - } - - private buildThreadOptions(modeId: string): ThreadOptions { - const options: ThreadOptions = { - workingDirectory: this.config.cwd, - skipGitRepoCheck: true, - }; - - const preset = MODE_PRESETS[modeId] ?? MODE_PRESETS[DEFAULT_CODEX_MODE_ID]; - if (preset) { - Object.assign(options, preset); - } - - if (this.config.model) { - options.model = this.config.model; - } - if (this.config.sandboxMode) { - options.sandboxMode = - coerceSandboxMode(this.config.sandboxMode) ?? options.sandboxMode; - } - if (this.config.approvalPolicy) { - options.approvalPolicy = - coerceApprovalMode(this.config.approvalPolicy) ?? - options.approvalPolicy; - } - if (typeof this.config.networkAccess === "boolean") { - options.networkAccessEnabled = this.config.networkAccess; - } - if (typeof this.config.webSearch === "boolean") { - options.webSearchEnabled = this.config.webSearch; - } - if (this.config.reasoningEffort) { - options.modelReasoningEffort = - coerceReasoningEffort(this.config.reasoningEffort) ?? - options.modelReasoningEffort; - } - - const extra = normalizeExtraOptions( - this.config.extra?.codex as Record | undefined - ); - if (extra) { - Object.assign(options, extra); - } - - // Codex CLI currently disables MCP tool availability when approval_policy is set. - // When injecting MCP servers, omit approvalPolicy/webSearchEnabled/networkAccessEnabled to ensure MCP tools are discoverable. - if (resolveAgentControlMcpUrl(this.config)) { - delete (options as Partial).approvalPolicy; - delete (options as Partial).webSearchEnabled; - delete (options as Partial).networkAccessEnabled; - } - - return options; - } - - private async loadReplayHistory( - handle: AgentPersistenceHandle - ): Promise { - const threadIdentifier = handle.sessionId ?? handle.nativeHandle; - if (!threadIdentifier || !this.codexSessionDir) { - return; - } - - let rolloutFile = this.rolloutPath; - if (rolloutFile && !(await fileExists(rolloutFile))) { - rolloutFile = null; - } - - if (!rolloutFile) { - rolloutFile = await findRolloutFile( - threadIdentifier, - this.codexSessionDir - ); - } - - if (!rolloutFile) { - return; - } - - this.rolloutPath = rolloutFile; - await this.refreshRuntimeInfoFromRollout(); - - try { - const events = await parseRolloutFile(rolloutFile, threadIdentifier); - if (events.length) { - this.historyEvents = events; - } - } catch (error) { - console.warn( - `[CodexAgentSession] Failed to parse rollout ${rolloutFile}:`, - error - ); - } - } - - private resolveCodexSessionDir( - handle?: AgentPersistenceHandle - ): string | null { - const metadataDir = readMetadataString(handle, "codexSessionDir"); - if (metadataDir) { - return metadataDir; - } - - const extraDir = readCodexExtraString(this.config, "sessionDir"); - if (extraDir) { - return extraDir; - } - - if (process.env.CODEX_SESSION_DIR) { - return process.env.CODEX_SESSION_DIR; - } - - const extraHome = readCodexExtraString(this.config, "home"); - const codexHome = - extraHome ?? process.env.CODEX_HOME ?? path.join(os.homedir(), ".codex"); - return path.join(codexHome, "sessions"); - } - - private readRolloutPath(handle?: AgentPersistenceHandle): string | null { - const metadataRollout = readMetadataString(handle, "codexRolloutPath"); - if (metadataRollout) { - return metadataRollout; - } - const extraRollout = readCodexExtraString(this.config, "rolloutPath"); - return extraRollout ?? null; - } - - private async refreshRuntimeInfoFromRollout(): Promise { - if (!this.rolloutPath) { - const threadIdentifier = this.threadId; - if (threadIdentifier && this.codexSessionDir) { - const rolloutFile = await findRolloutFile( - threadIdentifier, - this.codexSessionDir - ); - if (rolloutFile) { - this.rolloutPath = rolloutFile; - } - } - } - const modelFromRollout = await this.readRuntimeModelFromRollout(); - if (!modelFromRollout && this.cachedRuntimeInfo) { - return; - } - this.cachedRuntimeInfo = { - provider: "codex", - sessionId: this.threadId ?? this.pendingLocalId ?? null, - model: - modelFromRollout ?? - this.threadOptions.model ?? - null, - modeId: this.currentMode ?? null, - }; - } - - private async readRuntimeModelFromRollout(): Promise { - const rollout = this.rolloutPath; - if (!rollout) { - return null; - } - try { - return await readLatestTurnContextModel(rollout); - } catch { - return null; - } - } - - private buildTurnOptions(options?: AgentRunOptions): TurnOptions { - if (!options?.outputSchema) { - return {}; - } - return { outputSchema: options.outputSchema }; - } - - private toCodexInput(prompt: AgentPromptInput): CodexInput { - if (typeof prompt === "string") { - return prompt; - } - return prompt.map((chunk) => ({ type: "text", text: chunk.text })); - } - - private *translateEvent(event: ThreadEvent): Generator { - const permissionEvents = this.handlePermissionEvent(event); - if (permissionEvents) { - for (const permissionEvent of permissionEvents) { - yield permissionEvent; - } - return; - } - - switch (event.type) { - case "thread.started": - this.threadId = event.thread_id; - this.pendingLocalId = event.thread_id; - this.persistence = null; - yield { - type: "thread_started", - provider: "codex", - sessionId: event.thread_id, - }; - break; - case "turn.started": - yield { type: "turn_started", provider: "codex" }; - break; - case "turn.completed": - yield { - type: "turn_completed", - provider: "codex", - usage: this.convertUsage(event.usage), - }; - break; - case "turn.failed": - yield { - type: "turn_failed", - provider: "codex", - error: event.error?.message ?? "Codex turn failed", - }; - break; - case "item.started": - case "item.updated": - case "item.completed": { - const item = this.threadItemToTimeline(event.item); - if (item) { - yield { type: "timeline", provider: "codex", item }; - } - break; - } - case "error": { - const message = event.message ?? "Codex stream error"; - yield { - type: "timeline", - provider: "codex", - item: { type: "error", message }, - }; - yield { type: "turn_failed", provider: "codex", error: message }; - break; - } - default: - break; - } - } - - private handlePermissionEvent(event: ThreadEvent): AgentStreamEvent[] | null { - const eventType = - typeof (event as { type?: string }).type === "string" - ? (event as { type?: string }).type - : null; - if (!eventType) { - return null; - } - - if (eventType === "exec_approval_request") { - const request = this.buildExecPermissionRequest(event); - return this.enqueuePermissionRequest(request); - } - - if (eventType === "apply_patch_approval_request") { - const request = this.buildPatchPermissionRequest(event); - return this.enqueuePermissionRequest(request); - } - - return null; - } - - private enqueuePermissionRequest( - request: AgentPermissionRequest | null - ): AgentStreamEvent[] | null { - if (!request) { - return null; - } - - this.pendingPermissions.set(request.id, request); - - const events: AgentStreamEvent[] = [ - { - type: "timeline", - provider: "codex", - item: createToolCallTimelineItem({ - server: "permission", - tool: request.name, - status: "requested", - callId: request.id, - displayName: request.title ?? request.name, - kind: "permission", - input: request.input, - }), - }, - { type: "permission_requested", provider: "codex", request }, - ]; - - return events; - } - - private buildExecPermissionRequest( - rawInput: unknown - ): AgentPermissionRequest | null { - if (!rawInput || typeof rawInput !== "object") { - return null; - } - const raw = rawInput as ExecPermissionRequestPayload; - - const commandParts: string[] = Array.isArray(raw.command) - ? raw.command.filter((entry: unknown) => typeof entry === "string") - : typeof raw.command === "string" - ? [raw.command] - : []; - const commandText = commandParts.join(" ").trim(); - const cwd = - typeof raw.cwd === "string" && raw.cwd.length ? raw.cwd : undefined; - const requestId = - typeof raw.call_id === "string" && raw.call_id.length - ? raw.call_id - : randomUUID(); - const reason = - typeof raw.reason === "string" && raw.reason.length - ? raw.reason - : undefined; - const risk = - raw.risk && typeof raw.risk === "object" ? raw.risk : undefined; - const parsedCommand = Array.isArray(raw.parsed_cmd) - ? raw.parsed_cmd - : undefined; - - const metadata: Record = { - callId: raw.call_id, - command: commandText || undefined, - cwd, - reason, - }; - if (risk) { - metadata.risk = risk; - } - if (parsedCommand) { - metadata.parsedCommand = parsedCommand; - } - - const description = reason ?? (risk?.description as string | undefined); - - const request: AgentPermissionRequest = { - id: `permission-${requestId}`, - provider: "codex", - name: "exec_command", - kind: "tool", - title: commandText ? `Run command: ${commandText}` : "Run shell command", - description: description ?? undefined, - input: { - command: commandParts, - cwd, - reason, - risk, - parsedCommand, - }, - metadata: sanitizeMetadata(metadata), - }; - - return request; - } - - private buildPatchPermissionRequest( - rawInput: unknown - ): AgentPermissionRequest | null { - if (!rawInput || typeof rawInput !== "object") { - return null; - } - const raw = rawInput as PatchPermissionRequestPayload; - - const requestId = - typeof raw.call_id === "string" && raw.call_id.length - ? raw.call_id - : randomUUID(); - const changes = - raw.changes && typeof raw.changes === "object" ? raw.changes : undefined; - const filePaths = changes ? Object.keys(changes) : []; - const grantRoot = - typeof raw.grant_root === "string" && raw.grant_root.length - ? raw.grant_root - : undefined; - const reason = - typeof raw.reason === "string" && raw.reason.length - ? raw.reason - : undefined; - - const title = - filePaths.length > 0 - ? `Apply patch to ${filePaths.length} file${ - filePaths.length === 1 ? "" : "s" - }` - : "Apply patch"; - - const metadata: Record = { - callId: raw.call_id, - files: filePaths.length ? filePaths : undefined, - grantRoot, - reason, - }; - - const input: Record = { - changes, - files: filePaths, - grantRoot, - reason, - }; - - const request: AgentPermissionRequest = { - id: `permission-${requestId}`, - provider: "codex", - name: "apply_patch", - kind: "tool", - title, - description: reason ?? undefined, - input, - suggestions: grantRoot ? [{ grantRoot }] : undefined, - metadata: sanitizeMetadata(metadata), - }; - - return request; - } - - private threadItemToTimeline(item: ThreadItem): AgentTimelineItem | null { - const callId = extractThreadItemCallId(item); - switch (item.type) { - case "agent_message": - return { type: "assistant_message", text: item.text }; - case "reasoning": - return { type: "reasoning", text: item.text }; - case "command_execution": { - // Codex SDK uses aggregated_output and exit_code - const aggregatedOutput = (item as any)?.aggregated_output; - const exitCode = (item as any)?.exit_code; - const cwd = (item as any)?.cwd; - const commandValue = item.command; - const command = - typeof commandValue === "string" - ? commandValue - : Array.isArray(commandValue) - ? (commandValue as string[]).join(" ") - : "command"; - - // Build structured command result matching StructuredToolResult type - const structuredOutput = - typeof aggregatedOutput === "string" - ? { - type: "command" as const, - command, - output: aggregatedOutput, - exitCode: typeof exitCode === "number" ? exitCode : undefined, - cwd, - } - : undefined; - - return createToolCallTimelineItem({ - server: "command", - tool: "shell", - status: item.status, - callId, - displayName: buildCommandDisplayName(item.command), - kind: "execute", - input: { command: item.command, cwd }, - output: structuredOutput, - error: (item as any)?.error, - }); - } - case "file_change": { - const files = item.changes.map((change) => ({ - path: change.path, - kind: change.kind, - })); - return createToolCallTimelineItem({ - server: "file_change", - tool: "apply_patch", - status: "completed", - callId, - displayName: buildFileChangeSummary(files), - kind: "edit", - output: { files }, - }); - } - case "mcp_tool_call": - return createToolCallTimelineItem({ - server: item.server, - tool: item.tool, - status: item.status, - callId, - displayName: `${item.server}.${item.tool}`, - kind: "tool", - input: (item as any)?.input, - output: (item as any)?.output, - }); - case "web_search": - return createToolCallTimelineItem({ - server: "web_search", - tool: "web_search", - status: (item as any)?.status ?? "completed", - callId, - displayName: item.query ? `Web search: ${item.query}` : "Web search", - kind: "search", - input: { query: item.query }, - }); - case "todo_list": - return { type: "todo", items: item.items }; - case "error": - return { type: "error", message: item.message }; - default: - return null; - } - } - - private convertUsage(usage?: CodexUsage): AgentUsage | undefined { - if (!usage) { - return undefined; - } - return { - inputTokens: usage.input_tokens, - cachedInputTokens: usage.cached_input_tokens, - outputTokens: usage.output_tokens, - }; - } -} - -async function fileExists(filePath: string): Promise { - try { - await fs.access(filePath, fsConstants.F_OK); - return true; - } catch { - return false; - } -} - -async function findRolloutFile( - threadId: string, - root: string -): Promise { - const stack: { dir: string; depth: number }[] = [{ dir: root, depth: 0 }]; - while (stack.length > 0) { - const { dir, depth } = stack.pop()!; - let entries: Dirent[]; - try { - entries = await fs.readdir(dir, { withFileTypes: true }); - } catch { - continue; - } - for (const entry of entries) { - const entryPath = path.join(dir, entry.name); - if (entry.isFile()) { - const matchesThread = entry.name.includes(threadId); - const matchesPrefix = entry.name.startsWith("rollout-"); - const matchesExtension = - entry.name.endsWith(".json") || entry.name.endsWith(".jsonl"); - if (matchesThread && matchesPrefix && matchesExtension) { - return entryPath; - } - } else if (entry.isDirectory() && depth < MAX_ROLLOUT_SEARCH_DEPTH) { - stack.push({ dir: entryPath, depth: depth + 1 }); - } - } - } - return null; -} - -async function parseRolloutFile( - filePath: string, - threadId: string -): Promise { - const content = await fs.readFile(filePath, "utf8"); - const lines = content - .split(/\r?\n/) - .map((line) => line.trim()) - .filter(Boolean); - const events: AgentStreamEvent[] = [ - { type: "thread_started", provider: "codex", sessionId: threadId }, - ]; - const commandCalls = new Map(); - - for (const line of lines) { - const entry = parseRolloutEntry(line); - if (!entry) continue; - - if (entry.type === "response_item") { - handleRolloutResponseItem(entry.payload, events, commandCalls); - } else if (entry.type === "event_msg") { - handleRolloutEventMessage(entry.payload, events); - } - } - - return events; -} - -export async function readLatestTurnContextModel( - filePath: string -): Promise { - let content: string; - try { - content = await fs.readFile(filePath, "utf8"); - } catch { - return null; - } - const lines = content.split(/\r?\n/).filter(Boolean); - for (let i = lines.length - 1; i >= 0; i -= 1) { - const line = lines[i]; - try { - const parsed = JSON.parse(line); - if ( - parsed && - typeof parsed === "object" && - (parsed as { type?: string }).type === "turn_context" - ) { - const payload = (parsed as { payload?: unknown }).payload; - if (payload && typeof payload === "object") { - const model = (payload as Record).model; - if (typeof model === "string" && model.trim().length > 0) { - return model.trim(); - } - } - } - } catch { - continue; - } - } - return null; -} - -function parseRolloutEntry(line: string): RolloutEntry | null { - if (!line) { - return null; - } - try { - const parsed = JSON.parse(line); - if (isRolloutEntry(parsed)) { - return parsed; - } - if ( - parsed && - typeof parsed === "object" && - typeof (parsed as { output?: unknown }).output === "string" - ) { - return parseRolloutEntry((parsed as { output: string }).output); - } - } catch { - return null; - } - return null; -} - -function isRolloutEntry(value: unknown): value is RolloutEntry { - if (!value || typeof value !== "object" || !("type" in value)) { - return false; - } - const type = (value as { type?: unknown }).type; - return type === "response_item" || type === "event_msg"; -} - -function isSessionMetaEntry(value: unknown): value is SessionMetaEntry { - return Boolean( - value && - typeof value === "object" && - (value as { type?: unknown }).type === "session_meta" - ); -} - -function handleRolloutResponseItem( - payload: RolloutResponsePayload | undefined, - events: AgentStreamEvent[], - commandCalls: Map -): void { - if (!payload || typeof payload !== "object") { - return; - } - - switch (payload.type) { - case "message": { - const text = extractMessageText(payload.content); - if (text) { - if (payload.role === "assistant") { - events.push({ - type: "timeline", - provider: "codex", - item: { type: "assistant_message", text }, - }); - } else if (payload.role === "user") { - if (isSyntheticRolloutUserMessage(text)) { - break; - } - events.push({ - type: "timeline", - provider: "codex", - item: { type: "user_message", text }, - }); - } - } - break; - } - case "reasoning": { - const text = extractReasoningText(payload); - if (text) { - events.push({ - type: "timeline", - provider: "codex", - item: { type: "reasoning", text }, - }); - } - break; - } - case "function_call": { - handleRolloutFunctionCall(payload, events, commandCalls); - break; - } - case "function_call_output": { - finalizeRolloutFunctionCall(payload, events, commandCalls); - break; - } - case "custom_tool_call": { - handleRolloutCustomToolCall(payload, events); - break; - } - default: - break; - } -} - -function handleRolloutEventMessage( - payload: RolloutEventPayload | undefined, - events: AgentStreamEvent[] -): void { - if (!payload || typeof payload !== "object") { - return; - } - if (payload.type === "agent_reasoning" && typeof payload.text === "string") { - events.push({ - type: "timeline", - provider: "codex", - item: { type: "reasoning", text: payload.text }, - }); - } -} - -function handleRolloutFunctionCall( - payload: RolloutFunctionCallPayload, - events: AgentStreamEvent[], - commandCalls: Map -): void { - const name = typeof payload.name === "string" ? payload.name : undefined; - if (!name) { - return; - } - - if (SHELL_FUNCTION_NAMES.has(name)) { - const { command, cwd } = buildCommandContext(payload.arguments); - if (typeof payload.call_id === "string") { - commandCalls.set(payload.call_id, { command, cwd }); - events.push({ - type: "timeline", - provider: "codex", - item: createToolCallTimelineItem({ - server: "command", - tool: "shell", - status: payload.status ?? "in_progress", - callId: payload.call_id, - displayName: buildCommandDisplayName(command), - kind: "execute", - input: { command, cwd }, - }), - }); - } - return; - } - - if (name === "update_plan") { - const args = safeJsonParse<{ plan?: unknown }>(payload.arguments); - const planItems = parsePlanItems(args); - if (planItems.length) { - events.push({ - type: "timeline", - provider: "codex", - item: { type: "todo", items: planItems }, - }); - } - return; - } - - events.push({ - type: "timeline", - provider: "codex", - item: createToolCallTimelineItem({ - server: "codex", - tool: name, - status: payload.status ?? "in_progress", - callId: typeof payload.call_id === "string" ? payload.call_id : undefined, - displayName: `${name}`, - kind: "tool", - input: safeJsonParse(payload.arguments), - }), - }); -} - -function finalizeRolloutFunctionCall( - payload: RolloutFunctionCallOutputPayload, - events: AgentStreamEvent[], - commandCalls: Map -): void { - if (typeof payload.call_id !== "string") { - return; - } - const command = commandCalls.get(payload.call_id); - if (!command) { - return; - } - const { status, output } = buildCommandOutput(command, payload.output); - - events.push({ - type: "timeline", - provider: "codex", - item: createToolCallTimelineItem({ - server: "command", - tool: "shell", - status, - callId: payload.call_id, - displayName: buildCommandDisplayName(command.command), - kind: "execute", - input: { command: command.command, cwd: command.cwd }, - output, - }), - }); - commandCalls.delete(payload.call_id); -} - -function handleRolloutCustomToolCall( - payload: RolloutCustomToolCallPayload, - events: AgentStreamEvent[] -): void { - if (payload?.name === "apply_patch" && typeof payload.input === "string") { - const patchText = payload.input; - const files = parsePatchFiles(patchText); - if (files.length) { - // Build structured output with the patch/diff for each file - const parsedEdits = files.map((file) => ({ - filePath: file.path, - kind: file.kind, - // Include the full patch as diff - frontend will render it - diff: patchText, - })); - - events.push({ - type: "timeline", - provider: "codex", - item: createToolCallTimelineItem({ - server: "file_change", - tool: "apply_patch", - status: "completed", - displayName: buildFileChangeSummary(files), - kind: "edit", - output: { - type: "file_edit" as const, - filePath: files[0]?.path ?? "unknown", - diff: patchText, - parsedEdits, - }, - }), - }); - } - return; - } - - if (payload?.name && SHELL_FUNCTION_NAMES.has(payload.name)) { - const commandContext = buildCommandContext(payload.input); - const { status, output } = buildCommandOutput(commandContext, payload.output); - events.push({ - type: "timeline", - provider: "codex", - item: createToolCallTimelineItem({ - server: "command", - tool: "shell", - status: payload.status ?? status, - callId: - typeof payload.call_id === "string" ? payload.call_id : undefined, - displayName: buildCommandDisplayName(commandContext.command), - kind: "execute", - input: { command: commandContext.command, cwd: commandContext.cwd }, - output, - }), - }); - return; - } - - if (typeof payload?.name === "string") { - events.push({ - type: "timeline", - provider: "codex", - item: createToolCallTimelineItem({ - server: "codex", - tool: payload.name, - status: payload.status ?? "completed", - displayName: payload.name, - kind: "tool", - input: payload.input, - output: payload.output, - }), - }); - } -} - -type RolloutCandidate = { - path: string; - mtime: Date; -}; - -type RolloutContentBlock = { - text?: string; - message?: string; - [key: string]: unknown; -}; - -type RolloutMessagePayload = { - type: "message"; - role?: string; - content?: RolloutContentBlock[]; -}; - -type RolloutReasoningSummary = { - text?: string; -}; - -type RolloutReasoningPayload = { - type: "reasoning"; - summary?: RolloutReasoningSummary[]; - text?: string; -}; - -type RolloutFunctionCallPayload = { - type: "function_call"; - name?: string; - arguments?: string; - call_id?: string; - status?: string; -}; - -type RolloutFunctionCallOutputPayload = { - type: "function_call_output"; - call_id?: string; - output?: string; -}; - -type RolloutCustomToolCallPayload = { - type: "custom_tool_call"; - name?: string; - status?: string; - call_id?: string; - input?: string | Record; - output?: unknown; -}; - -type RolloutResponsePayload = - | RolloutMessagePayload - | RolloutReasoningPayload - | RolloutFunctionCallPayload - | RolloutFunctionCallOutputPayload - | RolloutCustomToolCallPayload; - -type RolloutEventPayload = { - type: string; - text?: string; - [key: string]: unknown; -}; - -type RolloutEntry = - | { type: "response_item"; payload?: RolloutResponsePayload } - | { type: "event_msg"; payload?: RolloutEventPayload }; - -type CommandExecutionResult = { - metadata?: { - exit_code?: number; - [key: string]: unknown; - }; - [key: string]: unknown; -}; - -type SessionMetaPayload = { - id?: string; - cwd?: string; - [key: string]: unknown; -}; - -type SessionMetaEntry = { - type: "session_meta"; - payload?: SessionMetaPayload; -}; - -async function collectRecentRolloutFiles( - rootDir: string, - limit: number -): Promise { - const candidates: RolloutCandidate[] = []; - async function traverse(dir: string): Promise { - let entries: Dirent[]; - try { - entries = await fs.readdir(dir, { withFileTypes: true }); - } catch { - return; - } - const files = entries.filter( - (entry) => - entry.isFile() && - entry.name.startsWith("rollout-") && - (entry.name.endsWith(".json") || entry.name.endsWith(".jsonl")) - ); - for (const entry of files) { - const fullPath = path.join(dir, entry.name); - try { - const stat = await fs.stat(fullPath); - candidates.push({ path: fullPath, mtime: stat.mtime }); - } catch { - // ignore stat failures - } - } - const directories = entries - .filter((entry) => entry.isDirectory()) - .sort((a, b) => b.name.localeCompare(a.name)); - for (const dirEntry of directories) { - await traverse(path.join(dir, dirEntry.name)); - } - } - await traverse(rootDir); - return candidates - .sort((a, b) => b.mtime.getTime() - a.mtime.getTime()) - .slice(0, limit); -} - -async function parseRolloutDescriptor( - filePath: string, - mtime: Date, - sessionRoot: string -): Promise { - let content: string; - try { - content = await fs.readFile(filePath, "utf8"); - } catch { - return null; - } - - let sessionId: string | null = null; - let cwd: string | null = null; - let title: string | null = null; - - for (const rawLine of content.split(/\r?\n/)) { - const line = rawLine.trim(); - if (!line) continue; - let entry: unknown; - try { - entry = JSON.parse(line); - } catch { - continue; - } - if (!sessionId && isSessionMetaEntry(entry) && entry.payload) { - sessionId = - typeof entry.payload.id === "string" ? entry.payload.id : null; - if ( - typeof entry.payload.cwd === "string" && - entry.payload.cwd.length > 0 - ) { - cwd = entry.payload.cwd; - } - } - if (sessionId && cwd) { - break; - } - } - - if (!sessionId || !cwd) { - return null; - } - - const timeline = await loadCodexPersistedTimeline(filePath, sessionId); - if (!title) { - title = timeline.find((item) => item.type === "user_message")?.text ?? null; - } - - const persistence: AgentPersistenceHandle = { - provider: "codex", - sessionId, - nativeHandle: sessionId, - metadata: sanitizeMetadata({ - provider: "codex", - cwd, - codexSessionDir: sessionRoot, - codexRolloutPath: filePath, - }), - }; - - return { - provider: "codex", - sessionId, - cwd, - title: (title ?? "").trim() || `Codex session ${sessionId.slice(0, 8)}`, - lastActivityAt: mtime, - persistence, - timeline, - }; -} - -function resolveCodexSessionRoot(): string | null { - if (process.env.CODEX_SESSION_DIR) { - return process.env.CODEX_SESSION_DIR; - } - const codexHome = process.env.CODEX_HOME ?? path.join(os.homedir(), ".codex"); - return path.join(codexHome, "sessions"); -} - -function extractMessageText(content: unknown): string { - if (!Array.isArray(content)) { - return ""; - } - const parts: string[] = []; - for (const block of content) { - if (!block || typeof block !== "object") { - continue; - } - const record = block as Record; - const text = typeof record.text === "string" ? record.text : undefined; - if (text && text.trim()) { - parts.push(text.trim()); - continue; - } - const message = - typeof record.message === "string" ? record.message : undefined; - if (message && message.trim()) { - parts.push(message.trim()); - } - } - return parts.join("\n").trim(); -} - -export function isSyntheticRolloutUserMessage(text: string): boolean { - const normalized = text.trim(); - if (!normalized) { - return false; - } - - const lower = normalized.toLowerCase(); - if ( - lower.startsWith("# agents.md instructions for") && - lower.includes("") - ) { - return true; - } - - if (lower.startsWith("")) { - return true; - } - - return false; -} - -function extractReasoningText(payload: RolloutReasoningPayload): string { - if (Array.isArray(payload?.summary)) { - const text = payload.summary - .map((item) => (item && typeof item.text === "string" ? item.text : "")) - .filter(Boolean) - .join("\n") - .trim(); - if (text) { - return text; - } - } - if (typeof payload?.text === "string") { - return payload.text; - } - return ""; -} - -async function loadCodexPersistedTimeline( - filePath: string, - threadId: string -): Promise { - try { - const events = await parseRolloutFile(filePath, threadId); - const timeline: AgentTimelineItem[] = []; - for (const event of events) { - if (event.type !== "timeline" || event.provider !== "codex") { - continue; - } - timeline.push(event.item); - if (timeline.length >= PERSISTED_TIMELINE_LIMIT) { - break; - } - } - return timeline; - } catch (error) { - console.warn( - `[CodexAgentSession] Failed to load persisted timeline for ${threadId}:`, - error - ); - return []; - } -} - -function parsePlanItems(args: unknown): { text: string; completed: boolean }[] { - const planValue = - typeof args === "object" && args - ? (args as { plan?: unknown }).plan - : undefined; - const plan = Array.isArray(planValue) ? planValue : []; - const items: { text: string; completed: boolean }[] = []; - for (const step of plan) { - if (!step || typeof step !== "object") { - continue; - } - const record = step as { step?: unknown; text?: unknown; status?: unknown }; - const text = - typeof record.step === "string" - ? record.step - : typeof record.text === "string" - ? record.text - : ""; - if (!text) { - continue; - } - const status = typeof record.status === "string" ? record.status : ""; - items.push({ text, completed: status === "completed" }); - } - return items; -} - -function parsePatchFiles(patch: string): { path: string; kind: string }[] { - const files: { path: string; kind: string }[] = []; - const seen = new Set(); - const lines = patch.split(/\r?\n/); - for (const line of lines) { - const trimmed = line.trim(); - let kind: string | null = null; - let pathValue: string | null = null; - if (trimmed.startsWith("*** Add File:")) { - kind = "add"; - pathValue = trimmed.replace("*** Add File:", "").trim(); - } else if (trimmed.startsWith("*** Delete File:")) { - kind = "delete"; - pathValue = trimmed.replace("*** Delete File:", "").trim(); - } else if (trimmed.startsWith("*** Update File:")) { - kind = "update"; - pathValue = trimmed.replace("*** Update File:", "").trim(); - } - if (kind && pathValue && !seen.has(`${kind}:${pathValue}`)) { - seen.add(`${kind}:${pathValue}`); - files.push({ path: pathValue, kind }); - } - } - return files; -} - -function safeJsonParse(value: unknown): T | null { - if (typeof value !== "string") { - return null; - } - try { - return JSON.parse(value) as T; - } catch { - return null; - } -} - -function parseJsonOrObject(value: unknown): T | null { - if (typeof value === "string") { - return safeJsonParse(value); - } - if (value && typeof value === "object") { - return value as T; - } - return null; -} - -function parseCommandOutputText( - output: string -): { exitCode?: number; stdout?: string } | null { - if (!output.trim()) { - return null; - } - - const exitCodeMatch = output.match(/Exit code:\s*(\d+)/i); - const exitCode = - exitCodeMatch && exitCodeMatch[1] - ? Number.parseInt(exitCodeMatch[1], 10) - : undefined; - - const outputMatch = output.match(/\nOutput:\s*\n([\s\S]*)$/); - const stdout = outputMatch ? outputMatch[1].replace(/\s+$/, "") : undefined; - - if (exitCode === undefined && stdout === undefined) { - return null; - } - - return { - exitCode: Number.isNaN(exitCode as number) ? undefined : exitCode, - stdout, - }; -} - -function buildCommandContext(input: unknown): { command: string; cwd?: string } { - const args = parseJsonOrObject>(input); - const commandFromArgs = formatCommand(args); - let command = commandFromArgs; - if (!command && typeof input === "string") { - const trimmed = input.trim(); - if (trimmed && !trimmed.startsWith("{") && !trimmed.startsWith("[")) { - command = trimmed; - } - } - const cwd = - args && - typeof args === "object" && - typeof (args as { workdir?: unknown }).workdir === "string" - ? ((args as { workdir?: unknown }).workdir as string) - : undefined; - return { command: command ?? "Command", cwd }; -} - -function buildCommandOutput( - command: { command: string; cwd?: string }, - rawOutput: unknown -): { status: "completed" | "failed"; output: unknown } { - const result = parseJsonOrObject(rawOutput); - const parsedOutput = - typeof rawOutput === "string" ? parseCommandOutputText(rawOutput) : null; - const exitCode = result?.metadata?.exit_code ?? parsedOutput?.exitCode; - const status = - exitCode === undefined || exitCode === 0 ? "completed" : "failed"; - const stdout = - result?.stdout ?? - (typeof result?.output === "string" ? result.output : undefined) ?? - parsedOutput?.stdout; - const metadata = - result?.metadata ?? - (typeof exitCode === "number" ? { exit_code: exitCode } : undefined); - - let output: unknown; - if (stdout !== undefined) { - output = { - type: "command" as const, - command: command.command, - output: stdout, - exitCode, - cwd: command.cwd, - metadata, - }; - } else if (result && typeof result === "object") { - output = { ...result, metadata: metadata ?? result.metadata }; - } else if (metadata) { - output = { metadata }; - } else { - output = result; - } - - return { status, output }; -} - -function formatCommand(args: unknown): string | null { - if (!args || typeof args !== "object") { - return null; - } - const record = args as { command?: unknown }; - if (Array.isArray(record.command)) { - return record.command.join(" "); - } - if (typeof record.command === "string") { - return record.command; - } - return null; -} - -function readCodexExtraString( - config: CodexAgentConfig, - key: string -): string | undefined { - const extras = config.extra?.codex; - if (!extras || typeof extras !== "object") { - return undefined; - } - const value = (extras as Record)[key]; - return typeof value === "string" && value.length > 0 ? value : undefined; -} - -function readMetadataString( - handle: AgentPersistenceHandle | undefined, - key: string -): string | undefined { - if (!handle?.metadata || typeof handle.metadata !== "object") { - return undefined; - } - const value = (handle.metadata as Record)[key]; - return typeof value === "string" && value.length > 0 ? value : undefined; -} - -function sanitizeMetadata( - metadata: Record -): Record | undefined { - const entries = Object.entries(metadata).filter( - ([_, value]) => value !== undefined && value !== null - ); - if (!entries.length) { - return undefined; - } - return Object.fromEntries(entries); -} diff --git a/packages/server/src/server/agent/providers/codex-agent.unit.test.ts b/packages/server/src/server/agent/providers/codex-agent.unit.test.ts deleted file mode 100644 index ccb8ff62f..000000000 --- a/packages/server/src/server/agent/providers/codex-agent.unit.test.ts +++ /dev/null @@ -1,28 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { promises as fs } from "node:fs"; -import path from "node:path"; -import os from "node:os"; - -import { readLatestTurnContextModel } from "./codex-agent.js"; - -describe("readLatestTurnContextModel", () => { - it("returns the most recent turn_context model from rollout", async () => { - const dir = await fs.mkdtemp(path.join(os.tmpdir(), "codex-rollout-")); - const file = path.join(dir, "rollout.jsonl"); - const lines = [ - JSON.stringify({ type: "session_meta", payload: { id: "thread-1" } }), - JSON.stringify({ - type: "turn_context", - payload: { model: "gpt-5.1-codex", cwd: "/tmp" }, - }), - JSON.stringify({ - type: "turn_context", - payload: { model: "gpt-5.2-codex", cwd: "/tmp" }, - }), - ]; - await fs.writeFile(file, lines.join("\n"), "utf8"); - - const model = await readLatestTurnContextModel(file); - expect(model).toBe("gpt-5.2-codex"); - }); -}); diff --git a/packages/server/test-results-after.txt b/packages/server/test-results-after.txt deleted file mode 100644 index 3fb37dacb..000000000 --- a/packages/server/test-results-after.txt +++ /dev/null @@ -1,16 +0,0 @@ - -> @paseo/server@0.1.0 test -> vitest run - - - RUN v3.2.4 /home/moboudra/dev/voice-dev/packages/server - -stdout | src/server/agent/providers/codex-agent.test.ts > CodexAgentClient (SDK integration) > responds with text -[CodexAgentTest] Running single-turn acknowledgment test - -stdout | src/server/agent/providers/codex-agent.test.ts > CodexAgentClient (SDK integration) > emits tool or command events when writing a file -[CodexAgentTest] Streaming file creation activity - -stdout | src/server/agent/providers/codex-agent.test.ts > CodexAgentClient (SDK integration) > emits Codex tool calls that hydrate into specific UI entries -[CodexAgentTest] Streaming Codex run for tool call hydration test - diff --git a/packages/server/test-sdk-spawn.ts b/packages/server/test-sdk-spawn.ts deleted file mode 100644 index abec5e2de..000000000 --- a/packages/server/test-sdk-spawn.ts +++ /dev/null @@ -1,34 +0,0 @@ -import { CodexAgentClient } from "@paseo/server/src/server/agent/providers/codex-agent.js"; - -async function test() { - console.log("Creating CodexAgentClient..."); - const client = new CodexAgentClient(); - - console.log("Creating session..."); - const session = await client.createSession({ - provider: "codex", - cwd: process.cwd(), - modeId: "full-access", - }); - - console.log("Session created:", session.id); - console.log("\nNow check ps to see which codex binary is running!\n"); - - // Run a simple prompt using stream to see all events - console.log("Running test prompt..."); - console.log("Sleeping 30s - check ps for the args now!"); - await new Promise((resolve) => setTimeout(resolve, 30000)); - - const events = session.stream('echo "test"'); - - for await (const event of events) { - if (event.type === "turn_completed") { - console.log("Turn completed"); - } - } - - await session.close(); - console.log("Session closed"); -} - -test().catch(console.error); diff --git a/plan.md b/plan.md index 1ac74a7f2..bd304e182 100644 --- a/plan.md +++ b/plan.md @@ -823,3 +823,22 @@ Build a new Codex MCP provider side‑by‑side with the existing Codex SDK prov 4. ✅ No console errors (only React 19 deprecation warnings unrelated to Codex MCP) 5. ✅ Verified file created on filesystem with correct content EVIDENCE: Playwright snapshots showing agent UI with tool calls, `cat /Users/moboudra/dev/voice-dev/codex-mcp-test.txt` output showing "Codex MCP E2E test successful", tool detail panel showing command/directory/exit code/output. + +- [x] **Deprecate**: Remove old Codex SDK provider completely. + + The old `codex-agent.ts` is DEPRECATED. Codex MCP is now the only Codex provider. + + **Tasks:** + 1. Verify `bootstrap.ts` only registers `CodexMcpAgentClient` for "codex" provider + 2. Delete `codex-agent.ts` (the old SDK provider) + 3. Delete `codex-agent.test.ts` (the old SDK tests) + 4. Remove any imports of `CodexAgentClient` from the codebase + 5. Search for any remaining references to the old SDK provider and remove them + 6. Run typecheck and tests to verify nothing is broken + + **Acceptance criteria:** + - Only ONE Codex provider exists: `codex-mcp-agent.ts` + - No references to `CodexAgentClient` or `codex-agent.ts` in the codebase + - Typecheck passes + - All tests pass (no skipped tests for deprecated code) + - **Done (2025-12-25 15:42)**: WHAT: Deleted `packages/server/src/server/agent/providers/codex-agent.ts`, `packages/server/src/server/agent/providers/codex-agent.test.ts`, `packages/server/src/server/agent/providers/codex-agent.unit.test.ts`, `packages/server/test-sdk-spawn.ts`, `test-sdk-spawn.ts`, and `packages/server/test-results-after.txt`. Verified `bootstrap.ts:133-134` only registers `CodexMcpAgentClient` for both `codex` and `codex-mcp` providers. RESULT: Only `codex-mcp-agent.ts` and `codex-mcp-agent.test.ts` remain as Codex providers. Zero references to `CodexAgentClient` in `packages/` directory. EVIDENCE: `ls packages/server/src/server/agent/providers/codex*.ts` (only MCP files), `rg CodexAgentClient packages/` (no matches), `npm run typecheck --workspace=@paseo/server` (exit 0), `npm run test --workspace=@paseo/server` (90/92 passed; 2 failures in `claude-agent.test.ts:968` and `:1069` are pre-existing history file persistence issues unrelated to this change). diff --git a/test-sdk-spawn.ts b/test-sdk-spawn.ts deleted file mode 100644 index abec5e2de..000000000 --- a/test-sdk-spawn.ts +++ /dev/null @@ -1,34 +0,0 @@ -import { CodexAgentClient } from "@paseo/server/src/server/agent/providers/codex-agent.js"; - -async function test() { - console.log("Creating CodexAgentClient..."); - const client = new CodexAgentClient(); - - console.log("Creating session..."); - const session = await client.createSession({ - provider: "codex", - cwd: process.cwd(), - modeId: "full-access", - }); - - console.log("Session created:", session.id); - console.log("\nNow check ps to see which codex binary is running!\n"); - - // Run a simple prompt using stream to see all events - console.log("Running test prompt..."); - console.log("Sleeping 30s - check ps for the args now!"); - await new Promise((resolve) => setTimeout(resolve, 30000)); - - const events = session.stream('echo "test"'); - - for await (const event of events) { - if (event.type === "turn_completed") { - console.log("Turn completed"); - } - } - - await session.close(); - console.log("Session closed"); -} - -test().catch(console.error);