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