refactor(file-explorer): add deep linking, path navigation, diff preview, and Codex session persistence
This commit is contained in:
Mohamed Boudra
@@ -1,6 +1,8 @@
|
||||
import { spawn } from "child_process";
|
||||
import { Writable, Readable } from "stream";
|
||||
import { access, constants } from "fs/promises";
|
||||
import { access, constants, mkdir } from "fs/promises";
|
||||
import os from "os";
|
||||
import path from "path";
|
||||
import {
|
||||
ClientSideConnection,
|
||||
ndJsonStream,
|
||||
@@ -249,6 +251,16 @@ class ACPClient implements Client {
|
||||
export class AgentManager {
|
||||
private agents = new Map<string, ManagedAgent>();
|
||||
private persistence = new AgentPersistence();
|
||||
private codexHomeDir: string;
|
||||
private codexSessionDir: string;
|
||||
|
||||
constructor() {
|
||||
const defaultCodexHome = path.join(os.homedir(), ".voice-dev", "codex");
|
||||
this.codexHomeDir = process.env.CODEX_HOME ?? defaultCodexHome;
|
||||
this.codexSessionDir =
|
||||
process.env.CODEX_SESSION_DIR ??
|
||||
path.join(this.codexHomeDir, "sessions");
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the agent manager and load persisted agents
|
||||
@@ -1088,9 +1100,21 @@ export class AgentManager {
|
||||
}
|
||||
|
||||
const definition = getAgentTypeDefinition(agent.options.type);
|
||||
const agentProcess = spawn(definition.spawn.command, definition.spawn.args, {
|
||||
const spawnArgs = [...definition.spawn.args];
|
||||
const spawnEnv: NodeJS.ProcessEnv = { ...process.env };
|
||||
|
||||
if (agent.options.type === "codex") {
|
||||
await mkdir(this.codexSessionDir, { recursive: true });
|
||||
spawnArgs.push("--session-persist", this.codexSessionDir);
|
||||
spawnEnv.CODEX_HOME = this.codexHomeDir;
|
||||
spawnEnv.CODEX_SESSION_PERSIST = "1";
|
||||
spawnEnv.CODEX_SESSION_DIR = this.codexSessionDir;
|
||||
}
|
||||
|
||||
const agentProcess = spawn(definition.spawn.command, spawnArgs, {
|
||||
stdio: ["pipe", "pipe", "pipe"],
|
||||
cwd,
|
||||
env: spawnEnv,
|
||||
});
|
||||
|
||||
const input = Writable.toWeb(agentProcess.stdin);
|
||||
|
||||
@@ -71,12 +71,12 @@ const agentTypeDefinitions: Record<AgentType, AgentTypeDefinition> = {
|
||||
codex: {
|
||||
id: "codex",
|
||||
label: "Codex",
|
||||
description: "Zed Codex ACP integration (no session resume)",
|
||||
description: "Zed Codex ACP integration with session persistence",
|
||||
spawn: {
|
||||
command: "npx",
|
||||
args: ["@zed-industries/codex-acp"],
|
||||
},
|
||||
supportsSessionPersistence: false,
|
||||
supportsSessionPersistence: true,
|
||||
availableModes: CODEX_MODES,
|
||||
defaultModeId: "auto",
|
||||
},
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
# Agent SDK Capability Notes
|
||||
|
||||
## @openai/codex-sdk (0.58.0)
|
||||
|
||||
- Entry point is `Codex`/`Thread` (`node_modules/@openai/codex-sdk/dist/index.d.ts`). `Codex.startThread()` spawns the bundled `codex` CLI and returns a `Thread` handle that can issue `run()` or `runStreamed()` calls.
|
||||
- `Thread.runStreamed()` yields JSONL events typed as `ThreadEvent` (agent messages, reasoning, file changes, MCP tool calls, command executions, web searches, todo lists). These map one-to-one with CLI telemetry, so we can stream reasoning, tool calls, and status without ACP.
|
||||
- Session persistence is provided via `Codex.resumeThread(id)`; the CLI stores rollouts in `~/.codex/sessions` and resuming passes `resume <threadId>` to the CLI.
|
||||
- Session configuration is passed via `ThreadOptions`: `model`, `sandboxMode` (`read-only|workspace-write|danger-full-access`), `approvalPolicy`, `modelReasoningEffort`, `networkAccessEnabled`, `webSearchEnabled`, `workingDirectory`, and `skipGitRepoCheck`.
|
||||
- **Streams reasoning & tool activity**: `ThreadEvent` includes `item.started/updated/completed` with `reasoning` items, file changes, MCP tool calls, command execution events, etc. We get full turn telemetry similar to ACP.
|
||||
- **Gaps**:
|
||||
- No public API to enumerate supported modes/models or switch modes mid-session; options are fixed per `Thread` instance.
|
||||
- MCP servers cannot be injected dynamically; MCP items appear in the stream, but the CLI discovers servers from `codex` config, not via SDK APIs.
|
||||
- Thread metadata (title, timestamps) isn’t exposed, so our persistence layer would still need to scrape Codex’s manifest if we want richer state.
|
||||
|
||||
## @anthropic-ai/claude-agent-sdk (0.1.37)
|
||||
|
||||
- Primary surface is the `query()` helper (`node_modules/@anthropic-ai/claude-agent-sdk/sdk.d.ts`). It returns an `AsyncGenerator<SDKMessage>` so we get incremental updates (`SDKAssistantMessage`, `SDKPartialAssistantMessage`, `SDKSystemMessage`, `SDKToolProgressMessage`, etc.).
|
||||
- Control plane APIs are built into the generator: `interrupt()`, `setPermissionMode()`, `setModel()`, `setMaxThinkingTokens()`, and query-time introspection helpers (`supportedCommands()`, `supportedModels()`, `mcpServerStatus()`, `accountInfo()`). Final `SDKResultMessage` entries contain usage/cost and `permission_denials` so we can mirror Claude’s session summary.
|
||||
- Reasoning/tool output arrives in-band via `SDKAssistantMessage.message` (Anthropic streaming delta structure) and can include `tool_use` blocks; there’s also `SDKToolProgressMessage` updates and `SDKPartialAssistantMessage` typed separately for partial thoughts.
|
||||
- Session lifecycle:
|
||||
- Create a session with `query({ prompt, options })`.
|
||||
- Resume via `options.resume` (session id) and optionally `forkSession` if we don’t want to mutate the original log.
|
||||
- Hooks (`hooks` option) let us observe `SessionStart`, `SessionEnd`, `PreToolUse`, etc., mirroring Claude Code’s plugin system.
|
||||
- Configuration options include `agents`, `permissionMode`, `allowedTools`/`disallowedTools`, `systemPrompt`, `plugins`, and `mcpServers`. MCP servers can be stdio/SSE/HTTP or in-process (`createSdkMcpServer`).
|
||||
- Tool gating is first-class: `canUseTool` callback receives tool invocations and can return allow/deny plus suggested permission persistence updates (`PermissionUpdate`).
|
||||
- Partial reasoning is delivered via `SDKPartialAssistantMessage` while the final `SDKResultMessage` includes usage, total cost, and permission denials.
|
||||
- **Gaps**:
|
||||
- Claude Agent SDK currently consumes prompts via `query()` calls rather than maintaining a long-lived object per session, so we need to wrap it ourselves to keep stateful handles.
|
||||
- While MCPs can be injected, there’s no baked-in manifest of sessions; persistence is up to us (the SDK exposes `session_id` on every message for us to store).
|
||||
- Slash command support exists via `supportedCommands()`, but there’s no concept of sandbox/approval tiers like Codex’s `sandboxMode`; we instead rely on `permissionMode` + permission rules.
|
||||
@@ -0,0 +1,148 @@
|
||||
import type { ThreadEvent as CodexThreadEvent } from "@openai/codex-sdk";
|
||||
import type {
|
||||
Options as ClaudeAgentOptions,
|
||||
SDKMessage as ClaudeStreamMessage,
|
||||
McpServerConfig as ClaudeMcpServerConfig,
|
||||
} from "@anthropic-ai/claude-agent-sdk";
|
||||
import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
|
||||
|
||||
export type AgentProvider = "codex" | "claude";
|
||||
|
||||
export type AgentMode = {
|
||||
id: string;
|
||||
label: string;
|
||||
description?: string;
|
||||
};
|
||||
|
||||
export type AgentCapabilityFlags = {
|
||||
supportsStreaming: boolean;
|
||||
supportsSessionPersistence: boolean;
|
||||
supportsDynamicModes: boolean;
|
||||
supportsMcpServers: boolean;
|
||||
supportsReasoningStream: boolean;
|
||||
supportsToolInvocations: boolean;
|
||||
};
|
||||
|
||||
export type AgentPersistenceHandle = {
|
||||
provider: AgentProvider;
|
||||
sessionId: string;
|
||||
/** Provider specific handle (Codex thread id, Claude resume token, etc). */
|
||||
nativeHandle?: string;
|
||||
metadata?: Record<string, unknown>;
|
||||
};
|
||||
|
||||
export type AgentPromptInput = string | { type: "text"; text: string }[];
|
||||
|
||||
export type AgentRunOptions = {
|
||||
outputSchema?: unknown;
|
||||
resumeFrom?: AgentPersistenceHandle;
|
||||
maxThinkingTokens?: number;
|
||||
};
|
||||
|
||||
export type AgentUsage = {
|
||||
inputTokens?: number;
|
||||
cachedInputTokens?: number;
|
||||
outputTokens?: number;
|
||||
totalCostUsd?: number;
|
||||
};
|
||||
|
||||
export type AgentTimelineItem =
|
||||
| { type: "assistant_message"; text: string; raw?: unknown }
|
||||
| { type: "reasoning"; text: string; raw?: unknown }
|
||||
| { type: "command"; command: string; status: string; raw?: unknown }
|
||||
| { type: "file_change"; files: { path: string; kind: string }[]; raw?: unknown }
|
||||
| { type: "mcp_tool"; server: string; tool: string; status: string; raw?: unknown }
|
||||
| { type: "web_search"; query: string; raw?: unknown }
|
||||
| { type: "todo"; items: { text: string; completed: boolean }[]; raw?: unknown }
|
||||
| { type: "error"; message: string; raw?: unknown };
|
||||
|
||||
export type AgentStreamEvent =
|
||||
| { type: "thread_started"; sessionId: string; provider: AgentProvider }
|
||||
| { type: "turn_started"; provider: AgentProvider }
|
||||
| { type: "turn_completed"; provider: AgentProvider; usage?: AgentUsage }
|
||||
| { type: "turn_failed"; provider: AgentProvider; error: string }
|
||||
| { type: "timeline"; item: AgentTimelineItem; provider: AgentProvider }
|
||||
| { type: "provider_event"; provider: AgentProvider; raw: CodexThreadEvent | ClaudeStreamMessage }
|
||||
| { type: "permission_requested"; provider: AgentProvider; request: AgentPermissionRequest }
|
||||
| {
|
||||
type: "permission_resolved";
|
||||
provider: AgentProvider;
|
||||
requestId: string;
|
||||
resolution: AgentPermissionResponse;
|
||||
};
|
||||
|
||||
export type AgentPermissionRequestKind = "tool" | "plan" | "mode" | "other";
|
||||
|
||||
export type AgentPermissionUpdate = Record<string, unknown>;
|
||||
|
||||
export type AgentPermissionRequest = {
|
||||
id: string;
|
||||
provider: AgentProvider;
|
||||
name: string;
|
||||
kind: AgentPermissionRequestKind;
|
||||
title?: string;
|
||||
description?: string;
|
||||
input?: Record<string, unknown>;
|
||||
suggestions?: AgentPermissionUpdate[];
|
||||
metadata?: Record<string, unknown>;
|
||||
raw?: unknown;
|
||||
};
|
||||
|
||||
export type AgentPermissionResponse =
|
||||
| {
|
||||
behavior: "allow";
|
||||
updatedInput?: Record<string, unknown>;
|
||||
updatedPermissions?: AgentPermissionUpdate[];
|
||||
}
|
||||
| {
|
||||
behavior: "deny";
|
||||
message?: string;
|
||||
interrupt?: boolean;
|
||||
};
|
||||
|
||||
export type AgentRunResult = {
|
||||
sessionId: string;
|
||||
finalText: string;
|
||||
usage?: AgentUsage;
|
||||
timeline: AgentTimelineItem[];
|
||||
};
|
||||
|
||||
export type AgentSessionConfig = {
|
||||
provider: AgentProvider;
|
||||
cwd: string;
|
||||
modeId?: string;
|
||||
model?: string;
|
||||
approvalPolicy?: string;
|
||||
sandboxMode?: string;
|
||||
networkAccess?: boolean;
|
||||
webSearch?: boolean;
|
||||
reasoningEffort?: string;
|
||||
extra?: {
|
||||
codex?: Record<string, unknown>;
|
||||
claude?: Partial<ClaudeAgentOptions>;
|
||||
};
|
||||
mcpServers?: Record<string, ClaudeMcpServerConfig | McpServer | { handler: () => Promise<unknown> }>;
|
||||
};
|
||||
|
||||
export interface AgentSession {
|
||||
readonly provider: AgentProvider;
|
||||
readonly id: string | null;
|
||||
readonly capabilities: AgentCapabilityFlags;
|
||||
run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult>;
|
||||
stream(prompt: AgentPromptInput, options?: AgentRunOptions): AsyncGenerator<AgentStreamEvent>;
|
||||
streamHistory(): AsyncGenerator<AgentStreamEvent>;
|
||||
getAvailableModes(): Promise<AgentMode[]>;
|
||||
getCurrentMode(): Promise<string | null>;
|
||||
setMode(modeId: string): Promise<void>;
|
||||
getPendingPermissions(): AgentPermissionRequest[];
|
||||
respondToPermission(requestId: string, response: AgentPermissionResponse): Promise<void>;
|
||||
describePersistence(): AgentPersistenceHandle | null;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
export interface AgentClient {
|
||||
readonly provider: AgentProvider;
|
||||
readonly capabilities: AgentCapabilityFlags;
|
||||
createSession(config: AgentSessionConfig): Promise<AgentSession>;
|
||||
resumeSession(handle: AgentPersistenceHandle, overrides?: Partial<AgentSessionConfig>): Promise<AgentSession>;
|
||||
}
|
||||
@@ -0,0 +1,261 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
import { ClaudeAgentClient } from "./claude-agent.js";
|
||||
import type { AgentSessionConfig, AgentStreamEvent } from "../agent-sdk-types.js";
|
||||
|
||||
function tmpCwd(): string {
|
||||
const dir = mkdtempSync(path.join(os.tmpdir(), "claude-agent-e2e-"));
|
||||
return dir;
|
||||
}
|
||||
|
||||
async function autoApprove(session: Awaited<ReturnType<ClaudeAgentClient["createSession"]>>, event: AgentStreamEvent) {
|
||||
if (event.type === "permission_requested") {
|
||||
await session.respondToPermission(event.request.id, { behavior: "allow" });
|
||||
}
|
||||
}
|
||||
|
||||
describe("ClaudeAgentClient (SDK integration)", () => {
|
||||
test(
|
||||
"responds with text",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const client = new ClaudeAgentClient();
|
||||
const config: AgentSessionConfig = {
|
||||
provider: "claude",
|
||||
cwd,
|
||||
extra: { claude: { maxThinkingTokens: 1024 } },
|
||||
};
|
||||
const session = await client.createSession(config);
|
||||
|
||||
const result = await session.run(
|
||||
"Reply with the single word ACK and then stop."
|
||||
);
|
||||
|
||||
expect(result.finalText.toLowerCase()).toContain("ack");
|
||||
|
||||
await session.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
},
|
||||
120_000
|
||||
);
|
||||
|
||||
test(
|
||||
"streams reasoning chunks",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const client = new ClaudeAgentClient();
|
||||
const config: AgentSessionConfig = {
|
||||
provider: "claude",
|
||||
cwd,
|
||||
extra: { claude: { maxThinkingTokens: 2048 } },
|
||||
};
|
||||
const session = await client.createSession(config);
|
||||
|
||||
const events = session.stream(
|
||||
"Think step by step about the pros and cons of single-file tests, but only share a short plan."
|
||||
);
|
||||
|
||||
let sawReasoning = false;
|
||||
|
||||
for await (const event of events) {
|
||||
await autoApprove(session, event);
|
||||
if (event.type === "timeline" && event.item.type === "reasoning") {
|
||||
sawReasoning = true;
|
||||
}
|
||||
if (event.type === "turn_completed" || event.type === "turn_failed") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
expect(sawReasoning).toBe(true);
|
||||
|
||||
await session.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
},
|
||||
120_000
|
||||
);
|
||||
|
||||
test(
|
||||
"tracks permission + tool lifecycle when editing a file",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const client = new ClaudeAgentClient();
|
||||
const config: AgentSessionConfig = {
|
||||
provider: "claude",
|
||||
cwd,
|
||||
extra: { claude: { maxThinkingTokens: 1024 } },
|
||||
};
|
||||
const session = await client.createSession(config);
|
||||
|
||||
const events = session.stream(
|
||||
"First run a Bash command to print the working directory, then use your editor tools (not the shell) to create a file named tool-test.txt in the current directory that contains exactly the text 'hello world'. Report 'done' after the write finishes."
|
||||
);
|
||||
|
||||
const timeline: any[] = [];
|
||||
let completed = false;
|
||||
|
||||
for await (const event of events) {
|
||||
await autoApprove(session, event);
|
||||
if (event.type === "timeline") {
|
||||
timeline.push(event.item);
|
||||
}
|
||||
if (event.type === "turn_completed") {
|
||||
completed = true;
|
||||
break;
|
||||
}
|
||||
if (event.type === "turn_failed") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
const toolEvent = timeline.find((item) => item.type === "mcp_tool");
|
||||
|
||||
expect(completed).toBe(true);
|
||||
expect(toolEvent).toBeTruthy();
|
||||
|
||||
const filePath = path.join(cwd, "tool-test.txt");
|
||||
expect(existsSync(filePath)).toBe(true);
|
||||
expect(readFileSync(filePath, "utf8")).toContain("hello world");
|
||||
|
||||
await session.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
},
|
||||
180_000
|
||||
);
|
||||
|
||||
test(
|
||||
"supports multi-turn conversations",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const client = new ClaudeAgentClient();
|
||||
const config: AgentSessionConfig = {
|
||||
provider: "claude",
|
||||
cwd,
|
||||
extra: { claude: { maxThinkingTokens: 2048 } },
|
||||
};
|
||||
const session = await client.createSession(config);
|
||||
|
||||
const first = await session.run("Respond only with the word alpha.");
|
||||
expect(first.finalText.toLowerCase()).toContain("alpha");
|
||||
|
||||
const second = await session.run(
|
||||
"Without adding any explanations, repeat exactly the same word you just said."
|
||||
);
|
||||
expect(second.finalText.toLowerCase()).toContain("alpha");
|
||||
|
||||
await session.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
},
|
||||
120_000
|
||||
);
|
||||
|
||||
test(
|
||||
"resumes a persisted session",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const client = new ClaudeAgentClient();
|
||||
const config: AgentSessionConfig = {
|
||||
provider: "claude",
|
||||
cwd,
|
||||
extra: { claude: { maxThinkingTokens: 1024 } },
|
||||
};
|
||||
const session = await client.createSession(config);
|
||||
|
||||
const first = await session.run("Say READY and then stop.");
|
||||
expect(first.finalText.toLowerCase()).toContain("ready");
|
||||
|
||||
const handle = session.describePersistence();
|
||||
expect(handle).toBeTruthy();
|
||||
|
||||
await session.close();
|
||||
|
||||
const resumed = await client.resumeSession(handle!, { cwd });
|
||||
const resumedResult = await resumed.run(
|
||||
"Respond with the single word RESUMED."
|
||||
);
|
||||
expect(resumedResult.finalText.toLowerCase()).toContain("resumed");
|
||||
await resumed.close();
|
||||
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
},
|
||||
150_000
|
||||
);
|
||||
|
||||
test(
|
||||
"updates session modes",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const client = new ClaudeAgentClient();
|
||||
const config: AgentSessionConfig = {
|
||||
provider: "claude",
|
||||
cwd,
|
||||
extra: { claude: { maxThinkingTokens: 1024 } },
|
||||
};
|
||||
const session = await client.createSession(config);
|
||||
|
||||
const modes = await session.getAvailableModes();
|
||||
expect(modes.map((m) => m.id)).toContain("plan");
|
||||
|
||||
await session.setMode("plan");
|
||||
expect(await session.getCurrentMode()).toBe("plan");
|
||||
|
||||
const result = await session.run(
|
||||
"Just reply with the word PLAN to confirm you're still responsive."
|
||||
);
|
||||
expect(result.finalText.toLowerCase()).toContain("plan");
|
||||
|
||||
await session.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
},
|
||||
120_000
|
||||
);
|
||||
|
||||
test(
|
||||
"handles plan mode approval flow",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const client = new ClaudeAgentClient();
|
||||
const config: AgentSessionConfig = {
|
||||
provider: "claude",
|
||||
cwd,
|
||||
extra: { claude: { maxThinkingTokens: 2048 } },
|
||||
};
|
||||
const session = await client.createSession(config);
|
||||
await session.setMode("plan");
|
||||
|
||||
const events = session.stream(
|
||||
"Devise a plan to create a file named dummy.txt containing the word plan-test. After planning, proceed to execute your plan."
|
||||
);
|
||||
|
||||
let sawPlan = false;
|
||||
for await (const event of events) {
|
||||
await autoApprove(session, event);
|
||||
if (
|
||||
event.type === "timeline" &&
|
||||
event.item.type === "todo" &&
|
||||
event.item.items.some((entry) => entry.text.includes("dummy.txt"))
|
||||
) {
|
||||
sawPlan = true;
|
||||
}
|
||||
|
||||
if (event.type === "turn_completed" || event.type === "turn_failed") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
expect(sawPlan).toBe(true);
|
||||
expect(await session.getCurrentMode()).toBe("acceptEdits");
|
||||
|
||||
const filePath = path.join(cwd, "dummy.txt");
|
||||
expect(existsSync(filePath)).toBe(true);
|
||||
expect(readFileSync(filePath, "utf8")).toContain("plan-test");
|
||||
|
||||
await session.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
},
|
||||
180_000
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,707 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import {
|
||||
query,
|
||||
type AgentDefinition,
|
||||
type CanUseTool,
|
||||
type McpServerConfig as ClaudeMcpServerConfig,
|
||||
type Options as ClaudeOptions,
|
||||
type PermissionMode,
|
||||
type PermissionResult,
|
||||
type PermissionUpdate,
|
||||
type Query,
|
||||
type SDKMessage,
|
||||
type SDKPartialAssistantMessage,
|
||||
type SDKResultMessage,
|
||||
type SDKSystemMessage,
|
||||
type SDKUserMessage,
|
||||
} from "@anthropic-ai/claude-agent-sdk";
|
||||
|
||||
import type {
|
||||
AgentCapabilityFlags,
|
||||
AgentClient,
|
||||
AgentMode,
|
||||
AgentPermissionRequest,
|
||||
AgentPermissionResponse,
|
||||
AgentPermissionUpdate,
|
||||
AgentPersistenceHandle,
|
||||
AgentPromptInput,
|
||||
AgentRunOptions,
|
||||
AgentRunResult,
|
||||
AgentSession,
|
||||
AgentSessionConfig,
|
||||
AgentStreamEvent,
|
||||
AgentTimelineItem,
|
||||
AgentUsage,
|
||||
} from "../agent-sdk-types.js";
|
||||
|
||||
const CLAUDE_CAPABILITIES: AgentCapabilityFlags = {
|
||||
supportsStreaming: true,
|
||||
supportsSessionPersistence: true,
|
||||
supportsDynamicModes: true,
|
||||
supportsMcpServers: true,
|
||||
supportsReasoningStream: true,
|
||||
supportsToolInvocations: true,
|
||||
};
|
||||
|
||||
const DEFAULT_MODES: AgentMode[] = [
|
||||
{
|
||||
id: "default",
|
||||
label: "Always Ask",
|
||||
description: "Prompts for permission the first time a tool is used",
|
||||
},
|
||||
{
|
||||
id: "acceptEdits",
|
||||
label: "Accept File Edits",
|
||||
description: "Automatically approves edit-focused tools without prompting",
|
||||
},
|
||||
{
|
||||
id: "plan",
|
||||
label: "Plan Mode",
|
||||
description: "Analyze the codebase without executing tools or edits",
|
||||
},
|
||||
{
|
||||
id: "bypassPermissions",
|
||||
label: "Bypass Permissions",
|
||||
description: "Skip all permission prompts (use with caution)",
|
||||
},
|
||||
];
|
||||
|
||||
type ClaudeAgentConfig = AgentSessionConfig & { provider: "claude" };
|
||||
|
||||
type ClaudeContentChunk = { type: string; [key: string]: any };
|
||||
|
||||
type PendingPermission = {
|
||||
request: AgentPermissionRequest;
|
||||
resolve: (result: PermissionResult) => void;
|
||||
reject: (error: Error) => void;
|
||||
cleanup?: () => void;
|
||||
};
|
||||
|
||||
export class ClaudeAgentClient implements AgentClient {
|
||||
readonly provider = "claude" as const;
|
||||
readonly capabilities = CLAUDE_CAPABILITIES;
|
||||
|
||||
constructor(private readonly defaults?: { agents?: Record<string, AgentDefinition> }) {}
|
||||
|
||||
async createSession(config: AgentSessionConfig): Promise<AgentSession> {
|
||||
const claudeConfig = this.assertConfig(config);
|
||||
return new ClaudeAgentSession(claudeConfig, this.defaults);
|
||||
}
|
||||
|
||||
async resumeSession(
|
||||
handle: AgentPersistenceHandle,
|
||||
overrides?: Partial<AgentSessionConfig>
|
||||
): Promise<AgentSession> {
|
||||
const metadata = (handle.metadata ?? {}) as Partial<AgentSessionConfig>;
|
||||
const merged = { ...metadata, ...overrides } as Partial<AgentSessionConfig>;
|
||||
if (!merged.cwd) {
|
||||
throw new Error("Claude resume requires the original working directory in metadata");
|
||||
}
|
||||
const mergedConfig = { ...merged, provider: "claude" } as AgentSessionConfig;
|
||||
const claudeConfig = this.assertConfig(mergedConfig);
|
||||
return new ClaudeAgentSession(claudeConfig, this.defaults, handle);
|
||||
}
|
||||
|
||||
private assertConfig(config: AgentSessionConfig): ClaudeAgentConfig {
|
||||
if (config.provider !== "claude") {
|
||||
throw new Error(`ClaudeAgentClient received config for provider '${config.provider}'`);
|
||||
}
|
||||
return config as ClaudeAgentConfig;
|
||||
}
|
||||
}
|
||||
|
||||
class ClaudeAgentSession implements AgentSession {
|
||||
readonly provider = "claude" as const;
|
||||
readonly capabilities = CLAUDE_CAPABILITIES;
|
||||
|
||||
private readonly config: ClaudeAgentConfig;
|
||||
private readonly defaults?: { agents?: Record<string, AgentDefinition> };
|
||||
private query: Query | null = null;
|
||||
private input: Pushable<SDKUserMessage> | null = null;
|
||||
private claudeSessionId: string | null;
|
||||
private pendingLocalId: string;
|
||||
private persistence: AgentPersistenceHandle | null;
|
||||
private currentMode: PermissionMode;
|
||||
private availableModes: AgentMode[] = DEFAULT_MODES;
|
||||
private toolUseCache = new Map<string, { name: string; server: string }>();
|
||||
private pendingPermissions = new Map<string, PendingPermission>();
|
||||
private eventQueue: Pushable<AgentStreamEvent> | null = null;
|
||||
|
||||
constructor(
|
||||
config: ClaudeAgentConfig,
|
||||
defaults?: { agents?: Record<string, AgentDefinition> },
|
||||
handle?: AgentPersistenceHandle
|
||||
) {
|
||||
this.config = config;
|
||||
this.defaults = defaults;
|
||||
this.claudeSessionId = handle?.sessionId ?? null;
|
||||
this.pendingLocalId = this.claudeSessionId ?? `claude-${randomUUID()}`;
|
||||
this.persistence = handle ?? null;
|
||||
this.currentMode = (config.modeId as PermissionMode) ?? "default";
|
||||
}
|
||||
|
||||
get id(): string | null {
|
||||
return this.claudeSessionId;
|
||||
}
|
||||
|
||||
async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
sessionId: this.claudeSessionId ?? this.pendingLocalId,
|
||||
finalText,
|
||||
usage,
|
||||
timeline,
|
||||
};
|
||||
}
|
||||
|
||||
async *stream(
|
||||
prompt: AgentPromptInput,
|
||||
_options?: AgentRunOptions
|
||||
): AsyncGenerator<AgentStreamEvent> {
|
||||
const sdkMessage = this.toSdkUserMessage(prompt);
|
||||
const queue = new Pushable<AgentStreamEvent>();
|
||||
this.eventQueue = queue;
|
||||
void this.forwardPromptEvents(sdkMessage, queue);
|
||||
|
||||
try {
|
||||
for await (const event of queue) {
|
||||
yield event;
|
||||
if (event.type === "turn_completed" || event.type === "turn_failed") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (this.eventQueue === queue) {
|
||||
this.eventQueue = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
|
||||
// Claude Agent SDK does not provide a rollout replay API yet.
|
||||
}
|
||||
|
||||
async getAvailableModes(): Promise<AgentMode[]> {
|
||||
return this.availableModes;
|
||||
}
|
||||
|
||||
async getCurrentMode(): Promise<string | null> {
|
||||
return this.currentMode ?? null;
|
||||
}
|
||||
|
||||
async setMode(modeId: string): Promise<void> {
|
||||
const normalized = modeId as PermissionMode;
|
||||
const query = await this.ensureQuery();
|
||||
await query.setPermissionMode(normalized);
|
||||
this.currentMode = normalized;
|
||||
}
|
||||
|
||||
getPendingPermissions(): AgentPermissionRequest[] {
|
||||
return Array.from(this.pendingPermissions.values()).map((entry) => entry.request);
|
||||
}
|
||||
|
||||
async respondToPermission(requestId: string, response: AgentPermissionResponse): Promise<void> {
|
||||
const pending = this.pendingPermissions.get(requestId);
|
||||
if (!pending) {
|
||||
throw new Error(`No pending permission request with id '${requestId}'`);
|
||||
}
|
||||
this.pendingPermissions.delete(requestId);
|
||||
pending.cleanup?.();
|
||||
|
||||
if (response.behavior === "allow") {
|
||||
if (pending.request.kind === "plan") {
|
||||
await this.setMode("acceptEdits");
|
||||
this.enqueueTimeline({
|
||||
type: "command",
|
||||
command: "plan:approved",
|
||||
status: "granted",
|
||||
raw: pending.request,
|
||||
});
|
||||
}
|
||||
const result: PermissionResult = {
|
||||
behavior: "allow",
|
||||
updatedInput: response.updatedInput ?? pending.request.input ?? {},
|
||||
updatedPermissions: this.normalizePermissionUpdates(response.updatedPermissions),
|
||||
};
|
||||
pending.resolve(result);
|
||||
this.enqueueTimeline({
|
||||
type: "command",
|
||||
command: `permission:${pending.request.name}`,
|
||||
status: "granted",
|
||||
raw: { request: pending.request, response },
|
||||
});
|
||||
} else {
|
||||
const result: PermissionResult = {
|
||||
behavior: "deny",
|
||||
message: response.message ?? "Permission request denied",
|
||||
interrupt: response.interrupt,
|
||||
};
|
||||
pending.resolve(result);
|
||||
this.enqueueTimeline({
|
||||
type: "command",
|
||||
command: `permission:${pending.request.name}`,
|
||||
status: "denied",
|
||||
raw: { request: pending.request, response },
|
||||
});
|
||||
}
|
||||
|
||||
this.pushEvent({
|
||||
type: "permission_resolved",
|
||||
provider: "claude",
|
||||
requestId,
|
||||
resolution: response,
|
||||
});
|
||||
}
|
||||
|
||||
describePersistence(): AgentPersistenceHandle | null {
|
||||
if (this.persistence) {
|
||||
return this.persistence;
|
||||
}
|
||||
if (!this.claudeSessionId) {
|
||||
return null;
|
||||
}
|
||||
this.persistence = {
|
||||
provider: "claude",
|
||||
sessionId: this.claudeSessionId,
|
||||
nativeHandle: this.claudeSessionId,
|
||||
metadata: this.config,
|
||||
};
|
||||
return this.persistence;
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
this.rejectAllPendingPermissions(new Error("Claude session closed"));
|
||||
this.input?.end();
|
||||
await this.query?.return?.();
|
||||
this.query = null;
|
||||
this.input = null;
|
||||
}
|
||||
|
||||
private async ensureQuery(): Promise<Query> {
|
||||
if (this.query) {
|
||||
return this.query;
|
||||
}
|
||||
|
||||
const input = new Pushable<SDKUserMessage>();
|
||||
const options = this.buildOptions();
|
||||
this.input = input;
|
||||
this.query = query({ prompt: input, options });
|
||||
return this.query;
|
||||
}
|
||||
|
||||
private buildOptions(): ClaudeOptions {
|
||||
const base: ClaudeOptions = {
|
||||
cwd: this.config.cwd,
|
||||
includePartialMessages: true,
|
||||
permissionMode: this.currentMode,
|
||||
agents: this.defaults?.agents,
|
||||
canUseTool: this.handlePermissionRequest,
|
||||
...this.config.extra?.claude,
|
||||
};
|
||||
|
||||
if (this.config.mcpServers) {
|
||||
base.mcpServers = this.normalizeMcpServers(this.config.mcpServers);
|
||||
}
|
||||
if (this.config.model) {
|
||||
base.model = this.config.model;
|
||||
}
|
||||
if (this.claudeSessionId) {
|
||||
base.resume = this.claudeSessionId;
|
||||
base.continue = true;
|
||||
}
|
||||
return base;
|
||||
}
|
||||
|
||||
private normalizeMcpServers(
|
||||
servers: ClaudeAgentConfig["mcpServers"]
|
||||
): Record<string, ClaudeMcpServerConfig> | undefined {
|
||||
if (!servers) {
|
||||
return undefined;
|
||||
}
|
||||
const result: Record<string, ClaudeMcpServerConfig> = {};
|
||||
for (const [name, config] of Object.entries(servers)) {
|
||||
if (!config) continue;
|
||||
if ("type" in config || "command" in (config as any) || "url" in (config as any)) {
|
||||
result[name] = config as ClaudeMcpServerConfig;
|
||||
}
|
||||
}
|
||||
return Object.keys(result).length ? result : undefined;
|
||||
}
|
||||
|
||||
private toSdkUserMessage(prompt: AgentPromptInput): SDKUserMessage {
|
||||
const text = Array.isArray(prompt)
|
||||
? prompt.map((chunk) => chunk.text).join("\n\n")
|
||||
: prompt;
|
||||
|
||||
return {
|
||||
type: "user",
|
||||
message: {
|
||||
role: "user",
|
||||
content: [{ type: "text", text }],
|
||||
},
|
||||
parent_tool_use_id: null,
|
||||
session_id: this.claudeSessionId ?? this.pendingLocalId,
|
||||
};
|
||||
}
|
||||
|
||||
private async *processPrompt(
|
||||
sdkMessage: SDKUserMessage
|
||||
): AsyncGenerator<SDKMessage, void, undefined> {
|
||||
const query = await this.ensureQuery();
|
||||
if (!this.input) {
|
||||
throw new Error("Claude session input stream not initialized");
|
||||
}
|
||||
|
||||
this.input.push(sdkMessage);
|
||||
|
||||
while (true) {
|
||||
const { value, done } = await query.next();
|
||||
if (done) {
|
||||
break;
|
||||
}
|
||||
if (!value) {
|
||||
continue;
|
||||
}
|
||||
yield value;
|
||||
if (value.type === "result") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async forwardPromptEvents(message: SDKUserMessage, queue: Pushable<AgentStreamEvent>) {
|
||||
try {
|
||||
for await (const sdkEvent of this.processPrompt(message)) {
|
||||
const events = this.translateMessageToEvents(sdkEvent);
|
||||
for (const event of events) {
|
||||
queue.push(event);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
queue.push({
|
||||
type: "turn_failed",
|
||||
provider: "claude",
|
||||
error: error instanceof Error ? error.message : "Claude stream failed",
|
||||
});
|
||||
} finally {
|
||||
queue.end();
|
||||
}
|
||||
}
|
||||
|
||||
private translateMessageToEvents(message: SDKMessage): AgentStreamEvent[] {
|
||||
const events: AgentStreamEvent[] = [{ type: "provider_event", provider: "claude", raw: message }];
|
||||
|
||||
switch (message.type) {
|
||||
case "system":
|
||||
if (message.subtype === "init") {
|
||||
this.handleSystemMessage(message as SDKSystemMessage);
|
||||
}
|
||||
break;
|
||||
case "assistant": {
|
||||
const timelineItems = this.mapBlocksToTimeline(message.message.content);
|
||||
for (const item of timelineItems) {
|
||||
events.push({ type: "timeline", item, provider: "claude" });
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "stream_event": {
|
||||
const timelineItems = this.mapPartialEvent(message.event);
|
||||
for (const item of timelineItems) {
|
||||
events.push({ type: "timeline", item, provider: "claude" });
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "result": {
|
||||
const usage = this.convertUsage(message);
|
||||
if (message.subtype === "success") {
|
||||
events.push({ type: "turn_completed", provider: "claude", usage });
|
||||
} else {
|
||||
const errorMessage =
|
||||
"errors" in message && Array.isArray(message.errors) && message.errors.length > 0
|
||||
? message.errors.join("\n")
|
||||
: "Claude run failed";
|
||||
events.push({ type: "turn_failed", provider: "claude", error: errorMessage });
|
||||
}
|
||||
break;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
|
||||
return events;
|
||||
}
|
||||
|
||||
private handleSystemMessage(message: SDKSystemMessage): void {
|
||||
if (message.subtype !== "init") {
|
||||
return;
|
||||
}
|
||||
this.claudeSessionId = message.session_id;
|
||||
this.availableModes = DEFAULT_MODES;
|
||||
this.currentMode = message.permissionMode;
|
||||
this.persistence = null;
|
||||
}
|
||||
|
||||
private convertUsage(message: SDKResultMessage): AgentUsage | undefined {
|
||||
if (!message.usage) {
|
||||
return undefined;
|
||||
}
|
||||
return {
|
||||
inputTokens: message.usage.input_tokens,
|
||||
cachedInputTokens: message.usage.cache_read_input_tokens,
|
||||
outputTokens: message.usage.output_tokens,
|
||||
totalCostUsd: message.total_cost_usd,
|
||||
};
|
||||
}
|
||||
|
||||
private handlePermissionRequest: CanUseTool = async (
|
||||
toolName,
|
||||
input,
|
||||
options
|
||||
): Promise<PermissionResult> => {
|
||||
if (toolName === "ExitPlanMode") {
|
||||
this.emitPlanTodoItems(input);
|
||||
}
|
||||
|
||||
const requestId = `permission-${randomUUID()}`;
|
||||
const metadata: Record<string, unknown> = {};
|
||||
const permissionOptions = options as { toolUseID?: string } | undefined;
|
||||
if (permissionOptions?.toolUseID) {
|
||||
metadata.toolUseId = permissionOptions.toolUseID;
|
||||
}
|
||||
if (toolName === "ExitPlanMode" && typeof input.plan === "string") {
|
||||
metadata.planText = input.plan;
|
||||
}
|
||||
|
||||
const request: AgentPermissionRequest = {
|
||||
id: requestId,
|
||||
provider: "claude",
|
||||
name: toolName,
|
||||
kind: toolName === "ExitPlanMode" ? "plan" : "tool",
|
||||
input,
|
||||
suggestions: options?.suggestions as AgentPermissionUpdate[] | undefined,
|
||||
metadata: Object.keys(metadata).length ? metadata : undefined,
|
||||
raw: { toolName, input, options },
|
||||
};
|
||||
|
||||
this.enqueueTimeline({
|
||||
type: "command",
|
||||
command: `permission:${toolName}`,
|
||||
status: "requested",
|
||||
raw: { toolName, input },
|
||||
});
|
||||
|
||||
this.pushEvent({ type: "permission_requested", provider: "claude", request });
|
||||
|
||||
return await new Promise<PermissionResult>((resolve, reject) => {
|
||||
let cleanup: (() => void) | undefined;
|
||||
const abortHandler = () => {
|
||||
cleanup?.();
|
||||
this.pendingPermissions.delete(requestId);
|
||||
reject(new Error("Permission request aborted"));
|
||||
};
|
||||
|
||||
if (options?.signal) {
|
||||
if (options.signal.aborted) {
|
||||
abortHandler();
|
||||
return;
|
||||
}
|
||||
options.signal.addEventListener("abort", abortHandler, { once: true });
|
||||
cleanup = () => options.signal?.removeEventListener("abort", abortHandler);
|
||||
}
|
||||
|
||||
this.pendingPermissions.set(requestId, {
|
||||
request,
|
||||
resolve,
|
||||
reject,
|
||||
cleanup,
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
private emitPlanTodoItems(input: Record<string, unknown>) {
|
||||
const planText = typeof input.plan === "string" ? input.plan : JSON.stringify(input);
|
||||
const todoItems = planText
|
||||
.split("\n")
|
||||
.map((line) => line.trim())
|
||||
.filter(Boolean)
|
||||
.map((text) => ({ text, completed: false }));
|
||||
this.enqueueTimeline({
|
||||
type: "todo",
|
||||
items: todoItems.length > 0 ? todoItems : [{ text: planText, completed: false }],
|
||||
raw: input,
|
||||
});
|
||||
}
|
||||
|
||||
private enqueueTimeline(item: AgentTimelineItem) {
|
||||
this.pushEvent({ type: "timeline", item, provider: "claude" });
|
||||
}
|
||||
|
||||
private pushEvent(event: AgentStreamEvent) {
|
||||
if (this.eventQueue) {
|
||||
this.eventQueue.push(event);
|
||||
}
|
||||
}
|
||||
|
||||
private normalizePermissionUpdates(
|
||||
updates?: AgentPermissionUpdate[]
|
||||
): PermissionUpdate[] | undefined {
|
||||
if (!updates || updates.length === 0) {
|
||||
return undefined;
|
||||
}
|
||||
return updates as PermissionUpdate[];
|
||||
}
|
||||
|
||||
private rejectAllPendingPermissions(error: Error) {
|
||||
for (const [id, pending] of this.pendingPermissions) {
|
||||
pending.cleanup?.();
|
||||
pending.reject(error);
|
||||
this.pendingPermissions.delete(id);
|
||||
}
|
||||
}
|
||||
|
||||
private mapBlocksToTimeline(content: string | ClaudeContentChunk[]): AgentTimelineItem[] {
|
||||
if (typeof content === "string") {
|
||||
if (!content || content === "[Request interrupted by user for tool use]") {
|
||||
return [];
|
||||
}
|
||||
return [{ type: "assistant_message", text: content, raw: content }];
|
||||
}
|
||||
|
||||
const items: AgentTimelineItem[] = [];
|
||||
for (const block of content) {
|
||||
switch (block.type) {
|
||||
case "text":
|
||||
case "text_delta":
|
||||
if (block.text && block.text !== "[Request interrupted by user for tool use]") {
|
||||
items.push({ type: "assistant_message", text: block.text, raw: block });
|
||||
}
|
||||
break;
|
||||
case "thinking":
|
||||
case "thinking_delta":
|
||||
if (block.thinking) {
|
||||
items.push({ type: "reasoning", text: block.thinking, raw: block });
|
||||
}
|
||||
break;
|
||||
case "tool_use":
|
||||
case "server_tool_use":
|
||||
case "mcp_tool_use": {
|
||||
const name = block.name ?? "tool";
|
||||
const server = block.server ?? name;
|
||||
this.toolUseCache.set(block.id, { name, server });
|
||||
this.enqueueTimeline({
|
||||
type: "command",
|
||||
command: `permission:${name}`,
|
||||
status: "granted",
|
||||
raw: block,
|
||||
});
|
||||
items.push({
|
||||
type: "mcp_tool",
|
||||
server,
|
||||
tool: name,
|
||||
status: "pending",
|
||||
raw: block,
|
||||
});
|
||||
break;
|
||||
}
|
||||
case "tool_result":
|
||||
case "mcp_tool_result":
|
||||
case "web_fetch_tool_result":
|
||||
case "web_search_tool_result":
|
||||
case "code_execution_tool_result":
|
||||
case "bash_code_execution_tool_result":
|
||||
case "text_editor_code_execution_tool_result": {
|
||||
const cached = block.tool_use_id
|
||||
? this.toolUseCache.get(block.tool_use_id)
|
||||
: undefined;
|
||||
const server = cached?.server ?? block.server ?? "tool";
|
||||
const tool = cached?.name ?? block.tool_name ?? "tool";
|
||||
items.push({
|
||||
type: "mcp_tool",
|
||||
server,
|
||||
tool,
|
||||
status: block.is_error ? "failed" : "completed",
|
||||
raw: block,
|
||||
});
|
||||
if (!block.is_error && block.tool_use_id) {
|
||||
this.toolUseCache.delete(block.tool_use_id);
|
||||
}
|
||||
break;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
return items;
|
||||
}
|
||||
|
||||
private mapPartialEvent(event: SDKPartialAssistantMessage["event"]): AgentTimelineItem[] {
|
||||
switch (event.type) {
|
||||
case "content_block_start":
|
||||
return this.mapBlocksToTimeline([event.content_block as ClaudeContentChunk]);
|
||||
case "content_block_delta":
|
||||
return this.mapBlocksToTimeline([event.delta as ClaudeContentChunk]);
|
||||
default:
|
||||
return [];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class Pushable<T> implements AsyncIterable<T> {
|
||||
private queue: T[] = [];
|
||||
private resolvers: ((value: IteratorResult<T>) => void)[] = [];
|
||||
private closed = false;
|
||||
|
||||
push(item: T) {
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
if (this.resolvers.length > 0) {
|
||||
const resolve = this.resolvers.shift()!;
|
||||
resolve({ value: item, done: false });
|
||||
} else {
|
||||
this.queue.push(item);
|
||||
}
|
||||
}
|
||||
|
||||
end() {
|
||||
this.closed = true;
|
||||
while (this.resolvers.length > 0) {
|
||||
const resolve = this.resolvers.shift()!;
|
||||
resolve({ value: undefined as any, done: true });
|
||||
}
|
||||
}
|
||||
|
||||
[Symbol.asyncIterator](): AsyncIterator<T> {
|
||||
return {
|
||||
next: (): Promise<IteratorResult<T>> => {
|
||||
if (this.queue.length > 0) {
|
||||
const value = this.queue.shift()!;
|
||||
return Promise.resolve({ value, done: false });
|
||||
}
|
||||
if (this.closed) {
|
||||
return Promise.resolve({ value: undefined as any, done: true });
|
||||
}
|
||||
return new Promise<IteratorResult<T>>((resolve) => {
|
||||
this.resolvers.push(resolve);
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,225 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
import { CodexAgentClient } from "./codex-agent.js";
|
||||
import type { AgentSessionConfig, AgentStreamEvent } from "../agent-sdk-types.js";
|
||||
|
||||
function tmpCwd(): string {
|
||||
return mkdtempSync(path.join(os.tmpdir(), "codex-agent-e2e-"));
|
||||
}
|
||||
|
||||
function useTempCodexSessionDir(): () => void {
|
||||
const dir = mkdtempSync(path.join(os.tmpdir(), "codex-agent-session-"));
|
||||
const previous = process.env.CODEX_SESSION_DIR;
|
||||
process.env.CODEX_SESSION_DIR = dir;
|
||||
return () => {
|
||||
if (previous === undefined) {
|
||||
delete process.env.CODEX_SESSION_DIR;
|
||||
} else {
|
||||
process.env.CODEX_SESSION_DIR = previous;
|
||||
}
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
};
|
||||
}
|
||||
|
||||
function log(message: string): void {
|
||||
console.info(`[CodexAgentTest] ${message}`);
|
||||
}
|
||||
|
||||
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<ReturnType<typeof client.createSession>> | 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<ReturnType<typeof client.createSession>> | 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 === "file_change" || event.item.type === "command" || event.item.type === "mcp_tool")
|
||||
) {
|
||||
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(
|
||||
"supports multiple turns within the same session",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const restoreSessionDir = useTempCodexSessionDir();
|
||||
const client = new CodexAgentClient();
|
||||
const config: AgentSessionConfig = { provider: "codex", cwd };
|
||||
let session: Awaited<ReturnType<typeof client.createSession>> | 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 the exact text ACK-TWO and then stop.");
|
||||
expect(second.finalText.toLowerCase()).toContain("ack-two");
|
||||
|
||||
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<ReturnType<typeof client.createSession>> | 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<ReturnType<typeof client.createSession>> | null = null;
|
||||
let resumed: Awaited<ReturnType<typeof client.resumeSession>> | 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
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,899 @@
|
||||
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 {
|
||||
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,
|
||||
} from "../agent-sdk-types.js";
|
||||
|
||||
type CodexAgentConfig = AgentSessionConfig & { provider: "codex" };
|
||||
|
||||
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 DEFAULT_CODEX_MODE_ID = "auto";
|
||||
|
||||
const MODE_PRESETS: Record<
|
||||
string,
|
||||
Partial<ThreadOptions> & { 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<ThreadOptions> & { skipGitRepoCheck?: boolean };
|
||||
|
||||
const MAX_ROLLOUT_SEARCH_DEPTH = 5;
|
||||
|
||||
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<string, unknown>): 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<string, unknown>)[key] = value;
|
||||
}
|
||||
return normalized;
|
||||
}
|
||||
|
||||
export class CodexAgentClient implements AgentClient {
|
||||
readonly provider = "codex" as const;
|
||||
readonly capabilities = CODEX_CAPABILITIES;
|
||||
private readonly codex: Codex;
|
||||
|
||||
constructor(options?: CodexOptions) {
|
||||
this.codex = new Codex(options);
|
||||
}
|
||||
|
||||
async createSession(config: AgentSessionConfig): Promise<AgentSession> {
|
||||
const codexConfig = this.assertConfig(config);
|
||||
return CodexAgentSession.create(this.codex, codexConfig);
|
||||
}
|
||||
|
||||
async resumeSession(handle: AgentPersistenceHandle, overrides?: Partial<AgentSessionConfig>): Promise<AgentSession> {
|
||||
const metadata = (handle.metadata ?? {}) as Partial<AgentSessionConfig>;
|
||||
const merged = { ...metadata, ...overrides } as Partial<AgentSessionConfig>;
|
||||
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);
|
||||
return CodexAgentSession.create(this.codex, codexConfig, handle);
|
||||
}
|
||||
|
||||
private assertConfig(config: AgentSessionConfig): CodexAgentConfig {
|
||||
if (config.provider !== "codex") {
|
||||
throw new Error(`CodexAgentClient received config for provider '${config.provider}'`);
|
||||
}
|
||||
return config as CodexAgentConfig;
|
||||
}
|
||||
}
|
||||
|
||||
class CodexAgentSession implements AgentSession {
|
||||
static async create(codex: Codex, config: CodexAgentConfig, handle?: AgentPersistenceHandle): Promise<CodexAgentSession> {
|
||||
const session = new CodexAgentSession(codex, config, handle);
|
||||
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 historyEvents: AgentStreamEvent[] = [];
|
||||
|
||||
constructor(codex: Codex, config: CodexAgentConfig, handle?: AgentPersistenceHandle) {
|
||||
this.codex = codex;
|
||||
this.config = { ...config };
|
||||
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;
|
||||
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 run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
sessionId: this.threadId ?? this.pendingLocalId,
|
||||
finalText,
|
||||
usage,
|
||||
timeline,
|
||||
};
|
||||
}
|
||||
|
||||
async *stream(prompt: AgentPromptInput, options?: AgentRunOptions): AsyncGenerator<AgentStreamEvent> {
|
||||
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);
|
||||
|
||||
for await (const rawEvent of events) {
|
||||
yield* this.translateEvent(rawEvent);
|
||||
if (rawEvent.type === "turn.completed" || rawEvent.type === "turn.failed" || rawEvent.type === "error") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
|
||||
const replayEvents = this.drainHistoryEvents();
|
||||
for (const event of replayEvents) {
|
||||
yield event;
|
||||
}
|
||||
}
|
||||
|
||||
async getAvailableModes(): Promise<AgentMode[]> {
|
||||
return this.availableModes;
|
||||
}
|
||||
|
||||
async getCurrentMode(): Promise<string | null> {
|
||||
return this.currentMode ?? null;
|
||||
}
|
||||
|
||||
async setMode(modeId: string): Promise<void> {
|
||||
if (modeId === this.currentMode) {
|
||||
return;
|
||||
}
|
||||
if (!this.availableModes.some((mode) => mode.id === modeId)) {
|
||||
throw new Error(`Mode '${modeId}' not supported by Codex`);
|
||||
}
|
||||
|
||||
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 [];
|
||||
}
|
||||
|
||||
async respondToPermission(_requestId: string, _response: AgentPermissionResponse): Promise<void> {
|
||||
throw new Error("Codex permission responses are not implemented yet");
|
||||
}
|
||||
|
||||
describePersistence(): AgentPersistenceHandle | null {
|
||||
if (this.persistence) {
|
||||
return this.persistence;
|
||||
}
|
||||
if (!this.threadId) {
|
||||
return null;
|
||||
}
|
||||
this.persistence = {
|
||||
provider: "codex",
|
||||
sessionId: this.threadId,
|
||||
nativeHandle: this.threadId,
|
||||
metadata: {
|
||||
...this.config,
|
||||
codexSessionDir: this.codexSessionDir ?? undefined,
|
||||
codexRolloutPath: this.rolloutPath ?? undefined,
|
||||
},
|
||||
};
|
||||
return this.persistence;
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
this.thread = null;
|
||||
}
|
||||
|
||||
private drainHistoryEvents(): AgentStreamEvent[] {
|
||||
if (!this.historyEvents.length) {
|
||||
return [];
|
||||
}
|
||||
const events = [...this.historyEvents];
|
||||
this.historyEvents = [];
|
||||
return events;
|
||||
}
|
||||
|
||||
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<string, unknown> | undefined);
|
||||
if (extra) {
|
||||
Object.assign(options, extra);
|
||||
}
|
||||
|
||||
return options;
|
||||
}
|
||||
|
||||
private async loadReplayHistory(handle: AgentPersistenceHandle): Promise<void> {
|
||||
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;
|
||||
|
||||
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 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<AgentStreamEvent> {
|
||||
yield { type: "provider_event", provider: "codex", raw: event };
|
||||
|
||||
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, raw: event },
|
||||
};
|
||||
yield { type: "turn_failed", provider: "codex", error: message };
|
||||
break;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private threadItemToTimeline(item: ThreadItem): AgentTimelineItem | null {
|
||||
switch (item.type) {
|
||||
case "agent_message":
|
||||
return { type: "assistant_message", text: item.text, raw: item };
|
||||
case "reasoning":
|
||||
return { type: "reasoning", text: item.text, raw: item };
|
||||
case "command_execution":
|
||||
return { type: "command", command: item.command, status: item.status, raw: item };
|
||||
case "file_change":
|
||||
return {
|
||||
type: "file_change",
|
||||
files: item.changes.map((change) => ({ path: change.path, kind: change.kind })),
|
||||
raw: item,
|
||||
};
|
||||
case "mcp_tool_call":
|
||||
return {
|
||||
type: "mcp_tool",
|
||||
server: item.server,
|
||||
tool: item.tool,
|
||||
status: item.status,
|
||||
raw: item,
|
||||
};
|
||||
case "web_search":
|
||||
return { type: "web_search", query: item.query, raw: item };
|
||||
case "todo_list":
|
||||
return { type: "todo", items: item.items, raw: item };
|
||||
case "error":
|
||||
return { type: "error", message: item.message, raw: item };
|
||||
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<boolean> {
|
||||
try {
|
||||
await fs.access(filePath, fsConstants.F_OK);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function findRolloutFile(threadId: string, root: string): Promise<string | null> {
|
||||
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<AgentStreamEvent[]> {
|
||||
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<string, { command: string }>();
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
function parseRolloutEntry(line: string): { type: string; payload?: any } | null {
|
||||
if (!line) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
const parsed = JSON.parse(line);
|
||||
if (parsed && typeof parsed === "object" && "type" in parsed) {
|
||||
return parsed as { type: string; payload?: any };
|
||||
}
|
||||
if (parsed && typeof parsed === "object" && typeof (parsed as any).output === "string") {
|
||||
return parseRolloutEntry((parsed as any).output);
|
||||
}
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function handleRolloutResponseItem(
|
||||
payload: any,
|
||||
events: AgentStreamEvent[],
|
||||
commandCalls: Map<string, { command: string }>
|
||||
): void {
|
||||
if (!payload || typeof payload !== "object") {
|
||||
return;
|
||||
}
|
||||
|
||||
switch (payload.type) {
|
||||
case "message": {
|
||||
if (payload.role !== "assistant") {
|
||||
return;
|
||||
}
|
||||
const text = extractMessageText(payload.content);
|
||||
if (text) {
|
||||
events.push({ type: "timeline", provider: "codex", item: { type: "assistant_message", text, raw: payload } });
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "reasoning": {
|
||||
const text = extractReasoningText(payload);
|
||||
if (text) {
|
||||
events.push({ type: "timeline", provider: "codex", item: { type: "reasoning", text, raw: payload } });
|
||||
}
|
||||
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: any, 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, raw: payload } });
|
||||
}
|
||||
}
|
||||
|
||||
function handleRolloutFunctionCall(
|
||||
payload: any,
|
||||
events: AgentStreamEvent[],
|
||||
commandCalls: Map<string, { command: string }>
|
||||
): void {
|
||||
const name = typeof payload.name === "string" ? payload.name : undefined;
|
||||
if (!name) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (name === "shell") {
|
||||
const args = safeJsonParse(payload.arguments);
|
||||
const command = formatCommand(args);
|
||||
if (command && typeof payload.call_id === "string") {
|
||||
commandCalls.set(payload.call_id, { command });
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (name === "update_plan") {
|
||||
const args = safeJsonParse(payload.arguments);
|
||||
const planItems = parsePlanItems(args);
|
||||
if (planItems.length) {
|
||||
events.push({ type: "timeline", provider: "codex", item: { type: "todo", items: planItems, raw: payload } });
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
events.push({
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
item: {
|
||||
type: "mcp_tool",
|
||||
server: "codex",
|
||||
tool: name,
|
||||
status: payload.status ?? "in_progress",
|
||||
raw: payload,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function finalizeRolloutFunctionCall(
|
||||
payload: any,
|
||||
events: AgentStreamEvent[],
|
||||
commandCalls: Map<string, { command: string }>
|
||||
): void {
|
||||
if (typeof payload.call_id !== "string") {
|
||||
return;
|
||||
}
|
||||
const command = commandCalls.get(payload.call_id);
|
||||
if (!command) {
|
||||
return;
|
||||
}
|
||||
const result = safeJsonParse(payload.output);
|
||||
const exitCode = result?.metadata?.exit_code;
|
||||
const status = exitCode === undefined || exitCode === 0 ? "completed" : "failed";
|
||||
events.push({
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
item: {
|
||||
type: "command",
|
||||
command: command.command,
|
||||
status,
|
||||
raw: { payload, result },
|
||||
},
|
||||
});
|
||||
commandCalls.delete(payload.call_id);
|
||||
}
|
||||
|
||||
function handleRolloutCustomToolCall(payload: any, events: AgentStreamEvent[]): void {
|
||||
if (payload?.name === "apply_patch" && typeof payload.input === "string") {
|
||||
const files = parsePatchFiles(payload.input);
|
||||
if (files.length) {
|
||||
events.push({
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
item: {
|
||||
type: "file_change",
|
||||
files,
|
||||
raw: payload,
|
||||
},
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (typeof payload?.name === "string") {
|
||||
events.push({
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
item: {
|
||||
type: "mcp_tool",
|
||||
server: "codex",
|
||||
tool: payload.name,
|
||||
status: payload.status ?? "completed",
|
||||
raw: payload,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function extractMessageText(content: unknown): string {
|
||||
if (!Array.isArray(content)) {
|
||||
return "";
|
||||
}
|
||||
const parts: string[] = [];
|
||||
for (const block of content) {
|
||||
if (block && typeof block === "object" && (block as any).type === "output_text" && typeof (block as any).text === "string") {
|
||||
parts.push((block as any).text);
|
||||
}
|
||||
}
|
||||
return parts.join("\n").trim();
|
||||
}
|
||||
|
||||
function extractReasoningText(payload: any): string {
|
||||
if (Array.isArray(payload?.summary)) {
|
||||
const text = payload.summary
|
||||
.map((item: any) => (typeof item?.text === "string" ? item.text : ""))
|
||||
.filter(Boolean)
|
||||
.join("\n")
|
||||
.trim();
|
||||
if (text) {
|
||||
return text;
|
||||
}
|
||||
}
|
||||
if (typeof payload?.text === "string") {
|
||||
return payload.text;
|
||||
}
|
||||
return "";
|
||||
}
|
||||
|
||||
function parsePlanItems(args: any): { text: string; completed: boolean }[] {
|
||||
const plan = Array.isArray(args?.plan) ? args.plan : [];
|
||||
const items: { text: string; completed: boolean }[] = [];
|
||||
for (const step of plan) {
|
||||
if (!step || typeof step !== "object") {
|
||||
continue;
|
||||
}
|
||||
const text = typeof step.step === "string" ? step.step : typeof step.text === "string" ? step.text : "";
|
||||
if (!text) {
|
||||
continue;
|
||||
}
|
||||
const status = typeof step.status === "string" ? step.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<string>();
|
||||
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<T = any>(value: unknown): T | null {
|
||||
if (typeof value !== "string") {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
return JSON.parse(value) as T;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function formatCommand(args: any): string | null {
|
||||
if (!args) {
|
||||
return null;
|
||||
}
|
||||
if (Array.isArray(args.command)) {
|
||||
return args.command.join(" ");
|
||||
}
|
||||
if (typeof args.command === "string") {
|
||||
return args.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<string, unknown>)[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<string, unknown>)[key];
|
||||
return typeof value === "string" && value.length > 0 ? value : undefined;
|
||||
}
|
||||
@@ -101,8 +101,10 @@ export async function listDirectoryEntries({
|
||||
);
|
||||
|
||||
entries.sort((a, b) => {
|
||||
if (a.kind !== b.kind) {
|
||||
return a.kind === "directory" ? -1 : 1;
|
||||
const modifiedComparison =
|
||||
new Date(b.modifiedAt).getTime() - new Date(a.modifiedAt).getTime();
|
||||
if (modifiedComparison !== 0) {
|
||||
return modifiedComparison;
|
||||
}
|
||||
return a.name.localeCompare(b.name);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user