diff --git a/packages/plugins/agent-codex/src/app-server-client.test.ts b/packages/plugins/agent-codex/src/app-server-client.test.ts
index 9cebc4a63..1221f2570 100644
--- a/packages/plugins/agent-codex/src/app-server-client.test.ts
+++ b/packages/plugins/agent-codex/src/app-server-client.test.ts
@@ -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);
+    });
+  });
 });
diff --git a/packages/plugins/agent-codex/src/app-server-client.ts b/packages/plugins/agent-codex/src/app-server-client.ts
index 869dbe6e3..3b533cc44 100644
--- a/packages/plugins/agent-codex/src/app-server-client.ts
+++ b/packages/plugins/agent-codex/src/app-server-client.ts
@@ -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
diff --git a/packages/plugins/agent-codex/src/index.test.ts b/packages/plugins/agent-codex/src/index.test.ts
index 86dfccf65..de5eea31f 100644
--- a/packages/plugins/agent-codex/src/index.test.ts
+++ b/packages/plugins/agent-codex/src/index.test.ts
@@ -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" });
diff --git a/packages/plugins/agent-codex/src/index.ts b/packages/plugins/agent-codex/src/index.ts
index d91f2cc18..bd4030102 100644
--- a/packages/plugins/agent-codex/src/index.ts
+++ b/packages/plugins/agent-codex/src/index.ts
@@ -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);