diff --git a/packages/provider-muse/src/server/adapter.test.ts b/packages/provider-muse/src/server/adapter.test.ts index 7950ef28cb93a..925f292f8dcea 100644 --- a/packages/provider-muse/src/server/adapter.test.ts +++ b/packages/provider-muse/src/server/adapter.test.ts @@ -1101,6 +1101,149 @@ 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("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/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"); + }).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..9755781e5a491 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* ( @@ -345,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) => @@ -675,12 +691,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 +721,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 +970,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" @@ -1060,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, @@ -1328,7 +1367,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); @@ -1346,7 +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); + 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; @@ -2021,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, @@ -2149,7 +2199,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, ), @@ -2233,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 || @@ -2262,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( 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),