bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #188 from suraj-markup/session/ao-6 

feat(agent-codex): mature Codex plugin with full Agent interface parity
This commit is contained in:
suraj_markup 2026-02-27 15:21:09 +05:30 committed by GitHub
parent 3028cff4ab 433189a79b
commit d1fbb09eaf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 2734 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1067
packages/plugins/agent-codex/src/app-server-client.test.ts Normal file
View File

File diff suppressed because it is too large Load Diff

504
packages/plugins/agent-codex/src/app-server-client.ts Normal file
View File

@ -0,0 +1,504 @@
/**
* Codex App-Server JSON-RPC Client
*
* Manages a `codex app-server` subprocess and communicates via
* newline-delimited JSON over stdin/stdout. Provides typed methods
* for thread management, turn execution, and conversation resume.
*
* Protocol reference: Codex app-server developer guide
* Implementation reference: codex-autorunner integrations/app_server/client.py
*/
import { spawn, type ChildProcess } from "node:child_process";
import { randomUUID } from "node:crypto";
import { createInterface, type Interface as ReadlineInterface } from "node:readline";
import { EventEmitter } from "node:events";
// =============================================================================
// Types
// =============================================================================
/** JSON-RPC request sent from client to server */
export interface JsonRpcRequest {
jsonrpc: "2.0";
id: string;
method: string;
params: Record<string, unknown>;
}
/** JSON-RPC response from server */
export interface JsonRpcResponse {
id: string;
result?: Record<string, unknown>;
error?: JsonRpcError;
}
/** JSON-RPC error object */
export interface JsonRpcError {
code: number;
message: string;
data?: unknown;
}
/** JSON-RPC notification from server (no id) */
export interface JsonRpcNotification {
method: string;
params: Record<string, unknown>;
}
/** Approval request from server (has id + method + params) */
export interface JsonRpcApprovalRequest {
id: string | number;
method: string;
params: Record<string, unknown>;
}
/** Parsed message from the server — could be response, notification, or approval request */
type ServerMessage = JsonRpcResponse | JsonRpcNotification | JsonRpcApprovalRequest;
/** Callback for server notifications */
export type NotificationHandler = (method: string, params: Record<string, unknown>) => void;
/** Callback for approval requests */
export type ApprovalHandler = (
id: string | number,
method: string,
params: Record<string, unknown>,
) => Promise<ApprovalDecision>;
/** Approval decision values */
export type ApprovalDecision =
| "accept"
| "acceptForSession"
| "decline"
| "cancel";
/** Options for creating a CodexAppServerClient */
export interface AppServerClientOptions {
/** Path to codex binary (default: "codex") */
binaryPath?: string;
/** Working directory for the app-server process */
cwd?: string;
/** Environment variables for the process */
env?: Record<string, string>;
/** Timeout for requests in ms (default: 60000) */
requestTimeout?: number;
/** Handler for server notifications */
onNotification?: NotificationHandler;
/** Handler for approval requests (auto-accepts if not provided) */
onApproval?: ApprovalHandler;
}
/** Thread start parameters */
export interface ThreadStartParams {
model?: string;
modelProvider?: string;
cwd?: string;
/** Codex approval policy: untrusted (ask for all), on-request, or never */
approvalPolicy?: "untrusted" | "on-request" | "never";
/** Codex sandbox mode */
sandbox?: "read-only" | "workspace-write" | "danger-full-access";
/** Personality/instruction preset */
personality?: string;
}
/** Turn start parameters */
export interface TurnStartParams {
threadId: string;
input: string;
cwd?: string;
model?: string;
}
/** Pending request waiting for a response */
interface PendingRequest {
resolve: (result: Record<string, unknown>) => void;
reject: (error: Error) => void;
timer: ReturnType<typeof setTimeout>;
}
// =============================================================================
// Client Implementation
// =============================================================================
/**
* JSON-RPC client for Codex's app-server mode.
*
* Usage:
* ```ts
* const client = new CodexAppServerClient({ cwd: "/my/project" });
* await client.connect();
*
* const thread = await client.threadStart({ model: "o3-mini" });
* const turn = await client.turnStart({ threadId: thread.id, input: "Fix the bug" });
*
* await client.close();
* ```
*/
export class CodexAppServerClient extends EventEmitter {
private process: ChildProcess | null = null;
private readline: ReadlineInterface | null = null;
private pending = new Map<string, PendingRequest>();
private initialized = false;
private closed = false;
private connecting = false;
private readonly binaryPath: string;
private readonly cwd: string | undefined;
private readonly env: Record<string, string> | undefined;
private readonly requestTimeout: number;
private readonly onNotification: NotificationHandler | undefined;
private readonly onApproval: ApprovalHandler | undefined;
constructor(options: AppServerClientOptions = {}) {
super();
this.binaryPath = options.binaryPath ?? "codex";
this.cwd = options.cwd;
this.env = options.env;
this.requestTimeout = options.requestTimeout ?? 60_000;
this.onNotification = options.onNotification;
this.onApproval = options.onApproval;
}
/** Whether the client is connected and initialized */
get isConnected(): boolean {
return this.initialized && !this.closed && this.process !== null;
}
/**
* Spawn the app-server process and perform the initialization handshake.
* Must be called before any other method.
*/
async connect(): Promise<void> {
if (this.closed) throw new Error("Client is closed");
if (this.initialized) throw new Error("Client is already connected");
if (this.connecting) throw new Error("Client is already connecting");
this.connecting = true;
try {
this.process = spawn(this.binaryPath, ["app-server"], {
cwd: this.cwd,
env: this.env ? { ...process.env, ...this.env } : undefined,
stdio: ["pipe", "pipe", "pipe"],
});
if (!this.process.stdout || !this.process.stdin) {
throw new Error("Failed to open stdio pipes for codex app-server");
}
// Drain stderr to prevent the child process from blocking when
// the pipe buffer fills up.
this.process.stderr?.resume();
// Set up line-based reading from stdout
this.readline = createInterface({ input: this.process.stdout });
this.readline.on("line", (line) => this.handleLine(line));
// Handle process exit
this.process.once("exit", (code, signal) => {
this.handleProcessExit(code, signal);
});
this.process.once("error", (err) => {
this.handleProcessError(err);
});
await this.initialize();
} catch (err) {
this.connecting = false;
await this.close();
// Reset closed flag so the client can retry connect() after a
// transient handshake failure. The guard on line 172 ensures
// this.closed is always false when we reach this point.
this.closed = false;
throw err;
}
this.connecting = false;
}
/**
* Gracefully close the app-server process.
* Sends SIGTERM first, then SIGKILL after timeout.
*/
async close(): Promise<void> {
if (this.closed) return;
this.closed = true;
this.initialized = false;
// Reject all pending requests
for (const [id, pending] of this.pending) {
clearTimeout(pending.timer);
pending.reject(new Error("Client closed"));
this.pending.delete(id);
}
if (this.readline) {
this.readline.close();
this.readline = null;
}
if (this.process && this.process.exitCode === null) {
const proc = this.process;
await new Promise<void>((resolve) => {
const killTimer = setTimeout(() => {
try {
proc.kill("SIGKILL");
} catch {
// Already dead
}
resolve();
}, 5_000);
proc.once("exit", () => {
clearTimeout(killTimer);
resolve();
});
try {
proc.kill("SIGTERM");
} catch {
clearTimeout(killTimer);
resolve();
}
});
}
this.process = null;
}
// ---------------------------------------------------------------------------
// Thread Management
// ---------------------------------------------------------------------------
/** Create a new conversation thread */
async threadStart(params: ThreadStartParams = {}): Promise<Record<string, unknown>> {
return this.sendRequest("thread/start", { ...params });
}
/** Resume an existing conversation thread by ID */
async threadResume(threadId: string): Promise<Record<string, unknown>> {
return this.sendRequest("thread/resume", { threadId });
}
/** List threads (with optional cursor-based pagination) */
async threadList(cursor?: string, limit?: number): Promise<Record<string, unknown>> {
const params: Record<string, unknown> = {};
if (cursor) params["cursor"] = cursor;
if (limit !== undefined) params["limit"] = limit;
return this.sendRequest("thread/list", params);
}
/** Archive a thread */
async threadArchive(threadId: string): Promise<Record<string, unknown>> {
return this.sendRequest("thread/archive", { threadId });
}
// ---------------------------------------------------------------------------
// Turn Management
// ---------------------------------------------------------------------------
/** Start a new turn (send a message to the agent) */
async turnStart(params: TurnStartParams): Promise<Record<string, unknown>> {
return this.sendRequest("turn/start", {
threadId: params.threadId,
input: [{ type: "text", text: params.input }],
...(params.cwd ? { cwd: params.cwd } : {}),
...(params.model ? { model: params.model } : {}),
});
}
/** Interrupt a running turn */
async turnInterrupt(threadId: string, turnId: string): Promise<Record<string, unknown>> {
return this.sendRequest("turn/interrupt", { threadId, turnId });
}
// ---------------------------------------------------------------------------
// Model Discovery
// ---------------------------------------------------------------------------
/** List available models */
async modelList(cursor?: string, limit?: number): Promise<Record<string, unknown>> {
const params: Record<string, unknown> = {};
if (cursor) params["cursor"] = cursor;
if (limit !== undefined) params["limit"] = limit;
return this.sendRequest("model/list", params);
}
// ---------------------------------------------------------------------------
// Low-level Protocol
// ---------------------------------------------------------------------------
/** Send a JSON-RPC request and wait for the response */
async sendRequest(method: string, params: Record<string, unknown> = {}): Promise<Record<string, unknown>> {
if (!this.initialized && method !== "initialize") {
throw new Error("Client not initialized — call connect() first");
}
if (this.closed) throw new Error("Client is closed");
if (!this.process?.stdin?.writable) {
throw new Error("stdin not writable — process may have exited");
}
const id = randomUUID();
const request: JsonRpcRequest = { jsonrpc: "2.0", id, method, params };
return new Promise<Record<string, unknown>>((resolve, reject) => {
const timer = setTimeout(() => {
this.pending.delete(id);
reject(new Error(`Request ${method} timed out after ${this.requestTimeout}ms`));
}, this.requestTimeout);
this.pending.set(id, { resolve, reject, timer });
this.writeLine(JSON.stringify(request));
});
}
/** Send a JSON-RPC notification (no response expected) */
sendNotification(method: string, params: Record<string, unknown> = {}): void {
if (this.closed) return;
if (!this.process?.stdin?.writable) return;
this.writeLine(JSON.stringify({ jsonrpc: "2.0", method, params }));
}
/** Respond to an approval request from the server */
sendApprovalResponse(id: string | number, decision: ApprovalDecision): void {
if (this.closed) return;
if (!this.process?.stdin?.writable) return;
this.writeLine(JSON.stringify({ jsonrpc: "2.0", id, result: { decision } }));
}
// ---------------------------------------------------------------------------
// Internal
// ---------------------------------------------------------------------------
private async initialize(): Promise<void> {
const result = await this.sendRequest("initialize", {
clientInfo: {
name: "ao-agent-codex",
title: "Agent Orchestrator — Codex Plugin",
version: "0.1.0",
},
});
// Send the initialized notification to complete the handshake
this.sendNotification("initialized", {});
this.initialized = true;
this.emit("connected", result);
}
private writeLine(line: string): void {
if (!this.process?.stdin?.writable) return;
this.process.stdin.write(line + "\n");
}
private handleLine(line: string): void {
const trimmed = line.trim();
if (!trimmed) return;
let msg: ServerMessage;
try {
const parsed: unknown = JSON.parse(trimmed);
if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) return;
msg = parsed as ServerMessage;
} catch {
// Skip malformed lines
return;
}
// Classify the message
if ("id" in msg && msg.id !== undefined) {
const id = String(msg.id);
// Check if this is a response to a pending request
const pending = this.pending.get(id);
if (pending) {
this.pending.delete(id);
clearTimeout(pending.timer);
if ("error" in msg && msg.error) {
pending.reject(
new Error(`JSON-RPC error ${msg.error.code}: ${msg.error.message}`),
);
} else {
pending.resolve((msg as JsonRpcResponse).result ?? {});
}
return;
}
// If not a pending response, it's a server-initiated request (approval)
if ("method" in msg && typeof msg.method === "string") {
this.handleApprovalRequest(msg as JsonRpcApprovalRequest);
return;
}
}
// Notification (no id, has method)
if ("method" in msg && typeof msg.method === "string" && !("id" in msg)) {
const notification = msg as JsonRpcNotification;
this.emit("notification", notification.method, notification.params);
if (this.onNotification) {
this.onNotification(notification.method, notification.params);
}
}
}
private async handleApprovalRequest(request: JsonRpcApprovalRequest): Promise<void> {
try {
this.emit("approval", request.id, request.method, request.params);
if (this.onApproval) {
const decision = await this.onApproval(request.id, request.method, request.params);
this.sendApprovalResponse(request.id, decision);
} else {
// Default: auto-accept all approvals
this.sendApprovalResponse(request.id, "accept");
}
} catch {
// On any error (listener throw or handler rejection), decline the request
this.sendApprovalResponse(request.id, "decline");
}
}
private handleProcessExit(code: number | null, signal: string | null): void {
this.initialized = false;
// Close readline to release the event listener on the closed stdout stream
if (this.readline) {
this.readline.close();
this.readline = null;
}
// Reject all pending requests
const exitMsg = `codex app-server exited (code=${code}, signal=${signal})`;
for (const [id, pending] of this.pending) {
clearTimeout(pending.timer);
pending.reject(new Error(exitMsg));
this.pending.delete(id);
}
this.emit("exit", code, signal);
}
private handleProcessError(err: Error): void {
this.initialized = false;
// Close readline to release the event listener on the closed stdout stream
if (this.readline) {
this.readline.close();
this.readline = null;
}
// Reject all pending requests before emitting "error" — emit("error")
// with no listeners throws synchronously, which would skip cleanup.
for (const [id, pending] of this.pending) {
clearTimeout(pending.timer);
pending.reject(err);
this.pending.delete(id);
}
this.emit("error", err);
}
}

801
packages/plugins/agent-codex/src/index.test.ts
View File

@ -9,14 +9,24 @@ const {
mockWriteFile,
mockMkdir,
mockReadFile,
mockReaddir,
mockRename,
mockStat,
mockLstat,
mockOpen,
mockCreateReadStream,
mockHomedir,
} = vi.hoisted(() => ({
mockExecFileAsync: vi.fn(),
mockWriteFile: vi.fn().mockResolvedValue(undefined),
mockMkdir: vi.fn().mockResolvedValue(undefined),
mockReadFile: vi.fn(),
mockReaddir: vi.fn(),
mockRename: vi.fn().mockResolvedValue(undefined),
mockStat: vi.fn(),
mockLstat: vi.fn(),
mockOpen: vi.fn(),
mockCreateReadStream: vi.fn(),
mockHomedir: vi.fn(() => "/mock/home"),
}));
@ -31,7 +41,11 @@ vi.mock("node:fs/promises", () => ({
writeFile: mockWriteFile,
mkdir: mockMkdir,
readFile: mockReadFile,
readdir: mockReaddir,
rename: mockRename,
stat: mockStat,
lstat: mockLstat,
open: mockOpen,
}));
vi.mock("node:crypto", () => ({
@ -40,13 +54,15 @@ vi.mock("node:crypto", () => ({
vi.mock("node:fs", () => ({
existsSync: vi.fn(() => false),
createReadStream: mockCreateReadStream,
}));
vi.mock("node:os", () => ({
homedir: mockHomedir,
}));
import { create, manifest, default as defaultExport } from "./index.js";
import { Readable } from "node:stream";
import { create, manifest, default as defaultExport, resolveCodexBinary, _resetSessionFileCache } from "./index.js";
// ---------------------------------------------------------------------------
// Test helpers
@ -108,9 +124,58 @@ function mockTmuxWithProcess(processName: string, found = true) {
});
}
/**
* Create a mock file handle for `open()` that returns `content` from `read()`.
* Used by sessionFileMatchesCwd which reads only the first 4 KB.
*/
function makeFakeFileHandle(content: string) {
const buf = Buffer.from(content, "utf-8");
return {
read: vi.fn().mockImplementation((buffer: Buffer, offset: number, length: number, _position: number) => {
const bytesToCopy = Math.min(length, buf.length);
buf.copy(buffer, offset, 0, bytesToCopy);
return Promise.resolve({ bytesRead: bytesToCopy, buffer });
}),
close: vi.fn().mockResolvedValue(undefined),
};
}
/**
* Set up mockOpen so that any `open(path, "r")` call returns a fake handle
* reading `content`. This is used by sessionFileMatchesCwd.
*/
function setupMockOpen(content: string) {
mockOpen.mockResolvedValue(makeFakeFileHandle(content));
}
/**
* Create a Readable stream from a string. Used to mock createReadStream
* for the streaming JSONL parser (streamCodexSessionData).
*/
function makeContentStream(content: string): Readable {
return Readable.from(Buffer.from(content, "utf-8"));
}
/**
* Set up mockCreateReadStream to return a readable stream with the given content.
* Used by getSessionInfo/getRestoreCommand which now stream files line-by-line.
*/
function setupMockStream(content: string) {
mockCreateReadStream.mockReturnValue(makeContentStream(content));
}
beforeEach(() => {
vi.clearAllMocks();
_resetSessionFileCache();
mockHomedir.mockReturnValue("/mock/home");
// Default: open() returns a handle with empty content (no session_meta match).
// Session tests call setupMockOpen(content) to override.
mockOpen.mockResolvedValue(makeFakeFileHandle(""));
// Default: lstat rejects (no subdirectories). Session tests override as needed.
mockLstat.mockRejectedValue(new Error("ENOENT"));
// Default: createReadStream returns an empty stream. Session tests call
// setupMockStream(content) to override.
mockCreateReadStream.mockReturnValue(makeContentStream(""));
});
// =========================================================================
@ -145,12 +210,35 @@ describe("getLaunchCommand", () => {
const agent = create();
it("generates base command", () => {
expect(agent.getLaunchCommand(makeLaunchConfig())).toBe("codex");
expect(agent.getLaunchCommand(makeLaunchConfig())).toBe("'codex'");
});
it("includes --full-auto when permissions=skip", () => {
it("includes --dangerously-bypass-approvals-and-sandbox when permissions=skip", () => {
const cmd = agent.getLaunchCommand(makeLaunchConfig({ permissions: "skip" }));
expect(cmd).toContain("--full-auto");
expect(cmd).toContain("--dangerously-bypass-approvals-and-sandbox");
expect(cmd).not.toContain("--full-auto");
});
it("includes --ask-for-approval never when permissions=auto-edit", () => {
// Cast needed: "auto-edit" not yet in AgentLaunchConfig type union
const cmd = agent.getLaunchCommand(
makeLaunchConfig({ permissions: "auto-edit" as AgentLaunchConfig["permissions"] }),
);
expect(cmd).toContain("--ask-for-approval never");
});
it("includes --ask-for-approval untrusted when permissions=suggest", () => {
const cmd = agent.getLaunchCommand(
makeLaunchConfig({ permissions: "suggest" as AgentLaunchConfig["permissions"] }),
);
expect(cmd).toContain("--ask-for-approval untrusted");
});
it("omits approval flags when permissions=default", () => {
const cmd = agent.getLaunchCommand(makeLaunchConfig({ permissions: "default" }));
expect(cmd).not.toContain("--dangerously-bypass-approvals-and-sandbox");
expect(cmd).not.toContain("--ask-for-approval");
expect(cmd).not.toContain("--full-auto");
});
it("includes --model with shell-escaped value", () => {
@ -167,7 +255,7 @@ describe("getLaunchCommand", () => {
const cmd = agent.getLaunchCommand(
makeLaunchConfig({ permissions: "skip", model: "o3", prompt: "Go" }),
);
expect(cmd).toBe("codex --full-auto --model 'o3' -- 'Go'");
expect(cmd).toBe("'codex' --dangerously-bypass-approvals-and-sandbox --model 'o3' -c model_reasoning_effort=high -- 'Go'");
});
it("escapes single quotes in prompt (POSIX shell escaping)", () => {
@ -203,9 +291,54 @@ describe("getLaunchCommand", () => {
it("omits optional flags when not provided", () => {
const cmd = agent.getLaunchCommand(makeLaunchConfig());
expect(cmd).not.toContain("--full-auto");
expect(cmd).not.toContain("--dangerously-bypass-approvals-and-sandbox");
expect(cmd).not.toContain("--ask-for-approval");
expect(cmd).not.toContain("--model");
expect(cmd).not.toContain("-c");
expect(cmd).not.toContain("model_reasoning_effort");
});
// -- Reasoning effort tests --
describe("reasoning effort", () => {
it("adds model_reasoning_effort=high for o3 model", () => {
const cmd = agent.getLaunchCommand(makeLaunchConfig({ model: "o3" }));
expect(cmd).toContain("-c model_reasoning_effort=high");
});
it("adds model_reasoning_effort=high for o3-mini model", () => {
const cmd = agent.getLaunchCommand(makeLaunchConfig({ model: "o3-mini" }));
expect(cmd).toContain("-c model_reasoning_effort=high");
});
it("adds model_reasoning_effort=high for o4-mini model", () => {
const cmd = agent.getLaunchCommand(makeLaunchConfig({ model: "o4-mini" }));
expect(cmd).toContain("-c model_reasoning_effort=high");
});
it("adds model_reasoning_effort=high for O3 (case-insensitive)", () => {
const cmd = agent.getLaunchCommand(makeLaunchConfig({ model: "O3" }));
expect(cmd).toContain("-c model_reasoning_effort=high");
});
it("adds model_reasoning_effort=high for O4-MINI (case-insensitive)", () => {
const cmd = agent.getLaunchCommand(makeLaunchConfig({ model: "O4-MINI" }));
expect(cmd).toContain("-c model_reasoning_effort=high");
});
it("does NOT add reasoning effort for gpt-4o model", () => {
const cmd = agent.getLaunchCommand(makeLaunchConfig({ model: "gpt-4o" }));
expect(cmd).not.toContain("model_reasoning_effort");
});
it("does NOT add reasoning effort for gpt-4.1 model", () => {
const cmd = agent.getLaunchCommand(makeLaunchConfig({ model: "gpt-4.1" }));
expect(cmd).not.toContain("model_reasoning_effort");
});
it("does NOT add reasoning effort when no model specified", () => {
const cmd = agent.getLaunchCommand(makeLaunchConfig());
expect(cmd).not.toContain("model_reasoning_effort");
});
});
});
@ -443,54 +576,650 @@ describe("getActivityState", () => {
it("returns exited when no runtimeHandle", async () => {
const session = makeSession({ runtimeHandle: null });
const result = await agent.getActivityState(session);
expect(result).toEqual({ state: "exited" });
expect(result?.state).toBe("exited");
expect(result?.timestamp).toBeInstanceOf(Date);
});
it("returns exited when process is not running", async () => {
mockExecFileAsync.mockRejectedValue(new Error("tmux not running"));
const session = makeSession({ runtimeHandle: makeTmuxHandle() });
const result = await agent.getActivityState(session);
expect(result).toEqual({ state: "exited" });
expect(result?.state).toBe("exited");
expect(result?.timestamp).toBeInstanceOf(Date);
});
it("returns null (unknown) when process is running", async () => {
it("returns null when process is running but no workspacePath", async () => {
mockTmuxWithProcess("codex");
const session = makeSession({ runtimeHandle: makeTmuxHandle() });
const session = makeSession({ runtimeHandle: makeTmuxHandle(), workspacePath: undefined });
expect(await agent.getActivityState(session)).toBeNull();
});
it("returns null when process is running but no session file found", async () => {
mockTmuxWithProcess("codex");
mockReaddir.mockRejectedValue(new Error("ENOENT"));
const session = makeSession({ runtimeHandle: makeTmuxHandle(), workspacePath: "/workspace/test" });
expect(await agent.getActivityState(session)).toBeNull();
});
it("returns active when session file was recently modified", async () => {
mockTmuxWithProcess("codex");
const content = '{"type":"session_meta","cwd":"/workspace/test"}\n';
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
// mtime = now (just modified)
mockStat.mockResolvedValue({ mtimeMs: Date.now(), mtime: new Date() });
const session = makeSession({ runtimeHandle: makeTmuxHandle(), workspacePath: "/workspace/test" });
const result = await agent.getActivityState(session);
expect(result?.state).toBe("active");
expect(result?.timestamp).toBeInstanceOf(Date);
});
it("returns idle when session file is stale", async () => {
mockTmuxWithProcess("codex");
const content = '{"type":"session_meta","cwd":"/workspace/test"}\n';
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
// mtime = 10 minutes ago (past the 5-minute threshold)
const staleTime = Date.now() - 600_000;
mockStat.mockResolvedValue({ mtimeMs: staleTime, mtime: new Date(staleTime) });
const session = makeSession({ runtimeHandle: makeTmuxHandle(), workspacePath: "/workspace/test" });
const result = await agent.getActivityState(session);
expect(result?.state).toBe("idle");
expect(result?.timestamp).toBeInstanceOf(Date);
});
it("returns exited when process handle has dead PID", async () => {
const killSpy = vi.spyOn(process, "kill").mockImplementation(() => {
throw new Error("ESRCH");
});
const session = makeSession({ runtimeHandle: makeProcessHandle(999) });
const result = await agent.getActivityState(session);
expect(result).toEqual({ state: "exited" });
expect(result?.state).toBe("exited");
expect(result?.timestamp).toBeInstanceOf(Date);
killSpy.mockRestore();
});
it("does not include timestamp in exited state", async () => {
const session = makeSession({ runtimeHandle: null });
const result = await agent.getActivityState(session);
// The Codex implementation returns { state: "exited" } without timestamp
expect(result).toEqual({ state: "exited" });
expect(result?.timestamp).toBeUndefined();
});
});
// =========================================================================
// getSessionInfo
// getSessionInfo — Codex JSONL parsing
// =========================================================================
describe("getSessionInfo", () => {
const agent = create();
it("always returns null (not implemented)", async () => {
expect(await agent.getSessionInfo(makeSession())).toBeNull();
expect(await agent.getSessionInfo(makeSession({ workspacePath: "/some/path" }))).toBeNull();
// Helper to build JSONL content from lines
function jsonl(...lines: Record<string, unknown>[]): string {
return lines.map((l) => JSON.stringify(l)).join("\n") + "\n";
}
it("returns null when workspacePath is null", async () => {
expect(await agent.getSessionInfo(makeSession({ workspacePath: null }))).toBeNull();
});
it("returns null even with null workspacePath", async () => {
expect(await agent.getSessionInfo(makeSession({ workspacePath: null }))).toBeNull();
it("returns null when workspacePath is undefined", async () => {
expect(await agent.getSessionInfo(makeSession({ workspacePath: undefined }))).toBeNull();
});
it("returns null when ~/.codex/sessions/ directory does not exist", async () => {
mockReaddir.mockRejectedValue(new Error("ENOENT"));
expect(await agent.getSessionInfo(makeSession())).toBeNull();
});
it("returns null when sessions directory is empty", async () => {
mockReaddir.mockResolvedValue([]);
expect(await agent.getSessionInfo(makeSession())).toBeNull();
});
it("returns null when no session files match the workspace cwd", async () => {
mockReaddir.mockResolvedValue(["session-abc.jsonl"]);
const content = jsonl({ type: "session_meta", cwd: "/other/workspace", model: "gpt-4o" });
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
expect(await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }))).toBeNull();
});
it("returns session info with cost and model when matching session found", async () => {
const sessionContent = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "o3-mini" },
{ type: "event_msg", msg: { type: "token_count", input_tokens: 1000, output_tokens: 500, cached_tokens: 200, reasoning_tokens: 100 } },
{ type: "event_msg", msg: { type: "token_count", input_tokens: 2000, output_tokens: 300, cached_tokens: 0, reasoning_tokens: 0 } },
);
mockReaddir.mockResolvedValue(["session-123.jsonl"]);
setupMockOpen(sessionContent);
setupMockStream(sessionContent);
mockReadFile.mockResolvedValue(sessionContent);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
expect(result).not.toBeNull();
expect(result!.agentSessionId).toBe("session-123");
expect(result!.summary).toBe("Codex session (o3-mini)");
expect(result!.summaryIsFallback).toBe(true);
expect(result!.cost).toBeDefined();
// cached_tokens/reasoning_tokens are subsets, not additive
// input: 1000 + 2000 = 3000
// output: 500 + 300 = 800
expect(result!.cost!.inputTokens).toBe(3000);
expect(result!.cost!.outputTokens).toBe(800);
expect(result!.cost!.estimatedCostUsd).toBeGreaterThan(0);
});
it("picks the most recently modified matching session file", async () => {
const oldContent = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
);
const newContent = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "o3" },
);
mockReaddir.mockResolvedValue(["old-session.jsonl", "new-session.jsonl"]);
mockOpen.mockImplementation(async (path: string) => {
if (path.includes("old-session")) return makeFakeFileHandle(oldContent);
if (path.includes("new-session")) return makeFakeFileHandle(newContent);
throw new Error("ENOENT");
});
mockReadFile.mockImplementation((path: string) => {
if (path.includes("old-session")) return Promise.resolve(oldContent);
if (path.includes("new-session")) return Promise.resolve(newContent);
return Promise.reject(new Error("ENOENT"));
});
mockCreateReadStream.mockImplementation((path: string) => {
if (path.includes("old-session")) return makeContentStream(oldContent);
if (path.includes("new-session")) return makeContentStream(newContent);
return makeContentStream("");
});
mockStat.mockImplementation((path: string) => {
if (path.includes("old-session")) return Promise.resolve({ mtimeMs: 1000 });
if (path.includes("new-session")) return Promise.resolve({ mtimeMs: 2000 });
return Promise.reject(new Error("ENOENT"));
});
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
expect(result).not.toBeNull();
expect(result!.agentSessionId).toBe("new-session");
expect(result!.summary).toBe("Codex session (o3)");
});
it("handles corrupt/malformed JSONL lines gracefully", async () => {
const content = '{"type":"session_meta","cwd":"/workspace/test","model":"gpt-4o"}\n' +
"not valid json\n" +
'{"type":"event_msg","msg":{"type":"token_count","input_tokens":500,"output_tokens":200}}\n';
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
setupMockStream(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
expect(result).not.toBeNull();
expect(result!.cost!.inputTokens).toBe(500);
expect(result!.cost!.outputTokens).toBe(200);
});
it("returns null summary when no model in session_meta", async () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test" },
{ type: "event_msg", msg: { type: "token_count", input_tokens: 100, output_tokens: 50 } },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
setupMockStream(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
expect(result).not.toBeNull();
expect(result!.summary).toBeNull();
// Verify cost was actually parsed from the stream (not just defaulting to undefined)
expect(result!.cost).toBeDefined();
expect(result!.cost!.inputTokens).toBe(100);
expect(result!.cost!.outputTokens).toBe(50);
});
it("returns undefined cost when no token_count events", async () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ type: "event_msg", msg: { type: "other_event" } },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
setupMockStream(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
expect(result).not.toBeNull();
expect(result!.cost).toBeUndefined();
// Verify model was actually parsed from the stream (not just defaulting to null)
expect(result!.summary).toContain("gpt-4o");
});
it("handles unreadable session files gracefully", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
// open() finds matching session_meta, but readFile fails for full parse
setupMockOpen(jsonl({ type: "session_meta", cwd: "/workspace/test" }));
mockStat.mockResolvedValue({ mtimeMs: 1000 });
mockReadFile.mockRejectedValue(new Error("EACCES"));
mockCreateReadStream.mockImplementation(() => { throw new Error("EACCES"); });
expect(await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }))).toBeNull();
});
it("skips session files when stat throws", async () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockRejectedValue(new Error("EACCES"));
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
// stat failed so no bestMatch can be established
expect(result).toBeNull();
});
it("returns null when session JSONL has only empty/malformed lines", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
// open() finds matching session_meta for cwd check
setupMockOpen(jsonl({ type: "session_meta", cwd: "/workspace/test" }));
// readFile (full parse) returns only garbage
mockReadFile.mockResolvedValue("not json\n\n \n");
setupMockStream("not json\n\n \n");
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
// With streaming parser, garbage lines are skipped gracefully and a result
// is returned with null summary and undefined cost (no valid data extracted).
expect(result).not.toBeNull();
expect(result!.summary).toBeNull();
expect(result!.cost).toBeUndefined();
});
it("finds session files in date-sharded subdirectories (YYYY/MM/DD)", async () => {
// Simulate ~/.codex/sessions/2026/02/24/rollout-abc.jsonl
mockReaddir.mockImplementation((dir: string) => {
if (dir.endsWith("sessions")) return Promise.resolve(["2026"]);
if (dir.endsWith("2026")) return Promise.resolve(["02"]);
if (dir.endsWith("02")) return Promise.resolve(["24"]);
if (dir.endsWith("24")) return Promise.resolve(["rollout-abc.jsonl"]);
return Promise.resolve([]);
});
// lstat is used by collectJsonlFiles to check subdirectories (avoids symlink cycles)
mockLstat.mockResolvedValue({ isDirectory: () => true });
// stat is used by findCodexSessionFile to get mtimeMs of matching JSONL files
mockStat.mockResolvedValue({ mtimeMs: 2000 });
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "o3-mini" },
);
setupMockOpen(content);
setupMockStream(content);
mockReadFile.mockResolvedValue(content);
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
expect(result).not.toBeNull();
expect(result!.agentSessionId).toBe("rollout-abc");
expect(result!.summary).toBe("Codex session (o3-mini)");
});
it("ignores non-JSONL files in sessions directory", async () => {
mockReaddir.mockResolvedValue(["notes.txt", "config.json", "sess.jsonl"]);
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
);
setupMockOpen(content);
setupMockStream(content);
mockReadFile.mockResolvedValue(content);
// Non-JSONL entries trigger lstat to check isDirectory()
mockLstat.mockResolvedValue({ isDirectory: () => false });
// stat is used to get mtimeMs for matching JSONL files
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
expect(result).not.toBeNull();
// With streaming parser, readFile is no longer used for full parse;
// cwd check uses open(), data extraction uses createReadStream.
const readFileCalls = mockReadFile.mock.calls.filter(
(call: string[]) => typeof call[0] === "string" && call[0].includes("sessions/"),
);
expect(readFileCalls.length).toBe(0); // streaming replaces readFile for full parse
});
});
// =========================================================================
// getRestoreCommand — conversation resume
// =========================================================================
describe("getRestoreCommand", () => {
const agent = create();
function jsonl(...lines: Record<string, unknown>[]): string {
return lines.map((l) => JSON.stringify(l)).join("\n") + "\n";
}
function makeProjectConfig(overrides: Record<string, unknown> = {}) {
return {
name: "test-project",
repo: "owner/repo",
path: "/workspace/repo",
defaultBranch: "main",
sessionPrefix: "test",
...overrides,
};
}
it("returns null when workspacePath is null", async () => {
const session = makeSession({ workspacePath: null });
expect(await agent.getRestoreCommand!(session, makeProjectConfig())).toBeNull();
});
it("returns null when workspacePath is undefined", async () => {
const session = makeSession({ workspacePath: undefined });
expect(await agent.getRestoreCommand!(session, makeProjectConfig())).toBeNull();
});
it("returns null when no matching session file found", async () => {
mockReaddir.mockRejectedValue(new Error("ENOENT"));
const session = makeSession({ workspacePath: "/workspace/test" });
expect(await agent.getRestoreCommand!(session, makeProjectConfig())).toBeNull();
});
it("returns null when session has no threadId", async () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ role: "user", content: "Some prompt" },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
setupMockStream(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
// Native resume requires a threadId
expect(await agent.getRestoreCommand!(session, makeProjectConfig())).toBeNull();
});
it("builds native resume command with codex resume <threadId>", async () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ threadId: "thread-abc-123" },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
setupMockStream(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
const cmd = await agent.getRestoreCommand!(session, makeProjectConfig());
expect(cmd).not.toBeNull();
expect(cmd).toContain("'codex' resume");
expect(cmd).toContain("thread-abc-123");
});
it("includes --dangerously-bypass-approvals-and-sandbox from project config", async () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ threadId: "thread-1" },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
setupMockStream(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
const cmd = await agent.getRestoreCommand!(session, makeProjectConfig({
agentConfig: { permissions: "skip" },
}));
expect(cmd).toContain("--dangerously-bypass-approvals-and-sandbox");
});
it("includes --ask-for-approval never from project config", async () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ threadId: "thread-1" },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
setupMockStream(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
const cmd = await agent.getRestoreCommand!(session, makeProjectConfig({
agentConfig: { permissions: "auto-edit" },
}));
expect(cmd).toContain("--ask-for-approval never");
expect(cmd).not.toContain("--dangerously-bypass-approvals-and-sandbox");
});
it("includes --ask-for-approval untrusted from project config", async () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ threadId: "thread-1" },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
setupMockStream(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
const cmd = await agent.getRestoreCommand!(session, makeProjectConfig({
agentConfig: { permissions: "suggest" },
}));
expect(cmd).toContain("--ask-for-approval untrusted");
});
it("places flags before positional threadId in resume command", async () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "o3-mini" },
{ threadId: "thread-order-test" },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
setupMockStream(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
const cmd = await agent.getRestoreCommand!(session, makeProjectConfig({
agentConfig: { permissions: "auto-edit", model: "o3-mini" },
}));
expect(cmd).not.toBeNull();
// threadId should come after all flags
const threadIdIdx = cmd!.indexOf("thread-order-test");
const flagIdx = cmd!.indexOf("--ask-for-approval");
const modelIdx = cmd!.indexOf("--model");
expect(flagIdx).toBeLessThan(threadIdIdx);
expect(modelIdx).toBeLessThan(threadIdIdx);
});
it("includes model from project config (overrides session model)", async () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ threadId: "thread-1" },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
setupMockStream(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
const cmd = await agent.getRestoreCommand!(session, makeProjectConfig({
agentConfig: { model: "o3-mini" },
}));
expect(cmd).toContain("--model 'o3-mini'");
expect(cmd).toContain("-c model_reasoning_effort=high");
});
it("falls back to session model when project config has no model", async () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "o4-mini" },
{ threadId: "thread-1" },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
setupMockStream(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
const cmd = await agent.getRestoreCommand!(session, makeProjectConfig());
expect(cmd).toContain("--model 'o4-mini'");
expect(cmd).toContain("-c model_reasoning_effort=high");
});
it("handles unreadable session files gracefully", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
// open() finds matching session_meta for cwd check
setupMockOpen(jsonl({ type: "session_meta", cwd: "/workspace/test" }));
// readFile (full parse) fails
mockReadFile.mockRejectedValue(new Error("EACCES"));
mockCreateReadStream.mockImplementation(() => { throw new Error("EACCES"); });
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
expect(await agent.getRestoreCommand!(session, makeProjectConfig())).toBeNull();
});
});
// =========================================================================
// resolveCodexBinary
// =========================================================================
describe("resolveCodexBinary", () => {
it("returns path from `which` when codex is found", async () => {
mockExecFileAsync.mockResolvedValue({ stdout: "/usr/local/bin/codex\n", stderr: "" });
const result = await resolveCodexBinary();
expect(result).toBe("/usr/local/bin/codex");
expect(mockExecFileAsync).toHaveBeenCalledWith("which", ["codex"], { timeout: 10_000 });
});
it("falls back to common locations when `which` fails", async () => {
mockExecFileAsync.mockRejectedValue(new Error("not found"));
mockStat.mockImplementation((path: string) => {
if (path === "/usr/local/bin/codex") {
return Promise.resolve({ mtimeMs: 1000 });
}
return Promise.reject(new Error("ENOENT"));
});
const result = await resolveCodexBinary();
expect(result).toBe("/usr/local/bin/codex");
});
it("checks /opt/homebrew/bin/codex as fallback", async () => {
mockExecFileAsync.mockRejectedValue(new Error("not found"));
mockStat.mockImplementation((path: string) => {
if (path === "/opt/homebrew/bin/codex") {
return Promise.resolve({ mtimeMs: 1000 });
}
return Promise.reject(new Error("ENOENT"));
});
const result = await resolveCodexBinary();
expect(result).toBe("/opt/homebrew/bin/codex");
});
it("checks ~/.cargo/bin/codex as fallback (Rust-based codex)", async () => {
mockExecFileAsync.mockRejectedValue(new Error("not found"));
mockStat.mockImplementation((path: string) => {
if (path === "/mock/home/.cargo/bin/codex") {
return Promise.resolve({ mtimeMs: 1000 });
}
return Promise.reject(new Error("ENOENT"));
});
const result = await resolveCodexBinary();
expect(result).toBe("/mock/home/.cargo/bin/codex");
});
it("checks ~/.npm/bin/codex as fallback", async () => {
mockExecFileAsync.mockRejectedValue(new Error("not found"));
mockStat.mockImplementation((path: string) => {
if (path === "/mock/home/.npm/bin/codex") {
return Promise.resolve({ mtimeMs: 1000 });
}
return Promise.reject(new Error("ENOENT"));
});
const result = await resolveCodexBinary();
expect(result).toBe("/mock/home/.npm/bin/codex");
});
it("returns 'codex' when not found anywhere", async () => {
mockExecFileAsync.mockRejectedValue(new Error("not found"));
mockStat.mockRejectedValue(new Error("ENOENT"));
const result = await resolveCodexBinary();
expect(result).toBe("codex");
});
it("returns 'codex' when `which` returns empty stdout", async () => {
mockExecFileAsync.mockResolvedValue({ stdout: "", stderr: "" });
mockStat.mockRejectedValue(new Error("ENOENT"));
const result = await resolveCodexBinary();
expect(result).toBe("codex");
});
});
// =========================================================================
// postLaunchSetup — binary resolution
// =========================================================================
describe("postLaunchSetup", () => {
it("has postLaunchSetup method", () => {
const agent = create();
expect(typeof agent.postLaunchSetup).toBe("function");
});
it("runs setup when session has workspacePath", async () => {
const agent = create();
// which fails, stat fails → resolves to "codex"
mockExecFileAsync.mockRejectedValue(new Error("not found"));
mockStat.mockRejectedValue(new Error("ENOENT"));
mockReadFile.mockRejectedValue(new Error("ENOENT"));
await agent.postLaunchSetup!(makeSession({ workspacePath: "/workspace/test" }));
expect(mockMkdir).toHaveBeenCalled();
});
it("returns early when session has no workspacePath", async () => {
const agent = create();
mockExecFileAsync.mockRejectedValue(new Error("not found"));
mockStat.mockRejectedValue(new Error("ENOENT"));
await agent.postLaunchSetup!(makeSession({ workspacePath: undefined }));
expect(mockMkdir).not.toHaveBeenCalled();
});
it("resolves binary and uses it in getLaunchCommand after postLaunchSetup", async () => {
const agent = create();
mockExecFileAsync.mockResolvedValue({ stdout: "/opt/bin/codex\n", stderr: "" });
mockReadFile.mockRejectedValue(new Error("ENOENT"));
// Before postLaunchSetup, binary is "codex"
expect(agent.getLaunchCommand(makeLaunchConfig())).toBe("'codex'");
// After postLaunchSetup resolves the binary
await agent.postLaunchSetup!(makeSession({ workspacePath: "/workspace/test" }));
// Now getLaunchCommand should use the resolved binary
expect(agent.getLaunchCommand(makeLaunchConfig())).toBe("'/opt/bin/codex'");
});
});
@ -747,28 +1476,6 @@ describe("setupWorkspaceHooks", () => {
});
});
// =========================================================================
// postLaunchSetup
// =========================================================================
describe("postLaunchSetup", () => {
const agent = create();
it("has postLaunchSetup method", () => {
expect(typeof agent.postLaunchSetup).toBe("function");
});
it("runs setup when session has workspacePath", async () => {
mockReadFile.mockRejectedValue(new Error("ENOENT"));
await agent.postLaunchSetup!(makeSession({ workspacePath: "/workspace/test" }));
expect(mockMkdir).toHaveBeenCalled();
});
it("returns early when session has no workspacePath", async () => {
await agent.postLaunchSetup!(makeSession({ workspacePath: undefined }));
expect(mockMkdir).not.toHaveBeenCalled();
});
});
// =========================================================================
// Shell wrapper content verification
// =========================================================================

435
packages/plugins/agent-codex/src/index.ts
View File

@ -1,19 +1,24 @@
import {
DEFAULT_READY_THRESHOLD_MS,
shellEscape,
type Agent,
type AgentSessionInfo,
type AgentLaunchConfig,
type ActivityState,
type ActivityDetection,
type CostEstimate,
type PluginModule,
type ProjectConfig,
type RuntimeHandle,
type Session,
type WorkspaceHooksConfig,
} from "@composio/ao-core";
import { execFile } from "node:child_process";
import { writeFile, mkdir, readFile, rename } from "node:fs/promises";
import { createReadStream } from "node:fs";
import { writeFile, mkdir, readFile, readdir, rename, stat, lstat, open } from "node:fs/promises";
import { homedir } from "node:os";
import { join } from "node:path";
import { basename, join } from "node:path";
import { createInterface } from "node:readline";
import { promisify } from "node:util";
import { randomBytes } from "node:crypto";
@ -270,25 +275,304 @@ async function setupCodexWorkspace(workspacePath: string): Promise<void> {
}
}
// =============================================================================
// Codex Session JSONL Parsing (for getSessionInfo)
// =============================================================================
/** Codex session directory: ~/.codex/sessions/ */
const CODEX_SESSIONS_DIR = join(homedir(), ".codex", "sessions");
/** Typed representation of a line in a Codex JSONL session file */
interface CodexJsonlLine {
type?: string;
cwd?: string;
model?: string;
// Thread ID from thread_started notifications
threadId?: string;
// User message content (from user input events)
content?: string;
role?: string;
// event_msg with token_count subtype
msg?: {
type?: string;
input_tokens?: number;
output_tokens?: number;
cached_tokens?: number;
reasoning_tokens?: number;
};
}
/**
* Collect all JSONL files under a directory, recursively.
* Codex stores sessions in date-sharded directories:
* ~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl
*
* Uses lstat (not stat) so symlinks to directories are never followed,
* preventing infinite loops from symlink cycles. Max depth is capped at 4
* (YYYY/MM/DD + 1 buffer) as an additional safety guard.
*/
const MAX_SESSION_SCAN_DEPTH = 4;
async function collectJsonlFiles(dir: string, depth = 0): Promise<string[]> {
if (depth > MAX_SESSION_SCAN_DEPTH) return [];
let entries: string[];
try {
entries = await readdir(dir);
} catch {
return [];
}
const results: string[] = [];
for (const entry of entries) {
const fullPath = join(dir, entry);
if (entry.endsWith(".jsonl")) {
results.push(fullPath);
} else {
// Recurse into subdirectories (YYYY/MM/DD structure).
// Use lstat to avoid following symlinks that could create cycles.
try {
const s = await lstat(fullPath);
if (s.isDirectory()) {
const nested = await collectJsonlFiles(fullPath, depth + 1);
results.push(...nested);
}
} catch {
// Skip inaccessible entries
}
}
}
return results;
}
/**
* Check if the first few lines of a JSONL file contain a session_meta
* entry matching the given workspace path. Reads only the first 4 KB
* to avoid loading large rollout files into memory.
*/
async function sessionFileMatchesCwd(
filePath: string,
workspacePath: string,
): Promise<boolean> {
try {
// Read only the first 4 KB — session_meta is always in the first few lines.
// Avoids loading large rollout files (100 MB+) into memory.
const handle = await open(filePath, "r");
let content: string;
try {
const buffer = Buffer.allocUnsafe(4096);
const { bytesRead } = await handle.read(buffer, 0, 4096, 0);
content = buffer.subarray(0, bytesRead).toString("utf-8");
} finally {
await handle.close();
}
const lines = content.split("\n").slice(0, 10);
for (const line of lines) {
const trimmed = line.trim();
if (!trimmed) continue;
try {
const parsed: unknown = JSON.parse(trimmed);
if (
typeof parsed === "object" &&
parsed !== null &&
!Array.isArray(parsed) &&
(parsed as CodexJsonlLine).type === "session_meta" &&
(parsed as CodexJsonlLine).cwd === workspacePath
) {
return true;
}
} catch {
// Skip malformed lines
}
}
} catch {
// Unreadable file
}
return false;
}
/**
* Find Codex session files whose `session_meta` cwd matches the given workspace path.
* Recursively scans ~/.codex/sessions/ (date-sharded: YYYY/MM/DD/rollout-*.jsonl).
* Returns the path to the most recently modified matching file, or null.
*/
async function findCodexSessionFile(workspacePath: string): Promise<string | null> {
const jsonlFiles = await collectJsonlFiles(CODEX_SESSIONS_DIR);
if (jsonlFiles.length === 0) return null;
let bestMatch: { path: string; mtime: number } | null = null;
for (const filePath of jsonlFiles) {
const matches = await sessionFileMatchesCwd(filePath, workspacePath);
if (matches) {
try {
const s = await stat(filePath);
if (!bestMatch || s.mtimeMs > bestMatch.mtime) {
bestMatch = { path: filePath, mtime: s.mtimeMs };
}
} catch {
// Skip if stat fails
}
}
}
return bestMatch?.path ?? null;
}
/** Aggregated data extracted from a Codex session file via streaming */
interface CodexSessionData {
model: string | null;
threadId: string | null;
inputTokens: number;
outputTokens: number;
}
/**
* Stream a Codex JSONL session file line-by-line and aggregate the data
* we need (model, threadId, token counts) without loading the entire file
* into memory. This is critical because Codex rollout files can be 100 MB+.
*/
async function streamCodexSessionData(filePath: string): Promise<CodexSessionData | null> {
try {
const data: CodexSessionData = { model: null, threadId: null, inputTokens: 0, outputTokens: 0 };
const rl = createInterface({
input: createReadStream(filePath, { encoding: "utf-8" }),
crlfDelay: Infinity,
});
for await (const line of rl) {
const trimmed = line.trim();
if (!trimmed) continue;
try {
const parsed: unknown = JSON.parse(trimmed);
if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) continue;
const entry = parsed as CodexJsonlLine;
if (entry.type === "session_meta" && typeof entry.model === "string") {
data.model = entry.model;
}
if (typeof entry.threadId === "string" && entry.threadId) {
data.threadId = entry.threadId;
}
if (entry.type === "event_msg" && entry.msg?.type === "token_count") {
data.inputTokens += entry.msg.input_tokens ?? 0;
data.outputTokens += entry.msg.output_tokens ?? 0;
}
} catch {
// Skip malformed lines
}
}
return data;
} catch {
return null;
}
}
// =============================================================================
// Binary Resolution
// =============================================================================
/**
* Resolve the Codex CLI binary path.
* Checks (in order): which, common fallback locations.
* Returns "codex" as final fallback (let the shell resolve it at runtime).
*/
export async function resolveCodexBinary(): Promise<string> {
// 1. Try `which codex`
try {
const { stdout } = await execFileAsync("which", ["codex"], { timeout: 10_000 });
const resolved = stdout.trim();
if (resolved) return resolved;
} catch {
// Not found via which
}
// 2. Check common locations (npm global, Homebrew, Cargo — Codex is now Rust-based)
const home = homedir();
const candidates = [
"/usr/local/bin/codex",
"/opt/homebrew/bin/codex",
join(home, ".cargo", "bin", "codex"),
join(home, ".npm", "bin", "codex"),
];
for (const candidate of candidates) {
try {
await stat(candidate);
return candidate;
} catch {
// Not found at this location
}
}
// 3. Fallback: let the shell resolve it
return "codex";
}
// =============================================================================
// Agent Implementation
// =============================================================================
/** Append approval-policy flags to a command parts array */
function appendApprovalFlags(parts: string[], permissions: string | undefined): void {
if (permissions === "skip") {
parts.push("--dangerously-bypass-approvals-and-sandbox");
} else if (permissions === "auto-edit") {
parts.push("--ask-for-approval", "never");
} else if (permissions === "suggest") {
parts.push("--ask-for-approval", "untrusted");
}
}
/** Append model and reasoning flags to a command parts array */
function appendModelFlags(parts: string[], model: string | undefined): void {
if (!model) return;
parts.push("--model", shellEscape(model));
// Auto-detect o-series models and enable reasoning via config override.
// Codex does not have a --reasoning flag; reasoning is controlled via
// the model_reasoning_effort config key.
if (/^o[34]/i.test(model)) {
parts.push("-c", "model_reasoning_effort=high");
}
}
/** TTL for session file path cache (ms). Prevents redundant filesystem scans
* when getActivityState and getSessionInfo are called in the same refresh cycle. */
const SESSION_FILE_CACHE_TTL_MS = 30_000;
/** Module-level session file cache shared across the agent instance lifetime.
* Keyed by workspace path, stores the resolved file path and an expiry timestamp. */
const sessionFileCache = new Map<string, { path: string | null; expiry: number }>();
/** Find session file with caching to avoid double scans per refresh cycle */
async function findCodexSessionFileCached(workspacePath: string): Promise<string | null> {
const cached = sessionFileCache.get(workspacePath);
if (cached && Date.now() < cached.expiry) {
return cached.path;
}
const result = await findCodexSessionFile(workspacePath);
sessionFileCache.set(workspacePath, { path: result, expiry: Date.now() + SESSION_FILE_CACHE_TTL_MS });
return result;
}
function createCodexAgent(): Agent {
/** Cached resolved binary path (populated by init or first getLaunchCommand) */
let resolvedBinary: string | null = null;
/** Guard against concurrent resolveCodexBinary() calls */
let resolvingBinary: Promise<string> | null = null;
return {
name: "codex",
processName: "codex",
getLaunchCommand(config: AgentLaunchConfig): string {
const parts: string[] = ["codex"];
const binary = resolvedBinary ?? "codex";
const parts: string[] = [shellEscape(binary)];
if (config.permissions === "skip") {
parts.push("--full-auto");
}
if (config.model) {
parts.push("--model", shellEscape(config.model));
}
appendApprovalFlags(parts, config.permissions as string | undefined);
appendModelFlags(parts, config.model);
if (config.systemPromptFile) {
// Codex reads developer instructions from a file via config override
@ -342,20 +626,38 @@ function createCodexAgent(): Agent {
return "active";
},
async getActivityState(session: Session, _readyThresholdMs?: number): Promise<ActivityDetection | null> {
// Check if process is running first
if (!session.runtimeHandle) return { state: "exited" };
const running = await this.isProcessRunning(session.runtimeHandle);
if (!running) return { state: "exited" };
async getActivityState(session: Session, readyThresholdMs?: number): Promise<ActivityDetection | null> {
const threshold = readyThresholdMs ?? DEFAULT_READY_THRESHOLD_MS;
// NOTE: Codex stores rollout files in a global ~/.codex/sessions/ directory
// without workspace-specific scoping. When multiple Codex sessions run in
// parallel, we cannot reliably determine which rollout file belongs to which
// session. Until Codex provides per-workspace session tracking, we return
// null (unknown) rather than guessing. See issue #13 for details.
//
// TODO: Implement proper per-session activity detection when Codex supports it.
return null;
// Check if process is running first
const exitedAt = new Date();
if (!session.runtimeHandle) return { state: "exited", timestamp: exitedAt };
const running = await this.isProcessRunning(session.runtimeHandle);
if (!running) return { state: "exited", timestamp: exitedAt };
// Use session file mtime as a proxy for activity. Codex continuously
// appends to its rollout JSONL file while working, so a recently
// modified file means the agent is active.
if (!session.workspacePath) return null;
const sessionFile = await findCodexSessionFileCached(session.workspacePath);
if (!sessionFile) return null;
try {
const s = await stat(sessionFile);
const timestamp = s.mtime;
const ageMs = Date.now() - s.mtimeMs;
if (ageMs <= threshold) {
// File was recently modified — agent is actively working
return { state: "active", timestamp };
}
// File is stale — agent finished or is idle
return { state: "idle", timestamp };
} catch {
return null;
}
},
async isProcessRunning(handle: RuntimeHandle): Promise<boolean> {
@ -409,9 +711,63 @@ function createCodexAgent(): Agent {
}
},
async getSessionInfo(_session: Session): Promise<AgentSessionInfo | null> {
// Codex doesn't have JSONL session files for introspection yet
return null;
async getSessionInfo(session: Session): Promise<AgentSessionInfo | null> {
if (!session.workspacePath) return null;
const sessionFile = await findCodexSessionFileCached(session.workspacePath);
if (!sessionFile) return null;
// Stream the file line-by-line to avoid loading potentially huge
// rollout files (100 MB+) entirely into memory.
const data = await streamCodexSessionData(sessionFile);
if (!data) return null;
const agentSessionId = basename(sessionFile, ".jsonl");
const cost: CostEstimate | undefined =
data.inputTokens === 0 && data.outputTokens === 0
? undefined
: {
inputTokens: data.inputTokens,
outputTokens: data.outputTokens,
estimatedCostUsd:
(data.inputTokens / 1_000_000) * 2.5 + (data.outputTokens / 1_000_000) * 10.0,
};
return {
summary: data.model ? `Codex session (${data.model})` : null,
summaryIsFallback: true,
agentSessionId,
cost,
};
},
async getRestoreCommand(session: Session, project: ProjectConfig): Promise<string | null> {
if (!session.workspacePath) return null;
// Find the Codex session file for this workspace
const sessionFile = await findCodexSessionFileCached(session.workspacePath);
if (!sessionFile) return null;
// Stream the file line-by-line to avoid loading potentially huge
// rollout files (100 MB+) entirely into memory.
const data = await streamCodexSessionData(sessionFile);
if (!data?.threadId) return null;
// Use Codex's native `resume` subcommand for proper conversation resume.
// This restores the full thread state, not just a text prompt re-injection.
// Flags are placed before the positional threadId for CLI parser compatibility.
const binary = resolvedBinary ?? "codex";
const parts: string[] = [shellEscape(binary), "resume"];
appendApprovalFlags(parts, project.agentConfig?.permissions as string | undefined);
const effectiveModel = (project.agentConfig?.model ?? data.model) as string | undefined;
appendModelFlags(parts, effectiveModel ?? undefined);
// Positional threadId goes last, after all flags
parts.push(shellEscape(data.threadId));
return parts.join(" ");
},
async setupWorkspaceHooks(workspacePath: string, _config: WorkspaceHooksConfig): Promise<void> {
@ -419,6 +775,18 @@ function createCodexAgent(): Agent {
},
async postLaunchSetup(session: Session): Promise<void> {
// Resolve binary path on first launch (cached for subsequent calls).
// Uses a promise guard to prevent concurrent calls from racing.
if (!resolvedBinary) {
if (!resolvingBinary) {
resolvingBinary = resolveCodexBinary();
}
try {
resolvedBinary = await resolvingBinary;
} finally {
resolvingBinary = null;
}
}
if (!session.workspacePath) return;
await setupCodexWorkspace(session.workspacePath);
},
@ -433,4 +801,19 @@ export function create(): Agent {
return createCodexAgent();
}
/** @internal Clear the session file cache. Exported for testing only. */
export function _resetSessionFileCache(): void {
sessionFileCache.clear();
}
export { CodexAppServerClient } from "./app-server-client.js";
export type {
AppServerClientOptions,
ThreadStartParams,
TurnStartParams,
NotificationHandler,
ApprovalHandler,
ApprovalDecision,
} from "./app-server-client.js";
export default { manifest, create } satisfies PluginModule<Agent>;