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

fix(agent-codex): address review comments — resource cleanup, concurrency, and memory 

Fixes 6 issues flagged by Cursor Bugbot on PR #188:

1. sessionFileMatchesCwd reads only first 4 KB via open()+read() instead of
   loading entire rollout file into memory (Medium)
2. Binary resolution race condition: added promise guard to prevent
   concurrent resolveCodexBinary() calls (Low)
3. Concurrent connect() calls: added connecting guard to prevent
   orphaned child processes (High)
4. connect() failure cleanup: wrap initialize() in try/catch that
   calls close() on failure (Medium)
5. readline cleanup on process exit/error: close readline in both
   handleProcessExit and handleProcessError (Medium)
6. stderr pipe drain: call stderr.resume() to prevent child blocking
   when stderr buffer fills (High)

All 177 tests pass, typecheck clean.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
suraj-markup 2026-02-25 00:38:58 +05:30
parent 82317cbdeb
commit 4e7129d50d
4 changed files with 235 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
packages/plugins/agent-codex/src/app-server-client.test.ts
View File

@ -912,4 +912,92 @@ describe("CodexAppServerClient", () => {
await closePromise;
});
});
// =========================================================================
// Concurrency & resource cleanup (review fixes)
// =========================================================================
describe("concurrency guards", () => {
it("throws if connect() is called while already connecting", async () => {
const proc = createFakeProcess();
const client = new CodexAppServerClient({ requestTimeout: 500 });
// Start connect but don't complete the handshake
const connectPromise = client.connect();
// Second connect() should throw immediately
await expect(client.connect()).rejects.toThrow("already connecting");
// Complete the handshake to clean up
await new Promise((r) => setTimeout(r, 10));
const initReq = findRequest(proc, "initialize");
proc.sendLine(JSON.stringify({ id: initReq!["id"], result: {} }));
await connectPromise;
await closeClient(client, proc);
});
it("cleans up process when initialize handshake fails", async () => {
const proc = createFakeProcess();
// Make kill trigger exit so close() doesn't hang
vi.spyOn(proc, "kill").mockImplementation((_signal?: string) => {
proc.simulateExit(1, "SIGTERM");
return true;
});
const client = new CodexAppServerClient({ requestTimeout: 200 });
// Start connect — don't answer the initialize request so it times out
await expect(client.connect()).rejects.toThrow("timed out");
// Client should have cleaned up — not connected, not connecting
expect(client.isConnected).toBe(false);
});
it("drains stderr without blocking the child process", async () => {
const proc = createFakeProcess();
const client = new CodexAppServerClient({ requestTimeout: 500 });
await connectClient(client, proc);
// Write a large amount of data to stderr — should not block
const largeChunk = "x".repeat(65536);
proc.stderr.write(largeChunk);
await new Promise((r) => setTimeout(r, 10));
expect(client.isConnected).toBe(true);
await closeClient(client, proc);
});
});
describe("readline cleanup on process exit/error", () => {
it("closes readline on unexpected process exit", async () => {
const proc = createFakeProcess();
const client = new CodexAppServerClient({ requestTimeout: 500 });
await connectClient(client, proc);
// Spy on readline close — the client stores it privately so we check
// indirectly: after exit, sending another line should not cause errors
proc.simulateExit(1, null);
await new Promise((r) => setTimeout(r, 10));
// Client should be disconnected
expect(client.isConnected).toBe(false);
// Sending data on stdout should not throw (readline is closed)
expect(() => proc.sendLine('{"id":"x","result":{}}')).not.toThrow();
});
it("closes readline on process error", async () => {
const proc = createFakeProcess();
const client = new CodexAppServerClient({ requestTimeout: 500 });
await connectClient(client, proc);
// Catch the error event on the client to prevent unhandled error
const errors: Error[] = [];
client.on("error", (err: Error) => errors.push(err));
proc.emit("error", new Error("process crashed"));
await new Promise((r) => setTimeout(r, 10));
expect(client.isConnected).toBe(false);
expect(errors).toHaveLength(1);
});
});
});

35
packages/plugins/agent-codex/src/app-server-client.ts
View File

@ -139,6 +139,7 @@ export class CodexAppServerClient extends EventEmitter {
private pending = new Map<string, PendingRequest>();
private initialized = false;
private closed = false;
private connecting = false;
private readonly binaryPath: string;
private readonly cwd: string | undefined;
@ -169,6 +170,9 @@ export class CodexAppServerClient extends EventEmitter {
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;
this.process = spawn(this.binaryPath, ["app-server"], {
cwd: this.cwd,
@ -177,9 +181,14 @@ export class CodexAppServerClient extends EventEmitter {
});
if (!this.process.stdout || !this.process.stdin) {
this.connecting = false;
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));
@ -193,8 +202,17 @@ export class CodexAppServerClient extends EventEmitter {
this.handleProcessError(err);
});
// Perform initialization handshake
await this.initialize();
// Perform initialization handshake. On failure, clean up the spawned
// process so the caller doesn't need to call close() explicitly.
try {
await this.initialize();
} catch (err) {
this.connecting = false;
await this.close();
throw err;
}
this.connecting = false;
}
/**
@ -445,6 +463,12 @@ export class CodexAppServerClient extends EventEmitter {
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) {
@ -458,6 +482,13 @@ export class CodexAppServerClient extends EventEmitter {
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;
}
this.emit("error", err);
// Reject all pending requests

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

@ -12,6 +12,7 @@ const {
mockReaddir,
mockRename,
mockStat,
mockOpen,
mockHomedir,
} = vi.hoisted(() => ({
mockExecFileAsync: vi.fn(),
@ -21,6 +22,7 @@ const {
mockReaddir: vi.fn(),
mockRename: vi.fn().mockResolvedValue(undefined),
mockStat: vi.fn(),
mockOpen: vi.fn(),
mockHomedir: vi.fn(() => "/mock/home"),
}));
@ -38,6 +40,7 @@ vi.mock("node:fs/promises", () => ({
readdir: mockReaddir,
rename: mockRename,
stat: mockStat,
open: mockOpen,
}));
vi.mock("node:crypto", () => ({
@ -114,9 +117,36 @@ 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));
}
beforeEach(() => {
vi.clearAllMocks();
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(""));
});
// =========================================================================
@ -583,9 +613,9 @@ describe("getSessionInfo", () => {
it("returns null when no session files match the workspace cwd", async () => {
mockReaddir.mockResolvedValue(["session-abc.jsonl"]);
mockReadFile.mockResolvedValue(
jsonl({ type: "session_meta", cwd: "/other/workspace", model: "gpt-4o" }),
);
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();
});
@ -597,6 +627,7 @@ describe("getSessionInfo", () => {
);
mockReaddir.mockResolvedValue(["session-123.jsonl"]);
setupMockOpen(sessionContent);
mockReadFile.mockResolvedValue(sessionContent);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
@ -623,6 +654,11 @@ describe("getSessionInfo", () => {
);
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);
@ -647,6 +683,7 @@ describe("getSessionInfo", () => {
'{"type":"event_msg","msg":{"type":"token_count","input_tokens":500,"output_tokens":200}}\n';
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
@ -664,6 +701,7 @@ describe("getSessionInfo", () => {
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
@ -680,6 +718,7 @@ describe("getSessionInfo", () => {
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
@ -691,6 +730,9 @@ describe("getSessionInfo", () => {
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"));
expect(await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }))).toBeNull();
@ -701,6 +743,7 @@ describe("getSessionInfo", () => {
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockRejectedValue(new Error("EACCES"));
@ -711,18 +754,10 @@ describe("getSessionInfo", () => {
it("returns null when session JSONL has only empty/malformed lines", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
// First read (cwd check) has session_meta matching
// Second read (full parse) returns only garbage
let callCount = 0;
mockReadFile.mockImplementation(() => {
callCount++;
if (callCount <= 1) {
return Promise.resolve(jsonl(
{ type: "session_meta", cwd: "/workspace/test" },
));
}
return Promise.resolve("not json\n\n \n");
});
// 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");
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
@ -746,6 +781,7 @@ describe("getSessionInfo", () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "o3-mini" },
);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
@ -759,17 +795,18 @@ describe("getSessionInfo", () => {
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
// Non-JSONL entries trigger stat to check isDirectory()
mockStat.mockResolvedValue({ mtimeMs: 1000, isDirectory: () => false });
const result = await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" }));
expect(result).not.toBeNull();
// readFile should only be called for the .jsonl file
// readFile should only be called once (full parse); cwd check uses open()
const readFileCalls = mockReadFile.mock.calls.filter(
(call: string[]) => typeof call[0] === "string" && call[0].includes("sessions/"),
);
expect(readFileCalls.length).toBe(2); // once for cwd check, once for full parse
expect(readFileCalls.length).toBe(1); // once for full parse (cwd check uses open())
});
});
@ -811,11 +848,13 @@ describe("getRestoreCommand", () => {
});
it("returns null when session has no threadId", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
mockReadFile.mockResolvedValue(jsonl(
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ role: "user", content: "Some prompt" },
));
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
@ -824,11 +863,13 @@ describe("getRestoreCommand", () => {
});
it("builds native resume command with codex resume <threadId>", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
mockReadFile.mockResolvedValue(jsonl(
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ threadId: "thread-abc-123" },
));
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
@ -840,11 +881,13 @@ describe("getRestoreCommand", () => {
});
it("includes --dangerously-bypass-approvals-and-sandbox from project config", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
mockReadFile.mockResolvedValue(jsonl(
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ threadId: "thread-1" },
));
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
@ -856,11 +899,13 @@ describe("getRestoreCommand", () => {
});
it("includes --ask-for-approval never from project config", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
mockReadFile.mockResolvedValue(jsonl(
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ threadId: "thread-1" },
));
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
@ -873,11 +918,13 @@ describe("getRestoreCommand", () => {
});
it("includes --ask-for-approval untrusted from project config", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
mockReadFile.mockResolvedValue(jsonl(
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ threadId: "thread-1" },
));
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
@ -889,11 +936,13 @@ describe("getRestoreCommand", () => {
});
it("includes model from project config (overrides session model)", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
mockReadFile.mockResolvedValue(jsonl(
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" },
{ threadId: "thread-1" },
));
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
@ -906,11 +955,13 @@ describe("getRestoreCommand", () => {
});
it("falls back to session model when project config has no model", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
mockReadFile.mockResolvedValue(jsonl(
const content = jsonl(
{ type: "session_meta", cwd: "/workspace/test", model: "o4-mini" },
{ threadId: "thread-1" },
));
);
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockReadFile.mockResolvedValue(content);
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });
@ -922,17 +973,10 @@ describe("getRestoreCommand", () => {
it("handles unreadable session files gracefully", async () => {
mockReaddir.mockResolvedValue(["sess.jsonl"]);
// First call for cwd check succeeds, second call for full parse fails
let callCount = 0;
mockReadFile.mockImplementation(() => {
callCount++;
if (callCount <= 1) {
return Promise.resolve(jsonl(
{ type: "session_meta", cwd: "/workspace/test" },
));
}
return Promise.reject(new Error("EACCES"));
});
// 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"));
mockStat.mockResolvedValue({ mtimeMs: 1000 });
const session = makeSession({ workspacePath: "/workspace/test" });

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

@ -13,7 +13,7 @@ import {
type WorkspaceHooksConfig,
} from "@composio/ao-core";
import { execFile } from "node:child_process";
import { writeFile, mkdir, readFile, readdir, rename, stat } from "node:fs/promises";
import { writeFile, mkdir, readFile, readdir, rename, stat, open } from "node:fs/promises";
import { homedir } from "node:os";
import { basename, join } from "node:path";
import { promisify } from "node:util";
@ -343,8 +343,17 @@ async function sessionFileMatchesCwd(
workspacePath: string,
): Promise<boolean> {
try {
const content = await readFile(filePath, "utf-8");
// Only parse the first few lines for the metadata header
// 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();
@ -508,6 +517,8 @@ export async function resolveCodexBinary(): Promise<string> {
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",
@ -742,9 +753,14 @@ function createCodexAgent(): Agent {
},
async postLaunchSetup(session: Session): Promise<void> {
// Resolve binary path on first launch (cached for subsequent calls)
// Resolve binary path on first launch (cached for subsequent calls).
// Uses a promise guard to prevent concurrent calls from racing.
if (!resolvedBinary) {
resolvedBinary = await resolveCodexBinary();
if (!resolvingBinary) {
resolvingBinary = resolveCodexBinary();
}
resolvedBinary = await resolvingBinary;
resolvingBinary = null;
}
if (!session.workspacePath) return;
await setupCodexWorkspace(session.workspacePath);