fix(coding-agent): use strict JSONL framing fixes #1911

This commit is contained in:
Mario Zechner
2026-03-07 14:20:44 +01:00
parent 841c95ac9c
commit e3adaf1bd9
7 changed files with 188 additions and 27 deletions

View File

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

View File

@@ -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<string, { resolve: (response: RpcResponse) => 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<void> {
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));
});
}

View File

@@ -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<never> {
const output = (obj: RpcResponse | RpcExtensionUIRequest | object) => {
console.log(JSON.stringify(obj));
process.stdout.write(serializeJsonLine(obj));
};
const success = <T extends RpcCommand["type"]>(
@@ -587,6 +587,8 @@ export async function runRpcMode(session: AgentSession): Promise<never> {
* 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<void> {
if (!shutdownRequested) return;
@@ -595,19 +597,12 @@ export async function runRpcMode(session: AgentSession): Promise<never> {
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<never> {
} 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