test(muse): preserve timeout reproductions and patch validation

Retain the controlled protocol reproductions for #6865 without changing
production behavior. Keep malformed patch responses distinguishable from
unavailable output when evaluating the proposed recovery behavior.

Co-authored-by: Blah <michaelaguilera12345@gmail.com>

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

@@ -1,4 +1,4 @@
-import { appendFileSync, readFileSync } from "node:fs";
+import { appendFileSync, existsSync, readFileSync } from "node:fs";
 import { createInterface } from "node:readline";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
@@ -298,7 +298,38 @@ function validateCommand(frame, recorded) {
   )
     throw new Error("Wrong approval stage");
 }
+function failedOrDelayedResponse(recordedMessage, recorded, frame) {
+  if (recordedMessage.method || recordedMessage.id !== recorded.id) return false;
+  if (frame.method === "item/readOutput" && process.env.MUSE_TEST_OUTPUT_FAILURE) {
+    if (process.env.MUSE_TEST_OUTPUT_FAILURE === "malformed") {
+      process.stdout.write(
+        JSON.stringify({
+          jsonrpc: "2.0",
+          id: frame.id,
+          result: { encoding: "utf8", eof: true, content: null },
+        }) + "\n",
+      );
+      return true;
+    }
+    // Only the patch response fails; subsequent turn notifications still arrive.
+    if (process.env.MUSE_TEST_OUTPUT_FAILURE === "error")
+      rpcError(frame, -32000, "outputUnavailable", "Stored patch unavailable");
+    return true;
+  }
+  if (frame.method === "session/resume" && process.env.MUSE_TEST_RESUME_GATE) {
+    const gate = process.env.MUSE_TEST_RESUME_GATE;
+    delete process.env.MUSE_TEST_RESUME_GATE;
+    function release() {
+      if (existsSync(gate)) emitFixtureMessage(recordedMessage, recorded, frame);
+      else setTimeout(release, 10);
+    }
+    release();
+    return true;
+  }
+  return false;
+}
 function emitFixtureMessage(recordedMessage, recorded, frame) {
+  if (failedOrDelayedResponse(recordedMessage, recorded, frame)) return;
   const message = replace(recordedMessage);
   if (!message.method && message.id === recorded.id) message.id = frame.id;
   if (

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

@@ -138,6 +138,14 @@ async function harness(
       .filter(Boolean)
       .map((line) => JSON.parse(line));
   }
+  async function waitForRequest(method: string) {
+    const deadline = Date.now() + 5000;
+    while (Date.now() < deadline) {
+      if ((await recorded()).some((frame) => frame.method === method)) return;
+      await delay(10);
+    }
+    throw new Error(`Missing request: ${method}`);
+  }
   return {
     provider,
     launch,
@@ -148,6 +156,7 @@ async function harness(
     open,
     prompt,
     recorded,
+    waitForRequest,
     root,
     usageSource,
   };
@@ -260,6 +269,58 @@ test("tool calls classify read, shell, and edit with fetched JSON patch converte
   );
 });
 
+test.each(["error", "timeout"])(
+  "patch fetch %s preserves the edit and completes the turn",
+  async (failure) => {
+    const h = await harness("tools-edit", { MUSE_TEST_OUTPUT_FAILURE: failure });
+    expect(await h.open()).toMatchObject({ type: "session.ready" });
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+    await h.prompt();
+    await h.waitForRequest("item/readOutput");
+    if (failure === "timeout") await vi.advanceTimersByTimeAsync(60000);
+    expect(
+      await h.wait(
+        (event) =>
+          event.type === "session.runtime_failed" ||
+          (event.type === "session.turn" && event.state === "completed"),
+      ),
+    ).toMatchObject({ type: "session.turn", state: "completed" });
+    const edit = h.events.findLast(
+      (event) =>
+        event.type === "timeline.item" &&
+        event.item.type === "tool_call" &&
+        event.item.name === "edit_file",
+    );
+    expect(edit).toMatchObject({
+      type: "timeline.item",
+      item: {
+        type: "tool_call",
+        name: "edit_file",
+        status: "completed",
+        error: null,
+        detail: {
+          type: "edit",
+          filePath: "f",
+          oldString: "beta",
+          newString: "gamma",
+          unifiedDiff: undefined,
+        },
+      },
+    });
+    expect(
+      h.events.findLast(
+        (event) => event.type === "timeline.item" && event.item.type === "assistant_message",
+      ),
+    ).toMatchObject({
+      item: { text: "Done. `f` had 2 lines; changed line 2 from `beta` to `gamma`.\n" },
+    });
+    expect(h.events.filter((event) => event.type === "session.runtime_failed")).toEqual([]);
+    expect(
+      h.events.filter((event) => event.type === "session.turn" && event.state === "completed"),
+    ).toHaveLength(1);
+  },
+);
+
 test("multi-stage approvals surface Stage 2 then Stage 3 and send the current requirement guard", async () => {
   const h = await harness("multi-stage-approval");
   await h.open();
@@ -377,6 +438,36 @@ test("a fresh connection restores earlier messages from persisted session state"
   await reopened.wait((event) => event.type === "session.turn" && event.state === "completed");
 });
 
+test("resume succeeds after more than ten seconds and can complete a new turn", async () => {
+  const root = await mkdtemp(path.join(os.tmpdir(), "muse-resume-gate-"));
+  roots.push(root);
+  const gate = path.join(root, "release");
+  const h = await harness("resume-without-cursor", { MUSE_TEST_RESUME_GATE: gate });
+  vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+  const opened = h.open({ version: 1, data: { sessionId: "saved-session" } });
+  await h.waitForRequest("session/resume");
+  // The child holds its real MSP response while the provider's request clock advances.
+  await vi.advanceTimersByTimeAsync(15000);
+  await writeFile(gate, "release");
+  expect(await opened).toMatchObject({ type: "session.ready" });
+  expect(h.events).toEqual(
+    expect.arrayContaining([
+      expect.objectContaining({
+        type: "timeline.item",
+        item: expect.objectContaining({ type: "assistant_message", text: "RESUME_MARKER_OK" }),
+      }),
+    ]),
+  );
+  await h.prompt();
+  await h.wait((event) => event.type === "session.turn" && event.state === "completed");
+  expect(
+    h.events.filter(
+      (event) => event.type === "request.failed" || event.type === "session.runtime_failed",
+    ),
+  ).toEqual([]);
+  expect((await h.recorded()).filter((frame) => frame.method === "session/start")).toEqual([]);
+});
+
 for (const cursor of [undefined, "v:old:7"]) {
   test(`resume ${cursor ? "with a legacy persistence cursor" : "without a persistence cursor"} replays inline history`, async () => {
     const h = await harness("resume-without-cursor");
@@ -1743,3 +1834,16 @@ test("Muse reports belong only to the session's account and remain available glo
   expect(await registry.listReports({ agentId: "muse-unknown" })).toEqual([]);
   expect((await registry.listReports()).map((report) => report.id)).toEqual(accountIds);
 });
+
+test("malformed patch responses remain protocol errors", async () => {
+  const h = await harness("tools-edit", { MUSE_TEST_OUTPUT_FAILURE: "malformed" });
+  expect(await h.open()).toMatchObject({ type: "session.ready" });
+  await h.prompt();
+  expect(
+    await h.wait(
+      (event) =>
+        event.type === "session.runtime_failed" ||
+        (event.type === "session.turn" && event.state === "completed"),
+    ),
+  ).toMatchObject({ type: "session.runtime_failed" });
+});