fix(rpc): 清掉剩余 readline NDJSON 读口

apps/desktop/electron/main/importers/codex.ts

@@ -1,8 +1,9 @@
 import { createReadStream } from "node:fs";
 import fs, { type FileHandle } from "node:fs/promises";
-import { createInterface } from "node:readline";
 import os from "node:os";
 import path from "node:path";
+import type { Readable } from "node:stream";
+import { readNdjsonLines } from "@pi-desktop/shared";
 import type {
   ExternalSessionSummary,
   ImportedSession,
@@ -13,6 +14,35 @@ import { importedSessionId, toIso, truncateTitle } from "./types";
 
 const SESSIONS_DIR = path.join(os.homedir(), ".codex", "sessions");
 
+function readLfJsonl(
+  stream: Readable,
+  onLine: (line: string) => boolean | void,
+): Promise<void> {
+  return new Promise((resolve, reject) => {
+    let settled = false;
+    const finish = (error?: unknown) => {
+      if (settled) return;
+      settled = true;
+      reader.close();
+      stream.off("error", onError);
+      stream.off("end", onEnd);
+      if (error !== undefined) reject(error);
+      else resolve();
+    };
+    const onError = (error: Error) => finish(error);
+    const onEnd = () => finish();
+    const reader = readNdjsonLines(stream, (line) => {
+      try {
+        if (onLine(line) === false) finish();
+      } catch (error) {
+        finish(error);
+      }
+    });
+    stream.once("error", onError);
+    stream.once("end", onEnd);
+  });
+}
+
 interface CodexItem {
   type?: string;
   role?: string;
@@ -311,16 +341,14 @@ async function scanLargeFile(
       start: headLastNewline === -1 ? 0 : headBytes,
       encoding: "utf8",
     });
-    const lines = createInterface({
-      input: stream,
-      crlfDelay: Infinity,
-    });
-    for await (const line of lines) {
-      applyCodexLine(line, meta);
-      if (meta.firstUserText !== null) break;
+    try {
+      await readLfJsonl(stream, (line) => {
+        applyCodexLine(line, meta);
+        if (meta.firstUserText !== null) return false;
+      });
+    } finally {
+      stream.destroy();
     }
-    lines.close();
-    stream.destroy();
   }
 
   if (meta.startedAt !== null && headBytes < size) {
@@ -346,12 +374,13 @@ async function scanFile(filePath: string): Promise<CodexScanMeta | null> {
       const meta = newScanMeta();
       meta.mtimeMs = stats.mtimeMs;
       const stream = createReadStream(filePath, { encoding: "utf8" });
-      const lines = createInterface({ input: stream, crlfDelay: Infinity });
-      for await (const line of lines) {
-        applyCodexLine(line, meta);
+      try {
+        await readLfJsonl(stream, (line) => {
+          applyCodexLine(line, meta);
+        });
+      } finally {
+        stream.destroy();
       }
-      lines.close();
-      stream.destroy();
       if (!meta.sawItem) return null;
       if (!meta.externalId) meta.externalId = path.basename(filePath, ".jsonl");
       return meta;

apps/desktop/test/importer-codex-scan.test.mjs

@@ -87,6 +87,27 @@ test("small archives keep the exact full-parse semantics", async () => {
   );
 });
 
+test("JSONL user text with U+2028/U+2029 still scans as one session", async () => {
+  const title = "hello\u2028world\u2029end";
+  await withArchive(
+    [
+      [
+        "2026/01/unicode.jsonl",
+        [
+          metaLine("uni-1", "/repo", "2026-01-01T00:00:00Z"),
+          responseItem("user", title, "2026-01-01T00:00:01Z"),
+        ].join("\n") + "\n",
+      ],
+    ],
+    async (dir) => {
+      const summaries = await scanCodexSessions(dir);
+      assert.equal(summaries.length, 1);
+      assert.equal(summaries[0].externalId, "uni-1");
+      assert.equal(summaries[0].title, "hello world end");
+    },
+  );
+});
+
 test("old-format headers and item counts are preserved", async () => {
   await withArchive(
     [

apps/desktop/test/ndjson-stream.test.mjs

@@ -0,0 +1,21 @@
+import assert from "node:assert/strict";
+import { once } from "node:events";
+import { PassThrough } from "node:stream";
+import test from "node:test";
+import { readNdjsonLines } from "../../../packages/shared/src/ndjson.ts";
+
+const payload = { id: "rpc", result: { content: "A\u2028中文🙂\u2029B\r\nC" } };
+const wire = Buffer.from(`${JSON.stringify(payload)}\n`);
+
+test("Node setEncoding preserves Unicode separators at every byte split", async () => {
+  for (let split = 0; split <= wire.length; split++) {
+    const input = new PassThrough();
+    const lines = [];
+    readNdjsonLines(input, (line) => lines.push(JSON.parse(line)));
+    const ended = once(input, "end");
+    input.write(wire.subarray(0, split));
+    input.end(wire.subarray(split));
+    await ended;
+    assert.deepEqual(lines, [payload], `split at byte ${split}`);
+  }
+});

apps/desktop/test/rpc-lifecycle-contract.test.mjs

@@ -31,6 +31,18 @@ const agentSidecarEntrySource = await readFile(
   new URL("../../../packages/agent-runtime/src/sidecar.ts", import.meta.url),
   "utf8",
 );
+const e2eHostSource = await readFile(
+  new URL("../../../scripts/e2e/host.mjs", import.meta.url),
+  "utf8",
+);
+const e2eSmokeSource = await readFile(
+  new URL("../../../scripts/e2e-smoke.mjs", import.meta.url),
+  "utf8",
+);
+const codexImporterSource = await readFile(
+  new URL("../electron/main/importers/codex.ts", import.meta.url),
+  "utf8",
+);
 const apiSource = await readFile(
   new URL("../src/lib/api.ts", import.meta.url),
   "utf8",
@@ -41,12 +53,20 @@ test("stdio RPC readers split frames on LF only", () => {
     ["host-process", hostSource],
     ["agent-sidecar", sidecarSource],
     ["runtime sidecar", agentSidecarEntrySource],
+    ["e2e host harness", e2eHostSource],
+    ["e2e smoke host", e2eSmokeSource],
+    ["codex importer", codexImporterSource],
   ]) {
     assert.match(source, /readNdjsonLines/, `${name} must use LF NDJSON framing`);
     assert.doesNotMatch(
       source,
       /from ["']node:readline["']/,
-      `${name} must not use readline on the RPC pipe`,
+      `${name} must not use readline`,
+    );
+    assert.doesNotMatch(
+      source,
+      /createInterface/,
+      `${name} must not call createInterface`,
     );
   }
 });

packages/shared/src/ndjson.test.ts

@@ -5,11 +5,13 @@ type DataListener = (chunk: string) => void;
 type SignalListener = () => void;
 
 class FakeNdjsonInput {
+  encoding: "utf8" | undefined;
   private data: DataListener[] = [];
   private end: SignalListener[] = [];
   private close: SignalListener[] = [];
 
-  setEncoding(_encoding: "utf8") {
+  setEncoding(encoding: "utf8") {
+    this.encoding = encoding;
     return this;
   }
 
@@ -63,6 +65,7 @@ describe("readNdjsonLines", () => {
     const input = new FakeNdjsonInput();
     const lines: unknown[] = [];
     readNdjsonLines(input, (line) => lines.push(JSON.parse(line)));
+    expect(input.encoding).toBe("utf8");
     input.finish(new TextDecoder().decode(wire));
     expect(lines).toEqual([payload]);
   });

scripts/e2e-agent-live.mjs

@@ -1,5 +1,4 @@
 import { spawn } from "node:child_process";
-import { createInterface } from "node:readline";
 import { randomUUID } from "node:crypto";
 import { mkdtempSync, rmSync, existsSync } from "node:fs";
 import { tmpdir } from "node:os";

scripts/e2e-capability-move.mjs

@@ -20,7 +20,7 @@
  * Deterministic: no live network access.
  */
 import { spawn } from "node:child_process";
-import { createInterface } from "node:readline";
+import { readNdjsonLines } from "../packages/shared/dist/ndjson.js";
 import { randomUUID } from "node:crypto";
 import {
   existsSync,
@@ -79,8 +79,7 @@ class Host {
     });
     this.pending = new Map();
     this.child.stderr.on("data", () => {});
-    const rl = createInterface({ input: this.child.stdout });
-    rl.on("line", (line) => {
+    readNdjsonLines(this.child.stdout, (line) => {
       let msg;
       try {
         msg = JSON.parse(line);

scripts/e2e-mcp-market.mjs

@@ -13,7 +13,7 @@
  * Deterministic: no live network access.
  */
 import { spawn } from "node:child_process";
-import { createInterface } from "node:readline";
+import { readNdjsonLines } from "../packages/shared/dist/ndjson.js";
 import { randomUUID } from "node:crypto";
 import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs";
 import { tmpdir } from "node:os";
@@ -69,8 +69,7 @@ class Host {
     });
     this.pending = new Map();
     this.child.stderr.on("data", () => {});
-    const rl = createInterface({ input: this.child.stdout });
-    rl.on("line", (line) => {
+    readNdjsonLines(this.child.stdout, (line) => {
       let msg;
       try {
         msg = JSON.parse(line);

scripts/e2e-skill-market.mjs

@@ -15,13 +15,13 @@
  * Deterministic: no live network access.
  */
 import { spawn } from "node:child_process";
-import { createInterface } from "node:readline";
 import { randomUUID } from "node:crypto";
 import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join, dirname, resolve } from "node:path";
 import { fileURLToPath } from "node:url";
 import { PROTOCOL_VERSION } from "../packages/shared/dist/protocol.js";
+import { readNdjsonLines } from "../packages/shared/dist/ndjson.js";
 import {
   BUILTIN_SKILL_CATALOG,
   GLOBAL_SCOPE,
@@ -73,8 +73,7 @@ class Host {
     });
     this.pending = new Map();
     this.child.stderr.on("data", () => {});
-    const rl = createInterface({ input: this.child.stdout });
-    rl.on("line", (line) => {
+    readNdjsonLines(this.child.stdout, (line) => {
       let msg;
       try {
         msg = JSON.parse(line);

scripts/e2e-smoke.mjs

@@ -10,13 +10,13 @@
  *  PI_DESKTOP_HOST_BIN (optional)
  */
 import { spawn } from "node:child_process";
-import { createInterface } from "node:readline";
 import { randomUUID } from "node:crypto";
 import { existsSync, mkdtempSync, rmSync, writeFileSync, readFileSync, mkdirSync, readdirSync, realpathSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join, dirname, resolve } from "node:path";
 import { fileURLToPath, pathToFileURL } from "node:url";
 import { PROTOCOL_VERSION } from "../packages/shared/dist/protocol.js";
+import { readNdjsonLines } from "../packages/shared/dist/ndjson.js";
 import {
   loadDevelopmentPlugin,
   resolvePluginExecution,
@@ -84,8 +84,7 @@ class Host {
       // keep quiet unless debugging
       if (process.env.DEBUG_HOST) process.stderr.write(b);
     });
-    const rl = createInterface({ input: this.child.stdout });
-    rl.on("line", (line) => {
+    readNdjsonLines(this.child.stdout, (line) => {
       if (process.env.DEBUG_HOST) console.error(`[e2e host stdout] ${line}`);
       let msg;
       try {

scripts/e2e-subagent-models.mjs

@@ -3,7 +3,7 @@
 import assert from "node:assert/strict";
 import { spawn } from "node:child_process";
 import { createServer } from "node:http";
-import { createInterface } from "node:readline";
+import { readNdjsonLines } from "../packages/shared/dist/ndjson.js";
 import { fileURLToPath } from "node:url";
 
 const requests = [];
@@ -104,8 +104,7 @@ const child = spawn(process.execPath, [fileURLToPath(new URL("../packages/agent-
 let stderr = "";
 child.stderr.on("data", (chunk) => { stderr += chunk; });
 const send = (message) => child.stdin.write(`${JSON.stringify({ jsonrpc: "2.0", ...message })}\n`);
-const lines = createInterface({ input: child.stdout });
-lines.on("line", (line) => {
+readNdjsonLines(child.stdout, (line) => {
   const message = JSON.parse(line);
   if (message.method === "host.proxy") {
     const { method, params } = message.params;

scripts/e2e/host.mjs

@@ -1,11 +1,11 @@
 import { spawn } from "node:child_process";
-import { createInterface } from "node:readline";
 import { randomUUID } from "node:crypto";
 import { existsSync } from "node:fs";
 import { dirname, join, resolve } from "node:path";
 import { fileURLToPath } from "node:url";
 import { setTimeout as delay } from "node:timers/promises";
 
+import { readNdjsonLines } from "../../packages/shared/dist/ndjson.js";
 import { assert, shortJson } from "./assert.mjs";
 
 const root = join(dirname(fileURLToPath(import.meta.url)), "../..");
@@ -65,7 +65,7 @@ export class Host {
     this.binary = binary;
     this.dataDir = dataDir;
     this.child = null;
-    this.readline = null;
+    this.stdoutReader = null;
     this.pending = new Map();
     this.notifications = [];
     this.stderr = "";
@@ -107,8 +107,7 @@ export class Host {
       this.stderr += String(chunk);
       if (process.env.DEBUG_HOST) process.stderr.write(chunk);
     });
-    this.readline = createInterface({ input: child.stdout });
-    this.readline.on("line", (line) => {
+    this.stdoutReader = readNdjsonLines(child.stdout, (line) => {
       let message;
       try {
         message = JSON.parse(line);
@@ -199,8 +198,8 @@ export class Host {
       }
       await Promise.race([this.exitPromise, delay(3_000)]);
     }
-    this.readline?.close();
-    this.readline = null;
+    this.stdoutReader?.close();
+    this.stdoutReader = null;
     this.child = null;
     if (!this.exited) throw new Error("host did not exit during cleanup");
   }