From 27ccc8a93c78aee24e052465bfec62fc64526c6e Mon Sep 17 00:00:00 2001 From: kb Date: Tue, 20 Oct 2026 11:21:34 -0400 Subject: [PATCH] fix(muse): split MSP frames on newline only node:readline also ends lines at U+2028 and U+2029, which JSON allows unescaped inside strings, so a frame quoting either character was cut in two and the session failed with invalidFrame. Read muse serve stdout with a small LF-only reader that decodes UTF-8 in the stream, drops a trailing CR and flushes a final unterminated line. Closes #7084 Co-authored-by: Karl Bitz Co-authored-by: pernakor <94871534+pernakor@users.noreply.github.com> --- plugins/muse-provider/server/connection.ts | 38 +++++++++++++++++++-- plugins/muse-provider/test/fake-muse.mjs | 21 +++++++++++- plugins/muse-provider/test/provider.test.ts | 22 ++++++++++++ 3 files changed, 77 insertions(+), 4 deletions(-) diff --git a/plugins/muse-provider/server/connection.ts b/plugins/muse-provider/server/connection.ts index 3fca09ef5e..a1782233df 100644 --- a/plugins/muse-provider/server/connection.ts +++ b/plugins/muse-provider/server/connection.ts @@ -1,7 +1,6 @@ import type { ProviderLaunch } from "@getpaseo/plugin/server/provider"; import { spawnProcess, terminateProcess } from "@getpaseo/plugin/server"; import { randomBytes } from "node:crypto"; -import { createInterface } from "node:readline"; import { setTimeout as delay } from "node:timers/promises"; import { z } from "zod"; import { MuseError, classifyExit } from "./errors.js"; @@ -16,6 +15,40 @@ export function commandId(): string { return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`; } +/** + * Splits MSP output on "\n" only. `node:readline` also ends lines at U+2028 and U+2029, which JSON + * allows unescaped inside strings, so it cut frames quoting either character in two. + */ +function readLines( + input: NodeJS.ReadableStream, + onLine: (line: string) => void, +): { close(): void } { + // Decoding in the stream keeps a multibyte character whole when it straddles two chunks. + input.setEncoding("utf8"); + let partial = ""; + const emit = (line: string) => onLine(line.endsWith("\r") ? line.slice(0, -1) : line); + const onData = (chunk: string) => { + let start = 0; + for (let end = chunk.indexOf("\n"); end !== -1; end = chunk.indexOf("\n", start)) { + emit(partial + chunk.slice(start, end)); + partial = ""; + start = end + 1; + } + partial += chunk.slice(start); + }; + const onEnd = () => { + if (partial) emit(partial); + }; + input.on("data", onData); + input.on("end", onEnd); + return { + close() { + input.off("data", onData); + input.off("end", onEnd); + }, + }; +} + interface PendingRequest { resolve(value: unknown): void; reject(error: Error): void; @@ -61,8 +94,7 @@ export class MspConnection { this.child.stderr.on("data", (chunk: string) => { this.stderr = (this.stderr + chunk).slice(-8192); }); - const lines = createInterface({ input: this.child.stdout }); - lines.on("line", (line) => { + const lines = readLines(this.child.stdout, (line) => { try { this.receive(frameSchema.parse(JSON.parse(line))); } catch (error) { diff --git a/plugins/muse-provider/test/fake-muse.mjs b/plugins/muse-provider/test/fake-muse.mjs index 16cc821b79..4f54f91d3e 100644 --- a/plugins/muse-provider/test/fake-muse.mjs +++ b/plugins/muse-provider/test/fake-muse.mjs @@ -341,6 +341,12 @@ function applyWireVariants(message) { message.result.session.modelId = null; message.result.session.providerId = null; } + if ( + message.method === "item/completed" && + message.params.item.kind === "agentMessage" && + process.env.MUSE_TEST_AGENT_TEXT + ) + message.params.item.text = process.env.MUSE_TEST_AGENT_TEXT; } function parityDelivery(message) { if (process.env.MUSE_TEST_WORKFLOW && message.method === "turn/completed") @@ -457,7 +463,20 @@ function send(frame) { if (frame.method === "item/completed" && process.env.MUSE_TEST_HISTORY) { appendFileSync(process.env.MUSE_TEST_HISTORY, JSON.stringify(frame.params.item) + "\n"); } - process.stdout.write(JSON.stringify(frame) + "\n"); + const line = Buffer.from(JSON.stringify(frame) + "\n"); + if ( + process.env.MUSE_TEST_SPLIT_WRITE && + frame.method === "item/completed" && + frame.params.item.kind === "agentMessage" + ) { + // Cut inside the first multibyte character and pause so Paseo reads the halves as two chunks. + const cut = line.findIndex((byte) => byte > 0x7f) + 1; + process.stdout.write(line.subarray(0, cut)); + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 50); + process.stdout.write(line.subarray(cut)); + return; + } + process.stdout.write(line); } function respond(frame, result) { send({ jsonrpc: "2.0", id: frame.id, result }); diff --git a/plugins/muse-provider/test/provider.test.ts b/plugins/muse-provider/test/provider.test.ts index 4868634d70..a0ebc65c14 100644 --- a/plugins/muse-provider/test/provider.test.ts +++ b/plugins/muse-provider/test/provider.test.ts @@ -1330,6 +1330,28 @@ test("text deltas with an omitted field stream as text", async () => { ), ).toBe(true); }); +for (const [variant, env] of [ + ["U+2028 and U+2029", { MUSE_TEST_AGENT_TEXT: "before
middle
after" }], + [ + "a multibyte character split across stdout writes", + { MUSE_TEST_AGENT_TEXT: "split \u{1F600} here", MUSE_TEST_SPLIT_WRITE: "1" }, + ], +] as const) { + test(`assistant text with ${variant} arrives in one MSP frame`, async () => { + const h = await harness("text-reasoning", env); + await h.open(); + await h.prompt(); + await h.wait( + (e) => + (e.type === "session.turn" && e.state === "completed") || + e.type === "session.runtime_failed", + ); + expect(h.events.filter((e) => e.type === "session.runtime_failed")).toEqual([]); + expect( + h.events.findLast((e) => e.type === "timeline.item" && e.item.type === "assistant_message"), + ).toMatchObject({ item: { text: env.MUSE_TEST_AGENT_TEXT } }); + }); +} test("sessions with a null model import and resume without a null persistence model", async () => { const h = await harness("resume-without-cursor", { MUSE_TEST_NULL_MODEL: "1" }); await h.send({