fix(agent-codex): close codex jsonl stream on abort

packages/plugins/agent-codex/src/index.test.ts

@@ -1,4 +1,5 @@
 import { describe, it, expect, vi, beforeEach } from "vitest";
+import type * as Readline from "node:readline";
 import {
   createActivitySignal,
   type Session,
@@ -21,6 +22,7 @@ const {
   mockLstat,
   mockOpen,
   mockCreateReadStream,
+  mockCreateInterface,
   mockHomedir,
   mockReadLastJsonlEntry,
   mockIsWindows,
@@ -35,6 +37,7 @@ const {
   mockLstat: vi.fn(),
   mockOpen: vi.fn(),
   mockCreateReadStream: vi.fn(),
+  mockCreateInterface: vi.fn(),
   mockHomedir: vi.fn(() => "/mock/home"),
   mockReadLastJsonlEntry: vi.fn(),
   mockIsWindows: vi.fn(() => false),
@@ -67,6 +70,17 @@ vi.mock("node:fs", () => ({
   createReadStream: mockCreateReadStream,
 }));
 
+vi.mock("node:readline", async (importOriginal) => {
+  const actual = await importOriginal<typeof Readline>();
+  mockCreateInterface.mockImplementation((...args: Parameters<typeof actual.createInterface>) =>
+    actual.createInterface(...args),
+  );
+  return {
+    ...actual,
+    createInterface: mockCreateInterface,
+  };
+});
+
 vi.mock("node:os", () => ({
   homedir: mockHomedir,
 }));
@@ -1212,6 +1226,31 @@ describe("getSessionInfo", () => {
     ).toBeNull();
   });
 
+  it("closes readline and destroys the stream when JSONL streaming is interrupted", async () => {
+    const content = jsonl({ type: "session_meta", cwd: "/workspace/test" });
+    mockReaddir.mockResolvedValue(["sess.jsonl"]);
+    setupMockOpen(content);
+    mockStat.mockResolvedValue({ mtimeMs: 1000 });
+
+    const stream = makeContentStream(content);
+    const destroySpy = vi.spyOn(stream, "destroy");
+    const closeSpy = vi.fn();
+    mockCreateReadStream.mockReturnValue(stream);
+    mockCreateInterface.mockImplementationOnce(() => ({
+      close: closeSpy,
+      async *[Symbol.asyncIterator]() {
+        yield JSON.stringify({ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" });
+        throw new Error("aborted");
+      },
+    }));
+
+    expect(
+      await agent.getSessionInfo(makeSession({ workspacePath: "/workspace/test" })),
+    ).toBeNull();
+    expect(closeSpy).toHaveBeenCalledTimes(1);
+    expect(destroySpy).toHaveBeenCalledTimes(1);
+  });
+
   it("skips session files when stat throws", async () => {
     const content = jsonl({ type: "session_meta", cwd: "/workspace/test", model: "gpt-4o" });
     mockReaddir.mockResolvedValue(["sess.jsonl"]);

packages/plugins/agent-codex/src/index.ts

@@ -268,6 +268,9 @@ interface CodexSessionData {
  * into memory. This is critical because Codex rollout files can be 100 MB+.
  */
 async function streamCodexSessionData(filePath: string): Promise<CodexSessionData | null> {
+  let stream: ReturnType<typeof createReadStream> | null = null;
+  let rl: ReturnType<typeof createInterface> | null = null;
+
   try {
     const data: CodexSessionData = {
       model: null,
@@ -277,8 +280,9 @@ async function streamCodexSessionData(filePath: string): Promise<CodexSessionDat
       cachedTokens: 0,
       reasoningTokens: 0,
     };
-    const rl = createInterface({
-      input: createReadStream(filePath, { encoding: "utf-8" }),
+    stream = createReadStream(filePath, { encoding: "utf-8" });
+    rl = createInterface({
+      input: stream,
       crlfDelay: Infinity,
     });
 
@@ -352,6 +356,9 @@ async function streamCodexSessionData(filePath: string): Promise<CodexSessionDat
     return data;
   } catch {
     return null;
+  } finally {
+    rl?.close();
+    stream?.destroy();
   }
 }
 
@@ -833,7 +840,10 @@ function createCodexAgent(): Agent {
       return formatLaunchCommand(parts);
     },
 
-    async setupWorkspaceHooks(_workspacePath: string, _config: WorkspaceHooksConfig): Promise<void> {
+    async setupWorkspaceHooks(
+      _workspacePath: string,
+      _config: WorkspaceHooksConfig,
+    ): Promise<void> {
       // PATH wrappers are installed by session-manager for all agents.
     },