1/**
2 * Pi RPC replay: swaps the `ChildProcessSpawner` that `PiAdapterV2` uses to
3 * spawn `pi --mode rpc` for one that answers from a recorded transcript. The
4 * real adapter and `PiRpc` framing run unchanged against recorded Pi output.
5 *
6 * Transcript entries are stdio records. `expect_outbound` is one JSONL record
7 * the adapter wrote to stdin and `emit_inbound` is one record Pi wrote to
8 * stdout. Pi echoes the adapter's request ids, so a recorded id is rebound to
9 * the id the replaying adapter actually sent. `<any>` in an outbound frame
10 * matches any value.
11 *
12 * Every spawned Pi process begins with a synthetic `process_start` outbound
13 * record carrying its argv. Records of the Nth process (N > 1) carry an `@pN`
14 * label suffix, because a native fork runs a short-lived `--fork` process
15 * next to the session process.
16 */
17import * as NodeServices from "@effect/platform-node/NodeServices";
18import {
19 ProviderReplayEntry,
20 type ProviderInstanceEnvironment,
21 type ProviderReplayTranscript,
22 type ProviderReplayEntry as ProviderReplayEntryType,
23} from "@t3tools/contracts";
24import * as Cause from "effect/Cause";
25import * as Effect from "effect/Effect";
26import * as Layer from "effect/Layer";
27import * as Queue from "effect/Queue";
28import * as Schema from "effect/Schema";
29import * as Sink from "effect/Sink";
30import * as Stream from "effect/Stream";
31import { ChildProcess, ChildProcessSpawner } from "effect/process";
32
33import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost";
34import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
35import { ProviderAdapterDriverCreateError } from "@t3tools/provider-core/server/adapterDriver";
36import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts";
37import type { OrchestratorV2ProviderReplayHarness } from "../testkit/ProviderReplayHarness.ts";
38import { PI_PROVIDER, PiAdapterV2Driver } from "@t3tools/provider-pi/server";
39
40export const PI_RPC_REPLAY_PROTOCOL = "pi.rpc-jsonl";
41export const PI_REPLAY_ANY = "<any>";
42
43const PiRpcReplayTranscript = Schema.Struct({
44 provider: Schema.Literal(PI_PROVIDER),
45 protocol: Schema.Literal(PI_RPC_REPLAY_PROTOCOL),
46 version: Schema.String,
47 scenario: Schema.String,
48 metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
49 entries: Schema.Array(ProviderReplayEntry),
50});
51type PiRpcReplayTranscript = typeof PiRpcReplayTranscript.Type;
52const decodePiRpcReplayTranscript = Schema.decodeUnknownEffect(PiRpcReplayTranscript);
53
54class PiReplayTranscriptDecodeError extends Schema.TaggedError<PiReplayTranscriptDecodeError>()(
55 "PiReplayTranscriptDecodeError",
56 {
57 driver: Schema.optional(Schema.String),
58 protocol: Schema.optional(Schema.String),
59 scenario: Schema.optional(Schema.String),
60 cause: Schema.Defect(),
61 },
62) {
63 override get message(): string {
64 return `Failed to decode Pi RPC replay transcript for scenario ${this.scenario ?? "<unknown>"}.`;
65 }
66}
67
68class PiReplayMismatchError extends Schema.TaggedError<PiReplayMismatchError>()(
69 "PiReplayMismatchError",
70 {
71 scenario: Schema.String,
72 cursor: Schema.Number,
73 process: Schema.Number,
74 label: Schema.optional(Schema.String),
75 actualType: Schema.optional(Schema.String),
76 expected: Schema.Unknown,
77 actual: Schema.Unknown,
78 },
79) {
80 override get message(): string {
81 return `Pi replay frame mismatch at cursor ${this.cursor} (expected ${this.label ?? "<none>"}, received ${this.actualType ?? "<untyped>"} from process ${this.process}) in scenario ${this.scenario}.`;
82 }
83}
84
85class PiReplayIncompleteError extends Schema.TaggedError<PiReplayIncompleteError>()(
86 "PiReplayIncompleteError",
87 {
88 scenario: Schema.String,
89 cursor: Schema.Number,
90 remaining: Schema.Number,
91 nextLabel: Schema.optional(Schema.String),
92 },
93) {
94 override get message(): string {
95 return `Pi replay ended with ${this.remaining} unconsumed entries at cursor ${this.cursor} (next ${this.nextLabel ?? "<none>"}) in scenario ${this.scenario}.`;
96 }
97}
98
99/** Label or, failing that, entry type: bounded context for a replay error. */
100function entryLabel(entry: ProviderReplayEntryType | undefined): string | undefined {
101 if (entry === undefined) return undefined;
102 return entry.type === "runtime_exit" ? "runtime_exit" : (entry.label ?? entry.type);
103}
104
105const PiOrchestratorReplayHarnessError = Schema.Union([
106 PiReplayTranscriptDecodeError,
107 ProviderAdapterDriverCreateError,
108]);
109type PiOrchestratorReplayHarnessError = typeof PiOrchestratorReplayHarnessError.Type;
110
111function isRecord(value: unknown): value is Record<string, unknown> {
112 return typeof value === "object" && value !== null && !Array.isArray(value);
113}
114
115function replayValueMatches(expected: unknown, actual: unknown): boolean {
116 if (expected === PI_REPLAY_ANY) return true;
117 if (Array.isArray(expected)) {
118 return (
119 Array.isArray(actual) &&
120 expected.length === actual.length &&
121 expected.every((entry, index) => replayValueMatches(entry, actual[index]))
122 );
123 }
124 if (isRecord(expected)) {
125 if (!isRecord(actual)) return false;
126 const keys = new Set([...Object.keys(expected), ...Object.keys(actual)]);
127 return [...keys].every((key) => replayValueMatches(expected[key], actual[key]));
128 }
129 return Object.is(expected, actual);
130}
131
132/** PiRpc numbers its correlated requests `t3-N` per process. */
133const ADAPTER_REQUEST_ID = /^t3-\d+$/u;
134
135function adapterRequestId(frame: unknown): string | undefined {
136 const id = isRecord(frame) ? frame.id : undefined;
137 return typeof id === "string" && ADAPTER_REQUEST_ID.test(id) ? id : undefined;
138}
139
140/**
141 * Compares a recorded request against the adapter's own request id, so the
142 * correlation Pi recorded holds without pinning counter values. Any other id,
143 * such as the Pi request id an `extension_ui_response` must echo, stays exact.
144 */
145function withRequestId(expected: unknown, actual: unknown): unknown {
146 const actualId = adapterRequestId(actual);
147 return isRecord(expected) && adapterRequestId(expected) !== undefined && actualId !== undefined
148 ? { ...expected, id: actualId }
149 : expected;
150}
151
152/** A write may not match past one of these: it would be a turn early. */
153function startsTurn(entry: ProviderReplayEntryType): boolean {
154 if (entry.type !== "expect_outbound") return false;
155 const type = isRecord(entry.frame) ? entry.frame.type : undefined;
156 return type === "prompt" || type === "compact" || type === "process_start";
157}
158
159function processStartFrame(command: ChildProcess.Command) {
160 const args = ChildProcess.isStandardCommand(command) ? command.args : [];
161 return { type: "process_start", args: [...args] };
162}
163
164/** `response:get_state@p2` belongs to the second spawned process; no suffix is the first. */
165function entryProcess(entry: ProviderReplayEntryType): number {
166 const label = entry.type === "runtime_exit" ? undefined : entry.label;
167 const match = label === undefined ? null : /@p(\d+)$/u.exec(label);
168 return match === null ? 1 : Number(match[1]);
169}
170
171function recordLabel(frame: unknown, process: number): string | undefined {
172 const base = !isRecord(frame)
173 ? undefined
174 : frame.type === "response" && typeof frame.command === "string"
175 ? `response:${frame.command}`
176 : typeof frame.type === "string"
177 ? frame.type
178 : undefined;
179 if (base === undefined || process === 1) return base;
180 return `${base}@p${process}`;
181}
182
183const decodeJsonLine = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
184const encodeJsonLine = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
185
186/** Splits a byte stream into LF-delimited records, as Pi's RPC framing requires. */
187function makeLineSplitter() {
188 let buffer = "";
189 const decoder = new TextDecoder();
190 return (chunk: Uint8Array): ReadonlyArray<string> => {
191 buffer += decoder.decode(chunk, { stream: true });
192 const lines = buffer.split("\n");
193 buffer = lines.pop() ?? "";
194 return lines.map((line) => line.replace(/\r$/u, "")).filter((line) => line.length > 0);
195 };
196}
197
198interface PiReplayProcess {
199 readonly ordinal: number;
200 readonly stdout: Queue.Queue<Uint8Array, Cause.Done>;
201 /** Recorded request id -> id the replaying adapter sent in its place. */
202 readonly ids: Map<string, string>;
203 open: boolean;
204}
205
206/**
207 * Drives every Pi process of one scenario from a single ordered transcript.
208 * Outbound records must match the recording; the inbound records that follow
209 * them are written to the stdout of the process that produced them.
210 *
211 * Independent adapter fibers (skill discovery beside thread setup, a settle
212 * probe beside a steer) can reach stdin in a different order than recorded, so
213 * a write that matches a later outbound record of the same turn is held until
214 * the transcript reaches it. A write that matches nothing before the next
215 * turn-starting record fails immediately.
216 */
217class PiReplayController {
218 private cursor = 0;
219 private failure: unknown = null;
220 private readonly consumed = new Set<number>();
221 private readonly held: Array<{ readonly process: PiReplayProcess; readonly actual: unknown }> =
222 [];
223 private readonly processes: Array<PiReplayProcess> = [];
224 private readonly transcript: PiRpcReplayTranscript;
225
226 constructor(transcript: PiRpcReplayTranscript) {
227 this.transcript = transcript;
228 }
229
230 spawn(
231 command: ChildProcess.Command,
232 stdout: Queue.Queue<Uint8Array, Cause.Done>,
233 ): PiReplayProcess {
234 const process: PiReplayProcess = {
235 ordinal: this.processes.length + 1,
236 stdout,
237 ids: new Map(),
238 open: true,
239 };
240 this.processes.push(process);
241 this.receive(process, processStartFrame(command));
242 return process;
243 }
244
245 receive(process: PiReplayProcess, actual: unknown): void {
246 if (this.failure !== null) return;
247 const entries = this.transcript.entries;
248 let possible = false;
249 for (let index = this.cursor; index < entries.length && !possible; index += 1) {
250 const entry = entries[index]!;
251 const pending = entry.type === "expect_outbound" && !this.consumed.has(index);
252 possible =
253 pending &&
254 entryProcess(entry) === process.ordinal &&
255 replayValueMatches(withRequestId(entry.frame, actual), actual);
256 if (!possible && pending && startsTurn(entry)) break;
257 }
258 if (!possible) {
259 const expected = entries[this.cursor];
260 const label = entryLabel(expected);
261 const actualType = isRecord(actual) ? actual.type : undefined;
262 this.fail(
263 new PiReplayMismatchError({
264 scenario: this.transcript.scenario,
265 cursor: this.cursor,
266 process: process.ordinal,
267 ...(label === undefined ? {} : { label }),
268 ...(typeof actualType === "string" ? { actualType } : {}),
269 expected: expected ?? null,
270 actual,
271 }),
272 );
273 return;
274 }
275 this.held.push({ process, actual });
276 this.advance();
277 }
278
279 /** The adapter released a process: its scope closed. */
280 close(process: PiReplayProcess): void {
281 process.open = false;
282 Queue.endUnsafe(process.stdout);
283 }
284
285 assertComplete(): void {
286 if (this.failure !== null) throw this.failure;
287 if (this.cursor !== this.transcript.entries.length || this.held.length > 0) {
288 const nextLabel = entryLabel(this.transcript.entries[this.cursor]);
289 throw new PiReplayIncompleteError({
290 scenario: this.transcript.scenario,
291 cursor: this.cursor,
292 remaining: this.transcript.entries.length - this.cursor,
293 ...(nextLabel === undefined ? {} : { nextLabel }),
294 });
295 }
296 }
297
298 /** Emits due inbound records and consumes held writes until neither can progress. */
299 private advance(): void {
300 let progressed = true;
301 while (progressed && this.failure === null) {
302 progressed = this.drainInbound();
303 for (const [heldIndex, write] of this.held.entries()) {
304 if (this.consumePending(write.process, write.actual)) {
305 this.held.splice(heldIndex, 1);
306 progressed = true;
307 break;
308 }
309 }
310 }
311 }
312
313 /** Matches a write against the outbound records due before the next inbound one. */
314 private consumePending(process: PiReplayProcess, actual: unknown): boolean {
315 const entries = this.transcript.entries;
316 for (let index = this.cursor; index < entries.length; index += 1) {
317 const entry = entries[index]!;
318 if (entry.type !== "expect_outbound") return false;
319 if (this.consumed.has(index) || entryProcess(entry) !== process.ordinal) continue;
320 if (!replayValueMatches(withRequestId(entry.frame, actual), actual)) continue;
321 const recordedId = adapterRequestId(entry.frame);
322 const actualId = adapterRequestId(actual);
323 if (recordedId !== undefined && actualId !== undefined) {
324 process.ids.set(recordedId, actualId);
325 }
326 this.consumed.add(index);
327 while (this.consumed.has(this.cursor)) this.cursor += 1;
328 return true;
329 }
330 return false;
331 }
332
333 private drainInbound(): boolean {
334 let emitted = false;
335 while (this.failure === null) {
336 const entry = this.transcript.entries[this.cursor];
337 if (entry?.type === "runtime_exit") {
338 // Pi exited on its own in the recording: end the newest process's
339 // stdout, which is the EOF the adapter observed.
340 this.cursor += 1;
341 emitted = true;
342 const process = this.processes.at(-1);
343 if (process !== undefined) this.close(process);
344 continue;
345 }
346 if (entry?.type !== "emit_inbound") return emitted;
347 const process = this.processes[entryProcess(entry) - 1];
348 if (process === undefined) return emitted;
349 this.cursor += 1;
350 emitted = true;
351 // Output a process wrote after the adapter released it was never observed.
352 if (!process.open) continue;
353 const frame = entry.frame;
354 const recordedId = isRecord(frame) ? frame.id : undefined;
355 const reboundId = typeof recordedId === "string" ? process.ids.get(recordedId) : undefined;
356 const rebound =
357 isRecord(frame) && reboundId !== undefined ? { ...frame, id: reboundId } : frame;
358 Queue.offerUnsafe(process.stdout, new TextEncoder().encode(`${encodeJsonLine(rebound)}\n`));
359 }
360 return emitted;
361 }
362
363 private fail(cause: unknown): void {
364 if (this.failure !== null) return;
365 this.failure = cause;
366 for (const process of this.processes) Queue.endUnsafe(process.stdout);
367 }
368}
369
370/** Deliberately outside the valid pid range so PiRpc's group kill never lands. */
371const REPLAY_PID = 999_999_999;
372
373function makePiReplaySpawner(
374 controller: PiReplayController,
375): ChildProcessSpawner.ChildProcessSpawner["Service"] {
376 return ChildProcessSpawner.make((command) =>
377 Effect.gen(function* () {
378 const process = controller.spawn(command, yield* Queue.unbounded<Uint8Array, Cause.Done>());
379 yield* Effect.addFinalizer(() => Effect.sync(() => controller.close(process)));
380 const split = makeLineSplitter();
381 return ChildProcessSpawner.makeHandle({
382 pid: ChildProcessSpawner.ProcessId(REPLAY_PID),
383 exitCode: Effect.never,
384 isRunning: Effect.succeed(true),
385 kill: () => Effect.void,
386 unref: Effect.succeed(Effect.void),
387 stdin: Sink.forEach((chunk: Uint8Array) =>
388 Effect.sync(() => {
389 for (const line of split(chunk)) controller.receive(process, decodeJsonLine(line));
390 }),
391 ),
392 stdout: Stream.fromQueue(process.stdout),
393 stderr: Stream.empty,
394 all: Stream.empty,
395 getInputFd: () => Sink.drain,
396 getOutputFd: () => Stream.empty,
397 });
398 }),
399 );
400}
401
402/**
403 * Registry with the real Pi driver on the given spawner. The recorder passes a
404 * spawner that tees a live `pi`; replay passes the transcript-backed one. The
405 * launch arguments are part of the recorded argv, so replay reuses them.
406 */
407export function layer<E, R>(input: {
408 readonly scenario: string;
409 readonly spawner: Layer.Layer<ChildProcessSpawner.ChildProcessSpawner, E, R>;
410 readonly binaryPath: string;
411 readonly launchArgs: string;
412 readonly environment?: ProviderInstanceEnvironment;
413}) {
414 return ProviderAdapterRegistry.layerFromDrivers({
415 drivers: [PiAdapterV2Driver],
416 configMap: {
417 [PI_PROVIDER]: {
418 driver: PI_PROVIDER,
419 enabled: true,
420 config: { binaryPath: input.binaryPath, launchArgs: input.launchArgs },
421 ...(input.environment === undefined ? {} : { environment: input.environment }),
422 },
423 },
424 }).pipe(
425 Layer.provide(input.spawner),
426 Layer.provide(
427 Layer.mergeAll(
428 TestProviderHost.layer().pipe(Layer.provide(NodeServices.layer)),
429 NodeServices.layer,
430 IdAllocator.layer,
431 ),
432 ),
433 );
434}
435
436function metadataString(transcript: PiRpcReplayTranscript, key: string): string {
437 const value = transcript.metadata?.[key];
438 return typeof value === "string" ? value : "";
439}
440
441export const PiOrchestratorReplayHarness: OrchestratorV2ProviderReplayHarness<
442 PiRpcReplayTranscript,
443 PiOrchestratorReplayHarnessError
444> = {
445 driver: PI_PROVIDER,
446 decodeTranscript: (transcript: ProviderReplayTranscript) =>
447 decodePiRpcReplayTranscript(transcript).pipe(
448 Effect.mapError(
449 (cause) =>
450 new PiReplayTranscriptDecodeError({
451 driver: transcript.provider,
452 protocol: transcript.protocol,
453 scenario: transcript.scenario,
454 cause,
455 }),
456 ),
457 ),
458 makeProviderAdapterRegistryLayer: (transcript) =>
459 layer({
460 scenario: transcript.scenario,
461 binaryPath: "pi",
462 launchArgs: metadataString(transcript, "launchArgs"),
463 spawner: Layer.effect(
464 ChildProcessSpawner.ChildProcessSpawner,
465 Effect.gen(function* () {
466 const controller = new PiReplayController(transcript);
467 yield* Effect.addFinalizer(() => Effect.sync(() => controller.assertComplete()));
468 return makePiReplaySpawner(controller);
469 }),
470 ),
471 }),
472};
473
474// ── recording ─────────────────────────────────────────────────
475
476/**
477 * Wraps a live spawner and tees every Pi process's stdin and stdout records
478 * into `entries` in the order the adapter observed them: writes as the adapter
479 * makes them, reads as the adapter pulls them.
480 */
481export function makePiRecordingSpawner(
482 live: ChildProcessSpawner.ChildProcessSpawner["Service"],
483 entries: Array<ProviderReplayEntryType>,
484): ChildProcessSpawner.ChildProcessSpawner["Service"] {
485 let spawned = 0;
486 return ChildProcessSpawner.make((command) =>
487 Effect.gen(function* () {
488 const handle = yield* live.spawn(command);
489 const process = ++spawned;
490 const record = (type: "expect_outbound" | "emit_inbound", frame: unknown) => {
491 const label = recordLabel(frame, process);
492 entries.push({ type, ...(label === undefined ? {} : { label }), frame });
493 };
494 const recordLines = (
495 type: "expect_outbound" | "emit_inbound",
496 lines: ReadonlyArray<string>,
497 ) => {
498 for (const line of lines) {
499 try {
500 record(type, decodeJsonLine(line));
501 } catch {
502 // PiRpc drops non-JSON stdout lines, so the transcript does too.
503 }
504 }
505 };
506 record("expect_outbound", processStartFrame(command));
507 const splitIn = makeLineSplitter();
508 const splitOut = makeLineSplitter();
509 return ChildProcessSpawner.makeHandle({
510 ...handle,
511 stdin: handle.stdin.pipe(
512 Sink.mapInput((chunk: Uint8Array) => {
513 recordLines("expect_outbound", splitIn(chunk));
514 return chunk;
515 }),
516 ),
517 stdout: handle.stdout.pipe(
518 Stream.tap((chunk) => Effect.sync(() => recordLines("emit_inbound", splitOut(chunk)))),
519 ),
520 });
521 }),
522 );
523}