diff --git a/plugins/muse-provider/server/children.ts b/plugins/muse-provider/server/children.ts index 0cd2b56bff..4a9f813484 100644 --- a/plugins/muse-provider/server/children.ts +++ b/plugins/muse-provider/server/children.ts @@ -22,6 +22,7 @@ interface Child { } export class Children { private readonly children = new Map(); + 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 { + // Each update reads fresh history after any older read, with serialized cursor advancement. + private async read(id: string): Promise { 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(); } } diff --git a/plugins/muse-provider/server/session.ts b/plugins/muse-provider/server/session.ts index dbf851a1c9..f6420d32de 100644 --- a/plugins/muse-provider/server/session.ts +++ b/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, diff --git a/plugins/muse-provider/test/fake-muse.mjs b/plugins/muse-provider/test/fake-muse.mjs index afdebcf98e..a71c246100 100644 --- a/plugins/muse-provider/test/fake-muse.mjs +++ b/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", diff --git a/plugins/muse-provider/test/provider.test.ts b/plugins/muse-provider/test/provider.test.ts index 8296b69957..b4fb10ea35 100644 --- a/plugins/muse-provider/test/provider.test.ts +++ b/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); +});