From 90e89cb1cd4d3b33cdd592741b506666bea26628 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:18:36 -0700 Subject: [PATCH 1/3] fix(muse): yielded commands keep the thread waiting until they finish Muse hands a long command back to the model once `yield_time_ms` passes, keeps it running, and wakes the agent when it exits. T3 marked the command done when the turn ended and dropped its real completion, so the thread looked idle and the wake read "Background activity updated". Muse tags a yielded call with the status line "terminal process running". Those calls now stay running past their turn, list as awaited command background work (holding the thread and the host's idle release), and name the wake: `Command "" finished`. Co-Authored-By: Claude Opus 5.5 --- .../provider-muse/src/server/adapter.test.ts | 71 +++++++++++++++++++ packages/provider-muse/src/server/adapter.ts | 51 ++++++++++--- .../src/server/itemPresentation.ts | 25 ++++--- packages/provider-muse/src/server/protocol.ts | 1 + 4 files changed, 129 insertions(+), 19 deletions(-) diff --git a/packages/provider-muse/src/server/adapter.test.ts b/packages/provider-muse/src/server/adapter.test.ts index 7950ef28cb93a..d53209865ca66 100644 --- a/packages/provider-muse/src/server/adapter.test.ts +++ b/packages/provider-muse/src/server/adapter.test.ts @@ -1101,6 +1101,77 @@ describe("MuseAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(testLayer)), ); + it.effect("keeps a yielded command running past its turn and names it in Muse's wake", () => + Effect.gen(function* () { + const fake = yield* makeFakeMuse(); + const offered = + yield* Deferred.make(); + const harness = yield* makeHarness( + fake, + INSTANCE_ID, + undefined, + undefined, + undefined, + runtimePolicy, + { continuationRequests: { offer: (request) => Deferred.succeed(offered, request) } }, + ); + const { nativeId } = yield* startConversation(harness, fake); + const bash = (revision: number, status: string, extra: Record = {}) => ({ + itemId: "bash-1", + kind: "toolCall", + tool: "bash", + revision, + status, + turnId: nativeId, + args: JSON.stringify({ + command: "vp test run", + description: "Run the test suite", + yield_time_ms: 300000, + }), + ...extra, + }); + const latestRoster = () => + harness.allEvents.findLast((event) => event.type === "provider_thread.updated") + ?.providerThread.pendingBackgroundTasks; + yield* fake.emit("item/started", { turnId: nativeId, item: bash(1, "inProgress") }); + yield* fake.emit("item/updated", { + turnId: nativeId, + item: bash(2, "inProgress", { statusLine: "terminal process running" }), + }); + yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "completed" }); + assert.strictEqual((yield* harness.takeEvent("turn.terminal")).status, "completed"); + const bashStatuses = harness.allEvents.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === "bash-1" + ? [event.turnItem.status] + : [], + ); + assert.strictEqual(bashStatuses.at(-1), "running"); + assert.deepStrictEqual(latestRoster(), [ + { taskId: "bash-1", kind: "command", awaited: true, description: "Run the test suite" }, + ]); + assert.isTrue(yield* harness.runtime.hasPendingBackgroundWork!); + + yield* fake.emit("item/completed", { + turnId: nativeId, + item: bash(3, "completed", { visibleOutput: "1 passed\n" }), + }); + yield* fake.emit("turn/started", { turnId: "wake-1" }); + const wake = yield* Deferred.await(offered); + assert.deepStrictEqual(wake.notification, { + source: { kind: "command" }, + outcome: "completed", + summary: 'Command "Run the test suite" finished', + }); + yield* harness.takeEvent( + "turn_item.updated", + (event) => + event.turnItem.nativeItemRef?.nativeId === "bash-1" && + event.turnItem.status === "completed", + ); + assert.deepStrictEqual(latestRoster(), []); + }).pipe(Effect.scoped, Effect.provide(testLayer)), + ); + it.effect.each([ { when: "after its turn ended", turnEnded: true }, { when: "while its turn runs", turnEnded: false }, diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index 40785df6bada1..6834ad01c8fd1 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -89,7 +89,7 @@ import { subagentThreadTitle, } from "@t3tools/provider-core/server/subagentProjection"; import { turnScopedSelectionTransition } from "@t3tools/provider-core/server/selectionTransition"; -import { museItemStatus, museToolPresentation } from "./itemPresentation.ts"; +import { museCommandLabel, museItemStatus, museToolPresentation } from "./itemPresentation.ts"; const MUSE_PROVIDER = ProviderDriverKind.make("muse"); const isOpenSessionError = Schema.is(ProviderAdapter.ProviderAdapterOpenSessionError); @@ -256,6 +256,20 @@ const INFORMATIONAL_NOTIFICATIONS = new Set([ "usage/changed", ]); const TOOLS_WITH_NATIVE_ROWS = new Set(["write_todos", "request_user_input", "workflow"]); +// Muse returns a command to the model once `yield_time_ms` passes, keeps it running, and wakes +// the agent when it exits. It marks the call with this status line. +const isYieldedCommand = (item: MuseItem) => + item.kind === "toolCall" && item.statusLine === "terminal process running"; +const backgroundKind = (item: MuseItem | undefined) => + item?.kind === "subagent" + ? ("subagent" as const) + : item?.kind === "toolCall" + ? ("command" as const) + : ("background_task" as const); +const backgroundLabel = (item: MuseItem | undefined) => + item?.kind === "toolCall" + ? museCommandLabel(item) + : (item?.objective ?? item?.entryId ?? item?.fallbackText); const responseAnswerSchema = Schema.Union([Schema.String, Schema.Array(Schema.String)]); export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( @@ -675,12 +689,14 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( yield* updateThread({ pendingBackgroundTasks: [...observedChildren].flatMap(([itemId, owner]) => { const item = owner.items.get(itemId); - const description = item?.objective ?? item?.entryId ?? item?.fallbackText; + const kind = backgroundKind(item); + const description = backgroundLabel(item); return [ { taskId: itemId, - kind: - item?.kind === "subagent" ? ("subagent" as const) : ("background_task" as const), + kind, + // Muse always wakes the agent when a yielded command exits. + ...(kind === "command" ? { awaited: true as const } : {}), ...(description ? { description } : {}), }, ]; @@ -703,13 +719,19 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( return; const time = yield* DateTime.now; const status = terminal ?? itemStatus(item); - if (item.kind === "subagent" || item.kind === "workflow") { + if ( + item.kind === "subagent" || + item.kind === "workflow" || + isYieldedCommand(item) || + observedChildren.has(item.itemId) + ) { const wasRunning = observedChildren.has(item.itemId); if (status === "running") observedChildren.set(item.itemId, turn); - else if (observedChildren.delete(item.itemId) && active !== turn) + // Work that ends during another run reaches that run, not a later wake. + else if (observedChildren.delete(item.itemId) && active === undefined) finishedBackground.push({ - kind: item.kind === "subagent" ? "subagent" : "background_task", - label: item.objective ?? item.entryId ?? item.fallbackText, + kind: backgroundKind(item), + label: backgroundLabel(item), outcome: status === "completed" ? "completed" @@ -946,7 +968,9 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( const completedAt = yield* DateTime.now; for (const item of turn.items.values()) { if ( - (item.kind === "subagent" || item.kind === "workflow") && + (item.kind === "subagent" || + item.kind === "workflow" || + observedChildren.has(item.itemId)) && item.status === "inProgress" && status === "completed" && disposition === "reusable" @@ -1328,7 +1352,12 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( : undefined; if (itemEvent) { const owner = observedChildren.get(itemEvent.itemId); - if (owner && owner !== active && itemEvent.turnId === owner.nativeId) { + if ( + owner && + owner !== active && + typeof itemEvent.turnId === "string" && + owns(owner, itemEvent.turnId) + ) { const previous = owner.items.get(itemEvent.itemId); if (!previous || itemEvent.revision > previous.revision) { owner.items.set(itemEvent.itemId, itemEvent); @@ -2149,7 +2178,7 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( providerSession: session, events: Stream.fromQueue(events), backgroundTasksEndWithProcess: true, - // A running workflow or a held Muse turn keeps the host: idle release must wait. + // Running background work or a held Muse turn keeps the host: idle release must wait. hasPendingBackgroundWork: Effect.sync( () => observedChildren.size > 0 || wake !== undefined, ), diff --git a/packages/provider-muse/src/server/itemPresentation.ts b/packages/provider-muse/src/server/itemPresentation.ts index d74bc7143ea76..439b6ae3ca6c0 100644 --- a/packages/provider-muse/src/server/itemPresentation.ts +++ b/packages/provider-muse/src/server/itemPresentation.ts @@ -46,18 +46,27 @@ export function museItemStatus(item: MuseItem): OrchestrationV2TurnItemStatus { return "failed"; } +function toolArgs(item: MuseItem) { + if (!item.args) return undefined; + try { + return decodeToolArgs(item.args); + } catch { + // Streaming model-authored arguments may not be valid JSON yet. + return undefined; + } +} + +export function museCommandLabel(item: MuseItem) { + const { description, command } = toolArgs(item) ?? {}; + if (typeof description === "string" && description.trim()) return description; + return item.commandText ?? (typeof command === "string" ? command : undefined); +} + export function museToolPresentation( item: MuseItem, status = museItemStatus(item), ): MuseToolPresentation { - let args: Record | undefined; - if (item.args) { - try { - args = decodeToolArgs(item.args); - } catch { - // Streaming model-authored arguments may not be valid JSON yet. - } - } + const args = toolArgs(item); const stringArg = (...keys: string[]) => { for (const key of keys) { const value = args?.[key]; diff --git a/packages/provider-muse/src/server/protocol.ts b/packages/provider-muse/src/server/protocol.ts index 9c3e7c40d957b..ed649b292a1ef 100644 --- a/packages/provider-muse/src/server/protocol.ts +++ b/packages/provider-muse/src/server/protocol.ts @@ -44,6 +44,7 @@ export const MuseItem = Schema.Struct({ failureReason: Schema.optional(Schema.String), fallbackText: Schema.optional(Schema.String), commandText: Schema.optional(Schema.String), + statusLine: Schema.optional(Schema.String), exitCode: Schema.optional(Schema.Int), durationMs: Schema.optional(Schema.Finite), subagentId: Schema.optional(Schema.String), From 5c159dfec6beb41cc40959d84f4c12cd1401f2e4 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:41:37 -0700 Subject: [PATCH 2/3] fix(muse): Stop ends Muse background work without waking the agent Stop on a thread waiting on a yielded command reached the adapter for a settled turn and failed with "This Muse turn is no longer active", so the command kept running and Stop retried until it gave up. User Stop now sends `task/stopAll`, on an idle thread and after interrupting a running turn, and settles tracked background work as interrupted. Muse starts a turn to report work it stopped; until the next run those turns are interrupted, so Stop does not restart the agent. Co-Authored-By: Claude Opus 5.5 --- .../provider-muse/src/server/adapter.test.ts | 70 +++++++++++++++++++ packages/provider-muse/src/server/adapter.ts | 33 +++++++++ 2 files changed, 103 insertions(+) diff --git a/packages/provider-muse/src/server/adapter.test.ts b/packages/provider-muse/src/server/adapter.test.ts index d53209865ca66..9856043d2a35f 100644 --- a/packages/provider-muse/src/server/adapter.test.ts +++ b/packages/provider-muse/src/server/adapter.test.ts @@ -1172,6 +1172,76 @@ describe("MuseAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(testLayer)), ); + it.effect("stops Muse's background work on an idle thread without waking the agent", () => + Effect.gen(function* () { + const fake = yield* makeFakeMuse(); + const offers: Array = []; + const harness = yield* makeHarness( + fake, + INSTANCE_ID, + undefined, + undefined, + undefined, + runtimePolicy, + { continuationRequests: { offer: (request) => Effect.sync(() => offers.push(request)) } }, + ); + const { nativeId, providerTurn } = yield* startConversation(harness, fake); + yield* fake.emit("item/updated", { + turnId: nativeId, + item: { + itemId: "bash-1", + kind: "toolCall", + tool: "bash", + revision: 2, + status: "inProgress", + turnId: nativeId, + statusLine: "terminal process running", + }, + }); + yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "completed" }); + yield* harness.takeEvent("turn.terminal"); + + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: providerTurn.id, + requestRuntimeRestart: true, + }); + yield* fake.takeCall("task/stopAll"); + yield* harness.takeEvent( + "turn_item.updated", + (event) => + event.turnItem.nativeItemRef?.nativeId === "bash-1" && + event.turnItem.status === "interrupted", + ); + assert.isFalse(yield* harness.runtime.hasPendingBackgroundWork!); + + // Muse wakes the agent to report the stopped command. + yield* fake.emit("turn/started", { turnId: "stop-report" }); + assert.strictEqual((yield* fake.takeCall("turn/interrupt")).params.turnId, "stop-report"); + assert.lengthOf(offers, 0); + }).pipe(Effect.scoped, Effect.provide(testLayer)), + ); + + it.effect("stops yielded commands when Stop ends a running turn", () => + Effect.gen(function* () { + const fake = yield* makeFakeMuse(); + const harness = yield* makeHarness(fake); + const { nativeId, providerTurn } = yield* startConversation(harness, fake); + const stopping = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: providerTurn.id, + requestRuntimeRestart: true, + }) + .pipe(Effect.forkScoped); + assert.strictEqual((yield* fake.takeCall("turn/interrupt")).params.turnId, nativeId); + yield* fake.takeCall("task/stopAll"); + yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "cancelled" }); + yield* Fiber.join(stopping); + assert.strictEqual((yield* harness.takeEvent("turn.terminal")).status, "interrupted"); + }).pipe(Effect.scoped, Effect.provide(testLayer)), + ); + it.effect.each([ { when: "after its turn ended", turnEnded: true }, { when: "while its turn runs", turnEnded: false }, diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index 6834ad01c8fd1..9c48fc7b43718 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -359,6 +359,8 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( let wake: | { readonly nativeId: string; readonly events: Array<[string, unknown]> } | undefined; + // Muse wakes the agent to report work a user Stop ended; until the next run, end those wakes. + let muteWakes = false; const emit = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); const decode = (schema: Schema.Codec, data: unknown) => @@ -1084,6 +1086,19 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( }); const owns = (turn: ActiveTurn, turnId: string) => turnId === turn.nativeId || turn.joined.has(turnId); + const stopBackgroundWork = Effect.fnUntraced(function* () { + muteWakes = true; + if (wake) yield* request("turn/interrupt", { turnId: wake.nativeId }).pipe(Effect.ignore); + yield* request("task/stopAll", {}); + }); + const settleStoppedWork = eventPermit.withPermits(1)( + Effect.gen(function* () { + wake = undefined; + yield* settleObservedChildren("interrupted"); + // Reports from stopped work must not name a later wake. + finishedBackground.length = 0; + }), + ); const decideApproval = Effect.fnUntraced(function* ( entry: PendingRequest, decision: ProviderApprovalDecision | undefined, @@ -1376,6 +1391,11 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( params.turnId !== wake?.nativeId ) { if (active && !active.compact) active.joined.add(params.turnId); + else if (!active && !wake && muteWakes) + yield* request("turn/interrupt", { turnId: params.turnId }).pipe( + Effect.ignore, + Effect.forkIn(scope), + ); else if (!active && !wake) { const held = { nativeId: params.turnId, events: [] }; wake = held; @@ -2050,6 +2070,7 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( yield* Effect.gen(function* () { thread = turnInput.providerThread; active = turn; + muteWakes = false; yield* emit({ type: "provider_turn.updated", driver: MUSE_PROVIDER, @@ -2262,6 +2283,15 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( interruptTurn: Effect.fn("MuseAdapter.interruptTurn")( function* (args) { const turn = active; + // Stop also reaches settled turns, for the background work Muse still runs. + if ( + !turn && + args.requestRuntimeRestart === true && + validateThread(args.providerThread) + ) { + yield* stopBackgroundWork(); + return yield* settleStoppedWork; + } if ( !turn || turn.providerTurn.id !== args.providerTurnId || @@ -2291,11 +2321,14 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( for (const joined of turn.joined) yield* request("turn/interrupt", { turnId: joined }).pipe(Effect.ignore); yield* request("turn/interrupt", { turnId: turn.nativeId }); + // User Stop also ends commands Muse yielded to the background. + if (args.requestRuntimeRestart === true) yield* stopBackgroundWork(); yield* Deferred.await(turn.done).pipe( Effect.timeout(options.requestTimeoutMs ?? 30_000), Effect.mapError((cause) => protocolError("Muse did not confirm interruption", cause)), Effect.onError((cause) => eventPermit.withPermits(1)(failHost(cause))), ); + if (args.requestRuntimeRestart === true) yield* settleStoppedWork; }, (effect, args) => effect.pipe( From 42a9245706834d6907e1112bb90572aef93706d1 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:53:53 -0700 Subject: [PATCH 3/3] fix(muse): end stop-report turns that start while Stop finishes a run A report turn Muse started after `task/stopAll`, while the stopped run was still settling, joined that run and kept going after it ended. Muted wakes are now interrupted before any joining. Co-Authored-By: Claude Opus 5.5 --- packages/provider-muse/src/server/adapter.test.ts | 2 ++ packages/provider-muse/src/server/adapter.ts | 4 ++-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/packages/provider-muse/src/server/adapter.test.ts b/packages/provider-muse/src/server/adapter.test.ts index 9856043d2a35f..925f292f8dcea 100644 --- a/packages/provider-muse/src/server/adapter.test.ts +++ b/packages/provider-muse/src/server/adapter.test.ts @@ -1236,6 +1236,8 @@ describe("MuseAdapterV2", () => { .pipe(Effect.forkScoped); assert.strictEqual((yield* fake.takeCall("turn/interrupt")).params.turnId, nativeId); yield* fake.takeCall("task/stopAll"); + yield* fake.emit("turn/started", { turnId: "stop-report" }); + assert.strictEqual((yield* fake.takeCall("turn/interrupt")).params.turnId, "stop-report"); yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "cancelled" }); yield* Fiber.join(stopping); assert.strictEqual((yield* harness.takeEvent("turn.terminal")).status, "interrupted"); diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index 9c48fc7b43718..9755781e5a491 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -1390,12 +1390,12 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( params.turnId !== active?.nativeId && params.turnId !== wake?.nativeId ) { - if (active && !active.compact) active.joined.add(params.turnId); - else if (!active && !wake && muteWakes) + if (muteWakes) yield* request("turn/interrupt", { turnId: params.turnId }).pipe( Effect.ignore, Effect.forkIn(scope), ); + else if (active && !active.compact) active.joined.add(params.turnId); else if (!active && !wake) { const held = { nativeId: params.turnId, events: [] }; wake = held;