fix(rpc): 修复特殊 Unicode 字符导致的会话读取和发送超时

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<typeof createInterface>;
+  private stdoutReader?: ReturnType<typeof readNdjsonLines>;
   // Tools served by Electron main itself (e.g. BrowserPreview drives the
   // work panel's WebContentsView) — host-core never sees these.
   private localTools = new Map<string, LocalToolHandler>();
@@ -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;
     }
 

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<void>;
   private resolveExit!: () => void;
   private disposePromise?: Promise<void>;
-  private readline?: ReturnType<typeof createInterface>;
+  private stdoutReader?: ReturnType<typeof readNdjsonLines>;
   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) {

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);
+});

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
 

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.

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 风格
 

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.

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",

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<unknown> {
   }
 }
 
-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

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";

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 };
+}