diff --git a/apps/desktop/electron/main/agent-sidecar.ts b/apps/desktop/electron/main/agent-sidecar.ts index caec3a3b4..bb486b7d1 100644 --- a/apps/desktop/electron/main/agent-sidecar.ts +++ b/apps/desktop/electron/main/agent-sidecar.ts @@ -1,11 +1,10 @@ import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; -import { createInterface } from "node:readline"; import { randomUUID } from "node:crypto"; import { join } from "node:path"; import { existsSync } from "node:fs"; import type { HostProcess, ProcessExitHandler, StderrHandler } from "./host-process"; import { redactValue } from "./logger"; -import { DEFAULT_RPC_TIMEOUT_MS, rpcTimeoutMs } from "@pi-desktop/shared"; +import { DEFAULT_RPC_TIMEOUT_MS, readNdjsonLines, rpcTimeoutMs } from "@pi-desktop/shared"; // stderr lines kept per sidecar so an unexpected exit can be reported with the // process's last words instead of a bare "agent sidecar exited". @@ -123,7 +122,7 @@ export class AgentSidecar { private host: HostProcess | null = null; private unsubscribeHost: (() => void) | null = null; private unsubscribeHostExit: (() => void) | null = null; - private readline?: ReturnType; + private stdoutReader?: ReturnType; // Tools served by Electron main itself (e.g. BrowserPreview drives the // work panel's WebContentsView) — host-core never sees these. private localTools = new Map(); @@ -179,9 +178,9 @@ export class AgentSidecar { this.notifyExit({ code: null, signal: null, intentional: this.disposed }); }); - const rl = createInterface({ input: this.child.stdout }); - this.readline = rl; - rl.on("line", (line) => void this.onLine(line)); + this.stdoutReader = readNdjsonLines(this.child.stdout, (line) => + void this.onLine(line), + ); } private recordStderr(text: string) { @@ -210,8 +209,8 @@ export class AgentSidecar { for (const timer of this.localToolTimers) clearTimeout(timer); this.localToolTimers.clear(); this.handlers.clear(); - this.readline?.close(); - this.readline = undefined; + this.stdoutReader?.close(); + this.stdoutReader = undefined; this.child.removeAllListeners("exit"); this.child.removeAllListeners("error"); this.child.stderr.removeAllListeners("data"); @@ -426,6 +425,9 @@ export class AgentSidecar { try { msg = JSON.parse(line); } catch { + console.warn( + `[RPC] Invalid agent-sidecar NDJSON frame (${Buffer.byteLength(line, "utf8")} bytes)`, + ); return; } diff --git a/apps/desktop/electron/main/host-process.ts b/apps/desktop/electron/main/host-process.ts index db9dd15fd..d9e1f8660 100644 --- a/apps/desktop/electron/main/host-process.ts +++ b/apps/desktop/electron/main/host-process.ts @@ -1,5 +1,4 @@ import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; -import { createInterface } from "node:readline"; import { randomUUID } from "node:crypto"; import { existsSync } from "node:fs"; import { join } from "node:path"; @@ -8,6 +7,7 @@ import { MAX_HOST_STDIN_LINE_BYTES, PROTOCOL_VERSION, rpcTimeoutMs, + readNdjsonLines, stripProxyEnv, } from "@pi-desktop/shared"; import { @@ -101,7 +101,7 @@ export class HostProcess { private exitPromise: Promise; private resolveExit!: () => void; private disposePromise?: Promise; - private readline?: ReturnType; + private stdoutReader?: ReturnType; private lastStderr = ""; readonly binaryPath: string; readonly generation = randomUUID(); @@ -169,9 +169,9 @@ export class HostProcess { if (this.exitObserved) this.cleanupProcessListeners(); }); - const rl = createInterface({ input: this.child.stdout }); - this.readline = rl; - rl.on("line", (line) => this.onLine(line)); + this.stdoutReader = readNdjsonLines(this.child.stdout, (line) => + this.onLine(line), + ); } private closeTransport(error: Error) { @@ -184,8 +184,8 @@ export class HostProcess { } this.pending.clear(); this.handlers.clear(); - this.readline?.close(); - this.readline = undefined; + this.stdoutReader?.close(); + this.stdoutReader = undefined; } private cleanupProcessListeners() { @@ -264,6 +264,9 @@ export class HostProcess { try { msg = JSON.parse(line); } catch { + console.warn( + `[RPC] Invalid host-process NDJSON frame (${Buffer.byteLength(line, "utf8")} bytes)`, + ); return; } if (msg.id !== undefined && msg.id !== null) { diff --git a/apps/desktop/test/ndjson.test.mjs b/apps/desktop/test/ndjson.test.mjs new file mode 100644 index 000000000..06b7598d2 --- /dev/null +++ b/apps/desktop/test/ndjson.test.mjs @@ -0,0 +1,63 @@ +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("NDJSON preserves Unicode separators at every possible 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}`); + } +}); + +test("NDJSON accepts consecutive LF/CRLF frames, blank lines, and EOF tail", async () => { + const input = new PassThrough(); + const lines = []; + readNdjsonLines(input, (line) => lines.push(line)); + const ended = once(input, "end"); + input.write('\n{"id":1}\r'); + assert.deepEqual(lines, [""], "CR alone does not terminate a frame"); + input.write('\n{"id":2}\n{"id":'); + input.end("3}"); + await ended; + assert.deepEqual(lines, ["", '{"id":1}', '{"id":2}', '{"id":3}']); + for (const event of ["data", "end", "close"]) assert.equal(input.listenerCount(event), 0); +}); + +test("closing inside a callback stops buffered frames and removes only owned listeners", () => { + const input = new PassThrough(); + const external = () => {}; + input.on("data", external); + const lines = []; + const reader = readNdjsonLines(input, (line) => { lines.push(line); reader.close(); }); + input.write("first\nsecond\npartial"); + reader.close(); + input.write("ignored\n"); + assert.deepEqual(lines, ["first"]); + assert.deepEqual(input.listeners("data"), [external]); + assert.equal(input.listenerCount("end"), 0); + assert.equal(input.listenerCount("close"), 0); + input.destroy(); +}); + +test("stream destruction discards a partial frame and detaches the reader", async () => { + const input = new PassThrough(); + const lines = []; + readNdjsonLines(input, (line) => lines.push(line)); + input.write('{"unfinished":'); + const closed = once(input, "close"); + input.destroy(); + await closed; + assert.deepEqual(lines, []); + for (const event of ["data", "end", "close"]) assert.equal(input.listenerCount(event), 0); +}); diff --git a/docs/spec/03-runtime/06-host-rpc-protocol.md b/docs/spec/03-runtime/06-host-rpc-protocol.md index 2c87d144b..df5ab860a 100644 --- a/docs/spec/03-runtime/06-host-rpc-protocol.md +++ b/docs/spec/03-runtime/06-host-rpc-protocol.md @@ -16,7 +16,14 @@ MVP transport decision (**D001**): - Process: Electron main spawns Rust host-core sidecar - Channel: child process stdin/stdout -- Framing: one JSON object per line (NDJSON) +- Framing: one JSON object per LF-delimited line (NDJSON); CRLF is accepted. + U+2028 and U+2029 inside JSON strings are payload, never frame delimiters. + All Node stdio readers preserve UTF-8 characters across input chunks and + release buffered fragments/listeners on transport close. A final unterminated + frame is accepted at EOF for compatibility. +- Invalid JSON frames produce a diagnostic containing only the byte length, + never payload text, before being discarded. Later complete frames remain + readable. Existing session text is not rewritten or migrated. - Encoding: UTF-8 - Request/response: JSON-RPC 2.0 style diff --git a/docs/spec/06-delivery/04-e2e-test-plan.md b/docs/spec/06-delivery/04-e2e-test-plan.md index 1cdd3699d..990d9add5 100644 --- a/docs/spec/06-delivery/04-e2e-test-plan.md +++ b/docs/spec/06-delivery/04-e2e-test-plan.md @@ -7503,6 +7503,7 @@ identify the platform validation still needed. | Acceptance | Scenarios | |---|---| +| A / C — Unicode stdio framing | E2E-RPC-unicode-separators | | C / G / Quality — Plugins navigation | E2E-NAV-plugins-button-goes-back | | C / D / Quality — Sidebar row states | E2E-LAYOUT-sidebar-row-states | | A / C / Quality — Sidebar material and settings return | E2E-LAYOUT-sidebar-settings | @@ -13058,3 +13059,22 @@ plugin-form fixtures in an isolated temporary directory at runtime. Renderer fixtures alone do not prove settings persistence. - **Specs:** 04-ux/06-settings-ia, 04-ux/08-component-spec, 04-ux/09-interaction-patterns; ADR turn-process-and-thinking-display. + +### E2E-RPC-unicode-separators + +- **Preconditions:** Built host-core, shared package and agent runtime; isolated + temporary data directory; loopback-only fixture provider. No live credentials. +- **Steps:** Append a user message containing U+2028/U+2029, CJK text, emoji and + escaped CR/LF; read it, restart the host, and read it again. Send a Unicode + prompt through AgentSidecar; restore history through the parent host proxy; + stream and persist a Unicode answer. Send an unknown method containing the + same characters, then a health request. +- **Expected:** Text survives unchanged across persistence and all stdio + directions. Requests settle without RPC timeouts; error replies and subsequent + requests remain usable. No migration of existing sessions is needed. +- **Automation:** `pnpm test:e2e:rpc-unicode`; `ndjson.test.mjs` additionally + checks every UTF-8 split boundary, consecutive frames, CRLF, EOF and disposal. +- **Specs:** 03-runtime/06-host-rpc-protocol §2. +- **Acceptance:** A (runtime), C (sessions). +- **Milestone:** M6+. +- **Status:** Automated; run against the task/PR integration candidate. diff --git a/docs/zh-CN/spec/03-runtime/06-host-rpc-protocol.md b/docs/zh-CN/spec/03-runtime/06-host-rpc-protocol.md index 4e96211ac..21aa0dbe7 100644 --- a/docs/zh-CN/spec/03-runtime/06-host-rpc-protocol.md +++ b/docs/zh-CN/spec/03-runtime/06-host-rpc-protocol.md @@ -19,7 +19,14 @@ MVP 传输决策 (**D001**)： - 流程：Electron 主要生成 Rust host-core sidecar - 通道：子进程 stdin/stdout -- 成帧：每行一个 JSON 对象 (NDJSON) +- Framing: one JSON object per LF-delimited line (NDJSON); CRLF is accepted. + U+2028 and U+2029 inside JSON strings are payload, never frame delimiters. + All Node stdio readers preserve UTF-8 characters across input chunks and + release buffered fragments/listeners on transport close. A final unterminated + frame is accepted at EOF for compatibility. +- Invalid JSON frames produce a diagnostic containing only the byte length, + never payload text, before being discarded. Later complete frames remain + readable. Existing session text is not rewritten or migrated. - 编码：UTF-8 - Request/response：JSON-RPC 2.0 风格 diff --git a/docs/zh-CN/spec/06-delivery/04-e2e-test-plan.md b/docs/zh-CN/spec/06-delivery/04-e2e-test-plan.md index ede2a73ae..1bd0e3173 100644 --- a/docs/zh-CN/spec/06-delivery/04-e2e-test-plan.md +++ b/docs/zh-CN/spec/06-delivery/04-e2e-test-plan.md @@ -5008,6 +5008,7 @@ IPC 请求无法关闭。 | 验收 | 应用场景 | |---|---| +| A / C — Unicode stdio framing | E2E-RPC-unicode-separators | | C / G / Quality — Plugins navigation | E2E-NAV-plugins-button-goes-back | | C / D / Quality — 侧边栏行状态 | E2E-LAYOUT-sidebar-row-states | | A / C / Quality — 侧栏材质与设置返回 | E2E-LAYOUT-sidebar-settings | @@ -7798,3 +7799,22 @@ runner 会在运行时的隔离临时目录中生成六个插件形态 fixture - **验收**：G（远程市场来源） - **里程碑**：M6+ - **状态**：草稿 + +### E2E-RPC-unicode-separators + +- **Preconditions:** Built host-core, shared package and agent runtime; isolated + temporary data directory; loopback-only fixture provider. No live credentials. +- **Steps:** Append a user message containing U+2028/U+2029, CJK text, emoji and + escaped CR/LF; read it, restart the host, and read it again. Send a Unicode + prompt through AgentSidecar; restore history through the parent host proxy; + stream and persist a Unicode answer. Send an unknown method containing the + same characters, then a health request. +- **Expected:** Text survives unchanged across persistence and all stdio + directions. Requests settle without RPC timeouts; error replies and subsequent + requests remain usable. No migration of existing sessions is needed. +- **Automation:** `pnpm test:e2e:rpc-unicode`; `ndjson.test.mjs` additionally + checks every UTF-8 split boundary, consecutive frames, CRLF, EOF and disposal. +- **Specs:** 03-runtime/06-host-rpc-protocol §2. +- **Acceptance:** A (runtime), C (sessions). +- **Milestone:** M6+. +- **Status:** Automated; run against the task/PR integration candidate. diff --git a/package.json b/package.json index a56f6b745..67feeb3e1 100644 --- a/package.json +++ b/package.json @@ -21,6 +21,7 @@ "test": "pnpm build:js && pnpm -r --if-present test && cargo test -p host-core", "test:host": "cargo test -p host-core", "test:e2e": "node scripts/e2e-smoke.mjs", + "test:e2e:rpc-unicode": "node scripts/e2e-rpc-unicode.mjs", "test:e2e:plan": "node scripts/e2e-plan.mjs", "test:e2e:plan-ui": "node scripts/e2e-plan-ui.mjs", "test:e2e:theme-surfaces": "node scripts/e2e-theme-surfaces.mjs", diff --git a/packages/agent-runtime/src/sidecar.ts b/packages/agent-runtime/src/sidecar.ts index 05bc691bf..a6348fbdc 100644 --- a/packages/agent-runtime/src/sidecar.ts +++ b/packages/agent-runtime/src/sidecar.ts @@ -3,7 +3,6 @@ * Protocol: NDJSON JSON-RPC on stdio with Electron main. * Host access is proxied through main (single host-core process). */ -import { createInterface } from "node:readline"; import { createHash, randomUUID } from "node:crypto"; import { constants as fsConstants } from "node:fs"; import { copyFile, mkdir, readFile, realpath, stat } from "node:fs/promises"; @@ -33,6 +32,7 @@ import { isCommandShellOption, MAX_INLINE_IMAGE_BYTES, normalizeMode, + readNdjsonLines, normalizeNetworkProxy, OAUTH_AUTH_KIND, } from "@pi-desktop/shared"; @@ -658,13 +658,15 @@ async function handle(method: string, params: any): Promise { } } -const rl = createInterface({ input: process.stdin }); -rl.on("line", async (line) => { +readNdjsonLines(process.stdin, async (line) => { if (!line.trim()) return; let msg: any; try { msg = JSON.parse(line); } catch { + process.stderr.write( + `[agent-sidecar] Invalid NDJSON frame (${Buffer.byteLength(line, "utf8")} bytes)\n`, + ); return; } // Responses to host.proxy requests from parent diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 357170568..ceff8bd9a 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -39,6 +39,7 @@ export * from "./command-shells.js"; export * from "./context-compaction.js"; export * from "./rpc-timeouts.js"; export * from "./rpc-limits.js"; +export * from "./ndjson.js"; export * from "./subagent-definition.js"; export * from "./subagent-presets.js"; export * from "./provider-presets.js"; diff --git a/packages/shared/src/ndjson.ts b/packages/shared/src/ndjson.ts new file mode 100644 index 000000000..8000ba567 --- /dev/null +++ b/packages/shared/src/ndjson.ts @@ -0,0 +1,56 @@ +/** The readable-stream surface needed by the stdio transport (no Node imports). */ +interface NdjsonInput { + setEncoding(encoding: "utf8"): unknown; + on(event: "data", listener: (chunk: string) => void): unknown; + on(event: "end" | "close", listener: () => void): unknown; + off(event: "data", listener: (chunk: string) => void): unknown; + off(event: "end" | "close", listener: () => void): unknown; +} + +/** + * Read LF-delimited JSON text, accepting CRLF and a final unterminated frame. + * Unlike readline, Unicode line/paragraph separators are ordinary payload. + * The input owns UTF-8 decoding, including characters split across byte chunks. + */ +export function readNdjsonLines(input: NdjsonInput, onLine: (line: string) => void) { + let closed = false; + let fragments: string[] = []; + const close = () => { + if (closed) return; + closed = true; + fragments = []; + input.off("data", onData); + input.off("end", onEnd); + input.off("close", close); + }; + const emit = (tail: string) => { + fragments.push(tail); + const line = fragments.join(""); + fragments = []; + onLine(line.endsWith("\r") ? line.slice(0, -1) : line); + }; + const onData = (chunk: string) => { + let start = 0; + while (!closed) { + const end = chunk.indexOf("\n", start); + if (end === -1) { + if (start < chunk.length) fragments.push(chunk.slice(start)); + return; + } + emit(chunk.slice(start, end)); + start = end + 1; + } + }; + const onEnd = () => { + try { + if (!closed && fragments.length > 0) emit(""); + } finally { + close(); + } + }; + input.setEncoding("utf8"); + input.on("data", onData); + input.on("end", onEnd); + input.on("close", close); + return { close }; +} diff --git a/scripts/e2e-rpc-unicode.mjs b/scripts/e2e-rpc-unicode.mjs new file mode 100644 index 000000000..ddf393397 --- /dev/null +++ b/scripts/e2e-rpc-unicode.mjs @@ -0,0 +1,134 @@ +#!/usr/bin/env node +/** Real stdio transports, isolated storage, and no external model calls. */ +import assert from "node:assert/strict"; +import { createServer } from "node:http"; +import { createRequire } from "node:module"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; + +const root = join(dirname(fileURLToPath(import.meta.url)), ".."); +const require = createRequire(join(root, "packages/agent-runtime/package.json")); +const { build } = require("esbuild"); +const temp = await mkdtemp(join(tmpdir(), "pi-rpc-unicode-")); +const previousBinary = process.env.PI_DESKTOP_HOST_BIN; +let host; +let sidecar; +let server; +const requests = []; +const text = "before\u2028中🙂\u2029after\r\nend"; +async function deadline(promise) { + let timer; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error("Unicode RPC did not settle within 10 seconds")), 10_000); + }), + ]); + } finally { + clearTimeout(timer); + } +} +try { + const bundle = join(temp, "transports.cjs"); + await build({ + stdin: { + contents: 'export { HostProcess } from "./apps/desktop/electron/main/host-process"; export { AgentSidecar } from "./apps/desktop/electron/main/agent-sidecar";', + resolveDir: root, + loader: "ts", + }, + outfile: bundle, + bundle: true, + platform: "node", + format: "cjs", + // Keep production development-path resolution pointed at this candidate. + define: { __dirname: JSON.stringify(join(root, "apps/desktop/electron/main")) }, + }); + const { HostProcess, AgentSidecar } = require(bundle); + process.env.PI_DESKTOP_HOST_BIN ??= join(root, "target/debug", `pi-desktop-host-core${process.platform === "win32" ? ".exe" : ""}`); + host = new HostProcess(join(temp, "data"), () => {}); + await deadline(host.handshake()); + const created = await deadline(host.call("session.create", { title: "Unicode regression", mode: "agent", projectPath: temp })); + const sessionId = created.session.id; + await deadline(host.call("session.appendMessage", { + sessionId, + message: { id: "unicode-message", role: "user", content: `history ${text}`, createdAt: new Date().toISOString(), status: "complete" }, + })); + const readSession = async () => { + const result = await deadline(host.call("session.get", { id: sessionId })); + assert.equal(result.session.messages[0].content, `history ${text}`); + }; + await readSession(); + await host.dispose(); + host = new HostProcess(join(temp, "data"), () => {}); + await deadline(host.handshake()); + await readSession(); + console.log("PASS session append/get/restart preserves Unicode separators"); + + sidecar = new AgentSidecar(() => {}); + sidecar.setHost(host); + const unknownMethod = `unknown.${text}`; + await assert.rejects(deadline(sidecar.call(unknownMethod)), (error) => error.message === `method not found: ${unknownMethod}`); + const events = []; + let finish; + const finished = new Promise((resolve) => { finish = resolve; }); + sidecar.onNotification((method, envelope) => { + if (method !== "agent.event" || envelope.sessionId !== sessionId) return; + events.push(envelope.event); + if (envelope.event.type === "agent_end") finish(); + }); + server = createServer(async (req, res) => { + try { + let body = ""; + for await (const chunk of req) body += chunk; + const payload = JSON.parse(body); + requests.push(payload); + const base = { id: "unicode-reply", object: "chat.completion.chunk", created: 1, model: payload.model }; + res.writeHead(200, { "content-type": "text/event-stream" }); + res.write(`data: ${JSON.stringify({ ...base, choices: [{ index: 0, delta: { role: "assistant", content: text }, finish_reason: null }] })}\n\n`); + res.write(`data: ${JSON.stringify({ ...base, choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 } })}\n\n`); + res.end("data: [DONE]\n\n"); + } catch { + res.writeHead(500); + res.end("Invalid fixture request"); + } + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const accepted = await deadline(sidecar.call("agent.prompt", { + sessionId, content: text, projectPath: temp, thinkingLevel: "off", + provider: { + id: "fixture", name: "Fixture", modelId: "fixture-model", apiKey: "", authKind: "none", + baseUrl: `http://127.0.0.1:${server.address().port}/v1`, apiStyle: "openai-chat", + supportsReasoning: false, supportedThinkingLevels: ["off"], + }, + commandShell: { id: "bash", label: "Bash", dialect: "posix", available: true, isDefault: true }, + })); + assert.equal(accepted.accepted, true); + await deadline(finished); + assert.equal(requests.length, 1); + const wireMessages = JSON.stringify(requests[0].messages); + assert.ok(wireMessages.includes(JSON.stringify(`history ${text}`).slice(1, -1)), "history crossed the parent host proxy unchanged"); + assert.ok(wireMessages.includes(JSON.stringify(text).slice(1, -1)), "prompt reached the local provider unchanged"); + assert.ok(JSON.stringify(events).includes(JSON.stringify(text).slice(1, -1)), "streamed Unicode answer reached main"); + const answer = events.find((event) => event.type === "message_end" && event.message?.role === "assistant")?.message; + assert.equal(answer?.content, text); + // Main owns persistence; exercise its HostProcess boundary with the received row. + await deadline(host.call("session.appendMessage", { sessionId, message: answer })); + const stored = await deadline(host.call("session.get", { id: sessionId })); + assert.ok(stored.session.messages.some((message) => message.role === "assistant" && message.content === text)); + console.log("PASS prompt, history proxy, streamed answer, and saved answer preserve Unicode"); + await deadline(sidecar.call("sidecar.health")); + console.log("PASS sidecar requests, error replies, and subsequent health request"); +} finally { + await sidecar?.dispose(); + await host?.dispose(); + if (server) { + server.closeAllConnections(); + await new Promise((resolve) => server.close(resolve)); + } + if (previousBinary === undefined) delete process.env.PI_DESKTOP_HOST_BIN; + else process.env.PI_DESKTOP_HOST_BIN = previousBinary; + await rm(temp, { recursive: true, force: true }); +}