diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 30902977..010c76d7 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,8 +2,13 @@ ## [Unreleased] +### Breaking Changes + +- RPC mode now uses strict LF-delimited JSONL framing. Clients must split records on `\n` only instead of using generic line readers such as Node `readline`, which also split on Unicode separators inside JSON payloads ([#1911](https://github.com/badlogic/pi-mono/issues/1911)) + ### Fixed +- Fixed RPC mode stdin/stdout framing to use strict LF-delimited JSONL instead of `readline`, so payloads containing `U+2028` or `U+2029` no longer corrupt command or event streams ([#1911](https://github.com/badlogic/pi-mono/issues/1911)) - Fixed `pi config` misclassifying `~/.agents/skills` as project-scoped in non-git directories under `$HOME`, so toggling those skills no longer writes project overrides to `.pi/settings.json` ([#1915](https://github.com/badlogic/pi-mono/issues/1915)) ## [0.56.3] - 2026-03-06 diff --git a/packages/coding-agent/README.md b/packages/coding-agent/README.md index c01af209..525091e7 100644 --- a/packages/coding-agent/README.md +++ b/packages/coding-agent/README.md @@ -395,6 +395,8 @@ For non-Node.js integrations, use RPC mode over stdin/stdout: pi --mode rpc ``` +RPC mode uses strict LF-delimited JSONL framing. Clients must split records on `\n` only. Do not use generic line readers like Node `readline`, which also split on Unicode separators inside JSON payloads. + See [docs/rpc.md](docs/rpc.md) for the protocol. --- diff --git a/packages/coding-agent/docs/rpc.md b/packages/coding-agent/docs/rpc.md index b41d60de..a037c04e 100644 --- a/packages/coding-agent/docs/rpc.md +++ b/packages/coding-agent/docs/rpc.md @@ -24,6 +24,17 @@ Common options: All commands support an optional `id` field for request/response correlation. If provided, the corresponding response will include the same `id`. +### Framing + +RPC mode uses strict JSONL semantics with LF (`\n`) as the only record delimiter. + +This matters for clients: +- Split records on `\n` only +- Accept optional `\r\n` input by stripping a trailing `\r` +- Do not use generic line readers that treat Unicode separators as newlines + +In particular, Node `readline` is not protocol-compliant for RPC mode because it also splits on `U+2028` and `U+2029`, which are valid inside JSON strings. + ## Commands ### Prompting @@ -1292,13 +1303,39 @@ For a complete example of handling the extension UI protocol, see [`examples/rpc ```javascript const { spawn } = require("child_process"); -const readline = require("readline"); +const { StringDecoder } = require("string_decoder"); const agent = spawn("pi", ["--mode", "rpc", "--no-session"]); -readline.createInterface({ input: agent.stdout }).on("line", (line) => { +function attachJsonlReader(stream, onLine) { + const decoder = new StringDecoder("utf8"); + let buffer = ""; + + stream.on("data", (chunk) => { + buffer += typeof chunk === "string" ? chunk : decoder.write(chunk); + + while (true) { + const newlineIndex = buffer.indexOf("\n"); + if (newlineIndex === -1) break; + + let line = buffer.slice(0, newlineIndex); + buffer = buffer.slice(newlineIndex + 1); + if (line.endsWith("\r")) line = line.slice(0, -1); + onLine(line); + } + }); + + stream.on("end", () => { + buffer += decoder.end(); + if (buffer.length > 0) { + onLine(buffer.endsWith("\r") ? buffer.slice(0, -1) : buffer); + } + }); +} + +attachJsonlReader(agent.stdout, (line) => { const event = JSON.parse(line); - + if (event.type === "message_update") { const { assistantMessageEvent } = event; if (assistantMessageEvent.type === "text_delta") { diff --git a/packages/coding-agent/src/modes/rpc/jsonl.ts b/packages/coding-agent/src/modes/rpc/jsonl.ts new file mode 100644 index 00000000..8962c734 --- /dev/null +++ b/packages/coding-agent/src/modes/rpc/jsonl.ts @@ -0,0 +1,58 @@ +import type { Readable } from "node:stream"; +import { StringDecoder } from "node:string_decoder"; + +/** + * Serialize a single strict JSONL record. + * + * Framing is LF-only. Payload strings may contain other Unicode separators such as + * U+2028 and U+2029. Clients must split records on `\n` only. + */ +export function serializeJsonLine(value: unknown): string { + return `${JSON.stringify(value)}\n`; +} + +/** + * Attach an LF-only JSONL reader to a stream. + * + * This intentionally does not use Node readline. Readline splits on additional + * Unicode separators that are valid inside JSON strings and therefore does not + * implement strict JSONL framing. + */ +export function attachJsonlLineReader(stream: Readable, onLine: (line: string) => void): () => void { + const decoder = new StringDecoder("utf8"); + let buffer = ""; + + const emitLine = (line: string) => { + onLine(line.endsWith("\r") ? line.slice(0, -1) : line); + }; + + const onData = (chunk: string | Buffer) => { + buffer += typeof chunk === "string" ? chunk : decoder.write(chunk); + + while (true) { + const newlineIndex = buffer.indexOf("\n"); + if (newlineIndex === -1) { + return; + } + + emitLine(buffer.slice(0, newlineIndex)); + buffer = buffer.slice(newlineIndex + 1); + } + }; + + const onEnd = () => { + buffer += decoder.end(); + if (buffer.length > 0) { + emitLine(buffer); + buffer = ""; + } + }; + + stream.on("data", onData); + stream.on("end", onEnd); + + return () => { + stream.off("data", onData); + stream.off("end", onEnd); + }; +} diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index b5c61dfb..e012ca05 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -5,12 +5,12 @@ */ import { type ChildProcess, spawn } from "node:child_process"; -import * as readline from "node:readline"; import type { AgentEvent, AgentMessage, ThinkingLevel } from "@mariozechner/pi-agent-core"; import type { ImageContent } from "@mariozechner/pi-ai"; import type { SessionStats } from "../../core/agent-session.js"; import type { BashResult } from "../../core/bash-executor.js"; import type { CompactionResult } from "../../core/compaction/index.js"; +import { attachJsonlLineReader, serializeJsonLine } from "./jsonl.js"; import type { RpcCommand, RpcResponse, RpcSessionState, RpcSlashCommand } from "./rpc-types.js"; // ============================================================================ @@ -53,7 +53,7 @@ export type RpcEventListener = (event: AgentEvent) => void; export class RpcClient { private process: ChildProcess | null = null; - private rl: readline.Interface | null = null; + private stopReadingStdout: (() => void) | null = null; private eventListeners: RpcEventListener[] = []; private pendingRequests: Map void; reject: (error: Error) => void }> = new Map(); @@ -94,13 +94,8 @@ export class RpcClient { this.stderr += data.toString(); }); - // Set up line reader for stdout - this.rl = readline.createInterface({ - input: this.process.stdout!, - terminal: false, - }); - - this.rl.on("line", (line) => { + // Set up strict JSONL reader for stdout. + this.stopReadingStdout = attachJsonlLineReader(this.process.stdout!, (line) => { this.handleLine(line); }); @@ -118,7 +113,8 @@ export class RpcClient { async stop(): Promise { if (!this.process) return; - this.rl?.close(); + this.stopReadingStdout?.(); + this.stopReadingStdout = null; this.process.kill("SIGTERM"); // Wait for process to exit @@ -135,7 +131,6 @@ export class RpcClient { }); this.process = null; - this.rl = null; this.pendingRequests.clear(); } @@ -493,7 +488,7 @@ export class RpcClient { }, }); - this.process!.stdin!.write(`${JSON.stringify(fullCommand)}\n`); + this.process!.stdin!.write(serializeJsonLine(fullCommand)); }); } diff --git a/packages/coding-agent/src/modes/rpc/rpc-mode.ts b/packages/coding-agent/src/modes/rpc/rpc-mode.ts index 89d35210..b38f43a1 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-mode.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-mode.ts @@ -12,7 +12,6 @@ */ import * as crypto from "node:crypto"; -import * as readline from "readline"; import type { AgentSession } from "../../core/agent-session.js"; import type { ExtensionUIContext, @@ -20,6 +19,7 @@ import type { ExtensionWidgetOptions, } from "../../core/extensions/index.js"; import { type Theme, theme } from "../interactive/theme/theme.js"; +import { attachJsonlLineReader, serializeJsonLine } from "./jsonl.js"; import type { RpcCommand, RpcExtensionUIRequest, @@ -44,7 +44,7 @@ export type { */ export async function runRpcMode(session: AgentSession): Promise { const output = (obj: RpcResponse | RpcExtensionUIRequest | object) => { - console.log(JSON.stringify(obj)); + process.stdout.write(serializeJsonLine(obj)); }; const success = ( @@ -587,6 +587,8 @@ export async function runRpcMode(session: AgentSession): Promise { * Check if shutdown was requested and perform shutdown if so. * Called after handling each command when waiting for the next command. */ + let detachInput = () => {}; + async function checkShutdownRequested(): Promise { if (!shutdownRequested) return; @@ -595,19 +597,12 @@ export async function runRpcMode(session: AgentSession): Promise { await currentRunner.emit({ type: "session_shutdown" }); } - // Close readline interface to stop waiting for input - rl.close(); + detachInput(); + process.stdin.pause(); process.exit(0); } - // Listen for JSON input - const rl = readline.createInterface({ - input: process.stdin, - output: process.stdout, - terminal: false, - }); - - rl.on("line", async (line: string) => { + const handleInputLine = async (line: string) => { try { const parsed = JSON.parse(line); @@ -632,6 +627,10 @@ export async function runRpcMode(session: AgentSession): Promise { } catch (e: any) { output(error(undefined, "parse", `Failed to parse command: ${e.message}`)); } + }; + + detachInput = attachJsonlLineReader(process.stdin, (line) => { + void handleInputLine(line); }); // Keep process alive forever diff --git a/packages/coding-agent/test/rpc-jsonl.test.ts b/packages/coding-agent/test/rpc-jsonl.test.ts new file mode 100644 index 00000000..f8ca4538 --- /dev/null +++ b/packages/coding-agent/test/rpc-jsonl.test.ts @@ -0,0 +1,65 @@ +import { Readable } from "node:stream"; +import { describe, expect, test } from "vitest"; +import { attachJsonlLineReader, serializeJsonLine } from "../src/modes/rpc/jsonl.js"; + +describe("RPC JSONL framing", () => { + test("serializes strict JSONL records without escaping Unicode separators", () => { + const line = serializeJsonLine({ text: "a\u2028b\u2029c" }); + + expect(line).toContain("a\u2028b\u2029c"); + expect(line.endsWith("\n")).toBe(true); + expect(JSON.parse(line.trim())).toEqual({ text: "a\u2028b\u2029c" }); + }); + + test("splits on LF only and preserves U+2028/U+2029 inside payloads", async () => { + const lines: string[] = []; + const stream = Readable.from([serializeJsonLine({ text: "a\u2028b\u2029c" })]); + + const done = new Promise((resolve) => { + stream.on("end", resolve); + }); + + attachJsonlLineReader(stream, (line) => { + lines.push(line); + }); + + await done; + + expect(lines).toHaveLength(1); + expect(JSON.parse(lines[0])).toEqual({ text: "a\u2028b\u2029c" }); + }); + + test("handles CRLF-delimited input", async () => { + const lines: string[] = []; + const stream = Readable.from([Buffer.from('{"a":1}\r\n{"b":2}\r\n')]); + + const done = new Promise((resolve) => { + stream.on("end", resolve); + }); + + attachJsonlLineReader(stream, (line) => { + lines.push(line); + }); + + await done; + + expect(lines).toEqual(['{"a":1}', '{"b":2}']); + }); + + test("emits a final line without trailing LF", async () => { + const lines: string[] = []; + const stream = Readable.from([Buffer.from('{"a":1}')]); + + const done = new Promise((resolve) => { + stream.on("end", resolve); + }); + + attachJsonlLineReader(stream, (line) => { + lines.push(line); + }); + + await done; + + expect(lines).toEqual(['{"a":1}']); + }); +});