test(muse): cover CRLF and final stdout frames

Preserve the framing contracts previously supplied by readline, including a CRLF split across pipe writes and a final frame delivered at stdout EOF.

plugins/muse-provider/test/fake-muse.mjs

@@ -460,10 +460,24 @@ 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");
   }
-  const line = Buffer.from(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" &&

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

@@ -1352,6 +1352,32 @@ for (const [variant, env] of [
     ).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({