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