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..854125283 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"; @@ -7,6 +6,7 @@ import { ErrorCodes, MAX_HOST_STDIN_LINE_BYTES, PROTOCOL_VERSION, + readNdjsonLines, rpcTimeoutMs, stripProxyEnv, } from "@pi-desktop/shared"; @@ -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/rpc-lifecycle-contract.test.mjs b/apps/desktop/test/rpc-lifecycle-contract.test.mjs index 65d556c0a..493a359fe 100644 --- a/apps/desktop/test/rpc-lifecycle-contract.test.mjs +++ b/apps/desktop/test/rpc-lifecycle-contract.test.mjs @@ -27,11 +27,30 @@ const rpcTimeoutSource = await readFile( new URL("../../../packages/shared/src/rpc-timeouts.ts", import.meta.url), "utf8", ); +const agentSidecarEntrySource = await readFile( + new URL("../../../packages/agent-runtime/src/sidecar.ts", import.meta.url), + "utf8", +); const apiSource = await readFile( new URL("../src/lib/api.ts", import.meta.url), "utf8", ); +test("stdio RPC readers split frames on LF only", () => { + for (const [name, source] of [ + ["host-process", hostSource], + ["agent-sidecar", sidecarSource], + ["runtime sidecar", agentSidecarEntrySource], + ]) { + 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`, + ); + } +}); + test("sidecar detaches host listeners and gates every child write", () => { assert.match(sidecarSource, /private closeTransport\(error: Error\)/); assert.match(sidecarSource, /unsubscribeHostExit/); 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..58ca9e363 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`; `packages/shared/src/ndjson.test.ts` + 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..d5323acc5 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,11 @@ MVP 传输决策 (**D001**)： - 流程：Electron 主要生成 Rust host-core sidecar - 通道：子进程 stdin/stdout -- 成帧：每行一个 JSON 对象 (NDJSON) +- 成帧：每行一个以 LF 分隔的 JSON 对象（NDJSON）；接受 CRLF。 + JSON 字符串内的 U+2028 与 U+2029 属于载荷，不是帧分隔符。 + 所有 Node stdio 读取器会跨输入块保留 UTF-8 字符，并在传输关闭时释放缓冲片段和监听器。 + 为兼容起见，EOF 时接受最后一帧未以换行结束的情况。 +- 非法 JSON 帧会先产出仅含字节长度、不含载荷文本的诊断，然后丢弃。后续完整帧仍可读。现有会话文本不会被改写或迁移。 - 编码：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..209bc54ce 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 成帧 | 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,14 @@ runner 会在运行时的隔离临时目录中生成六个插件形态 fixture - **验收**：G（远程市场来源） - **里程碑**：M6+ - **状态**：草稿 + +### E2E-RPC-unicode-separators + +- **前置条件**：已构建 host-core、shared 与 agent-runtime；隔离的临时数据目录；仅回环的 fixture provider。不使用真实凭证。 +- **步骤**：追加一条含 U+2028/U+2029、中文、emoji 以及转义 CR/LF 的用户消息；读取、重启 host 后再读。通过 AgentSidecar 发送 Unicode 提示；经 parent host proxy 恢复历史；流式返回并持久化 Unicode 回复。发送一条含相同字符的未知方法，再发健康检查。 +- **预期**：文本在持久化与所有 stdio 方向上保持不变。请求在 RPC 超时前完成；错误回复之后的请求仍可用。无需迁移现有会话。 +- **自动化**：`pnpm test:e2e:rpc-unicode`；`packages/shared/src/ndjson.test.ts` 额外覆盖每个 UTF-8 切分位置、连续帧、CRLF、EOF 与销毁。 +- **规格**：03-runtime/06-host-rpc-protocol §2。 +- **验收**：A（运行时），C（会话）。 +- **里程碑**：M6+。 +- **状态**：已自动化；针对 task/PR 集成候选运行。 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..a116af74b 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"; @@ -35,6 +34,7 @@ import { normalizeMode, normalizeNetworkProxy, OAUTH_AUTH_KIND, + readNdjsonLines, } from "@pi-desktop/shared"; import type { AgentEventEnvelope, @@ -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.test.ts b/packages/shared/src/ndjson.test.ts new file mode 100644 index 000000000..e69812feb --- /dev/null +++ b/packages/shared/src/ndjson.test.ts @@ -0,0 +1,126 @@ +import { describe, expect, it } from "vitest"; +import { readNdjsonLines } from "./ndjson.js"; + +type DataListener = (chunk: string) => void; +type SignalListener = () => void; + +class FakeNdjsonInput { + private data: DataListener[] = []; + private end: SignalListener[] = []; + private close: SignalListener[] = []; + + setEncoding(_encoding: "utf8") { + return this; + } + + on(event: "data", listener: DataListener): this; + on(event: "end" | "close", listener: SignalListener): this; + on(event: string, listener: DataListener | SignalListener) { + if (event === "data") this.data.push(listener as DataListener); + else if (event === "end") this.end.push(listener as SignalListener); + else if (event === "close") this.close.push(listener as SignalListener); + return this; + } + + off(event: "data", listener: DataListener): this; + off(event: "end" | "close", listener: SignalListener): this; + off(event: string, listener: DataListener | SignalListener) { + if (event === "data") this.data = this.data.filter((item) => item !== listener); + else if (event === "end") this.end = this.end.filter((item) => item !== listener); + else if (event === "close") this.close = this.close.filter((item) => item !== listener); + return this; + } + + write(chunk: string) { + for (const listener of this.data.slice()) listener(chunk); + } + + finish(chunk?: string) { + if (chunk) this.write(chunk); + for (const listener of this.end.slice()) listener(); + } + + destroy() { + for (const listener of this.close.slice()) listener(); + } + + listenerCount(event: "data" | "end" | "close") { + return this[event].length; + } + + listeners(event: "data" | "end" | "close") { + return this[event].slice(); + } +} + +const payload = { id: "rpc", result: { content: "A\u2028中文🙂\u2029B\r\nC" } }; +const wire = new TextEncoder().encode(`${JSON.stringify(payload)}\n`); + +describe("readNdjsonLines", () => { + it("keeps U+2028/U+2029 inside JSON.stringify output as payload", () => { + expect(JSON.stringify(payload)).toContain("\u2028"); + expect(JSON.stringify(payload)).toContain("\u2029"); + const input = new FakeNdjsonInput(); + const lines: unknown[] = []; + readNdjsonLines(input, (line) => lines.push(JSON.parse(line))); + input.finish(new TextDecoder().decode(wire)); + expect(lines).toEqual([payload]); + }); + + it("preserves Unicode separators at every possible byte split", () => { + for (let split = 0; split <= wire.length; split++) { + const input = new FakeNdjsonInput(); + const lines: unknown[] = []; + readNdjsonLines(input, (line) => lines.push(JSON.parse(line))); + const decoder = new TextDecoder(); + const head = decoder.decode(wire.subarray(0, split), { stream: true }); + if (head) input.write(head); + input.finish(decoder.decode(wire.subarray(split))); + expect(lines, `split at byte ${split}`).toEqual([payload]); + } + }); + + it("accepts consecutive LF/CRLF frames, blank lines, and an EOF tail", () => { + const input = new FakeNdjsonInput(); + const lines: string[] = []; + readNdjsonLines(input, (line) => lines.push(line)); + input.write('\n{"id":1}\r'); + expect(lines, "CR alone does not terminate a frame").toEqual([""]); + input.write('\n{"id":2}\n{"id":'); + input.finish("3}"); + expect(lines).toEqual(["", '{"id":1}', '{"id":2}', '{"id":3}']); + for (const event of ["data", "end", "close"] as const) { + expect(input.listenerCount(event)).toBe(0); + } + }); + + it("stops buffered frames on close and removes only owned listeners", () => { + const input = new FakeNdjsonInput(); + const external: DataListener = () => {}; + input.on("data", external); + const lines: string[] = []; + const reader = readNdjsonLines(input, (line) => { + lines.push(line); + reader.close(); + }); + input.write("first\nsecond\npartial"); + reader.close(); + input.write("ignored\n"); + expect(lines).toEqual(["first"]); + expect(input.listeners("data")).toEqual([external]); + expect(input.listenerCount("end")).toBe(0); + expect(input.listenerCount("close")).toBe(0); + }); + + it("discards a partial frame and detaches when the stream is destroyed", () => { + const input = new FakeNdjsonInput(); + const lines: string[] = []; + readNdjsonLines(input, (line) => lines.push(line)); + input.write('{"unfinished":'); + input.destroy(); + expect(lines).toEqual([]); + for (const event of ["data", "end", "close"] as const) { + expect(input.listenerCount(event)).toBe(0); + } + }); +}); diff --git a/packages/shared/src/ndjson.ts b/packages/shared/src/ndjson.ts new file mode 100644 index 000000000..254e36d3c --- /dev/null +++ b/packages/shared/src/ndjson.ts @@ -0,0 +1,61 @@ +/** 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 type NdjsonLineReader = { close: () => void }; + +export function readNdjsonLines( + input: NdjsonInput, + onLine: (line: string) => void, +): NdjsonLineReader { + 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 }); +}