1import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress";
2import { assert, describe, it } from "@effect/vitest";
3import * as NodeServices from "@effect/platform-node/NodeServices";
4import { MspError } from "@muse-code/sdk";
5import {
6 MUSE_DEFAULT_MODEL,
7 EnvironmentId,
8 MessageId,
9 NodeId,
10 ProjectId,
11 ProviderInstanceId,
12 ProviderDriverKind,
13 ProviderSessionId,
14 ProviderThreadId,
15 ProviderTurnId,
16 RunAttemptId,
17 RunId,
18 ThreadId,
19 type OrchestrationV2AppThread,
20 type OrchestrationV2ProviderThread,
21} from "@t3tools/contracts";
22import * as HostProcess from "@t3tools/shared/HostProcess";
23import { MuseSettings } from "../settings.ts";
24import * as DateTime from "effect/DateTime";
25import * as Deferred from "effect/Deferred";
26import * as Effect from "effect/Effect";
27import * as Option from "effect/Option";
28import type * as Exit from "effect/Exit";
29import * as Fiber from "effect/Fiber";
30import * as Layer from "effect/Layer";
31import * as Queue from "effect/Queue";
32import * as Schema from "effect/Schema";
33import * as Stream from "effect/Stream";
34
35import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost";
36import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
37import type { MuseItem } from "./protocol.ts";
38import type { MuseSdkHost } from "./sdk.ts";
39import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
40import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
41import type * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
42import { makeMuseAdapterV2, type MuseAdapterV2Options } from "./adapter.ts";
43
44const testLayer = Layer.mergeAll(
45 NodeServices.layer,
46 IdAllocator.layer,
47 McpProviderSessions.layer,
48 ScriptProgress.layerUnavailable,
49 TestProviderHost.layer().pipe(Layer.provide(NodeServices.layer)),
50);
51const MUSE_PROVIDER = ProviderDriverKind.make("muse");
52const INSTANCE_ID = ProviderInstanceId.make("muse_work");
53const THREAD_ID = ThreadId.make("thread-muse-test");
54const MODEL = "muse-spark-1.3-contributor";
55const museSettings = Schema.decodeSync(MuseSettings)({ enabled: true });
56const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({
57 runtimeMode: "full-access",
58 interactionMode: "default",
59 cwd: null,
60});
61const modelSelection = (instanceId = INSTANCE_ID) => ({
62 instanceId,
63 model: MODEL,
64 options: [{ id: "reasoningEffort", value: "max" }],
65});
66
67interface NativeCall {
68 readonly method: string;
69 readonly params: Record<string, unknown>;
70 readonly commandId?: string;
71}
72
73function pendingPromise<A>() {
74 let resolve!: (value: A) => void;
75 let reject!: (reason: unknown) => void;
76 const promise = new Promise<A>((done, fail) => {
77 resolve = done;
78 reject = fail;
79 });
80 return { promise, resolve, reject };
81}
82
83/** SDK-shaped transport fake: registration deliberately returns void, as the SDK does. */
84const makeFakeMuse = Effect.fnUntraced(function* (idPrefix = "native") {
85 const requests = yield* Queue.unbounded<NativeCall>();
86 const calls: NativeCall[] = [];
87 const closed = pendingPromise<void>();
88 const exited = pendingPromise<Awaited<MuseSdkHost["exited"]>>();
89 let notify: Parameters<MuseSdkHost["connection"]["onNotification"]>[0] = () => {};
90 let sequence = 0;
91 let sessionId = "";
92 let closeCount = 0;
93 let closeGate: Promise<void> | undefined;
94 let history: MuseItem[] = [];
95 const responses = new Map<
96 string,
97 Array<Promise<Record<string, unknown>> | Record<string, unknown> | Error>
98 >();
99 const respond = async (call: NativeCall): Promise<Record<string, unknown>> => {
100 calls.push(call);
101 Queue.offerUnsafe(requests, call);
102 if (["session/start", "session/resume", "session/read"].includes(call.method)) {
103 sessionId = String(call.params.sessionId);
104 }
105 const queued = responses.get(call.method)?.shift();
106 if (queued instanceof Error) throw queued;
107 if (queued) return queued;
108 switch (call.method) {
109 case "session/start":
110 case "session/resume":
111 case "session/read":
112 sessionId = String(call.params.sessionId);
113 return {
114 session: { sessionId, modelId: MODEL, activeTurnId: null },
115 history: { items: history },
116 };
117 case "turn/start":
118 return { turnId: call.commandId };
119 case "turn/steer":
120 return { turnId: call.params.expectedTurnId };
121 case "session/compact":
122 return { status: "accepted" };
123 default:
124 return {};
125 }
126 };
127 const host: MuseSdkHost = {
128 connection: {
129 mintCommandId: () => `${idPrefix}-${++sequence}`,
130 command: (method, params, options) =>
131 respond({
132 method,
133 params,
134 ...(options?.commandId ? { commandId: options.commandId } : {}),
135 }),
136 request: (method, params = {}) => respond({ method, params }),
137 onNotification: (handler) => {
138 notify = handler;
139 },
140 onProtocolError: () => {},
141 onServerRequest: () => {},
142 closed: closed.promise,
143 },
144 initializeResult: { grantedCapabilities: [] },
145 exited: exited.promise,
146 close: async () => {
147 closeCount += 1;
148 await closeGate;
149 closed.resolve();
150 exited.resolve({ code: 0, signal: null });
151 },
152 };
153 return {
154 host,
155 calls,
156 closeCount: () => closeCount,
157 holdClose: (gate: Promise<void>) => {
158 closeGate = gate;
159 },
160 disconnect: () =>
161 Effect.sync(() => {
162 closed.resolve();
163 exited.resolve({ code: 1, signal: null });
164 }),
165 setHistory: (items: MuseItem[]) => {
166 history = items;
167 },
168 queueResponse: (
169 method: string,
170 result: Record<string, unknown> | Promise<Record<string, unknown>> | Error,
171 ) => {
172 responses.set(method, [...(responses.get(method) ?? []), result]);
173 },
174 emit: (method: string, params: Record<string, unknown>) =>
175 Effect.sync(() => {
176 notify({ jsonrpc: "2.0", method, params: { sessionId, ...params } });
177 }),
178 takeCall: Effect.fnUntraced(function* (method: string) {
179 while (true) {
180 const call = yield* Queue.take(requests);
181 if (call.method === method) return call;
182 }
183 }),
184 };
185});
186
187const makeHarness = Effect.fnUntraced(function* (
188 fake: Effect.Success<ReturnType<typeof makeFakeMuse>>,
189 instanceId = INSTANCE_ID,
190 initialNativeThreadId?: string,
191 replacement?: Effect.Success<ReturnType<typeof makeFakeMuse>>,
192 existingProviderThread?: OrchestrationV2ProviderThread,
193 policy = runtimePolicy,
194 overrides: Pick<
195 MuseAdapterV2Options,
196 "createHost" | "nativeEventLogger" | "modelCatalog" | "continuationRequests"
197 > = {},
198) {
199 let hostCount = 0;
200 const adapter = yield* makeMuseAdapterV2({
201 instanceId,
202 settings: museSettings,
203 environment: { PATH: "/fake/bin" },
204 createHost: async () => (hostCount++ === 0 ? fake.host : (replacement ?? fake).host),
205 ...overrides,
206 });
207 const runtime = yield* adapter.openSession({
208 threadId: THREAD_ID,
209 providerSessionId: ProviderSessionId.make(`session-${instanceId}`),
210 modelSelection: modelSelection(instanceId),
211 runtimePolicy: policy,
212 ...(initialNativeThreadId ? { initialNativeThreadId } : {}),
213 });
214 const emitted = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
215 const allEvents: ProviderAdapter.ProviderAdapterV2Event[] = [];
216 const eventsEnded = yield* Deferred.make<Exit.Exit<void, Stream.Error<typeof runtime.events>>>();
217 yield* runtime.events.pipe(
218 Stream.runForEach((event) =>
219 Effect.gen(function* () {
220 allEvents.push(event);
221 yield* Queue.offer(emitted, event);
222 }),
223 ),
224 Effect.exit,
225 Effect.flatMap((exit) => Deferred.succeed(eventsEnded, exit)),
226 Effect.forkScoped,
227 );
228 const providerThread = yield* runtime.ensureThread({
229 threadId: THREAD_ID,
230 modelSelection: modelSelection(instanceId),
231 runtimePolicy: policy,
232 ...(existingProviderThread ? { existingProviderThread } : {}),
233 });
234 const takeEvent = Effect.fnUntraced(function* <
235 T extends ProviderAdapter.ProviderAdapterV2Event["type"],
236 >(
237 type: T,
238 predicate: (
239 event: Extract<ProviderAdapter.ProviderAdapterV2Event, { type: T }>,
240 ) => boolean = () => true,
241 ) {
242 while (true) {
243 const event = yield* Queue.take(emitted);
244 if (
245 event.type === type &&
246 predicate(event as Extract<ProviderAdapter.ProviderAdapterV2Event, { type: T }>)
247 ) {
248 return event as Extract<ProviderAdapter.ProviderAdapterV2Event, { type: T }>;
249 }
250 }
251 });
252 return { adapter, runtime, providerThread, takeEvent, allEvents, policy, eventsEnded };
253});
254
255const preallocatedProviderThread = Effect.fnUntraced(function* () {
256 const now = yield* DateTime.now;
257 return {
258 id: ProviderThreadId.make("preallocated-muse-thread"),
259 driver: MUSE_PROVIDER,
260 providerInstanceId: INSTANCE_ID,
261 providerSessionId: ProviderSessionId.make(`session-${INSTANCE_ID}`),
262 appThreadId: THREAD_ID,
263 ownerNodeId: null,
264 nativeThreadRef: null,
265 nativeConversationHeadRef: null,
266 status: "idle",
267 firstRunOrdinal: null,
268 lastRunOrdinal: null,
269 handoffIds: [],
270 forkedFrom: null,
271 createdAt: now,
272 updatedAt: now,
273 } satisfies OrchestrationV2ProviderThread;
274});
275
276const turnInput = Effect.fnUntraced(function* (
277 providerThread: OrchestrationV2ProviderThread,
278 runOrdinal = 1,
279 policy = runtimePolicy,
280): Effect.fn.Return<ProviderAdapter.ProviderAdapterV2TurnInput> {
281 const now = yield* DateTime.now;
282 const threadId = providerThread.appThreadId ?? THREAD_ID;
283 const selection = modelSelection(providerThread.providerInstanceId);
284 const appThread: OrchestrationV2AppThread = {
285 id: threadId,
286 projectId: ProjectId.make("project-muse-test"),
287 title: "Muse test",
288 createdBy: "user",
289 creationSource: "web",
290 providerInstanceId: providerThread.providerInstanceId,
291 modelSelection: selection,
292 runtimeMode: policy.runtimeMode,
293 interactionMode: policy.interactionMode,
294 branch: null,
295 worktreePath: null,
296 activeProviderThreadId: providerThread.id,
297 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
298 forkedFrom: null,
299 createdAt: now,
300 updatedAt: now,
301 archivedAt: null,
302 settledOverride: null,
303 settledAt: null,
304 lastVisitedAt: null,
305 deletedAt: null,
306 };
307 return {
308 appThread,
309 threadId,
310 runId: RunId.make(`run-${runOrdinal}`),
311 runOrdinal,
312 providerTurnOrdinal: runOrdinal,
313 attemptId: RunAttemptId.make(`attempt-${runOrdinal}`),
314 rootNodeId: NodeId.make(`root-${runOrdinal}`),
315 providerThread,
316 message: {
317 messageId: MessageId.make(`message-${runOrdinal}`),
318 text: "Hello Muse",
319 attachments: [],
320 createdBy: "user",
321 creationSource: "web",
322 },
323 modelSelection: selection,
324 runtimePolicy: policy,
325 };
326});
327
328const startConversation = Effect.fnUntraced(function* (
329 harness: Effect.Success<ReturnType<typeof makeHarness>>,
330 fake: Effect.Success<ReturnType<typeof makeFakeMuse>>,
331 ordinal = 1,
332) {
333 yield* harness.runtime.startTurn(
334 yield* turnInput(harness.providerThread, ordinal, harness.policy),
335 );
336 const call = yield* fake.takeCall("turn/start");
337 const event = yield* harness.takeEvent(
338 "provider_turn.updated",
339 (event) => event.providerTurn.status === "running",
340 );
341 return { nativeId: call.commandId!, providerTurn: event.providerTurn };
342});
343
344const approval = (turnId: string) => ({
345 approvalId: "approval-1",
346 turnId,
347 currentRequirementId: { approvalId: "approval-1", sourceIndex: 0 },
348 availableChoices: [
349 { choiceId: "once", label: "Allow once", decision: "approved", scope: "once" },
350 { choiceId: "deny", label: "Deny", decision: "denied", scope: "once" },
351 ],
352 subject: { kind: "shell", command: "touch result.txt" },
353});
354
355describe("MuseAdapterV2", () => {
356 it.effect("connects the thread's MCP credential on native start and resume", () =>
357 Effect.gen(function* () {
358 yield* (yield* McpProviderSessions.McpProviderSessions).set({
359 environmentId: EnvironmentId.make("test-environment"),
360 threadId: THREAD_ID,
361 providerSessionId: "test-session",
362 providerInstanceId: INSTANCE_ID,
363 endpoint: "http://127.0.0.1:43210/mcp",
364 authorizationHeader: "Bearer thread-scoped-test-token",
365 browserToolsAvailable: true,
366 });
367 for (const nativeId of [undefined, "saved-native-session"]) {
368 const fake = yield* makeFakeMuse();
369 fake.host.initializeResult.grantedCapabilities.push("sessionMcp");
370 yield* makeHarness(fake, INSTANCE_ID, nativeId);
371 const call = yield* fake.takeCall(nativeId ? "session/resume" : "session/start");
372 assert.deepStrictEqual(call.params.config, {
373 mcpServers: {
374 "t3-code": {
375 transport: "streamableHttp",
376 mode: "optional",
377 url: "http://127.0.0.1:43210/mcp",
378 headers: { Authorization: "Bearer thread-scoped-test-token" },
379 },
380 },
381 });
382 }
383 }).pipe(Effect.scoped, Effect.provide(testLayer)),
384 );
385
386 it.effect("selects the runtime mode's permission profile on native start and resume", () =>
387 Effect.gen(function* () {
388 for (const [runtimeMode, profile, approvalMode] of [
389 ["approval-required", ":ask-me", "promptUnmatched"],
390 ["full-access", ":unrestricted", "allowAll"],
391 ] as const) {
392 const policy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({
393 ...runtimePolicy,
394 runtimeMode,
395 });
396 const fresh = yield* makeFakeMuse();
397 yield* makeHarness(fresh, INSTANCE_ID, undefined, undefined, undefined, policy);
398 const started = yield* fresh.takeCall("session/start");
399 assert.strictEqual(started.params.permissionProfile, profile);
400 assert.strictEqual(started.params.approvalMode, approvalMode);
401
402 const saved = yield* makeFakeMuse();
403 yield* makeHarness(saved, INSTANCE_ID, "saved-session", undefined, undefined, policy);
404 const resumed = yield* saved.takeCall("session/resume");
405 assert.strictEqual(resumed.params.permissionProfile, profile);
406 const mode = yield* saved.takeCall("session/setApprovalMode");
407 assert.strictEqual(mode.params.mode, approvalMode);
408 }
409 }).pipe(Effect.scoped, Effect.provide(testLayer)),
410 );
411
412 it.effect("rejects a host without session MCP support before opening a native session", () =>
413 Effect.gen(function* () {
414 yield* (yield* McpProviderSessions.McpProviderSessions).set({
415 environmentId: EnvironmentId.make("test-environment"),
416 threadId: THREAD_ID,
417 providerSessionId: "test-session",
418 providerInstanceId: INSTANCE_ID,
419 endpoint: "http://127.0.0.1:43210/mcp",
420 authorizationHeader: "Bearer thread-scoped-test-token",
421 browserToolsAvailable: false,
422 });
423 const fake = yield* makeFakeMuse();
424 const outcome = yield* makeHarness(fake).pipe(Effect.result);
425 assert.strictEqual(outcome._tag, "Failure");
426 assert.isFalse(fake.calls.some((call) => call.method === "session/start"));
427 }).pipe(Effect.scoped, Effect.provide(testLayer)),
428 );
429
430 it.effect(
431 "starts fresh after recovery clears a native ref instead of reusing the startup hint",
432 () =>
433 Effect.gen(function* () {
434 const fake = yield* makeFakeMuse();
435 const cleared = yield* preallocatedProviderThread();
436 const harness = yield* makeHarness(
437 fake,
438 INSTANCE_ID,
439 "failed-old-session",
440 undefined,
441 cleared,
442 );
443 const started = yield* fake.takeCall("session/start");
444 assert.isFalse(fake.calls.some((call) => call.method === "session/resume"));
445 assert.notStrictEqual(started.params.sessionId, "failed-old-session");
446 assert.strictEqual(harness.providerThread.id, cleared.id);
447 yield* startConversation(harness, fake);
448 }).pipe(Effect.scoped, Effect.provide(testLayer)),
449 );
450
451 it.effect("retries a missing native session with a cleared ref on the same healthy host", () =>
452 Effect.gen(function* () {
453 for (const kind of ["sessionNotFound", "notFound"] as const) {
454 const fake = yield* makeFakeMuse();
455 const allocated = yield* preallocatedProviderThread();
456 const adapter = yield* makeMuseAdapterV2({
457 instanceId: INSTANCE_ID,
458 settings: museSettings,
459 environment: { PATH: "/fake/bin" },
460 createHost: async () => fake.host,
461 });
462 const runtime = yield* adapter.openSession({
463 threadId: THREAD_ID,
464 providerSessionId: ProviderSessionId.make(`missing-${kind}`),
465 modelSelection: modelSelection(),
466 runtimePolicy,
467 initialNativeThreadId: "missing-session",
468 });
469 fake.queueResponse(
470 "session/resume",
471 new MspError({ code: -32000, message: "Saved session is absent", data: { kind } }),
472 );
473 const failed = yield* runtime
474 .ensureThread({
475 threadId: THREAD_ID,
476 modelSelection: modelSelection(),
477 runtimePolicy,
478 existingProviderThread: {
479 ...allocated,
480 nativeThreadRef: {
481 driver: MUSE_PROVIDER,
482 nativeId: "missing-session",
483 strength: "strong",
484 },
485 },
486 })
487 .pipe(Effect.exit);
488 assert.strictEqual(failed._tag, "Failure");
489 assert.strictEqual(fake.closeCount(), 0);
490 const fresh = yield* runtime.ensureThread({
491 threadId: THREAD_ID,
492 modelSelection: modelSelection(),
493 runtimePolicy,
494 existingProviderThread: allocated,
495 });
496 assert.strictEqual(fresh.id, allocated.id);
497 assert.notStrictEqual(fresh.nativeThreadRef?.nativeId, "missing-session");
498 assert.strictEqual(fake.closeCount(), 0);
499 yield* runtime.startTurn(yield* turnInput(fresh));
500 const admitted = yield* fake.takeCall("turn/start");
501 assert.strictEqual(admitted.params.reasoningEffort, "max");
502 yield* fake.emit("turn/completed", { turnId: admitted.commandId, terminal: "completed" });
503 yield* Stream.runDrain(
504 runtime.events.pipe(Stream.takeUntil((event) => event.type === "turn.terminal")),
505 );
506 }
507 }).pipe(Effect.scoped, Effect.provide(testLayer)),
508 );
509
510 it.effect("rejects cross-instance ensure requests without closing the current valid host", () =>
511 Effect.gen(function* () {
512 const fake = yield* makeFakeMuse();
513 const harness = yield* makeHarness(fake);
514 const rejected = yield* harness.runtime
515 .ensureThread({
516 threadId: THREAD_ID,
517 modelSelection: modelSelection(),
518 runtimePolicy,
519 existingProviderThread: {
520 ...harness.providerThread,
521 providerInstanceId: ProviderInstanceId.make("other-account"),
522 },
523 })
524 .pipe(Effect.exit);
525 assert.strictEqual(rejected._tag, "Failure");
526 assert.strictEqual(fake.closeCount(), 0);
527 yield* startConversation(harness, fake);
528 }).pipe(Effect.scoped, Effect.provide(testLayer)),
529 );
530
531 it.effect("pins Windows turns to Muse's verbatim root and starts Muse in the plain path", () =>
532 Effect.gen(function* () {
533 const fake = yield* makeFakeMuse();
534 let hostCwd: string | undefined;
535 const policy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({
536 ...runtimePolicy,
537 cwd: "C:\\work\\repo",
538 });
539 const harness = yield* makeHarness(
540 fake,
541 INSTANCE_ID,
542 undefined,
543 undefined,
544 undefined,
545 policy,
546 {
547 createHost: async (options) => {
548 hostCwd = options.cwd;
549 return fake.host;
550 },
551 },
552 );
553 yield* startConversation(harness, fake);
554 const turnStart = fake.calls.find((call) => call.method === "turn/start");
555 // Muse 1.4.3 rejects any other form: "expected a canonical path (resolves to \\?\C:\…)".
556 assert.deepStrictEqual(turnStart?.params.workspaceRoots, ["\\\\?\\C:\\work\\repo"]);
557 // The muse.cmd launcher runs under cmd.exe, which cannot start in a verbatim directory.
558 assert.strictEqual(hostCwd, "C:\\work\\repo");
559 }).pipe(
560 Effect.provideService(HostProcess.Platform, "win32"),
561 Effect.scoped,
562 Effect.provide(testLayer),
563 ),
564 );
565
566 it.effect("skips an unreadable usage notification without ending the turn", () =>
567 Effect.gen(function* () {
568 const fake = yield* makeFakeMuse();
569 const harness = yield* makeHarness(fake);
570 const { nativeId } = yield* startConversation(harness, fake);
571 yield* fake.emit("session/tokenUsage", { turnId: nativeId, usage: "not usage" });
572 yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "completed" });
573 const terminal = yield* harness.takeEvent("turn.terminal");
574 assert.strictEqual(terminal.status, "completed");
575 assert.strictEqual(fake.closeCount(), 0);
576 }).pipe(Effect.scoped, Effect.provide(testLayer)),
577 );
578
579 it.effect("moves a resumed session on another model back to the catalog default", () =>
580 Effect.gen(function* () {
581 const fake = yield* makeFakeMuse();
582 // The saved session last ran MODEL; the catalog's default is another model.
583 const harness = yield* makeHarness(
584 fake,
585 INSTANCE_ID,
586 "saved-session",
587 undefined,
588 undefined,
589 runtimePolicy,
590 {
591 modelCatalog: Effect.succeed([
592 {
593 slug: "account-default",
594 name: "Default",
595 isCustom: false,
596 isDefault: true,
597 capabilities: null,
598 },
599 { slug: MODEL, name: "Other", isCustom: false, isDefault: false, capabilities: null },
600 ]),
601 },
602 );
603 const input = yield* turnInput(harness.providerThread, 1);
604 const selection = { ...input.modelSelection, model: MUSE_DEFAULT_MODEL };
605 yield* harness.runtime.startTurn({
606 ...input,
607 modelSelection: selection,
608 appThread: { ...input.appThread, modelSelection: selection },
609 });
610 const setModel = yield* fake.takeCall("session/setModel");
611 assert.deepStrictEqual(setModel.params.model, {
612 modelId: "account-default",
613 providerId: "meta",
614 });
615 yield* fake.takeCall("turn/start");
616 }).pipe(Effect.scoped, Effect.provide(testLayer)),
617 );
618
619 it.effect("settles partial messages and pending approvals before a provider failure", () =>
620 Effect.gen(function* () {
621 const fake = yield* makeFakeMuse();
622 const harness = yield* makeHarness(fake);
623 const { nativeId } = yield* startConversation(harness, fake);
624 yield* fake.emit("item/started", {
625 turnId: nativeId,
626 item: {
627 itemId: "partial",
628 turnId: nativeId,
629 kind: "agentMessage",
630 status: "inProgress",
631 revision: 1,
632 text: "Partial answer",
633 },
634 });
635 yield* fake.emit("approval/requested", approval(nativeId));
636 yield* harness.takeEvent(
637 "runtime_request.updated",
638 (event) => event.runtimeRequest.status === "pending",
639 );
640 yield* fake.emit("turn/completed", {
641 turnId: nativeId,
642 terminal: "failed",
643 error: { message: "Provider rejected this turn" },
644 });
645 const terminal = yield* harness.takeEvent("turn.terminal");
646 assert.strictEqual(terminal.status, "failed");
647 assert.strictEqual(terminal.threadDisposition, "reusable");
648 assert.isNotNull(terminal.failure);
649 assert.isTrue(
650 harness.allEvents.some(
651 (event) =>
652 event.type === "message.updated" &&
653 !event.message.streaming &&
654 event.message.text === "Partial answer",
655 ),
656 );
657 assert.isTrue(
658 harness.allEvents.some(
659 (event) =>
660 event.type === "runtime_request.updated" && event.runtimeRequest.status === "cancelled",
661 ),
662 );
663 yield* startConversation(harness, fake, 2);
664 }).pipe(Effect.scoped, Effect.provide(testLayer)),
665 );
666
667 it.effect("closes its SDK host exactly once when its session scope ends", () =>
668 Effect.gen(function* () {
669 const fake = yield* makeFakeMuse();
670 yield* makeHarness(fake).pipe(Effect.scoped);
671 assert.strictEqual(fake.closeCount(), 1);
672 }).pipe(Effect.provide(testLayer)),
673 );
674
675 it.effect(
676 "closes a newly opened host when native session registration returns another identity",
677 () =>
678 Effect.gen(function* () {
679 const fake = yield* makeFakeMuse();
680 fake.queueResponse("session/start", {
681 session: { sessionId: "unexpected-session", modelId: MODEL, activeTurnId: null },
682 history: { items: [] },
683 });
684 const result = yield* makeHarness(fake).pipe(Effect.exit);
685 assert.strictEqual(result._tag, "Failure");
686 // A failed host closes in the background so the failure shows at once.
687 yield* Effect.promise(() => fake.host.exited);
688 assert.strictEqual(fake.closeCount(), 1);
689 assert.strictEqual(fake.calls.filter((call) => call.method === "turn/start").length, 0);
690 }).pipe(Effect.scoped, Effect.provide(testLayer)),
691 );
692
693 it.effect("settles an active turn once when both the connection and host exit unexpectedly", () =>
694 Effect.gen(function* () {
695 const fake = yield* makeFakeMuse();
696 const harness = yield* makeHarness(fake);
697 const { nativeId } = yield* startConversation(harness, fake);
698 yield* fake.emit("approval/requested", approval(nativeId));
699 yield* harness.takeEvent(
700 "runtime_request.updated",
701 (event) => event.runtimeRequest.status === "pending",
702 );
703 yield* fake.disconnect();
704 const terminal = yield* harness.takeEvent("turn.terminal");
705 assert.strictEqual(terminal.status, "failed");
706 assert.strictEqual(terminal.threadDisposition, "broken");
707 assert.isTrue(
708 harness.allEvents.some(
709 (event) =>
710 event.type === "runtime_request.updated" && event.runtimeRequest.status === "cancelled",
711 ),
712 );
713 const result = yield* harness.runtime
714 .startTurn(yield* turnInput(harness.providerThread, 2))
715 .pipe(Effect.exit);
716 assert.strictEqual(result._tag, "Failure");
717 assert.strictEqual(
718 harness.allEvents.filter((event) => event.type === "turn.terminal").length,
719 1,
720 );
721 assert.strictEqual(fake.calls.filter((call) => call.method === "turn/start").length, 1);
722 }).pipe(Effect.scoped, Effect.provide(testLayer)),
723 );
724
725 it.effect("turns a delivery gap into a broken terminal and rejects subsequent work", () =>
726 Effect.gen(function* () {
727 const fake = yield* makeFakeMuse();
728 const harness = yield* makeHarness(fake);
729 yield* startConversation(harness, fake);
730 yield* fake.emit("view/gap", {});
731 const terminal = yield* harness.takeEvent("turn.terminal");
732 assert.strictEqual(terminal.status, "failed");
733 assert.strictEqual(terminal.threadDisposition, "broken");
734 yield* Effect.promise(() => fake.host.exited);
735 assert.strictEqual(fake.closeCount(), 1);
736 yield* Deferred.await(harness.eventsEnded);
737 const result = yield* harness.runtime
738 .startTurn(yield* turnInput(harness.providerThread, 2))
739 .pipe(Effect.exit);
740 assert.strictEqual(result._tag, "Failure");
741 assert.strictEqual(fake.calls.filter((call) => call.method === "turn/start").length, 1);
742 }).pipe(Effect.scoped, Effect.provide(testLayer)),
743 );
744
745 it.effect("uses the latest native approval requirement and waits for its resolution event", () =>
746 Effect.gen(function* () {
747 const fake = yield* makeFakeMuse();
748 const harness = yield* makeHarness(fake);
749 const { nativeId } = yield* startConversation(harness, fake);
750 yield* fake.emit("approval/requested", approval(nativeId));
751 const first = yield* harness.takeEvent(
752 "runtime_request.updated",
753 (event) => event.runtimeRequest.status === "pending",
754 );
755 yield* fake.emit("approval/updated", {
756 ...approval(nativeId),
757 currentRequirementId: { approvalId: "approval-1", sourceIndex: 1 },
758 });
759 const refreshed = yield* harness.takeEvent(
760 "runtime_request.updated",
761 (event) => event.runtimeRequest.status === "pending",
762 );
763 assert.strictEqual(refreshed.runtimeRequest.id, first.runtimeRequest.id);
764 yield* harness.runtime.respondToRuntimeRequest({
765 requestId: first.runtimeRequest.id,
766 decision: "accept",
767 });
768 const decision = yield* fake.takeCall("approval/decide");
769 assert.deepStrictEqual(decision.params.requirementId, {
770 approvalId: "approval-1",
771 sourceIndex: 1,
772 });
773 assert.strictEqual(decision.params.choiceId, "once");
774 assert.isFalse(
775 harness.allEvents.some(
776 (event) =>
777 event.type === "runtime_request.updated" && event.runtimeRequest.status === "resolved",
778 ),
779 );
780 yield* fake.emit("approval/resolved", { approvalId: "approval-1", turnId: nativeId });
781 const resolved = yield* harness.takeEvent(
782 "runtime_request.updated",
783 (event) => event.runtimeRequest.status === "resolved",
784 );
785 assert.strictEqual(resolved.runtimeRequest.id, first.runtimeRequest.id);
786 const repeated = yield* harness.runtime
787 .respondToRuntimeRequest({ requestId: first.runtimeRequest.id, decision: "accept" })
788 .pipe(Effect.exit);
789 assert.strictEqual(repeated._tag, "Failure");
790 }).pipe(Effect.scoped, Effect.provide(testLayer)),
791 );
792
793 it.effect("keeps the user's decision when Muse settles before acknowledging it", () =>
794 Effect.gen(function* () {
795 const fake = yield* makeFakeMuse();
796 const harness = yield* makeHarness(fake);
797 const { nativeId } = yield* startConversation(harness, fake);
798 yield* fake.emit("approval/requested", approval(nativeId));
799 const pending = yield* harness.takeEvent(
800 "runtime_request.updated",
801 (event) => event.runtimeRequest.status === "pending",
802 );
803 const acknowledgement = pendingPromise<Record<string, unknown>>();
804 fake.queueResponse("approval/decide", acknowledgement.promise);
805 const responding = yield* harness.runtime
806 .respondToRuntimeRequest({ requestId: pending.runtimeRequest.id, decision: "decline" })
807 .pipe(Effect.forkScoped);
808 yield* fake.takeCall("approval/decide");
809 // Muse's native decision for this choice maps to decline too; "cancel" proves it was ours.
810 yield* fake.emit("approval/resolved", {
811 approvalId: "approval-1",
812 turnId: nativeId,
813 decision: "abort",
814 });
815 const resolved = yield* harness.takeEvent(
816 "runtime_request.updated",
817 (event) => event.runtimeRequest.status === "resolved",
818 );
819 assert.strictEqual(resolved.runtimeRequest.decision, "decline");
820 acknowledgement.resolve({});
821 yield* Fiber.join(responding);
822 }).pipe(Effect.scoped, Effect.provide(testLayer)),
823 );
824
825 it.effect("lets a user turn take a held Muse report turn, approvals included", () =>
826 Effect.gen(function* () {
827 const fake = yield* makeFakeMuse();
828 const offers: Array<ProviderContinuationRequests.ProviderContinuationRequest> = [];
829 const harness = yield* makeHarness(
830 fake,
831 INSTANCE_ID,
832 undefined,
833 undefined,
834 undefined,
835 runtimePolicy,
836 { continuationRequests: { offer: (request) => Effect.sync(() => offers.push(request)) } },
837 );
838 // A finished workflow's report turn starts while T3 runs nothing: it is held.
839 yield* fake.emit("turn/started", { turnId: "report-1" });
840 yield* Effect.yieldNow;
841 assert.lengthOf(offers, 1);
842 // The user sends a message first; Muse runs the report before it.
843 const { nativeId } = yield* startConversation(harness, fake);
844 yield* fake.emit("approval/requested", approval("report-1"));
845 const asked = yield* harness.takeEvent(
846 "runtime_request.updated",
847 (event) => event.runtimeRequest.status === "pending",
848 );
849 yield* harness.runtime.respondToRuntimeRequest({
850 requestId: asked.runtimeRequest.id,
851 decision: "accept",
852 });
853 yield* fake.takeCall("approval/decide");
854 yield* fake.emit("approval/resolved", { approvalId: "approval-1", turnId: "report-1" });
855 yield* fake.emit("turn/completed", { turnId: "report-1", terminal: "completed" });
856 yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "completed" });
857 const terminal = yield* harness.takeEvent("turn.terminal");
858 assert.strictEqual(terminal.status, "completed");
859 // The user turn took it, so the continuation is not dispatched.
860 const dispatched = yield* offers[0]!.dispatchIfCurrent!(Effect.succeed("run"));
861 assert.isTrue(Option.isNone(dispatched));
862 }).pipe(Effect.scoped, Effect.provide(testLayer)),
863 );
864
865 it.effect("approves a workflow child's approval in full access after its turn ended", () =>
866 Effect.gen(function* () {
867 const fake = yield* makeFakeMuse();
868 const harness = yield* makeHarness(fake);
869 const { nativeId } = yield* startConversation(harness, fake);
870 yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "completed" });
871 yield* harness.takeEvent("turn.terminal");
872 // Muse runs workflow children without the session's allowAll mode.
873 const childApproval = { ...approval("child-run-1"), subagentOrigin: { subagentId: "a7" } };
874 yield* fake.emit("approval/requested", childApproval);
875 const asked = yield* harness.takeEvent("runtime_request.updated");
876 assert.strictEqual(asked.runtimeRequest.status, "pending");
877 assert.isNull(asked.runtimeRequest.providerTurnId);
878 const decision = yield* fake.takeCall("approval/decide");
879 assert.strictEqual(decision.params.choiceId, "once");
880 yield* fake.emit("approval/resolved", childApproval);
881 const resolved = yield* harness.takeEvent("runtime_request.updated");
882 assert.strictEqual(resolved.runtimeRequest.status, "resolved");
883 assert.strictEqual(resolved.runtimeRequest.decision, "accept");
884 }).pipe(Effect.scoped, Effect.provide(testLayer)),
885 );
886
887 it.effect("keeps a workflow child's approval pending for the user past the turn's end", () =>
888 Effect.gen(function* () {
889 const fake = yield* makeFakeMuse();
890 const policy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({
891 ...runtimePolicy,
892 runtimeMode: "approval-required",
893 });
894 const harness = yield* makeHarness(
895 fake,
896 INSTANCE_ID,
897 undefined,
898 undefined,
899 undefined,
900 policy,
901 );
902 const { nativeId } = yield* startConversation(harness, fake);
903 const childApproval = { ...approval("child-run-1"), subagentOrigin: { subagentId: "a7" } };
904 yield* fake.emit("approval/requested", childApproval);
905 const asked = yield* harness.takeEvent("runtime_request.updated");
906 assert.isNull(asked.runtimeRequest.providerTurnId);
907 yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "completed" });
908 yield* harness.takeEvent("turn.terminal");
909 assert.isFalse(
910 harness.allEvents.some(
911 (event) =>
912 event.type === "runtime_request.updated" && event.runtimeRequest.status !== "pending",
913 ),
914 );
915 yield* harness.runtime.respondToRuntimeRequest({
916 requestId: asked.runtimeRequest.id,
917 decision: "decline",
918 });
919 const decision = yield* fake.takeCall("approval/decide");
920 assert.strictEqual(decision.params.choiceId, "deny");
921 }).pipe(Effect.scoped, Effect.provide(testLayer)),
922 );
923
924 it.effect("validates structured answers and preserves selected labels with a custom note", () =>
925 Effect.gen(function* () {
926 const fake = yield* makeFakeMuse();
927 const harness = yield* makeHarness(fake);
928 const { nativeId } = yield* startConversation(harness, fake);
929 yield* fake.emit("userInput/requested", {
930 userInputId: "question-1",
931 turnId: nativeId,
932 questions: [
933 {
934 id: "approach",
935 header: "Approach",
936 question: "Which approach?",
937 options: [{ label: "Small" }, { label: "Large" }],
938 selection: { mode: "single" },
939 },
940 ],
941 });
942 const pending = yield* harness.takeEvent(
943 "runtime_request.updated",
944 (event) => event.runtimeRequest.kind === "user_input",
945 );
946 const invalid = yield* harness.runtime
947 .respondToRuntimeRequest({
948 requestId: pending.runtimeRequest.id,
949 answers: { approach: ["Small", "Large"] },
950 })
951 .pipe(Effect.exit);
952 assert.strictEqual(invalid._tag, "Failure");
953 assert.isFalse(fake.calls.some((call) => call.method === "userInput/answer"));
954 yield* harness.runtime.respondToRuntimeRequest({
955 requestId: pending.runtimeRequest.id,
956 answers: { approach: ["Small", "Keep tests focused"] },
957 });
958 const response = yield* fake.takeCall("userInput/answer");
959 assert.deepStrictEqual(response.params.answers, [
960 { questionId: "approach", selectedLabel: "Small", note: "Keep tests focused" },
961 ]);
962 yield* fake.emit("userInput/settled", {
963 userInputId: "question-1",
964 turnId: nativeId,
965 answers: response.params.answers,
966 });
967 yield* harness.takeEvent(
968 "runtime_request.updated",
969 (event) => event.runtimeRequest.status === "resolved",
970 );
971 }).pipe(Effect.scoped, Effect.provide(testLayer)),
972 );
973
974 it.effect("steers only the active native turn and waits for confirmed interruption", () =>
975 Effect.gen(function* () {
976 const fake = yield* makeFakeMuse();
977 const harness = yield* makeHarness(fake);
978 const turn = yield* startConversation(harness, fake);
979 const input = yield* turnInput(harness.providerThread);
980 const invalid = yield* harness.runtime.steerTurn!({
981 threadId: THREAD_ID,
982 runId: input.runId,
983 providerThread: harness.providerThread,
984 providerTurnId: ProviderTurnId.make("unrelated-turn"),
985 message: input.message,
986 }).pipe(Effect.exit);
987 assert.strictEqual(invalid._tag, "Failure");
988 assert.isFalse(fake.calls.some((call) => call.method === "turn/steer"));
989 yield* harness.runtime.steerTurn!({
990 threadId: THREAD_ID,
991 runId: input.runId,
992 providerThread: harness.providerThread,
993 providerTurnId: turn.providerTurn.id,
994 message: { ...input.message, text: "Focus on the parser" },
995 });
996 assert.strictEqual((yield* fake.takeCall("turn/steer")).params.expectedTurnId, turn.nativeId);
997 const interrupt = yield* harness.runtime
998 .interruptTurn({
999 providerThread: harness.providerThread,
1000 providerTurnId: turn.providerTurn.id,
1001 })
1002 .pipe(Effect.forkScoped);
1003 assert.strictEqual((yield* fake.takeCall("turn/interrupt")).params.turnId, turn.nativeId);
1004 yield* fake.emit("turn/completed", { turnId: turn.nativeId, terminal: "cancelled" });
1005 yield* Fiber.join(interrupt);
1006 assert.strictEqual((yield* harness.takeEvent("turn.terminal")).status, "interrupted");
1007 }).pipe(Effect.scoped, Effect.provide(testLayer)),
1008 );
1009
1010 it.effect("resumes the requested native session and reads the same session snapshot", () =>
1011 Effect.gen(function* () {
1012 const fake = yield* makeFakeMuse();
1013 const nativeItem: MuseItem = {
1014 itemId: "prior-message",
1015 turnId: "prior-turn",
1016 kind: "agentMessage",
1017 status: "completed",
1018 revision: 1,
1019 text: "Previous answer",
1020 };
1021 fake.setHistory([nativeItem]);
1022 const harness = yield* makeHarness(fake, INSTANCE_ID, "saved-session");
1023 const resumed = yield* fake.takeCall("session/resume");
1024 assert.strictEqual(resumed.params.sessionId, "saved-session");
1025 assert.strictEqual(resumed.params.excludeItems, true);
1026 assert.isFalse(fake.calls.some((call) => call.method === "session/start"));
1027 assert.strictEqual(harness.providerThread.nativeThreadRef?.nativeId, "saved-session");
1028 const snapshot = yield* harness.runtime.readThreadSnapshot!({
1029 providerThread: harness.providerThread,
1030 });
1031 assert.strictEqual(snapshot.providerThread.nativeThreadRef?.nativeId, "saved-session");
1032 assert.deepStrictEqual(snapshot.providerPayload, [nativeItem]);
1033 assert.strictEqual(snapshot.messages[0]?.text, "Previous answer");
1034 }).pipe(Effect.scoped, Effect.provide(testLayer)),
1035 );
1036
1037 it.effect("serializes interruption behind pending turn admission without losing the target", () =>
1038 Effect.gen(function* () {
1039 const fake = yield* makeFakeMuse();
1040 const harness = yield* makeHarness(fake);
1041 const admission = pendingPromise<Record<string, unknown>>();
1042 fake.queueResponse("turn/start", admission.promise);
1043 const starting = yield* harness.runtime
1044 .startTurn(yield* turnInput(harness.providerThread))
1045 .pipe(Effect.forkScoped);
1046 const call = yield* fake.takeCall("turn/start");
1047 const running = yield* harness.takeEvent(
1048 "provider_turn.updated",
1049 (event) => event.providerTurn.status === "running",
1050 );
1051 const stopping = yield* harness.runtime
1052 .interruptTurn({
1053 providerThread: harness.providerThread,
1054 providerTurnId: running.providerTurn.id,
1055 })
1056 .pipe(Effect.forkScoped);
1057 admission.resolve({ turnId: call.commandId });
1058 yield* Fiber.join(starting);
1059 assert.strictEqual((yield* fake.takeCall("turn/interrupt")).params.turnId, call.commandId);
1060 yield* fake.emit("turn/completed", { turnId: call.commandId, terminal: "cancelled" });
1061 yield* Fiber.join(stopping);
1062 assert.strictEqual((yield* harness.takeEvent("turn.terminal")).status, "interrupted");
1063 }).pipe(Effect.scoped, Effect.provide(testLayer)),
1064 );
1065
1066 it.effect("reports a no-op compaction as a visible failure and permits another turn", () =>
1067 Effect.gen(function* () {
1068 const fake = yield* makeFakeMuse();
1069 const harness = yield* makeHarness(fake);
1070 fake.queueResponse("session/compact", { status: "noop", reason: "no_compactable_history" });
1071 yield* harness.runtime.compactThread!(yield* turnInput(harness.providerThread));
1072 const terminal = yield* harness.takeEvent("turn.terminal");
1073 assert.strictEqual(terminal.status, "failed");
1074 assert.strictEqual(terminal.threadDisposition, "reusable");
1075 yield* startConversation(harness, fake, 2);
1076 }).pipe(Effect.scoped, Effect.provide(testLayer)),
1077 );
1078
1079 it.effect("projects native todos with stable identity and removes cancelled steps", () =>
1080 Effect.gen(function* () {
1081 const fake = yield* makeFakeMuse();
1082 const harness = yield* makeHarness(fake);
1083 const { nativeId } = yield* startConversation(harness, fake);
1084 // Neither the tool call that wrote the list nor Muse's per-step reminder
1085 // child may show as a generic tool row.
1086 yield* fake.emit("item/completed", {
1087 turnId: nativeId,
1088 item: {
1089 itemId: "todo-tool",
1090 turnId: nativeId,
1091 kind: "toolCall",
1092 tool: "write_todos",
1093 revision: 1,
1094 status: "completed",
1095 },
1096 });
1097 yield* fake.emit("item/started", {
1098 turnId: nativeId,
1099 item: {
1100 itemId: "reminder",
1101 turnId: nativeId,
1102 kind: "reminderChild",
1103 revision: 1,
1104 status: "inProgress",
1105 fallbackText: "Reminder child session",
1106 },
1107 });
1108 yield* fake.emit("session/todoListChanged", {
1109 items: [
1110 { text: " Investigate ", status: "completed" },
1111 { text: "Fix", status: "inProgress" },
1112 { text: "Discarded", status: "cancelled" },
1113 { text: " ", status: "pending" },
1114 ],
1115 });
1116 const first = yield* harness.takeEvent(
1117 "turn_item.updated",
1118 (event) => event.turnItem.type === "todo_list",
1119 );
1120 assert.strictEqual(first.turnItem.type, "todo_list");
1121 if (first.turnItem.type !== "todo_list") return;
1122 assert.deepStrictEqual(
1123 first.turnItem.steps.map(({ text, status }) => ({ text, status })),
1124 [
1125 { text: "Investigate", status: "completed" },
1126 { text: "Fix", status: "running" },
1127 ],
1128 );
1129 const planId = first.turnItem.planId;
1130 yield* fake.emit("session/todoListChanged", {
1131 items: [
1132 { text: "Investigate", status: "completed" },
1133 { text: "Fix", status: "completed" },
1134 ],
1135 });
1136 const next = yield* harness.takeEvent(
1137 "turn_item.updated",
1138 (event) => event.turnItem.type === "todo_list",
1139 );
1140 assert.strictEqual(next.turnItem.id, first.turnItem.id);
1141 assert.strictEqual(next.turnItem.nodeId, first.turnItem.nodeId);
1142 if (next.turnItem.type !== "todo_list") return;
1143 assert.strictEqual(next.turnItem.planId, first.turnItem.planId);
1144 assert.isTrue(
1145 harness.allEvents.some(
1146 (event) =>
1147 event.type === "plan.updated" &&
1148 event.plan.id === planId &&
1149 event.plan.status === "completed",
1150 ),
1151 );
1152 assert.isFalse(
1153 harness.allEvents.some(
1154 (event) => event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
1155 ),
1156 );
1157 assert.isTrue((yield* harness.adapter.getCapabilities()).planning.emitsPlanUpdated);
1158 assert.isTrue((yield* harness.adapter.getCapabilities()).planning.emitsTodoList);
1159 }).pipe(Effect.scoped, Effect.provide(testLayer)),
1160 );
1161
1162 it.effect("opens a prompt-and-result child thread for a native subagent", () =>
1163 Effect.gen(function* () {
1164 const fake = yield* makeFakeMuse();
1165 const harness = yield* makeHarness(fake);
1166 const { nativeId } = yield* startConversation(harness, fake);
1167 yield* fake.emit("item/started", {
1168 turnId: nativeId,
1169 item: {
1170 itemId: "sub-1",
1171 kind: "subagent",
1172 revision: 1,
1173 status: "inProgress",
1174 turnId: nativeId,
1175 subagentId: "subagent-1",
1176 controlStatus: "running",
1177 childSessionId: "child-session-1",
1178 objective: "Check the thing",
1179 role: "general-purpose",
1180 },
1181 });
1182 const created = yield* harness.takeEvent("app_thread.created");
1183 const childThreadId = created.appThread.id;
1184 yield* fake.emit("item/updated", {
1185 turnId: nativeId,
1186 item: {
1187 itemId: "sub-1",
1188 kind: "subagent",
1189 revision: 2,
1190 status: "inProgress",
1191 turnId: nativeId,
1192 subagentId: "subagent-1",
1193 controlStatus: "resultReady",
1194 childSessionId: "child-session-1",
1195 objective: "Check the thing",
1196 result: { summary: "All done" },
1197 },
1198 });
1199 // The revision without a role keeps the title the child thread opened with.
1200 const completed = yield* harness.takeEvent(
1201 "subagent.updated",
1202 (event) => event.subagent.status === "completed",
1203 );
1204 assert.strictEqual(completed.subagent.title, created.appThread.title);
1205 const result = yield* harness.takeEvent(
1206 "message.updated",
1207 (event) => event.message.threadId === childThreadId && event.message.role === "assistant",
1208 );
1209 assert.strictEqual(result.message.text, "All done");
1210 yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "completed" });
1211 const terminal = yield* harness.takeEvent("turn.terminal");
1212 assert.strictEqual(terminal.status, "completed");
1213 }).pipe(Effect.scoped, Effect.provide(testLayer)),
1214 );
1215
1216 it.effect("keeps a workflow child's first label after later revisions drop it", () =>
1217 Effect.gen(function* () {
1218 const fake = yield* makeFakeMuse();
1219 const harness = yield* makeHarness(fake);
1220 const { nativeId } = yield* startConversation(harness, fake);
1221 const workflowItem = (revision: number, children: unknown) => ({
1222 itemId: "wf-1",
1223 kind: "workflow",
1224 revision,
1225 status: "inProgress",
1226 turnId: nativeId,
1227 entryId: "Probe workflow",
1228 children,
1229 });
1230 yield* fake.emit("item/started", {
1231 turnId: nativeId,
1232 item: workflowItem(1, [{ childId: "c1", attempt: 1, status: "scheduled", label: "ping" }]),
1233 });
1234 const scheduled = yield* harness.takeEvent("subagent.updated");
1235 assert.strictEqual(scheduled.subagent.title, "ping");
1236 yield* fake.emit("item/updated", {
1237 turnId: nativeId,
1238 item: workflowItem(2, [{ childId: "c1", attempt: 1, status: "started" }]),
1239 });
1240 const started = yield* harness.takeEvent(
1241 "subagent.updated",
1242 (event) => event.subagent.status === "running",
1243 );
1244 assert.strictEqual(started.subagent.title, "ping");
1245 }).pipe(Effect.scoped, Effect.provide(testLayer)),
1246 );
1247
1248 it.effect("shows a workflow child that completed with a failure kind as failed", () =>
1249 Effect.gen(function* () {
1250 const fake = yield* makeFakeMuse();
1251 const harness = yield* makeHarness(fake);
1252 const { nativeId } = yield* startConversation(harness, fake);
1253 yield* fake.emit("item/started", {
1254 turnId: nativeId,
1255 item: {
1256 itemId: "wf-1",
1257 kind: "workflow",
1258 revision: 1,
1259 status: "inProgress",
1260 turnId: nativeId,
1261 children: [
1262 {
1263 childId: "c1",
1264 attempt: 1,
1265 status: "finished",
1266 terminal: "completed",
1267 failureKind: "budget_exhausted",
1268 },
1269 ],
1270 },
1271 });
1272 const child = yield* harness.takeEvent("subagent.updated");
1273 assert.strictEqual(child.subagent.status, "failed");
1274 assert.strictEqual(child.subagent.result, "budget_exhausted");
1275 }).pipe(Effect.scoped, Effect.provide(testLayer)),
1276 );
1277});