From 8da5a8739d0a070dbc1caa0423bd75d36d3d43c6 Mon Sep 17 00:00:00 2001 From: vastsa Date: Fri, 18 Sep 2026 01:41:01 +0800 Subject: [PATCH] fix(rpc): stop remaining readline NDJSON readers Switch Codex JSONL scanning and the e2e host harnesses off Node readline so U+2028/U+2029 stay inside a frame. Cover real Node setEncoding splits and lock createInterface out of the transport readers. --- apps/desktop/electron/main/importers/codex.ts | 59 ++++++++++++++----- .../desktop/test/importer-codex-scan.test.mjs | 21 +++++++ apps/desktop/test/ndjson-stream.test.mjs | 21 +++++++ .../test/rpc-lifecycle-contract.test.mjs | 22 ++++++- packages/shared/src/ndjson.test.ts | 5 +- scripts/e2e-agent-live.mjs | 1 - scripts/e2e-capability-move.mjs | 5 +- scripts/e2e-mcp-market.mjs | 5 +- scripts/e2e-skill-market.mjs | 5 +- scripts/e2e-smoke.mjs | 5 +- scripts/e2e-subagent-models.mjs | 5 +- scripts/e2e/host.mjs | 11 ++-- 12 files changed, 126 insertions(+), 39 deletions(-) create mode 100644 apps/desktop/test/ndjson-stream.test.mjs diff --git a/apps/desktop/electron/main/importers/codex.ts b/apps/desktop/electron/main/importers/codex.ts index 37b47a3bd..9ccf160d0 100644 --- a/apps/desktop/electron/main/importers/codex.ts +++ b/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 { + 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 { 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; diff --git a/apps/desktop/test/importer-codex-scan.test.mjs b/apps/desktop/test/importer-codex-scan.test.mjs index c53e70fba..2af4d7dd5 100644 --- a/apps/desktop/test/importer-codex-scan.test.mjs +++ b/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( [ diff --git a/apps/desktop/test/ndjson-stream.test.mjs b/apps/desktop/test/ndjson-stream.test.mjs new file mode 100644 index 000000000..67360fd7e --- /dev/null +++ b/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}`); + } +}); diff --git a/apps/desktop/test/rpc-lifecycle-contract.test.mjs b/apps/desktop/test/rpc-lifecycle-contract.test.mjs index 493a359fe..3a191b7fb 100644 --- a/apps/desktop/test/rpc-lifecycle-contract.test.mjs +++ b/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`, ); } }); diff --git a/packages/shared/src/ndjson.test.ts b/packages/shared/src/ndjson.test.ts index e69812feb..86f734e4c 100644 --- a/packages/shared/src/ndjson.test.ts +++ b/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]); }); diff --git a/scripts/e2e-agent-live.mjs b/scripts/e2e-agent-live.mjs index fdd7da7bf..8f4e77e11 100644 --- a/scripts/e2e-agent-live.mjs +++ b/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"; diff --git a/scripts/e2e-capability-move.mjs b/scripts/e2e-capability-move.mjs index 4888221a5..9abc0ec24 100644 --- a/scripts/e2e-capability-move.mjs +++ b/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); diff --git a/scripts/e2e-mcp-market.mjs b/scripts/e2e-mcp-market.mjs index c3b1675b9..2eb174911 100644 --- a/scripts/e2e-mcp-market.mjs +++ b/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); diff --git a/scripts/e2e-skill-market.mjs b/scripts/e2e-skill-market.mjs index 746e28b91..3236373f8 100644 --- a/scripts/e2e-skill-market.mjs +++ b/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); diff --git a/scripts/e2e-smoke.mjs b/scripts/e2e-smoke.mjs index 0ed06ca58..bfcad1b5f 100755 --- a/scripts/e2e-smoke.mjs +++ b/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 { diff --git a/scripts/e2e-subagent-models.mjs b/scripts/e2e-subagent-models.mjs index ed25b42db..081ea898d 100644 --- a/scripts/e2e-subagent-models.mjs +++ b/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; diff --git a/scripts/e2e/host.mjs b/scripts/e2e/host.mjs index ff41277b5..70ba92e68 100644 --- a/scripts/e2e/host.mjs +++ b/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"); }