test(server): drain Git traces before fixture cleanup

The missing-folder observation test deletes its trace directory while a queued diagnostic append can still recreate the trace file, producing ENOTEMPTY in CI. Drain the trace queue after service shutdown and before removing the fixture.

packages/server/src/server/workspace-git-service.observation.integration.test.ts

@@ -871,6 +871,7 @@ test.skipIf(process.platform !== "linux")(
     } finally {
       subscription.unsubscribe();
       await service.dispose();
+      await flushGitCommandTrace();
       if (previousTrace === undefined) delete process.env.PASEO_GIT_TRACE_FILE;
       else process.env.PASEO_GIT_TRACE_FILE = previousTrace;
       rmSync(root, { recursive: true, force: true });

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) {

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")
@@ -454,10 +460,37 @@ function replace(value) {
   return value;
 }
 function send(frame) {
+  if (process.stdout.writableEnded) return;
   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 text = JSON.stringify(frame);
+  if (frame.method === "item/completed" && frame.params.item.kind === "agentMessage") {
+    if (process.env.MUSE_TEST_FRAME_ENDING === "eof") {
+      process.stdout.end(text);
+      return;
+    }
+    if (process.env.MUSE_TEST_FRAME_ENDING === "split-crlf") {
+      process.stdout.write(text + "\r");
+      Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 50);
+      process.stdout.write("\n");
+      return;
+    }
+  }
+  const line = Buffer.from(text + "\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 });

plugins/muse-provider/test/provider.test.ts

@@ -1330,6 +1330,54 @@ 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 } });
+  });
+}
+for (const ending of ["split-crlf", "eof"] as const) {
+  test(`a final assistant frame with ${ending} is delivered exactly once`, async () => {
+    const text = `Final message with ${ending}`;
+    const h = await harness("text-reasoning", {
+      MUSE_TEST_AGENT_TEXT: text,
+      MUSE_TEST_FRAME_ENDING: ending,
+    });
+    await h.open();
+    await h.prompt();
+    const isFinalMessage = (event: ProviderEvent) =>
+      event.type === "timeline.item" &&
+      event.item.type === "assistant_message" &&
+      event.item.text === text;
+    await h.wait(isFinalMessage);
+    if (ending === "split-crlf") {
+      await h.wait(
+        (event) =>
+          (event.type === "session.turn" && event.state === "completed") ||
+          event.type === "session.runtime_failed",
+      );
+    }
+    await h.connection.close();
+    expect(h.events.filter(isFinalMessage)).toHaveLength(1);
+    expect(h.events.filter((event) => event.type === "session.runtime_failed")).toEqual([]);
+  });
+}
 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({