fix(muse): finish child refreshes and stop retries on failure

A final child update must read history newer than an in-flight retry. Serialize fresh reads instead of joining an older snapshot. Stop retry scheduling when the native runtime fails, including requests that reject during shutdown.

Co-authored-by: kb <kb@wardbitz.com>

plugins/muse-provider/server/children.ts

@@ -22,6 +22,7 @@ interface Child {
 }
 export class Children {
   private readonly children = new Map<string, Child>();
+  private stopped = false;
   constructor(
     private readonly host: MspConnection,
     private readonly parentId: string,
@@ -79,17 +80,18 @@ export class Children {
     });
   }
   // A failed final read must recover even after the parent finishes and stops its silence timer.
-  // Item updates and timer retries share one read so their cursors cannot race.
-  private read(id: string): Promise<void> {
+  // Each update reads fresh history after any older read, with serialized cursor advancement.
+  private async read(id: string): Promise<void> {
     const child = this.children.get(id);
-    if (!child) return Promise.resolve();
+    if (!child) return;
+    while (child.reading) await child.reading;
+    if (this.stopped || this.children.get(id) !== child) return;
     clearTimeout(child.retry);
     child.retry = undefined;
-    if (child.reading) return child.reading;
     child.reading = this.page(id)
       .catch((error: unknown) => {
         this.readFailed(id, error);
-        if (this.children.get(id) === child)
+        if (!this.stopped && this.children.get(id) === child)
           child.retry = setTimeout(() => {
             void this.read(id);
           }, READ_RETRY_MS);
@@ -151,11 +153,16 @@ export class Children {
       if (next !== null) child.cursor = next;
     } while (next !== null);
   }
-  close(): void {
-    for (const [sessionId, child] of this.children) {
+  stop(): void {
+    this.stopped = true;
+    for (const child of this.children.values()) {
       clearTimeout(child.retry);
-      this.emit({ type: "session.closed", sessionId });
+      child.retry = undefined;
     }
+  }
+  close(): void {
+    this.stop();
+    for (const sessionId of this.children.keys()) this.emit({ type: "session.closed", sessionId });
     this.children.clear();
   }
 }

plugins/muse-provider/server/session.ts

@@ -425,6 +425,7 @@ export class Session {
   }
   private runtimeFailed(error: unknown): void {
     this.live().recovery.close();
+    this.live().children.stop();
     this.options.emit({
       type: "session.runtime_failed",
       sessionId: this.options.id,

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

@@ -1,4 +1,4 @@
-import { appendFileSync, readFileSync } from "node:fs";
+import { appendFileSync, readFileSync, existsSync } from "node:fs";
 import { createInterface } from "node:readline";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
@@ -24,13 +24,14 @@ const rows = readFixture(scenario);
 if (process.env.MUSE_TEST_REQUESTS)
   appendFileSync(
     process.env.MUSE_TEST_REQUESTS,
-    JSON.stringify({ event: "hostStarted", args: process.argv.slice(2) }) + "\n",
+    JSON.stringify({ event: "hostStarted", pid: process.pid, args: process.argv.slice(2) }) + "\n",
   );
 const catalogRows = readFixture("catalog-controls");
 const replacements = new Map();
 const counts = new Map();
 const receipts = new Set();
 let pageHung = false;
+let deferredChildItem;
 const failedChildReads = new Set();
 const lines = createInterface({ input: process.stdin });
 lines.on("line", receive);
@@ -144,7 +145,34 @@ function scriptedResponse(frame) {
     respond(frame, { status: "accepted", commandId: frame.params.commandId });
     return true;
   }
-  return (frame.method === "view/page" && failedPage(frame)) || failedChildRead(frame);
+  return (
+    deferredChildPage(frame) ||
+    (frame.method === "view/page" && failedPage(frame)) ||
+    failedChildRead(frame)
+  );
+}
+// Hold an old child page until the consumer has received the final item update.
+function deferredChildPage(frame) {
+  const releasePath = process.env.MUSE_TEST_CHILD_PAGE_RELEASE;
+  if (
+    !releasePath ||
+    frame.method !== "view/page" ||
+    frame.params.sessionId !== "fixture-child" ||
+    !deferredChildItem ||
+    !failedChildReads.has("view/page")
+  )
+    return false;
+  send(deferredChildItem);
+  deferredChildItem = undefined;
+  function release() {
+    if (!existsSync(releasePath)) {
+      setTimeout(release, 5);
+      return;
+    }
+    respond(frame, { events: [], nextCursor: null });
+  }
+  release();
+  return true;
 }
 const childReadFailures = {
   "view/page": "MUSE_TEST_CHILD_PAGE_FAIL",
@@ -412,6 +440,14 @@ function applyWireVariants(message) {
   }
 }
 function parityDelivery(message) {
+  if (
+    process.env.MUSE_TEST_CHILD_PAGE_RELEASE &&
+    message.method === "item/completed" &&
+    message.params.item?.childSessionId === "fixture-child"
+  ) {
+    deferredChildItem = message;
+    return true;
+  }
   if (process.env.MUSE_TEST_WORKFLOW && message.method === "turn/completed")
     send({
       jsonrpc: "2.0",

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

@@ -1977,3 +1977,57 @@ test("closing a session cancels pending child history retries", async () => {
   await h.wait((e) => e.type === "session.closed" && e.sessionId === "paseo-session");
   expect(vi.getTimerCount()).toBe(0);
 });
+
+test("a final child update refreshes history after an older retry read", async () => {
+  const h = await harness("subagent", {
+    MUSE_TEST_CHILD: "dedicated",
+    MUSE_TEST_CHILD_PAGE_FAIL: "once",
+  });
+  const release = path.join(h.root, "release-child-page");
+  h.launch.env.MUSE_TEST_CHILD_PAGE_RELEASE = release;
+  await h.open();
+  vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+  await h.prompt();
+  await h.wait(
+    (e) => e.type === "session.turn" && e.sessionId === "paseo-session" && e.state === "completed",
+  );
+  await vi.advanceTimersByTimeAsync(120000);
+  await h.wait(
+    (e) =>
+      e.type === "timeline.item" &&
+      e.sessionId === "paseo-session" &&
+      e.item.type === "tool_call" &&
+      e.item.name === "subagent" &&
+      e.item.status === "completed",
+  );
+  await writeFile(release, "release");
+  await h.wait(
+    (e) => e.type === "session.turn" && e.sessionId === "fixture-child" && e.state === "completed",
+  );
+  expect(
+    h.events.some(
+      (e) =>
+        e.type === "timeline.item" &&
+        e.sessionId === "fixture-child" &&
+        e.item.type === "assistant_message" &&
+        e.item.text === "SUBAGENT_PHASE0_OK",
+    ),
+  ).toBe(true);
+}, 10000);
+
+test("native process exit stops child history retry timers", async () => {
+  const h = await harness("subagent", {
+    MUSE_TEST_CHILD: "dedicated",
+    MUSE_TEST_CHILD_PAGE_FAIL: "always",
+  });
+  await h.open();
+  vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+  await h.prompt();
+  await h.wait(
+    (e) => e.type === "session.turn" && e.sessionId === "paseo-session" && e.state === "completed",
+  );
+  const host = (await h.recorded()).find((f) => f.event === "hostStarted");
+  process.kill(host.pid, "SIGKILL");
+  await h.wait((e) => e.type === "session.runtime_failed");
+  expect(vi.getTimerCount()).toBe(0);
+});