1import * as NodeServices from "@effect/platform-node/NodeServices";
2import {
3 ProviderReplayEntry,
4 ProviderInstanceId,
5 type ResolvedProviderInstanceEnvironment,
6 type ProviderReplayTranscript,
7 type ProviderReplayEntry as ProviderReplayEntryType,
8} from "@t3tools/contracts";
9import * as Cause from "effect/Cause";
10import * as Effect from "effect/Effect";
11import * as Layer from "effect/Layer";
12import * as Queue from "effect/Queue";
13import * as Schema from "effect/Schema";
14import * as Sink from "effect/Sink";
15import * as Stream from "effect/Stream";
16import { ChildProcess, ChildProcessSpawner } from "effect/process";
17
18import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost";
19import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
20import { ProviderAdapterDriverCreateError } from "@t3tools/provider-core/server/adapterDriver";
21import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts";
22import type { OrchestratorV2ProviderReplayHarness } from "../testkit/ProviderReplayHarness.ts";
23import { PI_PROVIDER, PiAdapterV2Driver } from "@t3tools/provider-pi/server";
24
25export const PI_RPC_REPLAY_PROTOCOL = "pi.rpc-jsonl";
26export const PI_REPLAY_ANY = "<any>";
27
28const PiRpcReplayTranscript = Schema.Struct({
29 provider: Schema.Literal(PI_PROVIDER),
30 protocol: Schema.Literal(PI_RPC_REPLAY_PROTOCOL),
31 version: Schema.String,
32 scenario: Schema.String,
33 metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
34 entries: Schema.Array(ProviderReplayEntry),
35});
36type PiRpcReplayTranscript = typeof PiRpcReplayTranscript.Type;
37const decodePiRpcReplayTranscript = Schema.decodeUnknownEffect(PiRpcReplayTranscript);
38
39class PiReplayTranscriptDecodeError extends Schema.TaggedError<PiReplayTranscriptDecodeError>()(
40 "PiReplayTranscriptDecodeError",
41 {
42 driver: Schema.optional(Schema.String),
43 protocol: Schema.optional(Schema.String),
44 scenario: Schema.optional(Schema.String),
45 cause: Schema.Defect(),
46 },
47) {
48 override get message(): string {
49 return `Failed to decode Pi RPC replay transcript for scenario ${this.scenario ?? "<unknown>"}.`;
50 }
51}
52
53class PiReplayMismatchError extends Schema.TaggedError<PiReplayMismatchError>()(
54 "PiReplayMismatchError",
55 {
56 scenario: Schema.String,
57 cursor: Schema.Number,
58 process: Schema.Number,
59 label: Schema.optional(Schema.String),
60 actualType: Schema.optional(Schema.String),
61 expected: Schema.Unknown,
62 actual: Schema.Unknown,
63 },
64) {
65 override get message(): string {
66 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}.`;
67 }
68}
69
70class PiReplayIncompleteError extends Schema.TaggedError<PiReplayIncompleteError>()(
71 "PiReplayIncompleteError",
72 {
73 scenario: Schema.String,
74 cursor: Schema.Number,
75 remaining: Schema.Number,
76 nextLabel: Schema.optional(Schema.String),
77 },
78) {
79 override get message(): string {
80 return `Pi replay ended with ${this.remaining} unconsumed entries at cursor ${this.cursor} (next ${this.nextLabel ?? "<none>"}) in scenario ${this.scenario}.`;
81 }
82}
83
84function entryLabel(entry: ProviderReplayEntryType | undefined): string | undefined {
85 if (entry === undefined) return undefined;
86 return entry.type === "runtime_exit" ? "runtime_exit" : (entry.label ?? entry.type);
87}
88
89const PiOrchestratorReplayHarnessError = Schema.Union([
90 PiReplayTranscriptDecodeError,
91 ProviderAdapterDriverCreateError,
92]);
93type PiOrchestratorReplayHarnessError = typeof PiOrchestratorReplayHarnessError.Type;
94
95function isRecord(value: unknown): value is Record<string, unknown> {
96 return typeof value === "object" && value !== null && !Array.isArray(value);
97}
98
99function replayValueMatches(expected: unknown, actual: unknown): boolean {
100 if (expected === PI_REPLAY_ANY) return true;
101 if (Array.isArray(expected)) {
102 return (
103 Array.isArray(actual) &&
104 expected.length === actual.length &&
105 expected.every((entry, index) => replayValueMatches(entry, actual[index]))
106 );
107 }
108 if (isRecord(expected)) {
109 if (!isRecord(actual)) return false;
110 const keys = new Set([...Object.keys(expected), ...Object.keys(actual)]);
111 return [...keys].every((key) => replayValueMatches(expected[key], actual[key]));
112 }
113 return Object.is(expected, actual);
114}
115
116const ADAPTER_REQUEST_ID = /^t3-(?:prompt-)?\d+$/u;
117
118function adapterRequestId(frame: unknown): string | undefined {
119 const id = isRecord(frame) ? frame.id : undefined;
120 return typeof id === "string" && ADAPTER_REQUEST_ID.test(id) ? id : undefined;
121}
122
123/** Rebind adapter request ids without pinning counters; provider ids that responses must echo stay exact. */
124function withRequestId(expected: unknown, actual: unknown): unknown {
125 const actualId = adapterRequestId(actual);
126 return isRecord(expected) &&
127 actualId !== undefined &&
128 (adapterRequestId(expected) !== undefined ||
129 (expected.type === "prompt" && expected.id === undefined))
130 ? { ...expected, id: actualId }
131 : expected;
132}
133
134function startsTurn(entry: ProviderReplayEntryType): boolean {
135 if (entry.type !== "expect_outbound") return false;
136 const type = isRecord(entry.frame) ? entry.frame.type : undefined;
137 return type === "prompt" || type === "compact" || type === "process_start";
138}
139
140function processStartFrame(command: ChildProcess.Command) {
141 const args = ChildProcess.isStandardCommand(command) ? command.args : [];
142 return { type: "process_start", args: [...args] };
143}
144
145function entryProcess(entry: ProviderReplayEntryType): number {
146 const label = entry.type === "runtime_exit" ? undefined : entry.label;
147 const match = label === undefined ? null : /@p(\d+)$/u.exec(label);
148 return match === null ? 1 : Number(match[1]);
149}
150
151function recordLabel(frame: unknown, process: number): string | undefined {
152 const base = !isRecord(frame)
153 ? undefined
154 : frame.type === "response" && typeof frame.command === "string"
155 ? `response:${frame.command}`
156 : typeof frame.type === "string"
157 ? frame.type
158 : undefined;
159 if (base === undefined || process === 1) return base;
160 return `${base}@p${process}`;
161}
162
163const decodeJsonLine = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
164const encodeJsonLine = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
165
166function makeLineSplitter() {
167 let buffer = "";
168 const decoder = new TextDecoder();
169 return (chunk: Uint8Array): ReadonlyArray<string> => {
170 buffer += decoder.decode(chunk, { stream: true });
171 const lines = buffer.split("\n");
172 buffer = lines.pop() ?? "";
173 return lines.map((line) => line.replace(/\r$/u, "")).filter((line) => line.length > 0);
174 };
175}
176
177interface PiReplayProcess {
178 readonly ordinal: number;
179 readonly stdout: Queue.Queue<Uint8Array, Cause.Done>;
180 readonly ids: Map<string, string>;
181 open: boolean;
182}
183
184/** Adapter fibers can reorder stdin writes; hold matches within the turn until reached, and fail unmatched writes. */
185class PiReplayController {
186 private cursor = 0;
187 private failure: unknown = null;
188 private readonly consumed = new Set<number>();
189 private readonly held: Array<{ readonly process: PiReplayProcess; readonly actual: unknown }> =
190 [];
191 private readonly processes: Array<PiReplayProcess> = [];
192 private readonly transcript: PiRpcReplayTranscript;
193
194 constructor(transcript: PiRpcReplayTranscript) {
195 this.transcript = transcript;
196 }
197
198 spawn(
199 command: ChildProcess.Command,
200 stdout: Queue.Queue<Uint8Array, Cause.Done>,
201 ): PiReplayProcess {
202 const process: PiReplayProcess = {
203 ordinal: this.processes.length + 1,
204 stdout,
205 ids: new Map(),
206 open: true,
207 };
208 this.processes.push(process);
209 this.receive(process, processStartFrame(command));
210 return process;
211 }
212
213 receive(process: PiReplayProcess, actual: unknown): void {
214 if (this.failure !== null) return;
215 const entries = this.transcript.entries;
216 let possible = false;
217 for (let index = this.cursor; index < entries.length && !possible; index += 1) {
218 const entry = entries[index]!;
219 const pending = entry.type === "expect_outbound" && !this.consumed.has(index);
220 possible =
221 pending &&
222 entryProcess(entry) === process.ordinal &&
223 replayValueMatches(withRequestId(entry.frame, actual), actual);
224 if (!possible && pending && startsTurn(entry)) break;
225 }
226 if (!possible) {
227 const expected = entries[this.cursor];
228 const label = entryLabel(expected);
229 const actualType = isRecord(actual) ? actual.type : undefined;
230 this.fail(
231 new PiReplayMismatchError({
232 scenario: this.transcript.scenario,
233 cursor: this.cursor,
234 process: process.ordinal,
235 ...(label === undefined ? {} : { label }),
236 ...(typeof actualType === "string" ? { actualType } : {}),
237 expected: expected ?? null,
238 actual,
239 }),
240 );
241 return;
242 }
243 this.held.push({ process, actual });
244 this.advance();
245 }
246
247 close(process: PiReplayProcess): void {
248 process.open = false;
249 Queue.endUnsafe(process.stdout);
250 }
251
252 assertComplete(): void {
253 if (this.failure !== null) throw this.failure;
254 if (this.cursor !== this.transcript.entries.length || this.held.length > 0) {
255 const nextLabel = entryLabel(this.transcript.entries[this.cursor]);
256 throw new PiReplayIncompleteError({
257 scenario: this.transcript.scenario,
258 cursor: this.cursor,
259 remaining: this.transcript.entries.length - this.cursor,
260 ...(nextLabel === undefined ? {} : { nextLabel }),
261 });
262 }
263 }
264
265 private advance(): void {
266 let progressed = true;
267 while (progressed && this.failure === null) {
268 progressed = this.drainInbound();
269 for (const [heldIndex, write] of this.held.entries()) {
270 if (this.consumePending(write.process, write.actual)) {
271 this.held.splice(heldIndex, 1);
272 progressed = true;
273 break;
274 }
275 }
276 }
277 }
278
279 private consumePending(process: PiReplayProcess, actual: unknown): boolean {
280 const entries = this.transcript.entries;
281 for (let index = this.cursor; index < entries.length; index += 1) {
282 const entry = entries[index]!;
283 if (entry.type !== "expect_outbound") return false;
284 if (this.consumed.has(index) || entryProcess(entry) !== process.ordinal) continue;
285 if (!replayValueMatches(withRequestId(entry.frame, actual), actual)) continue;
286 const recordedId = adapterRequestId(entry.frame);
287 const actualId = adapterRequestId(actual);
288 if (recordedId !== undefined && actualId !== undefined) {
289 process.ids.set(recordedId, actualId);
290 }
291 this.consumed.add(index);
292 while (this.consumed.has(this.cursor)) this.cursor += 1;
293 return true;
294 }
295 return false;
296 }
297
298 private drainInbound(): boolean {
299 let emitted = false;
300 while (this.failure === null) {
301 const entry = this.transcript.entries[this.cursor];
302 if (entry?.type === "runtime_exit") {
303 this.cursor += 1;
304 emitted = true;
305 const process = this.processes.at(-1);
306 if (process !== undefined) this.close(process);
307 continue;
308 }
309 if (entry?.type !== "emit_inbound") return emitted;
310 const process = this.processes[entryProcess(entry) - 1];
311 if (process === undefined) return emitted;
312 this.cursor += 1;
313 emitted = true;
314 if (!process.open) continue;
315 const frame = entry.frame;
316 const recordedId = isRecord(frame) ? frame.id : undefined;
317 const reboundId = typeof recordedId === "string" ? process.ids.get(recordedId) : undefined;
318 const rebound =
319 isRecord(frame) && reboundId !== undefined ? { ...frame, id: reboundId } : frame;
320 Queue.offerUnsafe(process.stdout, new TextEncoder().encode(`${encodeJsonLine(rebound)}\n`));
321 }
322 return emitted;
323 }
324
325 private fail(cause: unknown): void {
326 if (this.failure !== null) return;
327 this.failure = cause;
328 for (const process of this.processes) Queue.endUnsafe(process.stdout);
329 }
330}
331
332/** Deliberately outside the valid pid range so PiRpc's group kill never lands. */
333const REPLAY_PID = 999_999_999;
334
335function makePiReplaySpawner(
336 controller: PiReplayController,
337): ChildProcessSpawner.ChildProcessSpawner["Service"] {
338 return ChildProcessSpawner.make((command) =>
339 Effect.gen(function* () {
340 const process = controller.spawn(command, yield* Queue.unbounded<Uint8Array, Cause.Done>());
341 yield* Effect.addFinalizer(() => Effect.sync(() => controller.close(process)));
342 const split = makeLineSplitter();
343 return ChildProcessSpawner.makeHandle({
344 pid: ChildProcessSpawner.ProcessId(REPLAY_PID),
345 exitCode: Effect.never,
346 isRunning: Effect.succeed(true),
347 kill: () => Effect.void,
348 unref: Effect.succeed(Effect.void),
349 stdin: Sink.forEach((chunk: Uint8Array) =>
350 Effect.sync(() => {
351 for (const line of split(chunk)) controller.receive(process, decodeJsonLine(line));
352 }),
353 ),
354 stdout: Stream.fromQueue(process.stdout),
355 stderr: Stream.empty,
356 all: Stream.empty,
357 getInputFd: () => Sink.drain,
358 getOutputFd: () => Stream.empty,
359 });
360 }),
361 );
362}
363
364export function layer<E, R>(input: {
365 readonly scenario: string;
366 readonly spawner: Layer.Layer<ChildProcessSpawner.ChildProcessSpawner, E, R>;
367 readonly binaryPath: string;
368 readonly launchArgs: string;
369 readonly environment?: ResolvedProviderInstanceEnvironment;
370}) {
371 return ProviderAdapterRegistry.layerFromAdaptersEffect(
372 PiAdapterV2Driver.create({
373 instanceId: ProviderInstanceId.make(PI_PROVIDER),
374 environment: input.environment ?? [],
375 displayName: undefined,
376 enabled: true,
377 config: {
378 ...PiAdapterV2Driver.defaultConfig(),
379 binaryPath: input.binaryPath,
380 launchArgs: input.launchArgs,
381 },
382 }).pipe(Effect.map((adapter) => [adapter])),
383 ).pipe(
384 Layer.provide(input.spawner),
385 Layer.provide(
386 Layer.mergeAll(
387 TestProviderHost.layer().pipe(Layer.provide(NodeServices.layer)),
388 NodeServices.layer,
389 IdAllocator.layer,
390 ),
391 ),
392 );
393}
394
395function metadataString(transcript: PiRpcReplayTranscript, key: string): string {
396 const value = transcript.metadata?.[key];
397 return typeof value === "string" ? value : "";
398}
399
400export const PiOrchestratorReplayHarness: OrchestratorV2ProviderReplayHarness<
401 PiRpcReplayTranscript,
402 PiOrchestratorReplayHarnessError
403> = {
404 driver: PI_PROVIDER,
405 decodeTranscript: (transcript: ProviderReplayTranscript) =>
406 decodePiRpcReplayTranscript(transcript).pipe(
407 Effect.mapError(
408 (cause) =>
409 new PiReplayTranscriptDecodeError({
410 driver: transcript.provider,
411 protocol: transcript.protocol,
412 scenario: transcript.scenario,
413 cause,
414 }),
415 ),
416 ),
417 makeProviderAdapterRegistryLayer: (transcript) =>
418 layer({
419 scenario: transcript.scenario,
420 binaryPath: "pi",
421 launchArgs: metadataString(transcript, "launchArgs"),
422 spawner: Layer.effect(
423 ChildProcessSpawner.ChildProcessSpawner,
424 Effect.gen(function* () {
425 const controller = new PiReplayController(transcript);
426 yield* Effect.addFinalizer(() => Effect.sync(() => controller.assertComplete()));
427 return makePiReplaySpawner(controller);
428 }),
429 ),
430 }),
431};
432
433export function makePiRecordingSpawner(
434 live: ChildProcessSpawner.ChildProcessSpawner["Service"],
435 entries: Array<ProviderReplayEntryType>,
436): ChildProcessSpawner.ChildProcessSpawner["Service"] {
437 let spawned = 0;
438 return ChildProcessSpawner.make((command) =>
439 Effect.gen(function* () {
440 const handle = yield* live.spawn(command);
441 const process = ++spawned;
442 const record = (type: "expect_outbound" | "emit_inbound", frame: unknown) => {
443 const label = recordLabel(frame, process);
444 entries.push({ type, ...(label === undefined ? {} : { label }), frame });
445 };
446 const recordLines = (
447 type: "expect_outbound" | "emit_inbound",
448 lines: ReadonlyArray<string>,
449 ) => {
450 for (const line of lines) {
451 try {
452 record(type, decodeJsonLine(line));
453 } catch {
454 // PiRpc drops non-JSON stdout lines, so the transcript does too.
455 }
456 }
457 };
458 record("expect_outbound", processStartFrame(command));
459 const splitIn = makeLineSplitter();
460 const splitOut = makeLineSplitter();
461 return ChildProcessSpawner.makeHandle({
462 ...handle,
463 stdin: handle.stdin.pipe(
464 Sink.mapInput((chunk: Uint8Array) => {
465 recordLines("expect_outbound", splitIn(chunk));
466 return chunk;
467 }),
468 ),
469 stdout: handle.stdout.pipe(
470 Stream.tap((chunk) => Effect.sync(() => recordLines("emit_inbound", splitOut(chunk)))),
471 ),
472 });
473 }),
474 );
475}