1// @effect-diagnostics nodeBuiltinImport:off globalTimers:off
2// The SDK's host factory is Promise-based and owns a raw child process, as production's does.
3/**
4 * Records a Muse MSP replay transcript by running a registered fixture's input
5 * through the real orchestrator and the real MuseAdapterV2 against a live
6 * `muse serve`. The host's stdio is teed line by line.
7 *
8 * node scripts/record-muse-msp-replay-fixture.ts --scenario simple
9 *
10 * Uses the machine's own `muse login`. `T3_MUSE_BIN` picks the binary. The
11 * recording replays itself through the fixture's assertions before it is
12 * written, so a run that fails them never replaces the existing transcript.
13 */
14import * as NodeChildProcess from "node:child_process";
15
16import * as NodeServices from "@effect/platform-node/NodeServices";
17import { createUuidV7Mint } from "@muse-code/sdk";
18import type { ProviderReplayEntry, ProviderReplayTranscript } from "@t3tools/contracts";
19import * as Console from "effect/Console";
20import * as Duration from "effect/Duration";
21import * as Effect from "effect/Effect";
22import * as FileSystem from "effect/FileSystem";
23import * as Path from "effect/Path";
24import * as Schema from "effect/Schema";
25import { TestClock } from "effect/testing";
26
27import {
28 connectMuseTransport,
29 layer as museReplayLayer,
30 materializeMuseReplayWorkspace,
31 MUSE_MSP_REPLAY_PROTOCOL,
32 MUSE_PROVIDER_KIND,
33 museRecordLabel,
34 MuseOrchestratorReplayHarness,
35} from "../src/orchestration-v2/Adapters/MuseAdapterV2.testkit.ts";
36import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
37import {
38 makeMuseEnvironment,
39 museLaunch,
40 museServeArgs,
41 museWorkspaceRoot,
42 parseMuseVersion,
43} from "@t3tools/provider-muse/testing";
44import { provideDeterministicTestRuntime } from "../src/orchestration-v2/testkit/DeterministicRuntime.ts";
45import { ORCHESTRATOR_REPLAY_FIXTURES } from "../src/orchestration-v2/testkit/fixtures/index.ts";
46import { materializeFixtureInput } from "../src/orchestration-v2/testkit/fixtures/shared.ts";
47import { runOrchestratorV2ProviderReplayScenario } from "../src/orchestration-v2/testkit/ProviderReplayHarness.ts";
48import {
49 checkpointWorkspace,
50 makeCheckpointWorkspace,
51} from "@t3tools/provider-testing/replayWorkspace";
52
53const CLOCK_TICK = Duration.millis(20);
54
55function readArgValue(name: string): string | undefined {
56 const index = process.argv.indexOf(name);
57 return index === -1 ? undefined : process.argv[index + 1];
58}
59
60const scenario = readArgValue("--scenario");
61const fixture = ORCHESTRATOR_REPLAY_FIXTURES.find((entry) => entry.name === scenario);
62const variant = fixture?.providers.find((provider) => provider.driver === MUSE_PROVIDER_KIND);
63if (fixture === undefined || variant === undefined) {
64 const names = ORCHESTRATOR_REPLAY_FIXTURES.filter((entry) =>
65 entry.providers.some((provider) => provider.driver === MUSE_PROVIDER_KIND),
66 ).map((entry) => entry.name);
67 throw new Error(`Pass --scenario with a fixture that registers Muse: ${names.join(", ")}`);
68}
69
70const museBinary = process.env.T3_MUSE_BIN ?? "muse";
71// Started as T3 starts it, so Windows needs no T3_MUSE_BIN for Muse's `muse.cmd` launcher.
72const museStart = Effect.runSync(museLaunch(museBinary));
73const home = process.env.HOME ?? "";
74
75function isRecord(value: unknown): value is Record<string, unknown> {
76 return typeof value === "object" && value !== null && !Array.isArray(value);
77}
78
79/**
80 * Makes a recording portable: the workspace becomes `<workspace>` and the home
81 * directory `~`, in both directions. Ids, timestamps and model output are kept
82 * as recorded; they are opaque to T3 and carry no secrets.
83 */
84function normalizeEntries(
85 entries: ReadonlyArray<ProviderReplayEntry>,
86 workspaces: ReadonlyArray<string>,
87): ReadonlyArray<ProviderReplayEntry> {
88 const normalizeString = (value: string): string => {
89 let next = value;
90 for (const workspace of workspaces) next = next.replaceAll(workspace, "<workspace>");
91 return home.length > 1 ? next.replaceAll(home, "~") : next;
92 };
93 const normalizeValue = (value: unknown): unknown =>
94 typeof value === "string"
95 ? normalizeString(value)
96 : Array.isArray(value)
97 ? value.map(normalizeValue)
98 : isRecord(value)
99 ? Object.fromEntries(
100 Object.entries(value).map(([key, entry]) => [key, normalizeValue(entry)]),
101 )
102 : value;
103 return entries.map((entry) =>
104 entry.type === "runtime_exit" ? entry : { ...entry, frame: normalizeValue(entry.frame) },
105 );
106}
107
108const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
109
110function encodeTranscriptNdjson(transcript: ProviderReplayTranscript): string {
111 const { entries, ...metadata } = transcript;
112 return [
113 encodeJson({ type: "transcript_start", ...metadata }),
114 ...entries.map((entry) => encodeJson(entry)),
115 "",
116 ].join("\n");
117}
118
119/** Recording follows wall time so a live Muse's own timeouts still elapse. */
120const followWallClock = Effect.gen(function* () {
121 while (true) {
122 yield* TestClock.withLive(Effect.sleep(CLOCK_TICK));
123 yield* TestClock.adjust(CLOCK_TICK);
124 }
125});
126
127/**
128 * Host factory that runs a live `muse serve` and tees every stdio line into
129 * `entries`, in the order the connection writes and reads them.
130 */
131function makeRecordingCreateHost(
132 entries: Array<ProviderReplayEntry>,
133 commandIds: Array<Array<string>>,
134): Parameters<typeof museReplayLayer>[0]["createHost"] {
135 return async (options) => {
136 const hostOrdinal = commandIds.length + 1;
137 const minted: Array<string> = [];
138 commandIds.push(minted);
139 const mint = createUuidV7Mint();
140 const args = museServeArgs(options);
141 const clientRequests = new Map<unknown, string>();
142 const serverRequests = new Map<unknown, string>();
143 const record = (type: "expect_outbound" | "emit_inbound", frame: unknown) => {
144 const label = museRecordLabel(
145 frame,
146 type,
147 type === "emit_inbound" ? clientRequests : serverRequests,
148 hostOrdinal,
149 );
150 entries.push({ type, ...(label === undefined ? {} : { label }), frame });
151 };
152 record("expect_outbound", { type: "host_start", args });
153 // The binary named by T3_MUSE_BIN, so the recorded version matches what ran.
154 const child = NodeChildProcess.spawn(museStart.command, [...museStart.args, ...args], {
155 cwd: options.cwd,
156 env: options.environment ?? makeMuseEnvironment(),
157 stdio: ["pipe", "pipe", "inherit"],
158 });
159 const exited = new Promise<{ code: number | null; signal: null }>((resolve) =>
160 child.once("exit", (code) => resolve({ code, signal: null })),
161 );
162 async function* incoming() {
163 let buffer = "";
164 // utf8 decoding keeps a character split across chunks intact.
165 for await (const chunk of child.stdout.setEncoding("utf8")) {
166 buffer += String(chunk);
167 const lines = buffer.split("\n");
168 buffer = lines.pop() ?? "";
169 for (const line of lines) {
170 if (!line.trim()) continue;
171 const frame: unknown = JSON.parse(line);
172 if (isRecord(frame) && typeof frame.method === "string" && "id" in frame)
173 serverRequests.set(frame.id, frame.method);
174 record("emit_inbound", frame);
175 yield `${line}\n`;
176 }
177 }
178 }
179 const close = async () => {
180 child.stdin.end();
181 const timer = setTimeout(() => child.kill("SIGKILL"), 10_000);
182 await exited;
183 clearTimeout(timer);
184 };
185 return connectMuseTransport({
186 transport: {
187 incoming: incoming(),
188 write: async (line) => {
189 for (const part of line.split("\n")) {
190 if (!part.trim()) continue;
191 const frame: unknown = JSON.parse(part);
192 if (isRecord(frame) && typeof frame.method === "string" && "id" in frame)
193 clientRequests.set(frame.id, frame.method);
194 record("expect_outbound", frame);
195 }
196 await new Promise<void>((resolve, reject) =>
197 child.stdin.write(line, (error) => (error ? reject(error) : resolve())),
198 );
199 },
200 close: () => void close(),
201 },
202 readOnly: options.readOnly,
203 mintCommandId: () => {
204 const id = mint();
205 minted.push(id);
206 return id;
207 },
208 exited,
209 close,
210 });
211 };
212}
213
214const readMuseVersion = Effect.sync(() => {
215 const output = NodeChildProcess.execFileSync(
216 museStart.command,
217 [...museStart.args, "--version"],
218 { encoding: "utf8" },
219 );
220 return parseMuseVersion(output) ?? output.trim();
221});
222
223const record = Effect.gen(function* () {
224 const fs = yield* FileSystem.FileSystem;
225 const path = yield* Path.Path;
226 const museVersion = yield* readMuseVersion;
227 const outputPath = readArgValue("--out") ?? (yield* path.fromFileUrl(variant.transcriptFile));
228 const fixtureInput = fixture.buildInput();
229 const workspace = yield* Effect.promise(() =>
230 makeCheckpointWorkspace(`muse-msp-record-${fixture.name}`, fixtureInput.workspaceFiles),
231 );
232 yield* Effect.addFinalizer(() =>
233 fs.remove(workspace, { recursive: true, force: true }).pipe(Effect.ignore),
234 );
235 const canonicalWorkspace = yield* fs.realPath(workspace);
236
237 const entries: Array<ProviderReplayEntry> = [];
238 const commandIds: Array<Array<string>> = [];
239 const placeholder: ProviderReplayTranscript = {
240 provider: MUSE_PROVIDER_KIND,
241 protocol: MUSE_MSP_REPLAY_PROTOCOL,
242 version: museVersion,
243 scenario: fixture.name,
244 entries: [],
245 };
246 const materializeInput = materializeFixtureInput({
247 scenario: fixture.name,
248 fixtureInput,
249 driver: variant.driver,
250 modelSelection: variant.modelSelection,
251 }).pipe(Effect.provide(IdAllocator.layer));
252 const continuationOptions =
253 variant.runContinuationWorker === true ? { runContinuationWorker: true } : {};
254
255 yield* Effect.gen(function* () {
256 yield* followWallClock.pipe(Effect.forkScoped);
257 const materialized = yield* materializeInput;
258 yield* runOrchestratorV2ProviderReplayScenario(
259 {
260 name: `${fixture.name}/muse:record`,
261 transcript: placeholder,
262 commands: materialized.commands,
263 steps: materialized.steps,
264 projectionThreadIds: materialized.projectionThreadIds,
265 runtimePolicyOverride: { ...variant.runtimePolicyOverride, cwd: workspace },
266 },
267 {
268 driver: MuseOrchestratorReplayHarness.driver,
269 decodeTranscript: Effect.succeed,
270 makeProviderAdapterRegistryLayer: () =>
271 museReplayLayer({
272 scenario: fixture.name,
273 createHost: makeRecordingCreateHost(entries, commandIds),
274 environment: makeMuseEnvironment(),
275 }),
276 },
277 continuationOptions,
278 );
279 }).pipe(Effect.scoped, provideDeterministicTestRuntime);
280
281 const transcript = {
282 ...placeholder,
283 metadata: {
284 generatedBy: "live-muse-recorder",
285 model: variant.modelSelection.model,
286 commandIds,
287 },
288 // The verbatim root first: it contains the canonical path.
289 entries: normalizeEntries(entries, [
290 yield* museWorkspaceRoot(canonicalWorkspace),
291 canonicalWorkspace,
292 workspace,
293 ]),
294 } satisfies ProviderReplayTranscript;
295
296 yield* Effect.gen(function* () {
297 const replayWorkspace = yield* checkpointWorkspace(fixture.name, fixtureInput.workspaceFiles);
298 const materialized = yield* materializeInput;
299 const replayResult = yield* runOrchestratorV2ProviderReplayScenario(
300 {
301 name: `${fixture.name}/muse:verify`,
302 transcript: yield* MuseOrchestratorReplayHarness.decodeTranscript(
303 yield* materializeMuseReplayWorkspace(transcript, replayWorkspace),
304 ),
305 commands: materialized.commands,
306 steps: materialized.steps,
307 projectionThreadIds: materialized.projectionThreadIds,
308 runtimePolicyOverride: { ...variant.runtimePolicyOverride, cwd: replayWorkspace },
309 },
310 MuseOrchestratorReplayHarness,
311 continuationOptions,
312 );
313 variant.assertOutput(replayResult, transcript);
314 }).pipe(Effect.scoped, provideDeterministicTestRuntime);
315
316 yield* fs.makeDirectory(path.dirname(outputPath), { recursive: true });
317 yield* fs.writeFileString(outputPath, encodeTranscriptNdjson(transcript));
318 yield* Console.log(`Wrote ${transcript.entries.length} Muse MSP replay entries to ${outputPath}`);
319});
320
321await Effect.runPromise(record.pipe(Effect.scoped, Effect.provide(NodeServices.layer)));