clover's git

the-knife

1	import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress";
2	import { MspError, type SendUserTurnOptions } from "@muse-code/sdk";
3	import {
4	MUSE_DEFAULT_MODEL,
5	ProviderDriverKind,
6	type ModelSelection,
7	type OrchestrationV2AppThread,
8	type OrchestrationV2ConversationMessage,
9	type OrchestrationV2ExecutionNode,
10	type OrchestrationV2ProviderCapabilities,
11	type OrchestrationV2ProviderSession,
12	type OrchestrationV2ProviderThread,
13	type OrchestrationV2ProviderTurn,
14	type OrchestrationV2RuntimeRequest,
15	type OrchestrationV2Subagent,
16	type OrchestrationV2TurnItemStatus,
17	type NodeId,
18	type OrchestrationV2TurnItem,
19	type ProviderInstanceId,
20	type PlanId,
21	type OrchestrationV2PlanStep,
22	type ProviderApprovalDecision,
23	type RuntimeRequestId,
24	type ServerProviderModel,
25	type ThreadId,
26	} from "@t3tools/contracts";
27	import type { MuseSettings } from "../settings.ts";
28	import { AgentScope } from "@t3tools/shared/AgentScope";
29	import { getModelSelectionStringOptionValue } from "@t3tools/shared/model";
30	import * as Cause from "effect/Cause";
31	import * as DateTime from "effect/DateTime";
32	import * as Deferred from "effect/Deferred";
33	import * as Effect from "effect/Effect";
34	import * as Option from "effect/Option";
35	import * as FileSystem from "effect/FileSystem";
36	import * as Queue from "effect/Queue";
37	import * as Schema from "effect/Schema";
38	import * as Semaphore from "effect/Semaphore";
39	import * as Scope from "effect/Scope";
40	import * as Stream from "effect/Stream";
41
42	import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost";
43	import * as McpProviderSession from "@t3tools/provider-core/server/mcpSession";
44	import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
45	import { buildRuntimeInstructions } from "@t3tools/provider-core/server/runtimeInstructions";
46	import { museModelCapabilities, resolveMuseReasoningEffort } from "./modelCatalog.ts";
47	import {
48	MuseApproval,
49	MuseCompactResult,
50	MuseContextUsage,
51	MuseDelta,
52	MuseItemEvent,
53	MuseSessionResult,
54	MuseTokenUsageEvent,
55	MuseTodoList,
56	MuseTurnRetryScheduled,
57	MuseTurnCompleted,
58	MuseTurnStartResult,
59	MuseUserInput,
60	museApprovalChoices,
61	museApprovalDecision,
62	museApprovalOptions,
63	type MuseItem,
64	} from "./protocol.ts";
65	import {
66	createMuseSdkHostEffect,
67	museApprovalMode,
68	musePermissionProfile,
69	museWorkspaceRoot,
70	type createMuseSdkHost,
71	type MuseSdkHost,
72	} from "./sdk.ts";
73	import type * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers";
74	import {
75	providerMessageTextWithAttachmentPaths,
76	isProviderNativeImageAttachment,
77	} from "@t3tools/provider-core/server/attachmentPrompt";
78	import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
79	import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
80	import {
81	backgroundWorkNotification,
82	type BackgroundWorkReport,
83	} from "@t3tools/provider-core/server/notification";
84	import type * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
85	import { makeProviderFailure } from "@t3tools/provider-core/server/failure";
86	import {
87	makeSubagentChildThread,
88	makeSubagentConversationArtifacts,
89	subagentThreadTitle,
90	} from "@t3tools/provider-core/server/subagentProjection";
91	import { turnScopedSelectionTransition } from "@t3tools/provider-core/server/selectionTransition";
92	import { museItemStatus, museToolPresentation } from "./itemPresentation.ts";
93
94	const MUSE_PROVIDER = ProviderDriverKind.make("muse");
95	const isOpenSessionError = Schema.is(ProviderAdapter.ProviderAdapterOpenSessionError);
96	const isProtocolError = Schema.is(ProviderAdapter.ProviderAdapterProtocolError);
97
98	const MuseProviderCapabilitiesV2 = {
99	runtimePolicy: { enforcement: "native" },
100	sessions: {
101	supportsMultipleProviderThreadsPerSession: false,
102	supportsModelSwitchInSession: true,
103	supportsProviderSwitchingViaHandoff: true,
104	supportsRuntimeModeSwitchInSession: false,
105	pendingRequestsSurviveRestart: false,
106	},
107	threads: {
108	canCreateEmptyThread: true,
109	canReadThreadSnapshot: true,
110	// Muse 1.2.1/1.3.0 reject tested completed-turn fork boundaries with InvalidCut.
111	// Public forks use portable context handoff until native boundaries are validated.
112	canRollbackThread: false,
113	canForkThread: false,
114	canForkFromTurn: false,
115	canForkFromSubagentThread: false,
116	exposesNativeThreadId: true,
117	},
118	turns: {
119	exposesNativeTurnId: true,
120	emitsTurnStarted: true,
121	emitsTurnCompleted: true,
122	supportsInterrupt: true,
123	supportsActiveSteering: true,
124	supportsSteeringByInterruptRestart: false,
125	supportsQueuedMessages: true,
126	terminalStatusQuality: "strong",
127	},
128	streaming: {
129	streamsAssistantText: true,
130	streamsReasoning: true,
131	streamsToolOutput: true,
132	streamsPlanText: false,
133	emitsMessageCompleted: true,
134	},
135	tools: {
136	exposesToolItemIds: true,
137	emitsToolStarted: true,
138	emitsToolCompleted: true,
139	emitsToolOutput: true,
140	supportsMcpTools: true,
141	supportsDynamicToolCallbacks: false,
142	},
143	approvals: {
144	supportsCommandApproval: true,
145	supportsFileReadApproval: true,
146	supportsFileChangeApproval: true,
147	supportsApplyPatchApproval: false,
148	approvalsHaveNativeRequestIds: true,
149	approvalCallbacksAreLiveOnly: true,
150	approvalsCanOriginateFromSubagents: true,
151	},
152	planning: {
153	emitsPlanUpdated: true,
154	emitsTodoList: true,
155	emitsProposedPlan: false,
156	supportsStructuredQuestions: true,
157	planDeltasHaveItemIds: false,
158	},
159	subagents: {
160	supportsSubagents: true,
161	exposesSubagentThreadIds: false,
162	emitsSubagentLifecycle: true,
163	canWaitForSubagents: false,
164	canCloseSubagents: false,
165	canForkSubagentThread: false,
166	},
167	context: {
168	acceptsSystemContext: false,
169	acceptsDeveloperContext: false,
170	acceptsSyntheticUserContext: true,
171	canGenerateSummaries: false,
172	canConsumeHandoffSummaries: true,
173	supportsDeltaHandoff: true,
174	supportsFullThreadHandoff: true,
175	maxRecommendedHandoffChars: null,
176	},
177	checkpointing: {
178	appCanCheckpointFilesystem: true,
179	supportsNestedCheckpointScopes: false,
180	providerCanRollbackConversation: false,
181	providerRollbackReturnsSnapshot: false,
182	providerCanReadConversationSnapshot: true,
183	},
184	identity: {
185	nativeThreadIds: "strong",
186	nativeTurnIds: "strong",
187	nativeItemIds: "strong",
188	nativeRequestIds: "strong",
189	},
190	} satisfies OrchestrationV2ProviderCapabilities;
191
192	export interface MuseAdapterV2Options {
193	readonly instanceId: ProviderInstanceId;
194	readonly settings: MuseSettings;
195	readonly environment: NodeJS.ProcessEnv;
196	readonly modelCatalog?: Effect.Effect<ReadonlyArray<ServerProviderModel>>;
197	readonly createHost?: typeof createMuseSdkHost;
198	readonly requestTimeoutMs?: number;
199	readonly nativeEventLogger?: ProviderEventLoggers.EventNdjsonLogger;
200	/** Where a turn Muse starts on its own (a finished workflow's report) asks for its run. */
201	readonly continuationRequests?: {
202	readonly offer: (
203	request: ProviderContinuationRequests.ProviderContinuationRequest,
204	) => Effect.Effect<void>;
205	};
206	}
207
208	interface ActiveTurn {
209	readonly input: ProviderAdapter.ProviderAdapterV2TurnInput;
210	providerTurn: OrchestrationV2ProviderTurn;
211	readonly nativeId: string;
212	/** Turns Muse started on its own while this run was active, which this run shows. */
213	readonly joined: Set<string>;
214	readonly items: Map<string, MuseItem>;
215	readonly ordinals: Map<string, number>;
216	readonly started: Map<string, DateTime.Utc>;
217	readonly dirty: Set<string>;
218	readonly settledRequests: Set<string>;
219	readonly done: Deferred.Deferred<void>;
220	todoPlanId?: PlanId;
221	nextOrdinal: number;
222	flushScheduled: boolean;
223	interruptRequested: boolean;
224	compact: boolean;
225	}
226
227	interface PendingRequest {
228	readonly native:
229	| { type: "approval"; value: MuseApproval }
230	| { type: "question"; value: MuseUserInput };
231	request: OrchestrationV2RuntimeRequest;
232	node: OrchestrationV2ExecutionNode;
233	item: OrchestrationV2TurnItem;
234	/** What T3 answered, recorded on the request once Muse confirms it. */
235	response?: Pick<OrchestrationV2RuntimeRequest, "decision" | "answers">;
236	}
237
238	/**
239	* One native subagent's child thread. Muse 1.4.4 cannot address an inline
240	* child session (`session/read` and `view/subscribe` answer sessionNotFound),
241	* so the thread carries the prompt, live status (on its root node) and result.
242	*/
243	interface ChildContext {
244	readonly childSessionId: string;
245	readonly childThreadId: ThreadId;
246	readonly appThread: OrchestrationV2AppThread;
247	readonly rootNodeId: NodeId;
248	readonly parentThreadId: ThreadId;
249	status: OrchestrationV2TurnItemStatus;
250	/** The result shown in the child thread, once the subagent settled. */
251	result?: string;
252	}
253
254	const nativeRef = (nativeId: string) => ({
255	driver: MUSE_PROVIDER,
256	nativeId,
257	strength: "strong" as const,
258	});
259	const recordSchema = Schema.Record(Schema.String, Schema.Unknown);
260	// Notifications that only add detail. A malformed one is skipped instead of ending the session.
261	const INFORMATIONAL_NOTIFICATIONS = new Set([
262	"session/contextUsage",
263	"session/tokenUsage",
264	"session/todoListChanged",
265	"turn/retryScheduled",
266	]);
267	// Tools whose results already show as their own rows: T3's todo list and question
268	// rows, and the workflow item a `workflow` call launches.
269	const TOOLS_WITH_NATIVE_ROWS = new Set(["write_todos", "request_user_input", "workflow"]);
270	const responseAnswerSchema = Schema.Union([Schema.String, Schema.Array(Schema.String)]);
271
272	/** One scoped Muse host owns one native session; the orchestrator owns app runs and queuing. */
273	export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* (
274	options: MuseAdapterV2Options,
275	) {
276	const idAllocator = yield* IdAllocator.IdAllocatorV2;
277	const providerHost = yield* ProviderHost.ProviderHost;
278	const fileSystem = yield* FileSystem.FileSystem;
279	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
280	const scriptProgress = yield* ScriptProgress.ScriptProgress;
281	const agentScope = yield* AgentScope;
282
283	const protocolError = (detail: string, payload?: unknown) =>
284	new ProviderAdapter.ProviderAdapterProtocolError({
285	driver: MUSE_PROVIDER,
286	detail,
287	...(payload === undefined ? {} : { payload }),
288	});
289	return ProviderAdapter.ProviderAdapterV2.of({
290	instanceId: options.instanceId,
291	driver: MUSE_PROVIDER,
292	getCapabilities: () => Effect.succeed(MuseProviderCapabilitiesV2),
293	planSelectionTransition: () => Effect.succeed(turnScopedSelectionTransition()),
294	openSession: Effect.fn("MuseAdapterV2.openSession")(function* (
295	input: ProviderAdapter.ProviderAdapterV2OpenSessionInput,
296	) {
297	if (!options.settings.enabled) return yield* protocolError("Muse Code is disabled");
298	if (input.runtimePolicy.interactionMode === "plan")
299	return yield* protocolError("Muse Code does not support dedicated Plan mode");
300	const scope = yield* Effect.scope;
301	// Muse rejects non-canonical workspace roots (for example macOS /tmp) and
302	// reports canonical paths in approvals, so resolve symlinks once here.
303	const requestedCwd = input.runtimePolicy.cwd ?? providerHost.paths.cwd;
304	const cwd = yield* fileSystem
305	.realPath(requestedCwd)
306	.pipe(Effect.orElseSucceed(() => requestedCwd));
307	const workspaceRoot = yield* museWorkspaceRoot(cwd);
308	const now = yield* DateTime.now;
309	let session: OrchestrationV2ProviderSession = {
310	id: input.providerSessionId,
311	driver: MUSE_PROVIDER,
312	providerInstanceId: options.instanceId,
313	status: "ready",
314	cwd,
315	model: input.modelSelection.model,
316	capabilities: MuseProviderCapabilitiesV2,
317	createdAt: now,
318	updatedAt: now,
319	lastError: null,
320	};
321	const events = yield* Queue.unbounded<
322	ProviderAdapter.ProviderAdapterV2Event,
323	ProviderAdapter.ProviderAdapterV2Error | Cause.Done
324	>();
325	type Inbox =
326	| { type: "notification"; method: string; params: unknown; epoch: number }
327	| { type: "failure"; cause: unknown; epoch: number }
328	| { type: "flush"; turn: ActiveTurn };
329	const inbox = yield* Queue.unbounded<Inbox>();
330	const commands = yield* Semaphore.make(1);
331	const eventPermit = yield* Semaphore.make(1);
332	let host: MuseSdkHost;
333	let hostEpoch = 0;
334	let closed = false;
335	let broken = false;
336	let nativeSessionId: string | undefined;
337	let thread: OrchestrationV2ProviderThread | undefined;
338	let active: ActiveTurn | undefined;
339	// The last finished turn, so context usage reported after it still lands on it.
340	let lastProviderTurn: OrchestrationV2ProviderTurn | undefined;
341	const agentEndedAt = new Map<string, DateTime.Utc>();
342	// "default" is the catalog's default model, from the provider snapshot. Without
343	// one, Muse keeps whatever model the session already runs.
344	const resolveModel = Effect.fnUntraced(function* (selection: ModelSelection) {
345	if (selection.model !== MUSE_DEFAULT_MODEL) return selection.model;
346	const catalog = options.modelCatalog ? yield* options.modelCatalog : [];
347	return catalog.find((model) => model.isDefault && !model.isCustom)?.slug;
348	});
349	const pending = new Map<RuntimeRequestId, PendingRequest>();
350	const seenTerminals = new Set<string>();
351	// Workflows and subagents that outlive their turn, by item id, with the turn that owns them.
352	const observedChildren = new Map<string, ActiveTurn>();
353	// Native subagents' child threads, by subagent item id.
354	const childrenByItemId = new Map<string, ChildContext>();
355	// Titles per native subagent and workflow child attempt; later revisions drop the label.
356	const agentTitles = new Map<string, string>();
357	// Background work that finished since the last turn, reported to the turn Muse starts for it.
358	const finishedBackground: Array<BackgroundWorkReport> = [];
359	// A turn Muse started on its own, held until the continuation run T3 opens for it takes it.
360	let wake:
361	| { readonly nativeId: string; readonly events: Array<[string, unknown]> }
362	| undefined;
363	const emit = (event: ProviderAdapter.ProviderAdapterV2Event) =>
364	Queue.offer(events, event).pipe(Effect.asVoid);
365	const decode = <A, I>(schema: Schema.Codec<A, I>, data: unknown) =>
366	Schema.decodeUnknownEffect(schema)(data).pipe(
367	Effect.mapError((cause) => protocolError("Invalid Muse protocol response", cause)),
368	);
369	const request = (
370	method: string,
371	params: Record<string, unknown>,
372	command = true,
373	commandId?: string,
374	) =>
375	Effect.tryPromise({
376	try: () =>
377	command
378	? host.connection.command(
379	method,
380	{ ...(nativeSessionId ? { sessionId: nativeSessionId } : {}), ...params },
381	{
382	...(method === "turn/start" ? {} : { maxAttempts: 1 }),
383	...(commandId ? { commandId } : {}),
384	},
385	)
386	: host.connection.request(method, {
387	...(nativeSessionId ? { sessionId: nativeSessionId } : {}),
388	...params,
389	}),
390	catch: (cause) =>
391	protocolError(
392	cause instanceof Error && cause.message
393	? `Muse ${method} failed: ${cause.message}`
394	: `Muse ${method} failed`,
395	cause,
396	),
397	}).pipe(
398	Effect.timeout(options.requestTimeoutMs ?? 30_000),
399	Effect.catchTags({
400	TimeoutError: (cause) =>
401	Effect.gen(function* () {
402	yield* eventPermit.withPermits(1)(failHost(cause));
403	return yield* protocolError(`Muse ${method} timed out; its host was closed`, cause);
404	}),
405	}),
406	);
407	const updateSession = Effect.fnUntraced(function* (
408	status: OrchestrationV2ProviderSession["status"],
409	lastError: string | null = null,
410	) {
411	session = { ...session, status, lastError, updatedAt: yield* DateTime.now };
412	yield* emit({
413	type: "provider_session.updated",
414	driver: MUSE_PROVIDER,
415	providerSession: session,
416	});
417	});
418	const updateThread = Effect.fnUntraced(function* (
419	patch: Partial<OrchestrationV2ProviderThread>,
420	) {
421	if (!thread) return;
422	thread = { ...thread, ...patch, updatedAt: yield* DateTime.now };
423	yield* emit({
424	type: "provider_thread.updated",
425	driver: MUSE_PROVIDER,
426	providerThread: thread,
427	});
428	});
429	const itemIdentity = (turn: ActiveTurn, nativeId: string) =>
430	`${options.instanceId}:${turn.input.providerThread.id}:${turn.nativeId}:${nativeId}`;
431	const baseItem = (turn: ActiveTurn, nativeId: string, time: DateTime.Utc) => {
432	if (!turn.ordinals.has(nativeId)) turn.ordinals.set(nativeId, turn.nextOrdinal++);
433	if (!turn.started.has(nativeId)) turn.started.set(nativeId, time);
434	const identity = itemIdentity(turn, nativeId);
435	return {
436	id: idAllocator.derive.turnItemFromProviderItem({
437	driver: MUSE_PROVIDER,
438	nativeItemId: identity,
439	}),
440	nodeId: idAllocator.derive.nodeFromProviderItem({
441	driver: MUSE_PROVIDER,
442	nativeItemId: identity,
443	}),
444	threadId: turn.input.threadId,
445	runId: turn.input.runId,
446	providerThreadId: turn.input.providerThread.id,
447	providerTurnId: turn.providerTurn.id,
448	nativeItemRef: nativeRef(nativeId),
449	parentItemId: null,
450	ordinal: turn.ordinals.get(nativeId)!,
451	startedAt: turn.started.get(nativeId)!,
452	updatedAt: time,
453	};
454	};
455	const childItemIdentity = (context: ChildContext, nativeId: string) =>
456	`${options.instanceId}:child:${context.childSessionId}:${nativeId}`;
457	/** A subagent with a ready result has finished its work even while Muse keeps it open. */
458	const itemStatus = (item: MuseItem): OrchestrationV2TurnItemStatus => {
459	const status = museItemStatus(item);
460	return status === "running" &&
461	item.kind === "subagent" &&
462	item.controlStatus === "resultReady"
463	? "completed"
464	: status;
465	};
466	const publishChildRoot = Effect.fnUntraced(function* (
467	context: ChildContext,
468	status: OrchestrationV2TurnItemStatus,
469	time: DateTime.Utc,
470	) {
471	context.status = status;
472	yield* emit({
473	type: "node.updated",
474	driver: MUSE_PROVIDER,
475	node: {
476	id: context.rootNodeId,
477	threadId: context.childThreadId,
478	runId: null,
479	parentNodeId: null,
480	rootNodeId: context.rootNodeId,
481	kind: "root_turn",
482	status,
483	countsForRun: false,
484	providerThreadId: null,
485	providerTurnId: null,
486	nativeItemRef: nativeRef(context.childSessionId),
487	runtimeRequestId: null,
488	checkpointScopeId: null,
489	startedAt: context.appThread.createdAt,
490	completedAt: status === "running" || status === "pending" ? null : time,
491	},
492	});
493	});
494	/** One synthesized conversation row in a child thread: its prompt or its result. */
495	const publishChildMessage = Effect.fnUntraced(function* (
496	context: ChildContext,
497	role: "user" | "assistant",
498	text: string,
499	suffix: "prompt" | "result",
500	time: DateTime.Utc,
501	) {
502	const identity = childItemIdentity(context, suffix);
503	const artifacts = makeSubagentConversationArtifacts({
504	messageId: idAllocator.derive.messageFromProviderItem({
505	driver: MUSE_PROVIDER,
506	nativeItemId: identity,
507	}),
508	...(role === "user" ? { senderThreadId: context.parentThreadId } : {}),
509	turnItemId: idAllocator.derive.turnItemFromProviderItem({
510	driver: MUSE_PROVIDER,
511	nativeItemId: identity,
512	}),
513	threadId: context.childThreadId,
514	rootNodeId: context.rootNodeId,
515	providerThreadId: null,
516	providerTurnId: null,
517	nativeItemRef: nativeRef(context.childSessionId),
518	role,
519	text,
520	ordinal: suffix === "prompt" ? 1 : 2,
521	now: time,
522	});
523	yield* emit({ type: "message.updated", driver: MUSE_PROVIDER, message: artifacts.message });
524	yield* emit({
525	type: "turn_item.updated",
526	driver: MUSE_PROVIDER,
527	turnItem: artifacts.turnItem,
528	});
529	});
530	/** Creates the child thread for a native subagent the first time its session id is known. */
531	const ensureChildContext = Effect.fnUntraced(function* (
532	turn: ActiveTurn,
533	item: MuseItem,
534	parentNodeId: NodeId,
535	title: string,
536	prompt: string,
537	status: OrchestrationV2TurnItemStatus,
538	time: DateTime.Utc,
539	) {
540	const existing = childrenByItemId.get(item.itemId);
541	const childSessionId = item.childSessionId;
542	if (existing || !childSessionId) return existing;
543	const childThreadId = idAllocator.derive.threadFromProviderThread({
544	driver: MUSE_PROVIDER,
545	providerInstanceId: options.instanceId,
546	nativeThreadId: childSessionId,
547	});
548	const appThread = makeSubagentChildThread({
549	parentThread: turn.input.appThread,
550	childThreadId,
551	parentNodeId,
552	activeProviderThreadId: null,
553	providerInstanceId: options.instanceId,
554	modelSelection: turn.input.modelSelection,
555	title,
556	now: time,
557	createdBy: "agent",
558	creationSource: "provider",
559	});
560	const context: ChildContext = {
561	childSessionId,
562	childThreadId,
563	appThread,
564	rootNodeId: idAllocator.derive.nodeFromProviderItem({
565	driver: MUSE_PROVIDER,
566	nativeItemId: `${options.instanceId}:child:${childSessionId}:thread-root`,
567	}),
568	parentThreadId: turn.input.threadId,
569	status,
570	};
571	childrenByItemId.set(item.itemId, context);
572	yield* emit({ type: "app_thread.created", driver: MUSE_PROVIDER, appThread });
573	yield* publishChildRoot(context, status, time);
574	if (prompt) yield* publishChildMessage(context, "user", prompt, "prompt", time);
575	return context;
576	});
577	/**
578	* Shows a native subagent on its parent thread and mirrors it into a child
579	* thread the user can open, like Claude's and Codex's subagents.
580	*/
581	const publishNativeSubagent = Effect.fnUntraced(function* (
582	turn: ActiveTurn,
583	item: MuseItem,
584	base: ReturnType<typeof baseItem> & {
585	readonly status: OrchestrationV2TurnItemStatus;
586	readonly completedAt: DateTime.Utc | null;
587	readonly title: string | null;
588	},
589	status: OrchestrationV2TurnItemStatus,
590	time: DateTime.Utc,
591	) {
592	const prompt = item.objective ?? item.fallbackText ?? "";
593	const settled = status !== "running" && status !== "pending";
594	const endKey = `subagent:${item.itemId}`;
595	if (settled && !agentEndedAt.has(endKey)) agentEndedAt.set(endKey, time);
596	if (!settled) agentEndedAt.delete(endKey);
597	const completedAt = settled ? (agentEndedAt.get(endKey) ?? time) : null;
598	if (!agentTitles.has(endKey))
599	agentTitles.set(
600	endKey,
601	subagentThreadTitle({
602	parentTitle: turn.input.appThread.title,
603	prompt,
604	title: item.role ?? null,
605	ordinal: childrenByItemId.size + 1,
606	}),
607	);
608	const title = agentTitles.get(endKey)!;
609	const context = yield* ensureChildContext(
610	turn,
611	item,
612	base.nodeId,
613	title,
614	prompt,
615	status,
616	time,
617	);
618	const resultText =
619	item.result?.text ??
620	item.result?.summary ??
621	(status === "failed" ? item.failureReason : undefined);
622	const result = resultText ?? null;
623	const nativeTaskRef = item.childSessionId
624	? nativeRef(item.childSessionId)
625	: nativeRef(item.subagentId ?? item.itemId);
626	const subagent: OrchestrationV2Subagent = {
627	id: base.nodeId,
628	threadId: base.threadId,
629	runId: base.runId,
630	parentNodeId: turn.input.rootNodeId,
631	origin: "provider_native",
632	createdBy: "agent",
633	driver: MUSE_PROVIDER,
634	providerInstanceId: options.instanceId,
635	providerThreadId: base.providerThreadId,
636	childThreadId: context?.childThreadId ?? null,
637	nativeTaskRef,
638	prompt,
639	title,
640	model: null,
641	status,
642	result,
643	startedAt: base.startedAt,
644	completedAt,
645	updatedAt: time,
646	};
647	yield* emit({ type: "subagent.updated", driver: MUSE_PROVIDER, subagent });
648	yield* emit({
649	type: "node.updated",
650	driver: MUSE_PROVIDER,
651	node: {
652	id: base.nodeId,
653	threadId: base.threadId,
654	runId: base.runId,
655	parentNodeId: turn.input.rootNodeId,
656	rootNodeId: turn.input.rootNodeId,
657	kind: "subagent",
658	status,
659	countsForRun: false,
660	providerThreadId: base.providerThreadId,
661	providerTurnId: base.providerTurnId,
662	nativeItemRef: nativeTaskRef,
663	runtimeRequestId: null,
664	checkpointScopeId: null,
665	startedAt: base.startedAt,
666	completedAt,
667	},
668	});
669	yield* emit({
670	type: "turn_item.updated",
671	driver: MUSE_PROVIDER,
672	turnItem: {
673	...base,
674	nativeItemRef: nativeTaskRef,
675	type: "subagent",
676	status,
677	title,
678	completedAt,
679	subagentId: subagent.id,
680	origin: "provider_native",
681	driver: MUSE_PROVIDER,
682	providerInstanceId: options.instanceId,
683	childThreadId: subagent.childThreadId,
684	prompt,
685	result,
686	},
687	});
688	if (context) {
689	if (context.status !== status) yield* publishChildRoot(context, status, time);
690	if (settled && result?.trim() && context.result !== result) {
691	context.result = result;
692	yield* publishChildMessage(context, "assistant", result, "result", time);
693	}
694	}
695	});
696	/** Lists running workflows and subagents on the thread, so it shows it still waits on them. */
697	const syncBackground = Effect.fnUntraced(function* () {
698	yield* updateThread({
699	pendingBackgroundTasks: [...observedChildren].flatMap(([itemId, owner]) => {
700	const item = owner.items.get(itemId);
701	const description = item?.objective ?? item?.entryId ?? item?.fallbackText;
702	return [
703	{
704	taskId: itemId,
705	kind:
706	item?.kind === "subagent" ? ("subagent" as const) : ("background_task" as const),
707	...(description ? { description } : {}),
708	},
709	];
710	}),
711	});
712	});
713	const publishItem = Effect.fnUntraced(function* (
714	turn: ActiveTurn,
715	item: MuseItem,
716	terminal?: OrchestrationV2TurnItem["status"],
717	) {
718	// Muse's skill-reminder child runs on every model step; it is housekeeping, not work.
719	if (item.kind === "userMessage" || item.kind === "reminderChild") return;
720	// A failed call keeps its row: it may have no native row to show its error.
721	if (
722	item.kind === "toolCall" &&
723	item.tool &&
724	TOOLS_WITH_NATIVE_ROWS.has(item.tool) &&
725	item.status !== "failed"
726	)
727	return;
728	const time = yield* DateTime.now;
729	const status = terminal ?? itemStatus(item);
730	if (item.kind === "subagent" || item.kind === "workflow") {
731	const wasRunning = observedChildren.has(item.itemId);
732	if (status === "running") observedChildren.set(item.itemId, turn);
733	else if (observedChildren.delete(item.itemId) && active !== turn)
734	finishedBackground.push({
735	kind: item.kind === "subagent" ? "subagent" : "background_task",
736	label: item.objective ?? item.entryId ?? item.fallbackText,
737	outcome:
738	status === "completed"
739	? "completed"
740	: status === "cancelled"
741	? "cancelled"
742	: "failed",
743	});
744	if (wasRunning !== observedChildren.has(item.itemId)) yield* syncBackground();
745	}
746	const streaming = status === "running";
747	const base = {
748	...baseItem(turn, item.itemId, time),
749	status,
750	completedAt: streaming ? null : time,
751	title: item.tool ?? null,
752	};
753	if (item.kind === "subagent") {
754	yield* publishNativeSubagent(turn, item, base, status, time);
755	return;
756	}
757	const text = item.text ?? item.summary?.join("\n") ?? item.fallbackText ?? "";
758	let turnItem: OrchestrationV2TurnItem;
759	if (item.kind === "agentMessage") {
760	const messageId = idAllocator.derive.messageFromProviderItem({
761	driver: MUSE_PROVIDER,
762	nativeItemId: itemIdentity(turn, item.itemId),
763	});
764	const message: OrchestrationV2ConversationMessage = {
765	id: messageId,
766	threadId: turn.input.threadId,
767	runId: turn.input.runId,
768	nodeId: base.nodeId,
769	role: "assistant",
770	text,
771	attachments: [],
772	streaming,
773	createdBy: "agent",
774	creationSource: "provider",
775	createdAt: base.startedAt,
776	updatedAt: time,
777	};
778	yield* emit({ type: "message.updated", driver: MUSE_PROVIDER, message });
779	turnItem = { ...base, type: "assistant_message", messageId, text, streaming };
780	} else if (item.kind === "reasoning") {
781	turnItem = { ...base, type: "reasoning", text, streaming };
782	} else if (item.kind === "compaction") {
783	turnItem = {
784	...base,
785	type: "compaction",
786	driver: MUSE_PROVIDER,
787	summary: text || item.reason,
788	};
789	} else {
790	turnItem = { ...base, ...museToolPresentation(item, status) };
791	}
792	yield* emit({
793	type: "node.updated",
794	driver: MUSE_PROVIDER,
795	node: {
796	id: base.nodeId,
797	threadId: base.threadId,
798	runId: base.runId,
799	parentNodeId: turn.input.rootNodeId,
800	rootNodeId: turn.input.rootNodeId,
801	kind:
802	item.kind === "agentMessage"
803	? "assistant_message"
804	: item.kind === "reasoning"
805	? "reasoning"
806	: "tool_call",
807	status,
808	countsForRun: false,
809	providerThreadId: base.providerThreadId,
810	providerTurnId: base.providerTurnId,
811	nativeItemRef: base.nativeItemRef,
812	runtimeRequestId: null,
813	checkpointScopeId: null,
814	startedAt: base.startedAt,
815	completedAt: base.completedAt,
816	},
817	});
818	yield* emit({ type: "turn_item.updated", driver: MUSE_PROVIDER, turnItem });
819	if (item.kind === "workflow") yield* publishWorkflowAgents(turn, item, base, status);
820	});
821	/**
822	* Shows each workflow child as a native subagent under the workflow row. Muse
823	* reports only their status, label and outcome, not their conversations, so
824	* they get no child thread.
825	*/
826	const publishWorkflowAgents = Effect.fnUntraced(function* (
827	turn: ActiveTurn,
828	item: MuseItem,
829	workflow: { readonly id: OrchestrationV2TurnItem["id"]; readonly nodeId: NodeId },
830	workflowStatus: OrchestrationV2TurnItemStatus,
831	) {
832	const latest = new Map((item.children ?? []).map((child) => [child.childId, child]));
833	for (const [index, child] of [...latest.values()].entries()) {
834	const time = yield* DateTime.now;
835	const base = baseItem(turn, `${item.itemId}:agent:${child.childId}`, time);
836	// Muse marks a failed attempt with `failureKind`, even when its terminal is `completed`.
837	const status: OrchestrationV2TurnItemStatus =
838	child.terminal === "completed" && child.failureKind === undefined
839	? "completed"
840	: child.terminal === "cancelled"
841	? "cancelled"
842	: child.terminal !== undefined
843	? "failed"
844	: workflowStatus !== "running"
845	? workflowStatus
846	: child.status === "scheduled"
847	? "pending"
848	: "running";
849	const settled = status !== "running" && status !== "pending";
850	// Children are re-sent on every workflow change; keep each one's first end time.
851	const endKey = `${item.itemId}:${child.childId}:${child.attempt}`;
852	if (settled && !agentEndedAt.has(endKey)) agentEndedAt.set(endKey, time);
853	// Later revisions drop the label; keep the last one seen.
854	if (child.label?.trim()) agentTitles.set(endKey, child.label.trim());
855	const title = agentTitles.get(endKey) || `Agent ${index + 1}`;
856	const result = child.failureReason ?? child.failureKind ?? null;
857	const subagent: OrchestrationV2Subagent = {
858	id: base.nodeId,
859	threadId: base.threadId,
860	runId: base.runId,
861	parentNodeId: workflow.nodeId,
862	origin: "provider_native",
863	createdBy: "agent",
864	driver: MUSE_PROVIDER,
865	providerInstanceId: options.instanceId,
866	providerThreadId: base.providerThreadId,
867	childThreadId: null,
868	nativeTaskRef: nativeRef(child.childId),
869	prompt: title,
870	title,
871	model: null,
872	status,
873	result,
874	startedAt: base.startedAt,
875	completedAt: settled ? (agentEndedAt.get(endKey) ?? time) : null,
876	updatedAt: time,
877	};
878	yield* emit({ type: "subagent.updated", driver: MUSE_PROVIDER, subagent });
879	yield* emit({
880	type: "node.updated",
881	driver: MUSE_PROVIDER,
882	node: {
883	id: base.nodeId,
884	threadId: base.threadId,
885	runId: base.runId,
886	parentNodeId: workflow.nodeId,
887	rootNodeId: turn.input.rootNodeId,
888	kind: "subagent",
889	status,
890	countsForRun: false,
891	providerThreadId: base.providerThreadId,
892	providerTurnId: base.providerTurnId,
893	nativeItemRef: subagent.nativeTaskRef,
894	runtimeRequestId: null,
895	checkpointScopeId: null,
896	startedAt: base.startedAt,
897	completedAt: subagent.completedAt,
898	},
899	});
900	yield* emit({
901	type: "turn_item.updated",
902	driver: MUSE_PROVIDER,
903	turnItem: {
904	...base,
905	nativeItemRef: subagent.nativeTaskRef,
906	parentItemId: workflow.id,
907	type: "subagent",
908	status,
909	title,
910	completedAt: subagent.completedAt,
911	subagentId: subagent.id,
912	origin: "provider_native",
913	driver: MUSE_PROVIDER,
914	providerInstanceId: options.instanceId,
915	childThreadId: null,
916	prompt: title,
917	result,
918	},
919	});
920	}
921	});
922	const settleObservedChildren = Effect.fnUntraced(function* (
923	status: "failed" | "cancelled" | "interrupted",
924	) {
925	for (const [itemId, owner] of observedChildren) {
926	const item = owner.items.get(itemId);
927	if (item) yield* publishItem(owner, item, status);
928	}
929	observedChildren.clear();
930	if (thread?.pendingBackgroundTasks?.length) yield* syncBackground();
931	// Only workflow children's requests outlive a turn; they end with the workflow.
932	for (const entry of pending.values()) yield* resolvePending(entry, "cancelled");
933	});
934	const resolvePending = Effect.fnUntraced(function* (
935	entry: PendingRequest,
936	status: "resolved" | "cancelled",
937	) {
938	const time = yield* DateTime.now;
939	entry.request = {
940	...entry.request,
941	...(status === "resolved" ? entry.response : undefined),
942	status,
943	resolvedAt: time,
944	responseCapability: { type: "not_resumable", reason: "This Muse request has ended." },
945	};
946	entry.node = {
947	...entry.node,
948	status: status === "resolved" ? "completed" : "cancelled",
949	completedAt: time,
950	};
951	entry.item = {
952	...entry.item,
953	status: status === "resolved" ? "completed" : "cancelled",
954	completedAt: time,
955	updatedAt: time,
956	};
957	pending.delete(entry.request.id);
958	yield* emit({
959	type: "runtime_request.updated",
960	driver: MUSE_PROVIDER,
961	threadId: input.threadId,
962	runtimeRequest: entry.request,
963	});
964	yield* emit({ type: "node.updated", driver: MUSE_PROVIDER, node: entry.node });
965	yield* emit({ type: "turn_item.updated", driver: MUSE_PROVIDER, turnItem: entry.item });
966	});
967	const finish = Effect.fnUntraced(function* (
968	turn: ActiveTurn,
969	status: "completed" | "cancelled" | "interrupted" | "failed",
970	detail?: string,
971	disposition: "reusable" | "broken" = "reusable",
972	) {
973	if (active !== turn || seenTerminals.has(turn.nativeId)) return;
974	seenTerminals.add(turn.nativeId);
975	const completedAt = yield* DateTime.now;
976	for (const item of turn.items.values()) {
977	// Workflows and subagents outlive a completed turn. A stopped or failed
978	// turn ends them in T3 too, as the orchestrator stops following them.
979	if (
980	(item.kind === "subagent" || item.kind === "workflow") &&
981	item.status === "inProgress" &&
982	status === "completed" &&
983	disposition === "reusable"
984	)
985	continue;
986	if (turn.dirty.has(item.itemId) || item.status === "inProgress")
987	yield* publishItem(turn, item, item.status === "inProgress" ? status : undefined);
988	}
989	turn.dirty.clear();
990	// A workflow child's request is runless and keeps waiting after the turn ends.
991	for (const entry of pending.values())
992	if (entry.request.providerTurnId !== null) yield* resolvePending(entry, "cancelled");
993	turn.providerTurn = { ...turn.providerTurn, status, completedAt };
994	lastProviderTurn = turn.providerTurn;
995	yield* emit({
996	type: "provider_turn.updated",
997	driver: MUSE_PROVIDER,
998	threadId: turn.input.threadId,
999	providerTurn: turn.providerTurn,
1000	});
1001	yield* updateThread({
1002	status: disposition === "broken" ? "error" : "idle",
1003	...(!turn.compact && status === "completed"
1004	? { nativeConversationHeadRef: nativeRef(turn.nativeId) }
1005	: {}),
1006	});
1007	yield* updateSession(disposition === "broken" ? "error" : "ready", detail ?? null);
1008	active = undefined;
1009	if (status === "failed") {
1010	const failure = makeProviderFailure({
1011	class: disposition === "broken" ? "transport_error" : "provider_error",
1012	message: detail ?? "Muse turn failed.",
1013	});
1014	const base = baseItem(turn, `failure:${turn.nativeId}`, completedAt);
1015	yield* emit({
1016	type: "turn_item.updated",
1017	driver: MUSE_PROVIDER,
1018	turnItem: {
1019	...base,
1020	type: "error",
1021	status: "failed",
1022	title: null,
1023	completedAt,
1024	failure,
1025	},
1026	});
1027	yield* emit({
1028	type: "turn.terminal",
1029	driver: MUSE_PROVIDER,
1030	providerThreadId: turn.providerTurn.providerThreadId,
1031	providerTurnId: turn.providerTurn.id,
1032	runOrdinal: turn.input.runOrdinal,
1033	status,
1034	failure,
1035	failureItemOrdinal: base.ordinal,
1036	threadDisposition: disposition,
1037	});
1038	} else {
1039	yield* emit({
1040	type: "turn.terminal",
1041	driver: MUSE_PROVIDER,
1042	providerThreadId: turn.providerTurn.providerThreadId,
1043	providerTurnId: turn.providerTurn.id,
1044	runOrdinal: turn.input.runOrdinal,
1045	status,
1046	failure: null,
1047	threadDisposition: disposition,
1048	});
1049	}
1050	yield* Deferred.succeed(turn.done, undefined);
1051	});
1052	const failHost = Effect.fnUntraced(function* (cause: unknown) {
1053	if (closed || broken) return;
1054	broken = true;
1055	const rootCause = Cause.isCause(cause) ? Cause.squash(cause) : cause;
1056	const failure = isOpenSessionError(rootCause) ? rootCause.cause : rootCause;
1057	const detail = isProtocolError(failure)
1058	? failure.detail
1059	: failure instanceof Error
1060	? failure.message
1061	: "Muse transport failed.";
1062	if (active) yield* finish(active, "failed", detail, "broken");
1063	else {
1064	yield* updateThread({ status: "error" });
1065	yield* updateSession("error", detail);
1066	}
1067	yield* settleObservedChildren("failed");
1068	yield* Queue.fail(events, protocolError(detail, cause));
1069	// Report the failure first; a hung host can take a while to close.
1070	const failed = host;
1071	yield* Effect.tryPromise(() => failed.close()).pipe(Effect.ignore, Effect.forkIn(scope));
1072	});
1073	const owns = (turn: ActiveTurn, turnId: string) =>
1074	turnId === turn.nativeId || turn.joined.has(turnId);
1075	/** Sends T3's decision for a pending approval to Muse. */
1076	const decideApproval = Effect.fnUntraced(function* (
1077	entry: PendingRequest,
1078	decision: ProviderApprovalDecision | undefined,
1079	) {
1080	if (entry.native.type !== "approval") return;
1081	const approval = entry.native.value;
1082	const choice = decision && museApprovalChoices(approval).get(decision);
1083	if (!choice) return yield* protocolError("Muse did not offer this approval decision");
1084	// Recorded first: Muse can settle the approval before it acknowledges the command.
1085	entry.response = { decision };
1086	yield* request("approval/decide", {
1087	approvalId: approval.approvalId,
1088	requirementId: approval.currentRequirementId,
1089	choiceId: choice.choiceId,
1090	}).pipe(Effect.tapError(() => Effect.sync(() => delete entry.response)));
1091	});
1092	/**
1093	* Shows a Muse approval or question as a T3 request. A workflow child's approval
1094	* is runless: the turn that started the workflow may end long before the child
1095	* asks, and T3 keeps a runless request on the thread until Muse settles it.
1096	*/
1097	const publishRequest = Effect.fnUntraced(function* (native: PendingRequest["native"]) {
1098	const child = native.type === "approval" && native.value.subagentOrigin !== undefined;
1099	const turn = child ? undefined : active;
1100	if (
1101	!thread ||
1102	(!child && !turn) ||
1103	native.value.sessionId !== nativeSessionId ||
1104	(turn && native.value.turnId && !owns(turn, native.value.turnId))
1105	)
1106	return;
1107	const nativeId =
1108	native.type === "approval" ? native.value.approvalId : native.value.userInputId;
1109	if (turn?.settledRequests.has(`${native.type}:${nativeId}`)) return;
1110	const previous = [...pending.values()].find(
1111	(entry) =>
1112	entry.native.type === native.type &&
1113	entry.request.nativeRequestRef?.nativeId === nativeId,
1114	);
1115	const providerTurnId = turn?.providerTurn.id ?? null;
1116	const threadId = turn?.input.threadId ?? thread.appThreadId ?? input.threadId;
1117	const requestId =
1118	previous?.request.id ??
1119	(yield* idAllocator.allocate
1120	.runtimeRequest({
1121	driver: MUSE_PROVIDER,
1122	...(providerTurnId ? { providerTurnId } : {}),
1123	nativeRequestId: nativeId,
1124	})
1125	.pipe(
1126	Effect.mapError((cause) => protocolError("Cannot allocate Muse request", cause)),
1127	));
1128	const time = yield* DateTime.now;
1129	const nodeId = idAllocator.derive.approvalNode({ requestId });
1130	const subject = native.type === "approval" ? native.value.subject : undefined;
1131	const kind = !subject
1132	? "user_input"
1133	: subject.kind === "shell"
1134	? "command"
1135	: subject.kind !== "fileAccess"
1136	? "permission"
1137	: subject.access === "read"
1138	? "file-read"
1139	: "file-change";
1140	const runtimeRequest: OrchestrationV2RuntimeRequest = {
1141	id: requestId,
1142	nodeId,
1143	providerTurnId,
1144	nativeRequestRef: nativeRef(nativeId),
1145	kind,
1146	status: "pending",
1147	responseCapability: { type: "live", providerSessionId: input.providerSessionId },
1148	createdAt: previous?.request.createdAt ?? time,
1149	resolvedAt: null,
1150	};
1151	const node: OrchestrationV2ExecutionNode = {
1152	id: nodeId,
1153	threadId,
1154	runId: turn?.input.runId ?? null,
1155	parentNodeId: turn?.input.rootNodeId ?? null,
1156	rootNodeId: turn?.input.rootNodeId ?? nodeId,
1157	kind: native.type === "approval" ? "approval_request" : "user_input_request",
1158	status: "waiting",
1159	countsForRun: false,
1160	providerThreadId: thread.id,
1161	providerTurnId,
1162	nativeItemRef: nativeRef(nativeId),
1163	runtimeRequestId: requestId,
1164	checkpointScopeId: null,
1165	startedAt: runtimeRequest.createdAt,
1166	completedAt: null,
1167	};
1168	const latestTurn = active?.providerTurn ?? lastProviderTurn;
1169	const base = {
1170	...(turn
1171	? baseItem(turn, `request:${nativeId}`, time)
1172	: {
1173	threadId,
1174	runId: null,
1175	providerThreadId: thread.id,
1176	providerTurnId: null,
1177	nativeItemRef: nativeRef(nativeId),
1178	parentItemId: null,
1179	// Sorts after the latest turn's items.
1180	ordinal: previous?.item.ordinal ?? ((latestTurn?.ordinal ?? 0) + 1) * 100,
1181	startedAt: runtimeRequest.createdAt,
1182	updatedAt: time,
1183	}),
1184	id: idAllocator.derive.approvalTurnItem({ requestId }),
1185	nodeId,
1186	status: "waiting" as const,
1187	title: null,
1188	completedAt: null,
1189	};
1190	const item: OrchestrationV2TurnItem =
1191	native.type === "approval"
1192	? {
1193	...base,
1194	type: "approval_request",
1195	requestId,
1196	requestKind: kind === "user_input" ? "command" : kind,
1197	prompt:
1198	native.value.subject.command ??
1199	native.value.subject.path ??
1200	native.value.subject.host ??
1201	native.value.subject.target ??
1202	native.value.toolName ??
1203	"Muse requests permission",
1204	options: museApprovalOptions(native.value),
1205	}
1206	: {
1207	...base,
1208	type: "user_input_request",
1209	requestId,
1210	questions: native.value.questions.map((question) => ({
1211	id: question.id,
1212	header: question.header || "Question",
1213	question: question.question,
1214	options: question.options.map((option) => ({
1215	label: option.label,
1216	description: option.description || option.label,
1217	})),
1218	multiSelect: question.selection.mode === "multiple",
1219	allowCustomAnswer: true,
1220	required: true,
1221	})),
1222	};
1223	const entry: PendingRequest = { native, request: runtimeRequest, node, item };
1224	pending.set(requestId, entry);
1225	yield* emit({
1226	type: "runtime_request.updated",
1227	driver: MUSE_PROVIDER,
1228	threadId,
1229	runtimeRequest,
1230	});
1231	yield* emit({ type: "node.updated", driver: MUSE_PROVIDER, node });
1232	yield* emit({ type: "turn_item.updated", driver: MUSE_PROVIDER, turnItem: item });
1233	yield* updateSession("waiting");
1234	// Muse does not give workflow children the session's allowAll mode, so in full
1235	// access T3 approves for them. If Muse rejects that, the request waits for the user.
1236	if (child && input.runtimePolicy.runtimeMode === "full-access")
1237	yield* decideApproval(entry, "accept").pipe(Effect.ignore, Effect.forkIn(scope));
1238	});
1239	/** Approval and question events; `turn` is undefined for a workflow child's approval. */
1240	const handleRequestEvent = Effect.fnUntraced(function* (
1241	method: string,
1242	params: Record<string, unknown>,
1243	turn: ActiveTurn | undefined,
1244	) {
1245	switch (method) {
1246	case "approval/requested":
1247	case "approval/updated":
1248	if (
1249	method === "approval/updated" &&
1250	![...pending.values()].some(
1251	(entry) =>
1252	entry.native.type === "approval" &&
1253	entry.native.value.approvalId === params.approvalId,
1254	)
1255	)
1256	return;
1257	return yield* publishRequest({
1258	type: "approval",
1259	value: yield* decode(MuseApproval, params),
1260	});
1261	case "userInput/requested":
1262	return yield* publishRequest({
1263	type: "question",
1264	value: yield* decode(MuseUserInput, params),
1265	});
1266	case "approval/resolved":
1267	case "userInput/settled": {
1268	const id = method === "approval/resolved" ? params.approvalId : params.userInputId;
1269	if (typeof id === "string")
1270	turn?.settledRequests.add(
1271	`${method === "approval/resolved" ? "approval" : "question"}:${id}`,
1272	);
1273	const entry = [...pending.values()].find(
1274	(candidate) => candidate.request.nativeRequestRef?.nativeId === id,
1275	);
1276	if (entry) {
1277	// An approval settled outside T3 still shows Muse's own decision.
1278	const native =
1279	method === "approval/resolved" && typeof params.decision === "string"
1280	? museApprovalDecision({ decision: params.decision, scope: "once" })
1281	: undefined;
1282	if (!entry.response && native) entry.response = { decision: native };
1283	yield* resolvePending(entry, "resolved");
1284	}
1285	if (!pending.size) yield* updateSession(active ? "running" : "ready");
1286	}
1287	}
1288	});
1289	const handleNotification = Effect.fnUntraced(function* (method: string, data: unknown) {
1290	const params = yield* decode(recordSchema, data);
1291	if (params.sessionId !== nativeSessionId) return;
1292	if (method === "view/gap")
1293	return yield* protocolError("Muse delivery gap requires session recovery");
1294	if (method === "session/contextUsage") {
1295	const usage = yield* decode(MuseContextUsage, params);
1296	const currentTurn = active?.providerTurn ?? lastProviderTurn;
1297	const updatedAt = DateTime.formatIso(yield* DateTime.now);
1298	const tokenUsage = {
1299	...currentTurn?.tokenUsage,
1300	usedTokens: usage.usedTokens,
1301	maxTokens: usage.windowTokens ?? currentTurn?.tokenUsage?.maxTokens ?? undefined,
1302	updatedAt,
1303	};
1304	yield* updateThread({ contextUsage: tokenUsage });
1305	if (currentTurn) {
1306	const updated = { ...currentTurn, tokenUsage };
1307	if (active) active.providerTurn = updated;
1308	else lastProviderTurn = updated;
1309	yield* emit({
1310	type: "provider_turn.updated",
1311	driver: MUSE_PROVIDER,
1312	providerTurn: updated,
1313	});
1314	}
1315	return;
1316	}
1317	const itemEvent = ["item/started", "item/updated", "item/completed"].includes(method)
1318	? (yield* decode(MuseItemEvent, params)).item
1319	: undefined;
1320	if (itemEvent) {
1321	const owner = observedChildren.get(itemEvent.itemId);
1322	if (owner && owner !== active && itemEvent.turnId === owner.nativeId) {
1323	const previous = owner.items.get(itemEvent.itemId);
1324	if (!previous || itemEvent.revision > previous.revision) {
1325	owner.items.set(itemEvent.itemId, itemEvent);
1326	yield* publishItem(owner, itemEvent);
1327	}
1328	return;
1329	}
1330	}
1331	// A workflow child's approval comes under the child's own run id, during a turn
1332	// or after it, so it skips turn routing.
1333	if (params.subagentOrigin !== undefined && method.startsWith("approval/"))
1334	return yield* handleRequestEvent(method, params, undefined);
1335	// Muse starts a turn on its own when a workflow finishes, to report its result.
1336	// During a run, that run shows it. Otherwise hold it and ask the orchestrator for
1337	// a run; that run takes the held events, unless a user turn takes them first.
1338	if (
1339	method === "turn/started" &&
1340	typeof params.turnId === "string" &&
1341	params.turnId !== active?.nativeId &&
1342	params.turnId !== wake?.nativeId
1343	) {
1344	if (active && !active.compact) active.joined.add(params.turnId);
1345	else if (!active && !wake) {
1346	const held = { nativeId: params.turnId, events: [] };
1347	wake = held;
1348	const notification = backgroundWorkNotification(finishedBackground.splice(0));
1349	if (thread?.appThreadId && options.continuationRequests)
1350	yield* options.continuationRequests.offer({
1351	threadId: thread.appThreadId,
1352	providerThreadId: thread.id,
1353	driver: MUSE_PROVIDER,
1354	detail: null,
1355	...(notification ? { notification } : {}),
1356	dispatchIfCurrent: (dispatch) =>
1357	wake === held ? Effect.map(dispatch, Option.some) : Effect.succeed(Option.none()),
1358	clearIfCurrent: () =>
1359	Effect.sync(() => {
1360	if (wake === held) wake = undefined;
1361	}),
1362	});
1363	}
1364	}
1365	if (wake && active?.nativeId !== wake.nativeId) {
1366	const turnId = typeof params.turnId === "string" ? params.turnId : itemEvent?.turnId;
1367	if (turnId === wake.nativeId) {
1368	wake.events.push([method, data]);
1369	return;
1370	}
1371	}
1372	const turn = active;
1373	if (!turn) return;
1374	if (typeof params.turnId === "string" && !owns(turn, params.turnId) && !turn.compact)
1375	return;
1376	switch (method) {
1377	case "item/started":
1378	case "item/updated":
1379	case "item/completed": {
1380	const item = itemEvent!;
1381	if (
1382	item.turnId &&
1383	!owns(turn, item.turnId) &&
1384	!(turn.compact && item.kind === "compaction")
1385	)
1386	return;
1387	const previous = turn.items.get(item.itemId);
1388	if (previous && previous.revision >= item.revision) return;
1389	turn.items.set(item.itemId, item);
1390	turn.dirty.delete(item.itemId);
1391	yield* publishItem(turn, item);
1392	if (turn.compact && item.kind === "compaction" && item.status !== "inProgress") {
1393	yield* finish(
1394	turn,
1395	item.status === "completed" && (!item.outcome || item.outcome === "compacted")
1396	? "completed"
1397	: item.status === "cancelled"
1398	? "cancelled"
1399	: "failed",
1400	item.failureReason ?? item.reason,
1401	);
1402	}
1403	break;
1404	}
1405	case "item/delta": {
1406	const delta = yield* decode(MuseDelta, params);
1407	const previous = turn.items.get(delta.itemId);
1408	if (!previous || previous.status !== "inProgress") return;
1409	const field = delta.field ?? "text";
1410	let item = previous;
1411	if (field === "text") item = { ...previous, text: (previous.text ?? "") + delta.delta };
1412	else if (field === "output" || field === "visibleOutput")
1413	item = { ...previous, visibleOutput: (previous.visibleOutput ?? "") + delta.delta };
1414	else if (field.startsWith("summary.")) {
1415	const index = Number(field.slice(8));
1416	if (Number.isSafeInteger(index) && index >= 0 && index < 100) {
1417	const summary = [...(previous.summary ?? [])];
1418	summary[index] = (summary[index] ?? "") + delta.delta;
1419	item = { ...previous, summary };
1420	}
1421	}
1422	if (item === previous) return;
1423	turn.items.set(item.itemId, item);
1424	turn.dirty.add(item.itemId);
1425	if (!turn.flushScheduled) {
1426	turn.flushScheduled = true;
1427	yield* Effect.sleep(50).pipe(
1428	Effect.andThen(Queue.offer(inbox, { type: "flush", turn })),
1429	Effect.forkIn(scope),
1430	);
1431	}
1432	break;
1433	}
1434	case "turn/completed": {
1435	if (turn.compact) return;
1436	const result = yield* decode(MuseTurnCompleted, params);
1437	// A joined Muse turn ending does not end this run; its own turn does.
1438	if (result.turnId !== turn.nativeId) break;
1439	if (result.usage) {
1440	const usage = result.usage;
1441	turn.providerTurn = {
1442	...turn.providerTurn,
1443	turnTokenUsage: {
1444	usageScope: "main_agent",
1445	usageStatus: "complete",
1446	hasSubagents: [...turn.items.values()].some((item) => item.kind === "subagent"),
1447	inputTokens: turn.providerTurn.turnTokenUsage?.inputTokens ?? usage.inputTokens,
1448	outputTokens: usage.outputTokens,
1449	cachedInputTokens: usage.cacheReadTokens ?? usage.cachedTokens,
1450	reasoningTokens: usage.reasoningTokens,
1451	...(usage.cacheWriteTokens === undefined
1452	? {}
1453	: { cacheCreationTokens: usage.cacheWriteTokens }),
1454	},
1455	};
1456	}
1457	const status =
1458	result.terminal === "completed"
1459	? "completed"
1460	: result.terminal === "cancelled"
1461	? turn.interruptRequested
1462	? "interrupted"
1463	: "cancelled"
1464	: "failed";
1465	yield* finish(turn, status, result.error?.message ?? result.reason);
1466	break;
1467	}
1468	case "approval/requested":
1469	case "approval/updated":
1470	case "approval/resolved":
1471	case "userInput/requested":
1472	case "userInput/settled":
1473	yield* handleRequestEvent(method, params, turn);
1474	break;
1475	case "session/tokenUsage": {
1476	const usage = yield* decode(MuseTokenUsageEvent, params);
1477	const previous = turn.providerTurn.turnTokenUsage;
1478	const contextUsage = thread?.contextUsage;
1479	turn.providerTurn = {
1480	...turn.providerTurn,
1481	...(contextUsage
1482	? {
1483	tokenUsage: {
1484	...turn.providerTurn.tokenUsage,
1485	usedTokens: contextUsage.usedTokens,
1486	inputTokens: usage.usage.inputTokens,
1487	outputTokens: usage.usage.outputTokens,
1488	cachedInputTokens: usage.usage.cacheReadTokens ?? usage.usage.cachedTokens,
1489	reasoningOutputTokens: usage.usage.reasoningTokens,
1490	updatedAt: DateTime.formatIso(yield* DateTime.now),
1491	},
1492	}
1493	: {}),
1494	turnTokenUsage: {
1495	usageScope: "main_agent",
1496	usageStatus: "partial",
1497	hasSubagents: [...turn.items.values()].some((item) => item.kind === "subagent"),
1498	inputTokens: (previous?.inputTokens ?? 0) + usage.promptTokens,
1499	outputTokens: (previous?.outputTokens ?? 0) + usage.usage.outputTokens,
1500	cachedInputTokens:
1501	(previous?.cachedInputTokens ?? 0) +
1502	(usage.usage.cacheReadTokens ?? usage.usage.cachedTokens),
1503	...(usage.usage.cacheWriteTokens === undefined &&
1504	previous?.cacheCreationTokens === undefined
1505	? {}
1506	: {
1507	cacheCreationTokens:
1508	(previous?.cacheCreationTokens ?? 0) + (usage.usage.cacheWriteTokens ?? 0),
1509	}),
1510	reasoningTokens: (previous?.reasoningTokens ?? 0) + usage.usage.reasoningTokens,
1511	},
1512	};
1513	yield* emit({
1514	type: "provider_turn.updated",
1515	driver: MUSE_PROVIDER,
1516	providerTurn: turn.providerTurn,
1517	});
1518	break;
1519	}
1520	case "session/todoListChanged": {
1521	const todo = yield* decode(MuseTodoList, params);
1522	const steps: ReadonlyArray<OrchestrationV2PlanStep> = todo.items
1523	.filter((item) => item.text.trim() && item.status !== "cancelled")
1524	.map((item, index) => ({
1525	id: `${turn.nativeId}:todo:${index}`,
1526	text: item.text.trim(),
1527	status:
1528	item.status === "completed"
1529	? "completed"
1530	: item.status === "inProgress"
1531	? "running"
1532	: "pending",
1533	}));
1534	const time = yield* DateTime.now;
1535	const base = baseItem(turn, "todo", time);
1536	const planId =
1537	turn.todoPlanId ??
1538	(yield* idAllocator.allocate
1539	.plan({
1540	threadId: turn.input.threadId,
1541	runId: turn.input.runId,
1542	driver: MUSE_PROVIDER,
1543	})
1544	.pipe(
1545	Effect.mapError((cause) =>
1546	protocolError("Cannot allocate Muse todo list", cause),
1547	),
1548	));
1549	turn.todoPlanId = planId;
1550	yield* emit({
1551	type: "plan.updated",
1552	driver: MUSE_PROVIDER,
1553	plan: {
1554	id: planId,
1555	threadId: turn.input.threadId,
1556	runId: turn.input.runId,
1557	nodeId: base.nodeId,
1558	kind: "todo_list",
1559	status: steps.every((step) => step.status === "completed") ? "completed" : "active",
1560	steps,
1561	},
1562	});
1563	yield* emit({
1564	type: "node.updated",
1565	driver: MUSE_PROVIDER,
1566	node: {
1567	id: base.nodeId,
1568	threadId: base.threadId,
1569	runId: base.runId,
1570	parentNodeId: turn.input.rootNodeId,
1571	rootNodeId: turn.input.rootNodeId,
1572	kind: "todo_list",
1573	status: "completed",
1574	countsForRun: false,
1575	providerThreadId: base.providerThreadId,
1576	providerTurnId: base.providerTurnId,
1577	nativeItemRef: null,
1578	runtimeRequestId: null,
1579	checkpointScopeId: null,
1580	startedAt: base.startedAt,
1581	completedAt: time,
1582	},
1583	});
1584	yield* emit({
1585	type: "turn_item.updated",
1586	driver: MUSE_PROVIDER,
1587	turnItem: {
1588	...base,
1589	nativeItemRef: null,
1590	type: "todo_list",
1591	status: "completed",
1592	title: null,
1593	completedAt: time,
1594	planId,
1595	steps,
1596	},
1597	});
1598	break;
1599	}
1600	case "turn/retryScheduled": {
1601	const retry = yield* decode(MuseTurnRetryScheduled, params);
1602	const time = yield* DateTime.now;
1603	yield* emit({
1604	type: "turn_item.updated",
1605	driver: MUSE_PROVIDER,
1606	turnItem: {
1607	...baseItem(turn, `retry:${retry.nextAttempt}`, time),
1608	type: "system_notice",
1609	status: "completed",
1610	title: "Muse retry",
1611	completedAt: time,
1612	message: `Muse is retrying this turn (attempt ${retry.nextAttempt} of ${retry.maxAttempts}): ${retry.reason}`,
1613	},
1614	});
1615	break;
1616	}
1617	}
1618	});
1619	yield* Stream.fromQueue(inbox).pipe(
1620	Stream.runForEach((entry) =>
1621	Effect.gen(function* () {
1622	if (entry.type === "flush") {
1623	const turn = entry.turn;
1624	turn.flushScheduled = false;
1625	if (active !== turn) return;
1626	for (const id of turn.dirty) {
1627	const item = turn.items.get(id);
1628	if (item) yield* publishItem(turn, item);
1629	}
1630	turn.dirty.clear();
1631	} else if (entry.epoch === hostEpoch) {
1632	if (entry.type === "failure") yield* failHost(entry.cause);
1633	else {
1634	if (options.nativeEventLogger && entry.method !== "item/delta")
1635	yield* options.nativeEventLogger.write(
1636	{ provider: "muse", method: entry.method, params: entry.params },
1637	input.threadId,
1638	);
1639	yield* handleNotification(entry.method, entry.params).pipe(
1640	Effect.catch((cause) =>
1641	INFORMATIONAL_NOTIFICATIONS.has(entry.method)
1642	? Effect.logWarning(`Skipped an unreadable Muse ${entry.method}`, cause)
1643	: failHost(cause),
1644	),
1645	);
1646	}
1647	}
1648	}).pipe(eventPermit.withPermits(1)),
1649	),
1650	Effect.forkIn(scope),
1651	);
1652	const launchHost = Effect.fnUntraced(function* () {
1653	const epoch = ++hostEpoch;
1654	const mcpSession = yield* mcpSessions.read(input.threadId);
1655	const environment = ScriptProgress.withScriptProgressEnvironment(
1656	McpProviderSession.withAgentDeviceEnvironment(options.environment, mcpSession),
1657	yield* scriptProgress.environmentForThread(input.threadId),
1658	);
1659	const launch = yield* agentScope.wrap({
1660	command: options.settings.binaryPath || "muse",
1661	args: [],
1662	name: "muse",
1663	threadId: input.threadId,
1664	env: environment,
1665	});
1666	const created = yield* Effect.acquireRelease(
1667	createMuseSdkHostEffect(
1668	{
1669	binaryPath: launch.command,
1670	launchArgs: launch.args,
1671	cwd,
1672	environment,
1673	runtimeMode: input.runtimePolicy.runtimeMode,
1674	},
1675	options.createHost,
1676	).pipe(
1677	Effect.mapError(
1678	(error) =>
1679	new ProviderAdapter.ProviderAdapterOpenSessionError({
1680	driver: MUSE_PROVIDER,
1681	providerSessionId: input.providerSessionId,
1682	cause: error.cause,
1683	}),
1684	),
1685	),
1686	(created) =>
1687	Effect.gen(function* () {
1688	if (host === created) {
1689	closed = true;
1690	hostEpoch++;
1691	}
1692	yield* Effect.tryPromise(() => created.close()).pipe(Effect.ignore);
1693	}),
1694	{ interruptible: true },
1695	).pipe(Effect.provideService(Scope.Scope, scope));
1696	host = created;
1697	created.connection.onNotification((notification) => {
1698	Queue.offerUnsafe(inbox, {
1699	type: "notification",
1700	method: notification.method,
1701	params: notification.params,
1702	epoch,
1703	});
1704	});
1705	created.connection.onProtocolError((cause) => {
1706	Queue.offerUnsafe(inbox, { type: "failure", cause, epoch });
1707	});
1708	// These are presentation receipts only. The matching notifications drive the UI,
1709	// and decisions travel as `approval/decide` and `userInput/answer` commands.
1710	created.connection.onServerRequest(async (request) => {
1711	if (request.method === "approval/request" || request.method === "userInput/request")
1712	return {};
1713	throw new Error(`Unsupported Muse server request: ${request.method}`);
1714	});
1715	void created.connection.closed.then(() => {
1716	if (!closed && epoch === hostEpoch)
1717	Queue.offerUnsafe(inbox, {
1718	type: "failure",
1719	cause: new Error("Muse connection closed unexpectedly"),
1720	epoch,
1721	});
1722	});
1723	void created.exited.then((exit) => {
1724	if (!closed && epoch === hostEpoch) {
1725	const reason = created
1726	.stderrTail?.()
1727	.findLast((line) => line.trim())
1728	?.trim();
1729	Queue.offerUnsafe(inbox, {
1730	type: "failure",
1731	cause: new Error(
1732	`Muse exited unexpectedly (${exit.code ?? exit.signal})${reason ? `: ${reason}` : ""}`,
1733	),
1734	epoch,
1735	});
1736	}
1737	});
1738	});
1739	yield* Effect.addFinalizer(() =>
1740	Effect.gen(function* () {
1741	closed = true;
1742	hostEpoch++;
1743	if (active) yield* finish(active, "cancelled", "Muse session closed", "broken");
1744	yield* settleObservedChildren("cancelled");
1745	yield* Queue.shutdown(inbox);
1746	yield* Queue.shutdown(events);
1747	}).pipe(eventPermit.withPermits(1)),
1748	);
1749	yield* launchHost();
1750
1751	const register = Effect.fnUntraced(function* (
1752	args: ProviderAdapter.ProviderAdapterV2EnsureThreadInput,
1753	) {
1754	if (broken || closed)
1755	return yield* protocolError("Muse host is unavailable; reopen the session");
1756	if (active)
1757	return yield* protocolError("Cannot change Muse sessions during an active turn");
1758	const existing = args.existingProviderThread;
1759	if (
1760	existing &&
1761	(existing.driver !== MUSE_PROVIDER || existing.providerInstanceId !== options.instanceId)
1762	)
1763	return yield* protocolError("Muse thread belongs to another provider instance");
1764	const requestedId = existing
1765	? (existing.nativeThreadRef?.nativeId ?? undefined)
1766	: input.initialNativeThreadId;
1767	if (
1768	thread &&
1769	(!requestedId || requestedId === nativeSessionId) &&
1770	thread.appThreadId === args.threadId
1771	)
1772	return thread;
1773	if (thread) return yield* protocolError("Open a new host to attach another Muse session");
1774	let missingNativeSession = false;
1775	return yield* Effect.gen(function* () {
1776	nativeSessionId = requestedId ?? host.connection.mintCommandId();
1777	const mcpSession = yield* mcpSessions.read(args.threadId);
1778	if (mcpSession && !host.initializeResult.grantedCapabilities.includes("sessionMcp"))
1779	return yield* protocolError(
1780	"Update Muse Code to a version that supports session MCP servers",
1781	);
1782	const config = mcpSession
1783	? {
1784	mcpServers: {
1785	// Muse defaults to "required", which fails the whole run when T3's
1786	// tools cannot be reached. The agent should still work without them.
1787	"t3-code": {
1788	transport: "streamableHttp",
1789	mode: "optional",
1790	url: mcpSession.endpoint,
1791	headers: { Authorization: mcpSession.authorizationHeader },
1792	},
1793	},
1794	}
1795	: undefined;
1796	const startModel = yield* resolveModel(args.modelSelection);
1797	// Resume selects it too, since the thread may predate it or its mode changed.
1798	// `session/setPermissionProfile` would need Muse's experimental API.
1799	const permissionProfile = musePermissionProfile(args.runtimePolicy.runtimeMode);
1800	const result = yield* request(
1801	requestedId ? "session/resume" : "session/start",
1802	requestedId
1803	? // T3 already has the transcript; skip Muse's history payload.
1804	{ excludeItems: true, ...(config ? { config } : {}), permissionProfile }
1805	: {
1806	workspaceRoot: cwd,
1807	...(config ? { config } : {}),
1808	...(startModel ? { modelId: startModel } : {}),
1809	providerId: "meta",
1810	approvalMode: museApprovalMode(args.runtimePolicy.runtimeMode),
1811	permissionProfile,
1812	},
1813	).pipe(
1814	Effect.catch((error) => {
1815	missingNativeSession =
1816	requestedId !== undefined &&
1817	error.payload instanceof MspError &&
1818	(error.payload.kind === "sessionNotFound" || error.payload.kind === "notFound");
1819	return Effect.fail(error);
1820	}),
1821	Effect.flatMap((value) => decode(MuseSessionResult, value)),
1822	);
1823	if (result.session.sessionId !== nativeSessionId)
1824	return yield* protocolError("Muse returned a different session identity");
1825	if (requestedId)
1826	yield* request("session/setApprovalMode", {
1827	mode: museApprovalMode(args.runtimePolicy.runtimeMode),
1828	});
1829	// A resumed session keeps the model it last ran with; the next turn
1830	// switches it when the selection differs.
1831	const createdAt = yield* DateTime.now;
1832	thread = existing
1833	? {
1834	...existing,
1835	providerSessionId: input.providerSessionId,
1836	nativeThreadRef: nativeRef(result.session.sessionId),
1837	status: "idle",
1838	updatedAt: createdAt,
1839	}
1840	: {
1841	id: idAllocator.derive.providerThread({
1842	driver: MUSE_PROVIDER,
1843	providerInstanceId: options.instanceId,
1844	nativeThreadId: nativeSessionId,
1845	}),
1846	driver: MUSE_PROVIDER,
1847	providerInstanceId: options.instanceId,
1848	providerSessionId: input.providerSessionId,
1849	appThreadId: args.threadId,
1850	ownerNodeId: null,
1851	nativeThreadRef: nativeRef(nativeSessionId),
1852	nativeConversationHeadRef: null,
1853	status: "idle",
1854	firstRunOrdinal: null,
1855	lastRunOrdinal: null,
1856	handoffIds: [],
1857	forkedFrom: null,
1858	createdAt,
1859	updatedAt: createdAt,
1860	};
1861	session = {
1862	...session,
1863	model: result.session.modelId ?? args.modelSelection.model,
1864	};
1865	yield* emit({
1866	type: "provider_thread.updated",
1867	driver: MUSE_PROVIDER,
1868	providerThread: thread,
1869	});
1870	return thread;
1871	}).pipe(
1872	Effect.onError((cause) =>
1873	missingNativeSession ? Effect.void : eventPermit.withPermits(1)(failHost(cause)),
1874	),
1875	);
1876	});
1877	const prompt = Effect.fnUntraced(function* (
1878	message: ProviderAdapter.ProviderAdapterV2TurnMessage,
1879	) {
1880	const parts: SendUserTurnOptions<unknown>["input"] = [];
1881	const text = providerMessageTextWithAttachmentPaths({
1882	text: message.text,
1883	attachments: message.attachments,
1884	resolveAttachmentPath: providerHost.resolveAttachmentPath,
1885	});
1886	if (text) parts.push({ type: "text", text });
1887	for (const attachment of message.attachments)
1888	if (isProviderNativeImageAttachment(attachment)) {
1889	const path = providerHost.resolveAttachmentPath(attachment);
1890	if (!path) return yield* protocolError("Muse image attachment is missing");
1891	const bytes = yield* fileSystem
1892	.readFile(path)
1893	.pipe(Effect.mapError((cause) => protocolError("Cannot read Muse image", cause)));
1894	parts.push({
1895	type: "image",
1896	base64Data: Buffer.from(bytes).toString("base64"),
1897	mediaType: attachment.mimeType,
1898	});
1899	}
1900	if (!parts.length) return yield* protocolError("Muse needs text or an image");
1901	return parts;
1902	});
1903	const validateThread = (candidate: OrchestrationV2ProviderThread) =>
1904	thread !== undefined &&
1905	candidate.driver === MUSE_PROVIDER &&
1906	candidate.providerInstanceId === options.instanceId &&
1907	candidate.appThreadId === thread.appThreadId &&
1908	candidate.nativeThreadRef?.nativeId === nativeSessionId;
1909	const selectionEffort = Effect.fnUntraced(function* (selection: ModelSelection) {
1910	if (selection.instanceId !== options.instanceId)
1911	return yield* protocolError("Model selection belongs to another Muse instance");
1912	const catalog = options.modelCatalog ? yield* options.modelCatalog : [];
1913	const model = yield* resolveModel(selection);
1914	const selected = getModelSelectionStringOptionValue(selection, "reasoningEffort");
1915	// With no saved choice, the model's own default applies.
1916	return resolveMuseReasoningEffort(
1917	catalog.find((entry) => entry.slug === model)?.capabilities ?? museModelCapabilities(),
1918	selected,
1919	);
1920	});
1921	const start = Effect.fnUntraced(function* (
1922	turnInput: ProviderAdapter.ProviderAdapterV2TurnInput,
1923	compact = false,
1924	) {
1925	if (!validateThread(turnInput.providerThread) || broken || closed)
1926	return yield* protocolError("Muse thread is not attached to this host");
1927	if (active)
1928	return yield* protocolError(
1929	"Muse already has an active turn; queue through the orchestrator",
1930	);
1931	if (turnInput.runtimePolicy.interactionMode === "plan")
1932	return yield* protocolError("Muse does not support dedicated Plan mode");
1933	if (turnInput.runtimePolicy.runtimeMode !== input.runtimePolicy.runtimeMode)
1934	return yield* protocolError("Runtime policy changes require a fresh Muse host");
1935	// A continuation run takes the turn Muse already started; it sends no prompt.
1936	const continuation =
1937	turnInput.message.createdBy === "agent" &&
1938	turnInput.message.creationSource === "provider";
1939	const adopted = continuation ? wake : undefined;
1940	const effort = continuation ? undefined : yield* selectionEffort(turnInput.modelSelection);
1941	const parts = compact || continuation ? [] : yield* prompt(turnInput.message);
1942	const targetModel = continuation
1943	? undefined
1944	: yield* resolveModel(turnInput.modelSelection);
1945	if (targetModel && session.model !== targetModel) {
1946	yield* request("session/setModel", {
1947	model: { modelId: targetModel, providerId: "meta" },
1948	});
1949	session = { ...session, model: targetModel };
1950	}
1951	const nativeId = adopted?.nativeId ?? host.connection.mintCommandId();
1952	const startedAt = yield* DateTime.now;
1953	const providerTurn: OrchestrationV2ProviderTurn = {
1954	id: idAllocator.derive.providerTurn({
1955	driver: MUSE_PROVIDER,
1956	nativeTurnId: `${options.instanceId}:${nativeSessionId}:${nativeId}`,
1957	}),
1958	providerThreadId: turnInput.providerThread.id,
1959	nodeId: turnInput.rootNodeId,
1960	runAttemptId: turnInput.attemptId,
1961	nativeTurnRef: compact
1962	? { ...nativeRef(nativeId), strength: "weak" }
1963	: nativeRef(nativeId),
1964	ordinal: turnInput.providerTurnOrdinal,
1965	status: "running",
1966	startedAt,
1967	completedAt: null,
1968	};
1969	const turn: ActiveTurn = {
1970	input: turnInput,
1971	providerTurn,
1972	nativeId,
1973	joined: new Set(),
1974	items: new Map(),
1975	ordinals: new Map(),
1976	started: new Map(),
1977	dirty: new Set(),
1978	settledRequests: new Set(),
1979	done: yield* Deferred.make<void>(),
1980	nextOrdinal: turnInput.providerTurnOrdinal * 100 + 1,
1981	flushScheduled: false,
1982	interruptRequested: false,
1983	compact,
1984	};
1985	yield* Effect.gen(function* () {
1986	thread = turnInput.providerThread;
1987	active = turn;
1988	yield* emit({
1989	type: "provider_turn.updated",
1990	driver: MUSE_PROVIDER,
1991	threadId: turnInput.threadId,
1992	providerTurn,
1993	});
1994	yield* updateThread({
1995	status: "active",
1996	firstRunOrdinal: thread.firstRunOrdinal ?? turnInput.runOrdinal,
1997	lastRunOrdinal: turnInput.runOrdinal,
1998	});
1999	yield* updateSession("running");
2000	// A user turn takes a held Muse turn: Muse runs it first, so this run shows it
2001	// (and its approvals), and the continuation asked for it is no longer needed.
2002	const held = wake;
2003	wake = undefined;
2004	if (continuation && !held) return yield* finish(turn, "completed");
2005	if (held && !continuation) turn.joined.add(held.nativeId);
2006	for (const [method, data] of held?.events ?? [])
2007	yield* handleNotification(method, data).pipe(Effect.catch(failHost));
2008	}).pipe(eventPermit.withPermits(1));
2009	if (continuation) return;
2010	yield* Effect.gen(function* () {
2011	if (compact) {
2012	const result = yield* request("session/compact", {}, true, nativeId).pipe(
2013	Effect.flatMap((value) => decode(MuseCompactResult, value)),
2014	);
2015	if (result.status === "noop")
2016	yield* eventPermit.withPermits(1)(
2017	finish(turn, "failed", result.reason ?? "Muse has no context to compact."),
2018	);
2019	else if (result.status !== "accepted")
2020	return yield* protocolError(
2021	`Muse returned an unsupported compaction status: ${result.status}`,
2022	);
2023	} else {
2024	const result = yield* request(
2025	"turn/start",
2026	{
2027	input: [
2028	{
2029	type: "text",
2030	text: buildRuntimeInstructions({
2031	harness: "Muse Code",
2032	model: session.model ?? undefined,
2033	reasoningEffort: effort,
2034	}),
2035	},
2036	...parts,
2037	],
2038	displayText: turnInput.message.text || "Image attachment",
2039	// A resumed session keeps the root it was created with; pin it to
2040	// this thread's current checkout so edits land where T3 tracks them.
2041	workspaceRoots: [workspaceRoot],
2042	ifBusy: "queue",
2043	...(effort ? { reasoningEffort: effort } : {}),
2044	},
2045	true,
2046	nativeId,
2047	).pipe(Effect.flatMap((value) => decode(MuseTurnStartResult, value)));
2048	if (result.turnId !== nativeId) {
2049	yield* eventPermit.withPermits(1)(
2050	failHost(new Error("Muse admitted an unowned turn")),
2051	);
2052	return yield* protocolError("Muse admitted the turn under an unexpected identity");
2053	}
2054	}
2055	}).pipe(
2056	Effect.onError((cause) => {
2057	const error = Cause.squash(cause);
2058	const detail = isProtocolError(error)
2059	? error.detail
2060	: error instanceof Error
2061	? error.message
2062	: "Muse could not start the turn.";
2063	return eventPermit.withPermits(1)(finish(turn, "failed", detail));
2064	}),
2065	);
2066	});
2067	// Only messages are read back; T3 owns turn and item history.
2068	const snapshot = Effect.fnUntraced(function* (
2069	current: OrchestrationV2ProviderThread,
2070	): Effect.fn.Return<
2071	ProviderAdapter.ProviderAdapterV2ThreadSnapshot,
2072	ProviderAdapter.ProviderAdapterV2Error
2073	> {
2074	const result = yield* request("session/read", { excludeItems: false }, false).pipe(
2075	Effect.flatMap((value) => decode(MuseSessionResult, value)),
2076	);
2077	if (result.session.sessionId !== nativeSessionId)
2078	return yield* protocolError("Muse snapshot belongs to a different session");
2079	const items = result.history?.items ?? result.history?.snapshot?.state.items ?? [];
2080	const now = yield* DateTime.now;
2081	return {
2082	providerThread: current,
2083	providerTurns: [],
2084	messages: items.flatMap((item): OrchestrationV2ConversationMessage[] =>
2085	item.kind === "agentMessage" || item.kind === "userMessage"
2086	? [
2087	{
2088	id: idAllocator.derive.messageFromProviderItem({
2089	driver: MUSE_PROVIDER,
2090	nativeItemId: `${options.instanceId}:${current.id}:${item.turnId ?? "history"}:${item.itemId}`,
2091	}),
2092	threadId: current.appThreadId ?? input.threadId,
2093	runId: null,
2094	nodeId: null,
2095	role: item.kind === "agentMessage" ? "assistant" : "user",
2096	text: item.text ?? "",
2097	attachments: [],
2098	streaming: item.status === "inProgress",
2099	createdBy: item.kind === "agentMessage" ? "agent" : "user",
2100	creationSource: "provider",
2101	createdAt: now,
2102	updatedAt: now,
2103	},
2104	]
2105	: [],
2106	),
2107	runtimeRequests: [...pending.values()].map((entry) => entry.request),
2108	providerPayload: items,
2109	};
2110	});
2111	const runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime = {
2112	instanceId: options.instanceId,
2113	driver: MUSE_PROVIDER,
2114	providerSessionId: input.providerSessionId,
2115	providerSession: session,
2116	events: Stream.fromQueue(events),
2117	// Workflows and subagents run in this session's own `muse` host.
2118	backgroundTasksEndWithProcess: true,
2119	// A running workflow or a held Muse turn keeps the host: idle release must wait.
2120	hasPendingBackgroundWork: Effect.sync(
2121	() => observedChildren.size > 0 || wake !== undefined,
2122	),
2123	ensureThread: (args) =>
2124	register(args).pipe(
2125	commands.withPermits(1),
2126	Effect.mapError(
2127	(cause) =>
2128	new ProviderAdapter.ProviderAdapterEnsureThreadError({
2129	driver: MUSE_PROVIDER,
2130	threadId: args.threadId,
2131	cause,
2132	}),
2133	),
2134	),
2135	resumeThread: (args) =>
2136	register({
2137	threadId: args.threadId ?? args.providerThread.appThreadId ?? input.threadId,
2138	modelSelection: args.modelSelection ?? input.modelSelection,
2139	runtimePolicy: args.runtimePolicy ?? input.runtimePolicy,
2140	existingProviderThread: args.providerThread,
2141	}).pipe(
2142	commands.withPermits(1),
2143	Effect.mapError(
2144	(cause) =>
2145	new ProviderAdapter.ProviderAdapterResumeThreadError({
2146	driver: MUSE_PROVIDER,
2147	providerSessionId: input.providerSessionId,
2148	providerThreadId: args.providerThread.id,
2149	cause,
2150	}),
2151	),
2152	),
2153	startTurn: (args) =>
2154	start(args).pipe(
2155	commands.withPermits(1),
2156	Effect.mapError(
2157	(cause) =>
2158	new ProviderAdapter.ProviderAdapterTurnStartError({
2159	driver: MUSE_PROVIDER,
2160	threadId: args.threadId,
2161	providerThreadId: args.providerThread.id,
2162	runId: args.runId,
2163	cause,
2164	}),
2165	),
2166	),
2167	compactThread: (args) => start(args, true).pipe(commands.withPermits(1)),
2168	steerTurn: (args) =>
2169	Effect.gen(function* () {
2170	const turn = active;
2171	if (
2172	!turn ||
2173	turn.providerTurn.id !== args.providerTurnId ||
2174	!validateThread(args.providerThread) ||
2175	turn.compact
2176	)
2177	return yield* protocolError("This Muse turn cannot be steered");
2178	const parts = yield* prompt(args.message);
2179	const result = yield* request("turn/steer", {
2180	expectedTurnId: turn.nativeId,
2181	input: parts,
2182	}).pipe(Effect.flatMap((value) => decode(MuseTurnStartResult, value)));
2183	if (result.turnId !== turn.nativeId)
2184	return yield* protocolError("Muse steering targeted a different turn");
2185	}).pipe(
2186	commands.withPermits(1),
2187	Effect.mapError(
2188	(cause) =>
2189	new ProviderAdapter.ProviderAdapterSteerRunError({
2190	driver: MUSE_PROVIDER,
2191	providerThreadId: args.providerThread.id,
2192	providerTurnId: args.providerTurnId,
2193	cause,
2194	}),
2195	),
2196	),
2197	interruptTurn: (args) =>
2198	Effect.gen(function* () {
2199	const turn = active;
2200	if (
2201	!turn ||
2202	turn.providerTurn.id !== args.providerTurnId ||
2203	!validateThread(args.providerThread)
2204	)
2205	return yield* protocolError("This Muse turn is no longer active");
2206	turn.interruptRequested = true;
2207	if (turn.compact) {
2208	yield* eventPermit.withPermits(1)(
2209	Effect.gen(function* () {
2210	if (active !== turn) return;
2211	hostEpoch++;
2212	yield* Effect.tryPromise(() => host.close()).pipe(
2213	Effect.mapError((cause) => protocolError("Cannot stop Muse compaction", cause)),
2214	Effect.onError(failHost),
2215	);
2216	broken = true;
2217	yield* finish(turn, "interrupted", undefined, "broken");
2218	yield* settleObservedChildren("interrupted");
2219	yield* updateSession("stopped");
2220	yield* Queue.end(events);
2221	}),
2222	);
2223	return;
2224	}
2225	// Muse runs a joined turn first, so Stop must end it too.
2226	for (const joined of turn.joined)
2227	yield* request("turn/interrupt", { turnId: joined }).pipe(Effect.ignore);
2228	yield* request("turn/interrupt", { turnId: turn.nativeId });
2229	yield* Deferred.await(turn.done).pipe(
2230	Effect.timeout(options.requestTimeoutMs ?? 30_000),
2231	Effect.mapError((cause) => protocolError("Muse did not confirm interruption", cause)),
2232	Effect.onError((cause) => eventPermit.withPermits(1)(failHost(cause))),
2233	);
2234	}).pipe(
2235	commands.withPermits(1),
2236	Effect.mapError(
2237	(cause) =>
2238	new ProviderAdapter.ProviderAdapterInterruptError({
2239	driver: MUSE_PROVIDER,
2240	providerThreadId: args.providerThread.id,
2241	providerTurnId: args.providerTurnId,
2242	cause,
2243	}),
2244	),
2245	),
2246	respondToRuntimeRequest: (args) =>
2247	Effect.gen(function* () {
2248	const entry = pending.get(args.requestId);
2249	if (!entry) return yield* protocolError("This Muse request is no longer pending");
2250	if (entry.native.type === "approval") yield* decideApproval(entry, args.decision);
2251	else {
2252	const questionRequest = entry.native.value;
2253	const answers = [];
2254	for (const question of questionRequest.questions) {
2255	const raw = yield* decode(responseAnswerSchema, args.answers?.[question.id]);
2256	const values = typeof raw === "string" ? [raw] : [...new Set(raw)];
2257	const selected = values.filter((value) =>
2258	question.options.some((option) => option.label === value),
2259	);
2260	const freeText = values.filter((value) => !selected.includes(value)).join("\n");
2261	if (freeText.length > 500 || (!freeText && !selected.length))
2262	return yield* protocolError(
2263	"Muse needs a nonempty answer with at most 500 custom characters",
2264	);
2265	if (
2266	(question.selection.mode === "single" && selected.length > 1) ||
2267	(selected.length > 0 &&
2268	(selected.length < (question.selection.minSelections ?? 0) ||
2269	selected.length > (question.selection.maxSelections ?? Infinity)))
2270	)
2271	return yield* protocolError("Muse question selection count is invalid");
2272	answers.push({
2273	questionId: question.id,
2274	...(selected.length
2275	? question.selection.mode === "single"
2276	? { selectedLabel: selected[0]! }
2277	: { selectedLabels: selected }
2278	: {}),
2279	...(freeText ? (selected.length ? { note: freeText } : { freeText }) : {}),
2280	});
2281	}
2282	if (args.answers) entry.response = { answers: args.answers };
2283	yield* request("userInput/answer", {
2284	userInputId: questionRequest.userInputId,
2285	answers,
2286	}).pipe(Effect.tapError(() => Effect.sync(() => delete entry.response)));
2287	}
2288	}).pipe(
2289	Effect.mapError(
2290	(cause) =>
2291	new ProviderAdapter.ProviderAdapterRuntimeRequestResponseError({
2292	driver: MUSE_PROVIDER,
2293	requestId: args.requestId,
2294	cause,
2295	}),
2296	),
2297	),
2298	readThreadSnapshot: (args) =>
2299	(!validateThread(args.providerThread)
2300	? protocolError("Snapshot requested for a different Muse thread")
2301	: snapshot(args.providerThread)
2302	).pipe(
2303	commands.withPermits(1),
2304	Effect.mapError(
2305	(cause) =>
2306	new ProviderAdapter.ProviderAdapterReadThreadSnapshotError({
2307	driver: MUSE_PROVIDER,
2308	providerThreadId: args.providerThread.id,
2309	cause,
2310	}),
2311	),
2312	),
2313	// Muse rejects forks at completed-turn boundaries (InvalidCut), so T3 uses
2314	// portable context handoff for forks and does not rewind Muse conversations.
2315	rollbackThread: (args) =>
2316	Effect.fail(
2317	new ProviderAdapter.ProviderAdapterRollbackThreadError({
2318	driver: MUSE_PROVIDER,
2319	providerThreadId: args.providerThread.id,
2320	checkpointId: args.target.checkpointId,
2321	cause: "Muse Code does not support conversation rollback in T3 Code.",
2322	}),
2323	),
2324	forkThread: (args) =>
2325	Effect.fail(
2326	new ProviderAdapter.ProviderAdapterForkThreadError({
2327	driver: MUSE_PROVIDER,
2328	providerThreadId: args.sourceProviderThread.id,
2329	cause: "Muse Code does not support native forks in T3 Code.",
2330	}),
2331	),
2332	};
2333	return runtime;
2334	}),
2335	});
2336	});