1/**
2 * Muse MSP replay: swaps the host factory `MuseAdapterV2` uses to launch
3 * `muse serve` for one that answers from a recorded transcript. The real
4 * adapter and the real `@muse-code/sdk` `Connection` run unchanged; only the
5 * stdio pipe underneath is replayed.
6 *
7 * Transcript entries are MSP JSON-RPC lines. `expect_outbound` is one line T3
8 * wrote to Muse's stdin and `emit_inbound` is one line Muse wrote to stdout.
9 * Every host begins with a synthetic `host_start` outbound record carrying its
10 * `muse serve` arguments; records of the Nth host (N > 1) carry an `@hN` label
11 * suffix. The SDK numbers its requests, so a recorded request id is rebound to
12 * the id the replaying connection sent. Command and session ids come from the
13 * SDK's minter, so replay mints the ids the recording minted, in order.
14 * `<any>` in an outbound frame matches any value.
15 */
16import * as NodeServices from "@effect/platform-node/NodeServices";
17import { Connection, type DuplexTransport } from "@muse-code/sdk";
18import {
19 ProviderDriverKind,
20 ProviderInstanceId,
21 ProviderReplayEntry,
22 type ProviderReplayEntry as ProviderReplayEntryType,
23 type ProviderReplayTranscript,
24} from "@t3tools/contracts";
25import { MuseSettings } from "@t3tools/provider-muse/settings";
26import * as Effect from "effect/Effect";
27import * as FileSystem from "effect/FileSystem";
28import * as Layer from "effect/Layer";
29import * as Schema from "effect/Schema";
30
31import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost";
32import {
33 museInitializeParams,
34 museServeArgs,
35 museWorkspaceRoot,
36 type MuseSdkHost,
37 type MuseSdkHostOptions,
38} from "@t3tools/provider-muse/testing";
39import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
40import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts";
41import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
42import type { OrchestratorV2ProviderReplayHarness } from "../testkit/ProviderReplayHarness.ts";
43import { materializeReplayTranscriptWorkspace } from "@t3tools/provider-testing/replayTranscript";
44import { makeMuseAdapterV2 } from "@t3tools/provider-muse/server";
45
46export const MUSE_PROVIDER_KIND = "muse";
47export const MUSE_MSP_REPLAY_PROTOCOL = "muse.msp-jsonl";
48const MUSE_REPLAY_ANY = "<any>";
49
50const MuseReplayTranscript = Schema.Struct({
51 provider: Schema.Literal(MUSE_PROVIDER_KIND),
52 protocol: Schema.Literal(MUSE_MSP_REPLAY_PROTOCOL),
53 version: Schema.String,
54 scenario: Schema.String,
55 metadata: Schema.Struct({
56 /** Ids each host's SDK minted, in order: session ids, then one per command. */
57 commandIds: Schema.Array(Schema.Array(Schema.String)),
58 }),
59 entries: Schema.Array(ProviderReplayEntry),
60});
61type MuseReplayTranscript = typeof MuseReplayTranscript.Type;
62const decodeMuseReplayTranscript = Schema.decodeUnknownEffect(MuseReplayTranscript);
63
64class MuseReplayTranscriptDecodeError extends Schema.TaggedError<MuseReplayTranscriptDecodeError>()(
65 "MuseReplayTranscriptDecodeError",
66 { scenario: Schema.optional(Schema.String), cause: Schema.Defect() },
67) {
68 override get message(): string {
69 return `Failed to decode Muse MSP replay transcript for scenario ${this.scenario ?? "<unknown>"}.`;
70 }
71}
72
73class MuseReplayMismatchError extends Schema.TaggedError<MuseReplayMismatchError>()(
74 "MuseReplayMismatchError",
75 {
76 scenario: Schema.String,
77 cursor: Schema.Number,
78 host: Schema.Number,
79 expected: Schema.Unknown,
80 actual: Schema.Unknown,
81 },
82) {
83 override get message(): string {
84 return `Muse replay frame mismatch at cursor ${this.cursor} from host ${this.host} in scenario ${this.scenario}.`;
85 }
86}
87
88class MuseReplayIncompleteError extends Schema.TaggedError<MuseReplayIncompleteError>()(
89 "MuseReplayIncompleteError",
90 {
91 scenario: Schema.String,
92 cursor: Schema.Number,
93 remaining: Schema.Number,
94 nextLabel: Schema.optional(Schema.String),
95 },
96) {
97 override get message(): string {
98 return `Muse replay ended with ${this.remaining} unconsumed entries at cursor ${this.cursor} (next ${this.nextLabel ?? "<none>"}) in scenario ${this.scenario}.`;
99 }
100}
101
102function isRecord(value: unknown): value is Record<string, unknown> {
103 return typeof value === "object" && value !== null && !Array.isArray(value);
104}
105
106function replayValueMatches(expected: unknown, actual: unknown): boolean {
107 if (expected === MUSE_REPLAY_ANY) return true;
108 if (Array.isArray(expected)) {
109 return (
110 Array.isArray(actual) &&
111 expected.length === actual.length &&
112 expected.every((entry, index) => replayValueMatches(entry, actual[index]))
113 );
114 }
115 if (isRecord(expected)) {
116 if (!isRecord(actual)) return false;
117 const keys = new Set([...Object.keys(expected), ...Object.keys(actual)]);
118 return [...keys].every((key) => replayValueMatches(expected[key], actual[key]));
119 }
120 return Object.is(expected, actual);
121}
122
123/** A client request: the SDK's own numbered id plus a method. */
124function clientRequestId(frame: unknown): number | undefined {
125 return isRecord(frame) && typeof frame.method === "string" && typeof frame.id === "number"
126 ? frame.id
127 : undefined;
128}
129
130function withRequestId(expected: unknown, actual: unknown): unknown {
131 const actualId = clientRequestId(actual);
132 return isRecord(expected) && clientRequestId(expected) !== undefined && actualId !== undefined
133 ? { ...expected, id: actualId }
134 : expected;
135}
136
137/** A write may not match past one of these: it would be a turn early. */
138function startsTurn(entry: ProviderReplayEntryType): boolean {
139 if (entry.type !== "expect_outbound" || !isRecord(entry.frame)) return false;
140 const method = entry.frame.method ?? entry.frame.type;
141 return method === "turn/start" || method === "session/compact" || method === "host_start";
142}
143
144function entryHost(entry: ProviderReplayEntryType): number {
145 const label = entry.type === "runtime_exit" ? undefined : entry.label;
146 const match = label === undefined ? null : /@h(\d+)$/u.exec(label);
147 return match === null ? 1 : Number(match[1]);
148}
149
150function entryLabel(entry: ProviderReplayEntryType | undefined): string | undefined {
151 if (entry === undefined) return undefined;
152 return entry.type === "runtime_exit" ? "runtime_exit" : (entry.label ?? entry.type);
153}
154
155/** `turn/start`, `response:turn/start`, `request:approval/request`, or `reply:approval/request`. */
156export function museRecordLabel(
157 frame: unknown,
158 direction: "expect_outbound" | "emit_inbound",
159 requestMethods: ReadonlyMap<unknown, string>,
160 host: number,
161): string | undefined {
162 if (!isRecord(frame)) return undefined;
163 const method = typeof frame.method === "string" ? frame.method : undefined;
164 const base =
165 frame.type === "host_start"
166 ? "host_start"
167 : method !== undefined
168 ? direction === "emit_inbound" && "id" in frame
169 ? `request:${method}`
170 : method
171 : requestMethods.has(frame.id)
172 ? `${direction === "emit_inbound" ? "response" : "reply"}:${requestMethods.get(frame.id)}`
173 : undefined;
174 return base === undefined || host === 1 ? base : `${base}@h${host}`;
175}
176
177/** Async line source the SDK `Connection` reads, fed by the replay controller. */
178class LineQueue implements AsyncIterable<string> {
179 private readonly lines: Array<string> = [];
180 private waiting: ((result: IteratorResult<string>) => void) | undefined;
181 private ended = false;
182
183 push(line: string): void {
184 if (this.ended) return;
185 const waiting = this.waiting;
186 if (waiting) {
187 this.waiting = undefined;
188 waiting({ value: line, done: false });
189 } else this.lines.push(line);
190 }
191
192 end(): void {
193 this.ended = true;
194 this.waiting?.({ value: undefined, done: true });
195 this.waiting = undefined;
196 }
197
198 [Symbol.asyncIterator](): AsyncIterator<string> {
199 return {
200 next: () => {
201 const line = this.lines.shift();
202 if (line !== undefined) return Promise.resolve({ value: line, done: false });
203 if (this.ended) return Promise.resolve({ value: undefined, done: true });
204 return new Promise((resolve) => {
205 this.waiting = resolve;
206 });
207 },
208 };
209 }
210}
211
212interface MuseReplayHost {
213 readonly ordinal: number;
214 readonly incoming: LineQueue;
215 /** Recorded request id -> id the replaying connection sent in its place. */
216 readonly ids: Map<number, number>;
217 readonly minted: Array<string>;
218 readonly exited: Promise<{ readonly code: number | null; readonly signal: null }>;
219 readonly exit: () => void;
220 open: boolean;
221}
222
223/**
224 * Drives every Muse host of one scenario from a single ordered transcript.
225 * Outbound lines must match the recording; the inbound lines that follow them
226 * go to the host that produced them. A write that matches a later outbound
227 * record of the same turn is held until the transcript reaches it, because
228 * independent adapter fibers can reach stdin in a different order than
229 * recorded. A write that matches nothing before the next turn fails at once.
230 */
231class MuseReplayController {
232 private cursor = 0;
233 private failure: unknown = null;
234 private readonly consumed = new Set<number>();
235 private readonly held: Array<{ readonly host: MuseReplayHost; readonly actual: unknown }> = [];
236 private readonly hosts: Array<MuseReplayHost> = [];
237 private readonly transcript: MuseReplayTranscript;
238
239 constructor(transcript: MuseReplayTranscript) {
240 this.transcript = transcript;
241 }
242
243 spawn(args: ReadonlyArray<string>): MuseReplayHost {
244 let exit = () => {};
245 const exited = new Promise<{ readonly code: number | null; readonly signal: null }>(
246 (resolve) => {
247 exit = () => resolve({ code: 0, signal: null });
248 },
249 );
250 const host: MuseReplayHost = {
251 ordinal: this.hosts.length + 1,
252 incoming: new LineQueue(),
253 ids: new Map(),
254 minted: [...(this.transcript.metadata.commandIds[this.hosts.length] ?? [])],
255 exited,
256 exit,
257 open: true,
258 };
259 this.hosts.push(host);
260 this.receive(host, { type: "host_start", args: [...args] });
261 return host;
262 }
263
264 mint(host: MuseReplayHost): string {
265 const id = host.minted.shift();
266 if (id === undefined) {
267 throw new Error(`Muse replay host ${host.ordinal} minted more ids than were recorded.`);
268 }
269 return id;
270 }
271
272 receive(host: MuseReplayHost, actual: unknown): void {
273 if (this.failure !== null) return;
274 const entries = this.transcript.entries;
275 let possible = false;
276 for (let index = this.cursor; index < entries.length && !possible; index += 1) {
277 const entry = entries[index]!;
278 const pending = entry.type === "expect_outbound" && !this.consumed.has(index);
279 possible =
280 pending &&
281 entryHost(entry) === host.ordinal &&
282 replayValueMatches(withRequestId(entry.frame, actual), actual);
283 if (!possible && pending && startsTurn(entry)) break;
284 }
285 if (!possible) {
286 this.fail(
287 new MuseReplayMismatchError({
288 scenario: this.transcript.scenario,
289 cursor: this.cursor,
290 host: host.ordinal,
291 expected: entries[this.cursor] ?? null,
292 actual,
293 }),
294 );
295 return;
296 }
297 this.held.push({ host, actual });
298 this.advance();
299 }
300
301 close(host: MuseReplayHost): void {
302 host.open = false;
303 host.incoming.end();
304 host.exit();
305 }
306
307 assertComplete(): void {
308 if (this.failure !== null) throw this.failure;
309 if (this.cursor !== this.transcript.entries.length || this.held.length > 0) {
310 const nextLabel = entryLabel(this.transcript.entries[this.cursor]);
311 throw new MuseReplayIncompleteError({
312 scenario: this.transcript.scenario,
313 cursor: this.cursor,
314 remaining: this.transcript.entries.length - this.cursor,
315 ...(nextLabel === undefined ? {} : { nextLabel }),
316 });
317 }
318 }
319
320 private advance(): void {
321 let progressed = true;
322 while (progressed && this.failure === null) {
323 progressed = this.drainInbound();
324 for (const [heldIndex, write] of this.held.entries()) {
325 if (this.consumePending(write.host, write.actual)) {
326 this.held.splice(heldIndex, 1);
327 progressed = true;
328 break;
329 }
330 }
331 }
332 }
333
334 private consumePending(host: MuseReplayHost, actual: unknown): boolean {
335 const entries = this.transcript.entries;
336 for (let index = this.cursor; index < entries.length; index += 1) {
337 const entry = entries[index]!;
338 if (entry.type !== "expect_outbound") return false;
339 if (this.consumed.has(index) || entryHost(entry) !== host.ordinal) continue;
340 if (!replayValueMatches(withRequestId(entry.frame, actual), actual)) continue;
341 const recordedId = clientRequestId(entry.frame);
342 const actualId = clientRequestId(actual);
343 if (recordedId !== undefined && actualId !== undefined) host.ids.set(recordedId, actualId);
344 this.consumed.add(index);
345 while (this.consumed.has(this.cursor)) this.cursor += 1;
346 return true;
347 }
348 return false;
349 }
350
351 private drainInbound(): boolean {
352 let emitted = false;
353 while (this.failure === null) {
354 const entry = this.transcript.entries[this.cursor];
355 if (entry?.type === "runtime_exit") {
356 // Muse exited on its own in the recording: end the newest host.
357 this.cursor += 1;
358 emitted = true;
359 const host = this.hosts.at(-1);
360 if (host !== undefined) this.close(host);
361 continue;
362 }
363 if (entry?.type !== "emit_inbound") return emitted;
364 const host = this.hosts[entryHost(entry) - 1];
365 if (host === undefined) return emitted;
366 this.cursor += 1;
367 emitted = true;
368 // Output a host wrote after T3 released it was never observed.
369 if (!host.open) continue;
370 const frame = entry.frame;
371 // Responses carry the client's request id; server requests carry their own.
372 const reboundId =
373 isRecord(frame) && frame.method === undefined && typeof frame.id === "number"
374 ? host.ids.get(frame.id)
375 : undefined;
376 const rebound =
377 isRecord(frame) && reboundId !== undefined ? { ...frame, id: reboundId } : frame;
378 host.incoming.push(`${JSON.stringify(rebound)}\n`);
379 }
380 return emitted;
381 }
382
383 private fail(cause: unknown): void {
384 if (this.failure !== null) return;
385 this.failure = cause;
386 for (const host of this.hosts) host.incoming.end();
387 }
388}
389
390/**
391 * Builds a `MuseSdkHost` over any line transport, doing the same `initialize`
392 * handshake the SDK's spawn path does. Replay and recording both use it.
393 */
394export async function connectMuseTransport(input: {
395 readonly transport: DuplexTransport;
396 readonly readOnly: boolean | undefined;
397 readonly mintCommandId?: () => string;
398 readonly exited: MuseSdkHost["exited"];
399 readonly close: () => Promise<void>;
400}): Promise<MuseSdkHost> {
401 const connection = new Connection(
402 input.transport,
403 input.mintCommandId ? { mintCommandId: input.mintCommandId } : {},
404 );
405 const result = await connection.request("initialize", museInitializeParams(input.readOnly));
406 connection.notify("initialized");
407 const granted = result.grantedCapabilities;
408 return {
409 connection,
410 initializeResult: {
411 grantedCapabilities: Array.isArray(granted)
412 ? granted.filter((capability) => typeof capability === "string")
413 : [],
414 },
415 exited: input.exited,
416 close: input.close,
417 };
418}
419
420function makeReplayCreateHost(controller: MuseReplayController) {
421 return async (options: MuseSdkHostOptions): Promise<MuseSdkHost> => {
422 const host = controller.spawn(museServeArgs(options));
423 return connectMuseTransport({
424 transport: {
425 incoming: host.incoming,
426 write: async (line) => {
427 for (const part of line.split("\n")) {
428 if (part.trim()) controller.receive(host, JSON.parse(part));
429 }
430 },
431 close: () => controller.close(host),
432 },
433 readOnly: options.readOnly,
434 mintCommandId: () => controller.mint(host),
435 exited: host.exited,
436 close: async () => controller.close(host),
437 });
438 };
439}
440
441const DEFAULT_MUSE_SETTINGS = Schema.decodeUnknownSync(MuseSettings)({ enabled: true });
442
443/** Registry with the real Muse adapter on the given host factory. */
444export function layer(input: {
445 readonly scenario: string;
446 readonly createHost: NonNullable<Parameters<typeof makeMuseAdapterV2>[0]["createHost"]>;
447 /** Replay needs none; a live recording passes the host environment. */
448 readonly environment?: NodeJS.ProcessEnv;
449}) {
450 return ProviderAdapterRegistry.layerFromAdaptersEffect(
451 Effect.gen(function* () {
452 return [
453 yield* makeMuseAdapterV2({
454 instanceId: ProviderInstanceId.make(MUSE_PROVIDER_KIND),
455 settings: DEFAULT_MUSE_SETTINGS,
456 environment: input.environment ?? {},
457 createHost: input.createHost,
458 // Same queue the continuation worker drains when the fixture runs it.
459 continuationRequests: yield* ProviderContinuationRequests.ProviderContinuationRequests,
460 }),
461 ];
462 }),
463 ).pipe(
464 Layer.provide(
465 Layer.mergeAll(
466 TestProviderHost.layer().pipe(Layer.provide(NodeServices.layer)),
467 NodeServices.layer,
468 IdAllocator.layer,
469 ),
470 ),
471 );
472}
473
474/**
475 * Fills a recording's `<workspace>` with what the adapter sends on this host:
476 * the canonical path (macOS /var -> /private/var), which `turn/start` names in
477 * museWorkspaceRoot's form (verbatim on Windows).
478 */
479export const materializeMuseReplayWorkspace = Effect.fn("materializeMuseReplayWorkspace")(
480 function* (transcript: ProviderReplayTranscript, workspace: string) {
481 const canonical = yield* FileSystem.FileSystem.pipe(
482 Effect.flatMap((fs) => fs.realPath(workspace)),
483 Effect.provide(NodeServices.layer),
484 );
485 const plain = materializeReplayTranscriptWorkspace(transcript, canonical);
486 const turnStarts = materializeReplayTranscriptWorkspace(
487 transcript,
488 yield* museWorkspaceRoot(canonical),
489 );
490 return {
491 ...plain,
492 entries: plain.entries.map((entry, index) =>
493 entry.type === "expect_outbound" &&
494 isRecord(entry.frame) &&
495 entry.frame.method === "turn/start"
496 ? turnStarts.entries[index]!
497 : entry,
498 ),
499 };
500 },
501);
502
503export const MuseOrchestratorReplayHarness: OrchestratorV2ProviderReplayHarness<
504 MuseReplayTranscript,
505 MuseReplayTranscriptDecodeError
506> = {
507 driver: ProviderDriverKind.make(MUSE_PROVIDER_KIND),
508 decodeTranscript: (transcript: ProviderReplayTranscript) =>
509 decodeMuseReplayTranscript(transcript).pipe(
510 Effect.mapError(
511 (cause) => new MuseReplayTranscriptDecodeError({ scenario: transcript.scenario, cause }),
512 ),
513 ),
514 makeProviderAdapterRegistryLayer: (transcript) =>
515 Layer.unwrap(
516 Effect.gen(function* () {
517 const controller = new MuseReplayController(transcript);
518 yield* Effect.addFinalizer(() => Effect.sync(() => controller.assertComplete()));
519 return layer({
520 scenario: transcript.scenario,
521 createHost: makeReplayCreateHost(controller),
522 });
523 }),
524 ),
525};