fix(agent-codex): close codex JSONL streams on abort (#1977)

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

* fix(agent-kimicode): close wire stream on interrupted summary reads

---------

Co-authored-by: i-trytoohard <193449657+i-trytoohard@users.noreply.github.com>

eslint.config.js

@@ -57,6 +57,21 @@ export default tseslint.config(
       "no-debugger": "error",
       "no-duplicate-imports": "error",
       "no-template-curly-in-string": "warn",
+      "no-restricted-syntax": [
+        "error",
+        {
+          selector:
+            "CallExpression[callee.name='createInterface'] Property[key.name='input'] > CallExpression[callee.name='createReadStream']",
+          message:
+            "Do not pass createReadStream() inline to createInterface(); keep the stream in a variable and close/destroy it in a finally block.",
+        },
+        {
+          selector:
+            "CallExpression[callee.name='createInterface'] Property[key.name='input'] > CallExpression[callee.property.name='createReadStream']",
+          message:
+            "Do not pass createReadStream() inline to createInterface(); keep the stream in a variable and close/destroy it in a finally block.",
+        },
+      ],
       "prefer-const": "error",
       "no-var": "error",
       eqeqeq: ["error", "always"],

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.
     },
 

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

@@ -52,9 +52,12 @@ async function extractKimiSummary(sessionDir: string): Promise<string | null> {
   // still be planted as symlinks pointing at /etc/passwd or /dev/zero.
   if (!(await isKimiSessionFile(wirePath))) return null;
   let summary: string | null = null;
+  let stream: ReturnType<typeof createReadStream> | null = null;
+  let rl: ReturnType<typeof createInterface> | null = null;
   try {
-    const rl = createInterface({
-      input: createReadStream(wirePath, { encoding: "utf-8" }),
+    stream = createReadStream(wirePath, { encoding: "utf-8" });
+    rl = createInterface({
+      input: stream,
       crlfDelay: Infinity,
     });
     let bytes = 0;
@@ -85,9 +88,11 @@ async function extractKimiSummary(sessionDir: string): Promise<string | null> {
         // Skip malformed line
       }
     }
-    rl.close();
   } catch {
     return null;
+  } finally {
+    rl?.close();
+    stream?.destroy();
   }
   return summary;
 }