diff --git a/plugins/muse-provider/server/children.ts b/plugins/muse-provider/server/children.ts index 69e0f4a855..73ba117143 100644 --- a/plugins/muse-provider/server/children.ts +++ b/plugins/muse-provider/server/children.ts @@ -1,6 +1,6 @@ import type { ProviderEvent } from "@getpaseo/plugin/server/provider"; import { MspConnection, type Notification } from "./connection.js"; -import { Timeline } from "./timeline.js"; +import { Timeline, subagentOutcome } from "./timeline.js"; import { childSessionSchema, pageSchema, @@ -13,6 +13,7 @@ import { interface Child { timeline: Timeline; cursor: string | undefined; + ended: boolean; } export class Children { private readonly children = new Map(); @@ -54,9 +55,29 @@ export class Children { this.children.set(id, { timeline: new Timeline(this.host, id, id, this.emit), cursor: undefined, + ended: false, }); } await this.page(id); + this.end(id, item); + } + // Muse sends no turn events for a child, so the parent's item is the only record that it finished. + // A subagent_* tool call completes when the spawn is accepted, so only subagent and workflow items count. + private end(id: string, item: WireItem): void { + const child = this.children.get(id)!; + const outcome = + item.kind === "subagent" || item.kind === "workflow" ? subagentOutcome(item) : undefined; + if (!outcome || child.ended) return; + child.ended = true; + this.emit({ + type: "session.turn", + sessionId: id, + turnId: item.itemId, + state: outcome, + ...(outcome === "failed" + ? { error: { message: item.result?.errorKind ?? "Muse subagent failed" } } + : {}), + }); } async notify(id: string, event: Notification): Promise { const child = this.children.get(id); diff --git a/plugins/muse-provider/server/session.ts b/plugins/muse-provider/server/session.ts index 31066c5b04..dbf851a1c9 100644 --- a/plugins/muse-provider/server/session.ts +++ b/plugins/muse-provider/server/session.ts @@ -164,16 +164,18 @@ export class Session { await live.recovery.backfill(); return live.children.refresh(); }) - .catch((error) => this.runtimeFailed(error)); + .catch((error: unknown) => this.repairFailed(error)); }, ), }; this.providerId = response.session.providerId ?? undefined; + const configured = { model: this.config.model, mode: this.config.mode }; this.config = { ...this.config, model: response.session.modelId ?? this.config.model, mode: response.session.approvalMode?.mode ?? this.config.mode ?? "onRequest", }; + if (input.persistence) await this.restoreControls(nativeId, configured); this.options.emit({ type: "session.opened", requestId: input.requestId, @@ -205,6 +207,26 @@ export class Session { }); } + // session/resume takes no model or approval mode, and Muse can reload a session on its defaults. + private async restoreControls( + sessionId: string, + configured: Pick, + ): Promise { + if (configured.model && configured.model !== this.config.model) { + await this.host.command( + "session/setModel", + { sessionId, model: { providerId: this.providerId, modelId: configured.model } }, + ackSchema, + ); + this.config = { ...this.config, model: configured.model }; + } + if (configured.mode && configured.mode !== this.config.mode) { + const mode = approvalModeSchema.parse(configured.mode); + await this.host.command("session/setApprovalMode", { sessionId, mode }, ackSchema); + this.config = { ...this.config, mode }; + } + } + private live(): Live { if (!this.liveState) throw new MuseError("sessionNotOpen", "Muse session is not open"); return this.liveState; @@ -393,6 +415,14 @@ export class Session { .then(() => this.notify(notification)) .catch((error: unknown) => this.runtimeFailed(error)); } + // The silence repair is best effort: the live view still delivers, so a failed page waits for the + // next silence instead of failing the session. Host death reaches runtimeFailed through onExit. + private repairFailed(error: unknown): void { + process.stderr.write( + `Muse silence repair failed: ${error instanceof Error ? error.message : String(error)}\n`, + ); + this.live().recovery.activity(undefined, this.activeTurnIds().length > 0); + } private runtimeFailed(error: unknown): void { this.live().recovery.close(); this.options.emit({ @@ -410,7 +440,9 @@ export class Session { return; } await this.processNotification(notification); - this.live().recovery.activity(envelope.viewCursor, this.activeTurnIds().length > 0); + // A projected subagent event carries the child's view cursor, which this session's view rejects. + const cursor = envelope.subagentOrigin === undefined ? envelope.viewCursor : undefined; + this.live().recovery.activity(cursor, this.activeTurnIds().length > 0); } private async processNotification(notification: Notification): Promise { switch (notification.method) { diff --git a/plugins/muse-provider/server/timeline.ts b/plugins/muse-provider/server/timeline.ts index df786df125..6b1c1cb8ae 100644 --- a/plugins/muse-provider/server/timeline.ts +++ b/plugins/muse-provider/server/timeline.ts @@ -144,6 +144,16 @@ function toolCallItem(item: WireItem, patch?: string): ProviderTimelineItem { } : toolDetail(item, patch), } as const; + if (item.kind === "subagent" || item.kind === "workflow") { + const outcome = subagentOutcome(item); + if (outcome === "failed") + return { + ...base, + status: outcome, + error: item.result?.errorKind ?? item.failureReason ?? item.status, + }; + return { ...base, status: outcome ?? "running", error: null }; + } if (item.status === "inProgress") return { ...base, status: "running", error: null }; if (item.status === "cancelled") return { ...base, status: "canceled", error: null }; if (item.status !== "completed") @@ -151,6 +161,18 @@ function toolCallItem(item: WireItem, patch?: string): ProviderTimelineItem { return { ...base, status: "completed", error: null }; } +// Muse 1.4.7 leaves a finished subagent's item `inProgress`; its control status says the child is done. +export function subagentOutcome(item: WireItem): "completed" | "failed" | "canceled" | undefined { + if (item.status === "cancelled") return "canceled"; + if (item.status !== "inProgress" && item.status !== "completed") return "failed"; + const done = + item.status === "completed" || + item.controlStatus === "resultReady" || + item.controlStatus === "closed"; + if (!done) return undefined; + return item.result?.errorKind ? "failed" : "completed"; +} + export function appendDelta( previous: WireItem, delta: z.infer, diff --git a/plugins/muse-provider/server/wire.ts b/plugins/muse-provider/server/wire.ts index 45aac6fe4f..9bbe64f038 100644 --- a/plugins/muse-provider/server/wire.ts +++ b/plugins/muse-provider/server/wire.ts @@ -28,6 +28,8 @@ export const itemSchema = z.object({ entryId: z.string().optional(), kind: z.string(), status: z.string(), + controlStatus: z.string().optional(), + result: z.object({ errorKind: z.string().optional() }).nullish(), turnId: z.string().nullable().optional(), commandId: z.string().optional(), recordedAt: z.string().optional(), @@ -84,6 +86,8 @@ export type WireApproval = z.infer; export const notificationSchema = z.object({ sessionId: z.string().optional(), viewCursor: z.string().optional(), + // Marks a subagent's event projected onto the parent's plane; its viewCursor is the child's. + subagentOrigin: z.unknown().optional(), }); export const frameSchema = z.object({ jsonrpc: z.literal("2.0"), diff --git a/plugins/muse-provider/test/fake-muse.mjs b/plugins/muse-provider/test/fake-muse.mjs index 16cc821b79..e4f4706a10 100644 --- a/plugins/muse-provider/test/fake-muse.mjs +++ b/plugins/muse-provider/test/fake-muse.mjs @@ -30,6 +30,7 @@ const catalogRows = readFixture("catalog-controls"); const replacements = new Map(); const counts = new Map(); const receipts = new Set(); +let pageFailed = false; const lines = createInterface({ input: process.stdin }); lines.on("line", receive); function receive(line) { @@ -78,6 +79,9 @@ function receiveFrame(line) { for (const row of terminalFirst) emitFixtureMessage(row.msg, recorded, frame); setTimeout(() => emitFixtureMessage(response.msg, recorded, frame), 50); } else for (const row of messages) emitFixtureMessage(row.msg, recorded, frame); + // The parent's last live notifications before it waits on the subagent. + if (process.env.MUSE_TEST_CHILD_APPROVAL && frame.method === "turn/start") + projectChildApproval(frame.params.sessionId); } function controlResponse(frame) { if (frame.method === "session/resume" && process.env.MUSE_TEST_HISTORY) { @@ -110,10 +114,7 @@ function controlResponse(frame) { return true; } if (process.env.MUSE_TEST_HANG === frame.method) return true; - if (frame.method === "session/setReasoningEffort") { - respond(frame, { status: "accepted", commandId: frame.params.commandId }); - return true; - } + if (scriptedResponse(frame)) return true; if (frame.method === "account/read") { respond(frame, { credentialRequired: true, state: process.env.MUSE_TEST_ACCOUNT || "envKey" }); return true; @@ -131,6 +132,45 @@ function controlResponse(frame) { } return false; } +// Commands accepted without a recorded exchange, and the view/page failures some tests script. +function scriptedResponse(frame) { + if ( + frame.method === "session/setReasoningEffort" || + (process.env.MUSE_TEST_CHILD_APPROVAL && frame.method === "approval/decide") || + (process.env.MUSE_TEST_RESUMED_MODEL && + ["session/setModel", "session/setApprovalMode"].includes(frame.method)) + ) { + respond(frame, { status: "accepted", commandId: frame.params.commandId }); + return true; + } + return frame.method === "view/page" && rejectedPage(frame); +} +function rejectedPage(frame) { + const anchor = frame.params.cursor; + // Real Muse rejects an anchor from another session's view the same way (live probe, Muse 1.4.7). + if ( + process.env.MUSE_TEST_CHILD_APPROVAL && + anchor && + anchor.split(":")[1] !== frame.params.sessionId + ) { + send({ + jsonrpc: "2.0", + id: frame.id, + error: { + code: -32004, + message: "unknown cursor anchor", + data: { kind: "notFound", reason: "missingAnchor", anchor }, + }, + }); + return true; + } + if (process.env.MUSE_TEST_PAGE_FAIL_ONCE && !pageFailed) { + pageFailed = true; + rpcError(frame, -32603, "internal", "view range read failed"); + return true; + } + return false; +} function parityResponse(frame) { if (workflowResponse(frame)) return true; if (frame.method === "view/page") { @@ -341,6 +381,15 @@ function applyWireVariants(message) { message.result.session.modelId = null; message.result.session.providerId = null; } + // Muse 1.4.7 can resume a session on its default model and approval mode. + if (message.result?.session && message.result.history && process.env.MUSE_TEST_RESUMED_MODEL) { + message.result.session.modelId = process.env.MUSE_TEST_RESUMED_MODEL; + message.result.session.approvalMode = { + lastCommandId: null, + mode: "onRequest", + source: "replay", + }; + } } function parityDelivery(message) { if (process.env.MUSE_TEST_WORKFLOW && message.method === "turn/completed") @@ -379,6 +428,42 @@ function parityDelivery(message) { }); return false; } +// A subagent's approval projected onto the parent's plane, shaped like Muse 1.4.7's: the parent's +// sessionId, but a viewCursor and sourceRange from the child's stream. +function projectChildApproval(sessionId) { + const child = "fixture-child"; + const subagentOrigin = { subagentId: "subagent-1", childSessionId: child, childRunId: child }; + const sourceRange = { + stream: { kind: "session", id: child }, + first: { id: "fixture-child-record", sequence: 2 }, + last: { id: "fixture-child-record", sequence: 2 }, + }; + const approvalId = "fixture-child-approval"; + send({ + jsonrpc: "2.0", + method: "approval/requested", + params: { + sessionId, + approvalId, + itemId: "fixture-child-tool", + toolName: "bash", + subject: { kind: "shell", command: "FOO=1 ls ~" }, + currentRequirementId: { approvalId, sourceIndex: 0 }, + availableChoices: [ + { choiceId: "approve-once", decision: "approved", scope: "once", label: "Approve once" }, + { choiceId: "deny-once", decision: "denied", scope: "once", label: "Deny" }, + ], + sourceRange, + subagentOrigin, + viewCursor: `v:${child}:2`, + }, + }); + send({ + jsonrpc: "2.0", + method: "approval/resolved", + params: { sessionId, approvalId, sourceRange, subagentOrigin, viewCursor: `v:${child}:3` }, + }); +} function applyTestVariants(message) { applyParityVariants(message); if (message.method === "turn/completed" && process.env.MUSE_TEST_INTERRUPT_TERMINAL) @@ -413,6 +498,14 @@ function applyParityVariants(message) { message.params.item.role = "worker"; message.params.item.objective = "Child objective"; } + // Muse 1.4.7 leaves a finished subagent `inProgress` and reports the end in controlStatus. + if (process.env.MUSE_TEST_CHILD_OUTCOME && message.method === "item/completed") { + message.params.item.status = "inProgress"; + message.params.item.controlStatus = "resultReady"; + message.params.item.result = { summary: "Child done", evidenceRefs: [], artifactRefs: [] }; + if (process.env.MUSE_TEST_CHILD_OUTCOME === "failed") + message.params.item.result.errorKind = "childFailed"; + } } if (message.params?.availableChoices && process.env.MUSE_TEST_DENIED) message.params.availableChoices.push({ diff --git a/plugins/muse-provider/test/provider.test.ts b/plugins/muse-provider/test/provider.test.ts index 4868634d70..dc1edceba1 100644 --- a/plugins/muse-provider/test/provider.test.ts +++ b/plugins/muse-provider/test/provider.test.ts @@ -406,6 +406,39 @@ for (const cursor of [undefined, "v:old:7"]) { ).not.toHaveProperty("cursor"); }); } +for (const source of ["agent config", "saved session"] as const) { + test(`resume re-applies the ${source} model and approval mode when Muse reports its defaults`, async () => { + const h = await harness("resume-without-cursor", { + MUSE_TEST_RESUMED_MODEL: "meta/muse-default", + }); + await h.open( + { version: 1, data: { sessionId: "saved-session", model: "meta/muse-spark-1.3" } }, + "allowAll", + source === "agent config" ? "meta/muse-spark-1.3" : null, + ); + const frames = await h.recorded(); + expect(frames.find((f) => f.method === "session/setModel")?.params.model).toMatchObject({ + modelId: "meta/muse-spark-1.3", + }); + expect(frames.find((f) => f.method === "session/setApprovalMode")?.params.mode).toBe( + "allowAll", + ); + expect(h.events.findLast((e) => e.type === "session.config")).toMatchObject({ + config: { model: "meta/muse-spark-1.3", mode: "allowAll" }, + }); + expect(h.events.findLast((e) => e.type === "session.persistence")).toMatchObject({ + persistence: { data: { model: "meta/muse-spark-1.3" } }, + }); + }); +} +test("resume sends no model or approval change when Muse already matches", async () => { + const h = await harness("resume-without-cursor"); + await h.open({ version: 1, data: { sessionId: "saved-session" } }); + const frames = await h.recorded(); + expect( + frames.filter((f) => ["session/setModel", "session/setApprovalMode"].includes(f.method)), + ).toEqual([]); +}); test("steer sends ifBusy steer and joins the running turn", async () => { const h = await harness("steer"); await h.open(); @@ -1173,10 +1206,17 @@ for (const variant of ["tool", "dedicated", "workflow"]) { e.item.type === "assistant_message", ), ).toMatchObject({ item: { text: "SUBAGENT_PHASE0_OK" } }); - await h.wait((e) => e.type === "session.turn" && e.state === "completed"); + await h.wait( + (e) => + e.type === "session.turn" && e.sessionId === "paseo-session" && e.state === "completed", + ); expect( h.events.filter((e) => e.type === "session.opened" && e.sessionId === "fixture-child"), ).toHaveLength(1); + // A completed subagent_spawn tool call only means the spawn was accepted. + expect( + h.events.filter((e) => e.type === "session.turn" && e.sessionId === "fixture-child"), + ).toHaveLength(variant === "tool" ? 0 : 1); expect( (await h.recorded()) .filter((f) => f.method === "session/read") @@ -1192,6 +1232,38 @@ for (const variant of ["tool", "dedicated", "workflow"]) { ).toHaveLength(1); }); } +for (const outcome of ["ready", "failed"] as const) { + test(`a subagent Muse leaves in progress with a ${outcome} result ends its child and its card`, async () => { + const h = await harness("subagent", { + MUSE_TEST_CHILD: "dedicated", + MUSE_TEST_CHILD_OUTCOME: outcome, + }); + await h.open(); + await h.prompt(); + const state = outcome === "ready" ? "completed" : "failed"; + expect( + await h.wait((e) => e.type === "session.turn" && e.sessionId === "fixture-child"), + ).toMatchObject({ + state, + ...(outcome === "failed" ? { error: { message: "childFailed" } } : {}), + }); + await h.wait( + (e) => + e.type === "session.turn" && e.sessionId === "paseo-session" && e.state === "completed", + ); + const cards = h.events.filter( + (e) => + e.type === "timeline.item" && + e.sessionId === "paseo-session" && + e.item.type === "tool_call" && + e.item.name === "subagent", + ); + expect(cards.at(-1)).toMatchObject({ item: { status: state } }); + expect( + h.events.filter((e) => e.type === "session.turn" && e.sessionId === "fixture-child"), + ).toHaveLength(1); + }); +} test("gap backfill repairs a missing completion without duplicating item revisions", async () => { const h = await harness("text-reasoning", { MUSE_TEST_GAP: "1" }); await h.open(); @@ -1220,6 +1292,55 @@ test("silent active turns page after two minutes and stop polling on recovered t await vi.advanceTimersByTimeAsync(240000); expect((await h.recorded()).filter((f) => f.method === "view/page")).toHaveLength(1); }); +test("a subagent approval does not move the parent's silence repair onto the child's cursor", async () => { + const h = await harness("text-reasoning", { + MUSE_TEST_SILENT: "1", + MUSE_TEST_CHILD_APPROVAL: "1", + }); + await h.open(undefined, "allowAll"); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + await h.prompt(); + await h.wait((e) => e.type === "session.permission_resolved"); + await vi.advanceTimersByTimeAsync(120000); + 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([]); + const frames = await h.recorded(); + expect(frames.find((f) => f.method === "approval/decide").params.approvalId).toBe( + "fixture-child-approval", + ); + const pages = frames.filter((f) => f.method === "view/page"); + expect(pages).toHaveLength(1); + expect(pages[0].params.cursor.split(":")[1]).toBe(pages[0].params.sessionId); +}); +test("a failed silence repair keeps the session and retries at the next silence", async () => { + const h = await harness("text-reasoning", { + MUSE_TEST_SILENT: "1", + MUSE_TEST_PAGE_FAIL_ONCE: "1", + }); + await h.open(); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + await h.prompt(); + await h.wait((e) => e.type === "session.prompt_result"); + await vi.advanceTimersByTimeAsync(120000); + const deadline = Date.now() + 5000; + while ((await h.recorded()).filter((f) => f.method === "view/page").length === 0) { + if (Date.now() > deadline) throw new Error("The first repair never paged"); + await delay(10); + } + await delay(50); + expect(h.events.filter((e) => e.type === "session.runtime_failed")).toEqual([]); + expect(h.events.filter((e) => e.type === "session.turn" && e.state === "completed")).toEqual([]); + await vi.advanceTimersByTimeAsync(120000); + await h.wait((e) => e.type === "session.turn" && e.state === "completed"); + expect((await h.recorded()).filter((f) => f.method === "view/page")).toHaveLength(2); + expect(h.events.filter((e) => e.type === "session.runtime_failed")).toEqual([]); + expect(h.events.filter((e) => e.type === "session.turn" && e.state === "completed")).toHaveLength( + 1, + ); +}); test("session readiness accepts initialization beyond the old event polling window", async () => { const h = await harness("text-reasoning", { MUSE_TEST_INITIALIZE_DELAY_MS: "3500" }); expect(await h.open()).toMatchObject({ type: "session.ready" });