1/**
2 * Records a Pi RPC replay transcript by running a registered fixture's input
3 * through the real orchestrator and the real PiAdapterV2 against a live
4 * `pi --mode rpc`. The adapter's spawner tees every stdin/stdout record.
5 *
6 * node scripts/record-pi-rpc-replay-fixture.ts --scenario simple
7 *
8 * The model is pinned to the fixture's Pi model selection through launch
9 * arguments and `set_model`. Sessions go to a temporary directory, never the
10 * user's Pi session store.
11 */
12import * as NodeServices from "@effect/platform-node/NodeServices";
13import type { ProviderReplayEntry, ProviderReplayTranscript } from "@t3tools/contracts";
14import * as Console from "effect/Console";
15import * as DateTime from "effect/DateTime";
16import * as Duration from "effect/Duration";
17import * as Effect from "effect/Effect";
18import * as FileSystem from "effect/FileSystem";
19import * as Layer from "effect/Layer";
20import * as Path from "effect/Path";
21import * as Schema from "effect/Schema";
22import { TestClock } from "effect/testing";
23import { ChildProcess, ChildProcessSpawner } from "effect/process";
24
25import {
26 makePiRecordingSpawner,
27 PI_REPLAY_ANY,
28 PI_RPC_REPLAY_PROTOCOL,
29 PiOrchestratorReplayHarness,
30} from "../src/orchestration-v2/Adapters/PiAdapterV2.testkit.ts";
31import * as PiAdapterV2Testkit from "../src/orchestration-v2/Adapters/PiAdapterV2.testkit.ts";
32import { PI_PROVIDER } from "@t3tools/provider-pi/server";
33import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
34import { provideDeterministicTestRuntime } from "../src/orchestration-v2/testkit/DeterministicRuntime.ts";
35import { ORCHESTRATOR_REPLAY_FIXTURES } from "../src/orchestration-v2/testkit/fixtures/index.ts";
36import { materializeFixtureInput } from "../src/orchestration-v2/testkit/fixtures/shared.ts";
37import { runOrchestratorV2ProviderReplayScenario } from "../src/orchestration-v2/testkit/ProviderReplayHarness.ts";
38import {
39 checkpointWorkspace,
40 makeCheckpointWorkspace,
41} from "@t3tools/provider-testing/replayWorkspace";
42
43/**
44 * Keeps the user's skills, templates and context files out of the recording.
45 * `--approve` trusts the throwaway workspace so its `.pi/settings.json` loads.
46 */
47const HERMETIC_LAUNCH_ARGS =
48 "--no-extensions --no-skills --no-prompt-templates --no-context-files --approve";
49/**
50 * Pi keeps the last ~20k tokens out of a compaction, so a short fixture
51 * conversation has nothing to compact. A tiny budget makes `/compact` real.
52 */
53const WORKSPACE_PI_SETTINGS = { compaction: { keepRecentTokens: 50 } };
54const CLOCK_TICK = Duration.millis(20);
55
56function readArgValue(name: string): string | undefined {
57 const index = process.argv.indexOf(name);
58 return index === -1 ? undefined : process.argv[index + 1];
59}
60
61const scenario = readArgValue("--scenario");
62const fixture = ORCHESTRATOR_REPLAY_FIXTURES.find((entry) => entry.name === scenario);
63const variant = fixture?.providers.find((provider) => provider.driver === PI_PROVIDER);
64if (fixture === undefined || variant === undefined) {
65 const names = ORCHESTRATOR_REPLAY_FIXTURES.filter((entry) =>
66 entry.providers.some((provider) => provider.driver === PI_PROVIDER),
67 ).map((entry) => entry.name);
68 throw new Error(`Pass --scenario with a fixture that registers Pi: ${names.join(", ")}`);
69}
70
71const piBinary = process.env.T3_PI_BIN ?? "pi";
72const [modelProvider, ...modelId] = variant.modelSelection.model.split("/");
73const launchArgs = `--provider ${modelProvider} --model ${modelId.join("/")} ${HERMETIC_LAUNCH_ARGS}`;
74const home = process.env.HOME ?? "";
75
76// ── normalization ─────────────────────────────────────────────
77
78const UUID_PATTERN = /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/giu;
79const ISO_TIMESTAMP_PATTERN = /\b\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z\b/gu;
80/** Recorded wall-clock times are shifted to start here, keeping their spacing. */
81const NORMALIZED_EPOCH_MS = DateTime.toEpochMillis(DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"));
82
83function escapeRegExp(value: string): string {
84 return value.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&");
85}
86
87function isRecord(value: unknown): value is Record<string, unknown> {
88 return typeof value === "object" && value !== null && !Array.isArray(value);
89}
90
91/**
92 * Makes a live recording portable and reviewable. Session files, the
93 * workspace, the home directory, UUIDs and wall-clock times become stable
94 * placeholders applied consistently to both directions, so values the adapter
95 * echoes back (a session path in `switch_session`, `--fork <file>`) still line
96 * up on replay. Opaque reasoning signatures are dropped, and the model catalog
97 * is trimmed to the pinned model: the adapter only reads context windows there.
98 */
99function normalizeEntries(input: {
100 readonly entries: ReadonlyArray<ProviderReplayEntry>;
101 readonly workspace: string;
102 readonly sessionDir: string;
103 readonly modelSlug: string;
104}): ReadonlyArray<ProviderReplayEntry> {
105 const sessionFiles = new Map<string, string>();
106 const uuids = new Map<string, string>();
107 let timeAnchorMs: number | undefined;
108 const shiftTime = (ms: number): number => {
109 timeAnchorMs ??= ms;
110 return NORMALIZED_EPOCH_MS + (ms - timeAnchorMs);
111 };
112 const sessionFilePattern = new RegExp(
113 `${escapeRegExp(input.sessionDir)}/[^\\s"']+?\\.jsonl`,
114 "gu",
115 );
116 const normalizeString = (value: string): string =>
117 value
118 .replace(sessionFilePattern, (file) => {
119 const existing = sessionFiles.get(file);
120 if (existing !== undefined) return existing;
121 const next = `/pi-sessions/session-${sessionFiles.size + 1}.jsonl`;
122 sessionFiles.set(file, next);
123 return next;
124 })
125 .replaceAll(input.sessionDir, "/pi-sessions")
126 .replaceAll(input.workspace, "<workspace>")
127 .replaceAll(home.length > 1 ? home : "\0", "~")
128 .replace(ISO_TIMESTAMP_PATTERN, (iso) =>
129 DateTime.formatIso(
130 DateTime.makeUnsafe(shiftTime(DateTime.toEpochMillis(DateTime.makeUnsafe(iso)))),
131 ),
132 )
133 .replace(UUID_PATTERN, (uuid) => {
134 const existing = uuids.get(uuid.toLowerCase());
135 if (existing !== undefined) return existing;
136 const next = `00000000-0000-4000-8000-${String(uuids.size + 1).padStart(12, "0")}`;
137 uuids.set(uuid.toLowerCase(), next);
138 return next;
139 });
140 const normalizeValue = (value: unknown): unknown => {
141 if (typeof value === "string") return normalizeString(value);
142 if (Array.isArray(value)) return value.map(normalizeValue);
143 if (!isRecord(value)) return value;
144 // Pi's system prompt lists the machine's skills, docs and cwd, and the
145 // adapter never reads system messages. Keep the shape, drop the text.
146 if (value.role === "system" && isRecord(value.sections)) {
147 return {
148 ...value,
149 sections: Object.fromEntries(
150 Object.keys(value.sections).map((key) => [key, "<system-prompt-section>"]),
151 ),
152 };
153 }
154 return Object.fromEntries(
155 Object.entries(value).map(([key, entry]) => [
156 key,
157 key === "thinkingSignature" || key === "textSignature"
158 ? "<signature>"
159 : key === "timestamp" && typeof entry === "number"
160 ? shiftTime(entry)
161 : normalizeValue(entry),
162 ]),
163 );
164 };
165 const normalizeFrame = (frame: unknown): unknown => {
166 if (!isRecord(frame)) return normalizeValue(frame);
167 const args = frame.args;
168 if (frame.type === "process_start" && Array.isArray(args)) {
169 // The injected extension lives in the server's per-run cache directory.
170 return normalizeValue({
171 ...frame,
172 args: args.map((arg, index) => (args[index - 1] === "--extension" ? PI_REPLAY_ANY : arg)),
173 });
174 }
175 const data = frame.data;
176 if (frame.command === "get_available_models" && isRecord(data) && Array.isArray(data.models)) {
177 return normalizeValue({
178 ...frame,
179 data: {
180 ...data,
181 models: data.models.filter(
182 (model) => isRecord(model) && `${model.provider}/${model.id}` === input.modelSlug,
183 ),
184 },
185 });
186 }
187 return normalizeValue(frame);
188 };
189 return input.entries.map((entry) =>
190 entry.type === "runtime_exit" ? entry : { ...entry, frame: normalizeFrame(entry.frame) },
191 );
192}
193
194const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
195
196function encodeTranscriptNdjson(transcript: ProviderReplayTranscript): string {
197 const { entries, ...metadata } = transcript;
198 return [
199 encodeJson({ type: "transcript_start", ...metadata }),
200 ...entries.map((entry) => encodeJson(entry)),
201 "",
202 ].join("\n");
203}
204
205// ── recording run ─────────────────────────────────────────────
206
207/**
208 * Replay runs on a frozen TestClock; recording shares that runtime so fixture
209 * `advance_clock` steps behave identically, but lets it follow wall time so a
210 * live Pi's timeouts and termination grace periods still elapse.
211 */
212const followWallClock = Effect.gen(function* () {
213 while (true) {
214 yield* TestClock.withLive(Effect.sleep(CLOCK_TICK));
215 yield* TestClock.adjust(CLOCK_TICK);
216 }
217});
218
219const readPiVersion = Effect.gen(function* () {
220 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
221 return (yield* spawner.string(ChildProcess.make(piBinary, ["--version"]))).trim();
222});
223
224const record = Effect.gen(function* () {
225 const fs = yield* FileSystem.FileSystem;
226 const path = yield* Path.Path;
227 const piVersion = yield* readPiVersion;
228 const outputPath = readArgValue("--out") ?? (yield* path.fromFileUrl(variant.transcriptFile));
229 const fixtureInput = fixture.buildInput();
230 const workspace = yield* Effect.promise(() =>
231 makeCheckpointWorkspace(`pi-rpc-record-${fixture.name}`, fixtureInput.workspaceFiles),
232 );
233 yield* fs.makeDirectory(path.join(workspace, ".pi"));
234 yield* fs.writeFileString(
235 path.join(workspace, ".pi", "settings.json"),
236 encodeJson(WORKSPACE_PI_SETTINGS),
237 );
238 const sessionDir = yield* fs.makeTempDirectory({ prefix: `t3-pi-record-sessions-` });
239 yield* Effect.addFinalizer(() =>
240 Effect.all([
241 fs.remove(workspace, { recursive: true, force: true }),
242 fs.remove(sessionDir, { recursive: true, force: true }),
243 ]).pipe(Effect.ignore),
244 );
245
246 const entries: Array<ProviderReplayEntry> = [];
247 const placeholder: ProviderReplayTranscript = {
248 provider: PI_PROVIDER,
249 protocol: PI_RPC_REPLAY_PROTOCOL,
250 version: piVersion,
251 scenario: fixture.name,
252 entries: [],
253 };
254 const layerRecordingSpawner = Layer.effect(
255 ChildProcessSpawner.ChildProcessSpawner,
256 Effect.map(Effect.service(ChildProcessSpawner.ChildProcessSpawner), (live) =>
257 makePiRecordingSpawner(live, entries),
258 ),
259 ).pipe(Layer.provide(NodeServices.layer));
260
261 const materializeInput = materializeFixtureInput({
262 scenario: fixture.name,
263 fixtureInput,
264 driver: variant.driver,
265 modelSelection: variant.modelSelection,
266 }).pipe(Effect.provide(IdAllocator.layer));
267 const continuationOptions =
268 variant.runContinuationWorker === true ? { runContinuationWorker: true } : {};
269
270 yield* Effect.gen(function* () {
271 yield* followWallClock.pipe(Effect.forkScoped);
272 const materialized = yield* materializeInput;
273 yield* runOrchestratorV2ProviderReplayScenario(
274 {
275 name: `${fixture.name}/pi:record`,
276 transcript: placeholder,
277 commands: materialized.commands,
278 steps: materialized.steps,
279 projectionThreadIds: materialized.projectionThreadIds,
280 runtimePolicyOverride: { ...variant.runtimePolicyOverride, cwd: workspace },
281 },
282 {
283 driver: PI_PROVIDER,
284 decodeTranscript: Effect.succeed,
285 makeProviderAdapterRegistryLayer: () =>
286 PiAdapterV2Testkit.layer({
287 scenario: fixture.name,
288 binaryPath: piBinary,
289 launchArgs,
290 environment: [
291 { name: "PI_CODING_AGENT_SESSION_DIR", value: sessionDir, sensitive: false },
292 { name: "PI_SKIP_VERSION_CHECK", value: "1", sensitive: false },
293 ],
294 spawner: layerRecordingSpawner,
295 }),
296 },
297 continuationOptions,
298 );
299 }).pipe(Effect.scoped, provideDeterministicTestRuntime);
300
301 const transcript = {
302 ...placeholder,
303 metadata: {
304 generatedBy: "live-pi-recorder",
305 piVersion,
306 model: variant.modelSelection.model,
307 launchArgs,
308 },
309 entries: normalizeEntries({
310 entries,
311 workspace,
312 sessionDir,
313 modelSlug: variant.modelSelection.model,
314 }),
315 } satisfies ProviderReplayTranscript;
316
317 // The live projection still holds raw session paths, so the fixture's
318 // assertions run on a replay of the normalized transcript. A recording that
319 // fails them never replaces the existing fixture.
320 yield* Effect.gen(function* () {
321 const replayWorkspace = yield* checkpointWorkspace(fixture.name, fixtureInput.workspaceFiles);
322 const materialized = yield* materializeInput;
323 const replayResult = yield* runOrchestratorV2ProviderReplayScenario(
324 {
325 name: `${fixture.name}/pi:verify`,
326 transcript: yield* PiOrchestratorReplayHarness.decodeTranscript(transcript),
327 commands: materialized.commands,
328 steps: materialized.steps,
329 projectionThreadIds: materialized.projectionThreadIds,
330 runtimePolicyOverride: { ...variant.runtimePolicyOverride, cwd: replayWorkspace },
331 },
332 PiOrchestratorReplayHarness,
333 continuationOptions,
334 );
335 variant.assertOutput(replayResult, transcript);
336 }).pipe(Effect.scoped, provideDeterministicTestRuntime);
337
338 yield* fs.makeDirectory(path.dirname(outputPath), { recursive: true });
339 yield* fs.writeFileString(outputPath, encodeTranscriptNdjson(transcript));
340 yield* Console.log(`Wrote ${transcript.entries.length} Pi RPC replay entries to ${outputPath}`);
341});
342
343await Effect.runPromise(record.pipe(Effect.scoped, Effect.provide(NodeServices.layer)));