1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { type ProviderReplayEntry } from "@t3tools/contracts";
4import { GrokSettings } from "@t3tools/provider-grok/settings";
5import * as HostProcess from "@t3tools/shared/HostProcess";
6import { resolveSelfInvocation } from "@t3tools/shared/nodeRuntime";
7import * as Clock from "effect/Clock";
8import * as Console from "effect/Console";
9import * as Config from "effect/Config";
10import * as Option from "effect/Option";
11import * as Effect from "effect/Effect";
12import * as FileSystem from "effect/FileSystem";
13import * as Layer from "effect/Layer";
14import * as Path from "effect/Path";
15import * as Schema from "effect/Schema";
16import type * as EffectAcpProtocol from "effect-acp/protocol";
17
18import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost";
19import {
20 GROK_DEFAULT_INSTANCE_ID,
21 GROK_PROVIDER,
22 grokLaunchRuntimeMode,
23 makeGrokAdapterV2,
24} from "@t3tools/provider-grok/testing";
25import { ACP_PROTOCOL } from "@t3tools/provider-acp/server/adapter";
26import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
27import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
28import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
29import * as ProviderAdapterRegistry from "../src/orchestration-v2/ProviderAdapterRegistry.ts";
30import { provideDeterministicTestRuntime } from "../src/orchestration-v2/testkit/DeterministicRuntime.ts";
31import { ORCHESTRATOR_REPLAY_FIXTURES } from "../src/orchestration-v2/testkit/fixtures/index.ts";
32import { materializeFixtureInput } from "../src/orchestration-v2/testkit/fixtures/shared.ts";
33import { runOrchestratorV2Scenario } from "../src/orchestration-v2/testkit/OrchestratorScenario.ts";
34import * as ProviderReplayHarness from "../src/orchestration-v2/testkit/ProviderReplayHarness.ts";
35import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
36import { makeGrokAcpRuntime } from "@t3tools/provider-grok/testing";
37import { buildRuntimeInstructions } from "@t3tools/provider-core/server/runtimeInstructions";
38import * as Result from "effect/Result";
39
40const wallClock = Clock.Clock.defaultValue();
41const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
42const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
43
44const DROPPED_INBOUND_METHODS = new Set(["_x.ai/settings/update", "_x.ai/announcements/update"]);
45const HOME_PLACEHOLDER = "/home/grok-replay";
46
47interface JsonRpcMessage {
48 readonly id?: string | number | null;
49 readonly method?: string;
50 readonly params?: unknown;
51 readonly result?: unknown;
52 readonly error?: unknown;
53}
54
55interface WireMessage {
56 readonly direction: "incoming" | "outgoing";
57 readonly message: JsonRpcMessage;
58}
59
60function readArgValues(name: string): ReadonlyArray<string> {
61 const args = process.argv.slice(2);
62 return args.flatMap((arg, index) => (arg === name && args[index + 1] ? [args[index + 1]!] : []));
63}
64
65function isRecord(value: unknown): value is Record<string, unknown> {
66 return typeof value === "object" && value !== null && !Array.isArray(value);
67}
68
69function mapStrings(value: unknown, map: (text: string) => string): unknown {
70 if (typeof value === "string") return map(value);
71 if (Array.isArray(value)) return value.map((entry) => mapStrings(entry, map));
72 if (!isRecord(value)) return value;
73 return Object.fromEntries(
74 Object.entries(value).map(([key, entry]) => [key, mapStrings(entry, map)]),
75 );
76}
77
78function makeWireTee() {
79 const wire: Array<WireMessage> = [];
80 let runtimeCount = 0;
81 const attachRuntime = () => {
82 runtimeCount += 1;
83 if (runtimeCount > 1) {
84 throw new Error("The Grok recording spawned a second ACP process; replay drives one.");
85 }
86 let incomingBuffer = "";
87 const push = (direction: WireMessage["direction"], line: string) => {
88 if (line.trim().length === 0) return;
89 wire.push({ direction, message: decodeJson(line) as JsonRpcMessage });
90 };
91 return {
92 logIncoming: true,
93 logOutgoing: true,
94 logger: (event: EffectAcpProtocol.AcpProtocolLogEvent) =>
95 Effect.sync(() => {
96 if (event.stage !== "raw" || typeof event.payload !== "string") return;
97 if (event.direction === "outgoing") {
98 for (const line of event.payload.split("\n")) push("outgoing", line);
99 return;
100 }
101 incomingBuffer += event.payload;
102 const lines = incomingBuffer.split("\n");
103 incomingBuffer = lines.pop() ?? "";
104 for (const line of lines) push("incoming", line);
105 }),
106 };
107 };
108 // Grok can report idle before its wake turns; wait for prompt queues, tasks and subagents to finish.
109 const grokHasPendingWork = () => {
110 const queues = new Map<unknown, Record<string, unknown>>();
111 const tasks = new Map<unknown, ReadonlyArray<unknown>>();
112 const completedPrompts = new Set<unknown>();
113 const liveSubagents = new Set<unknown>();
114 for (const { direction, message } of wire) {
115 if (direction !== "incoming" || !isRecord(message.params)) continue;
116 const params = message.params;
117 const update = isRecord(params.update) ? params.update : undefined;
118 if (message.method === "_x.ai/queue/changed") queues.set(params.sessionId, params);
119 if (update?.sessionUpdate === "turn_completed") completedPrompts.add(update.prompt_id);
120 if (update?.sessionUpdate === "background_tasks" && Array.isArray(update.tasks)) {
121 tasks.set(params.sessionId, update.tasks);
122 }
123 if (update?.sessionUpdate === "subagent_spawned") liveSubagents.add(update.subagent_id);
124 if (update?.sessionUpdate === "subagent_finished") liveSubagents.delete(update.subagent_id);
125 }
126 const promptPending = [...queues.values()].some(
127 (queue) =>
128 (Array.isArray(queue.entries) && queue.entries.length > 0) ||
129 (queue.runningPromptId !== undefined && !completedPrompts.has(queue.runningPromptId)),
130 );
131 const taskRunning = [...tasks.values()].some((list) =>
132 list.some((task) => isRecord(task) && task.status === "running"),
133 );
134 return promptPending || taskRunning || liveSubagents.size > 0;
135 };
136 return { wire, attachRuntime, grokHasPendingWork };
137}
138
139function frameLabel(kind: string, method: string, params: unknown): string {
140 const update = isRecord(params) && isRecord(params.update) ? params.update : undefined;
141 const updateType = typeof update?.sessionUpdate === "string" ? `:${update.sessionUpdate}` : "";
142 const meta = isRecord(params) && isRecord(params._meta) ? params._meta : undefined;
143 const promptId =
144 update?.sessionUpdate === "turn_completed" && typeof update.prompt_id === "string"
145 ? update.prompt_id
146 : typeof meta?.promptId === "string"
147 ? meta.promptId
148 : undefined;
149 return `${kind}:${method}${updateType}${promptId === undefined ? "" : `:${promptId}`}`;
150}
151
152function wireToEntries(wire: ReadonlyArray<WireMessage>): {
153 readonly entries: Array<ProviderReplayEntry>;
154 readonly droppedFrames: number;
155} {
156 const entries: Array<ProviderReplayEntry> = [];
157 const t3Requests = new Map<string, string>();
158 const agentRequests = new Map<string, string>();
159 let droppedFrames = 0;
160 for (const { direction, message } of wire) {
161 const type = direction === "outgoing" ? "expect_outbound" : "emit_inbound";
162 if (typeof message.method === "string") {
163 if (direction === "incoming" && DROPPED_INBOUND_METHODS.has(message.method)) {
164 droppedFrames += 1;
165 continue;
166 }
167 const isRequest = message.id !== undefined && message.id !== null;
168 if (isRequest) {
169 (direction === "outgoing" ? t3Requests : agentRequests).set(
170 String(message.id),
171 message.method,
172 );
173 }
174 const kind = isRequest ? "request" : "notification";
175 entries.push({
176 type,
177 label: frameLabel(kind, message.method, message.params),
178 frame: {
179 kind,
180 method: message.method,
181 ...(message.params === undefined ? {} : { params: message.params }),
182 },
183 });
184 continue;
185 }
186 const pending = direction === "outgoing" ? agentRequests : t3Requests;
187 const method = pending.get(String(message.id));
188 if (method === undefined) {
189 // Grok answers internal requests on this stream; T3 drops them before replay.
190 droppedFrames += 1;
191 continue;
192 }
193 pending.delete(String(message.id));
194 entries.push({
195 type,
196 label: `response:${method}`,
197 frame: {
198 kind: "response",
199 method,
200 ...(message.result === undefined ? {} : { result: message.result }),
201 ...(message.error === undefined ? {} : { error: message.error }),
202 },
203 });
204 }
205 return { entries, droppedFrames };
206}
207
208const T3_INSTRUCTIONS_BODY = /<t3_code_instructions>\n[\s\S]*?\n<\/t3_code_instructions>/u;
209
210function normalizeOutboundFrame(frame: Record<string, unknown>, runtimeInstructions: string) {
211 const params = isRecord(frame.params) ? frame.params : undefined;
212 if (params === undefined) return frame;
213 switch (frame.method) {
214 case "initialize":
215 return {
216 ...frame,
217 params: Object.fromEntries(
218 Object.keys(params).map((key) => [
219 key,
220 key === "protocolVersion" || key === "clientCapabilities" || key === "_meta"
221 ? params[key]
222 : "<any>",
223 ]),
224 ),
225 };
226 case "session/new":
227 case "session/load":
228 case "session/resume":
229 return { ...frame, params: { ...params, mcpServers: "<any>" } };
230 case "session/prompt": {
231 if (!Array.isArray(params.prompt)) return frame;
232 const prompt = params.prompt.filter(
233 (part) => !(isRecord(part) && part.type === "text" && part.text === runtimeInstructions),
234 );
235 return {
236 ...frame,
237 params: {
238 ...params,
239 prompt: prompt.map((part) =>
240 isRecord(part) && typeof part.text === "string"
241 ? {
242 ...part,
243 text: part.text.replace(
244 T3_INSTRUCTIONS_BODY,
245 "<t3_code_instructions>\n<any>\n</t3_code_instructions>",
246 ),
247 }
248 : part,
249 ),
250 },
251 };
252 }
253 default:
254 return frame;
255 }
256}
257
258function normalizeInboundFrame(frame: Record<string, unknown>): Record<string, unknown> {
259 const params = isRecord(frame.params) ? frame.params : undefined;
260 const update = isRecord(params?.update) ? params.update : undefined;
261 if (
262 update?.sessionUpdate === "available_commands_update" &&
263 Array.isArray(update.availableCommands)
264 ) {
265 return {
266 ...frame,
267 params: {
268 ...params,
269 update: {
270 ...update,
271 availableCommands: update.availableCommands.filter(
272 (command) =>
273 !(isRecord(command) && isRecord(command._meta) && command._meta.scope === "user"),
274 ),
275 },
276 },
277 };
278 }
279 const result = isRecord(frame.result) ? frame.result : undefined;
280 if (frame.method === "initialize" && isRecord(result?._meta)) {
281 const meta = result._meta;
282 return {
283 ...frame,
284 result: {
285 ...result,
286 _meta: {
287 ...meta,
288 agentId: "00000000-0000-4000-8000-000000000000",
289 agentInstanceId: "00000000-0000-4000-8000-000000000000",
290 hostname: "grok-replay",
291 ...(Array.isArray(meta.availableCommands)
292 ? {
293 availableCommands: meta.availableCommands.filter(
294 (command) =>
295 !(
296 isRecord(command) &&
297 isRecord(command._meta) &&
298 command._meta.scope === "user"
299 ),
300 ),
301 }
302 : {}),
303 },
304 },
305 };
306 }
307 return frame;
308}
309
310const UUID = /^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/iu;
311
312function collectSessionIds(entries: ReadonlyArray<ProviderReplayEntry>): ReadonlyArray<string> {
313 const ids: Array<string> = [];
314 const add = (value: unknown) => {
315 if (typeof value === "string" && value.length > 0 && !ids.includes(value)) ids.push(value);
316 };
317 for (const entry of entries) {
318 if (entry.type === "runtime_exit" || !isRecord(entry.frame)) continue;
319 const frame = entry.frame;
320 if (frame.kind === "response" && frame.method === "session/new" && isRecord(frame.result)) {
321 add(frame.result.sessionId);
322 }
323 if (entry.type === "emit_inbound" && isRecord(frame.params)) {
324 add(frame.params.sessionId);
325 const update = isRecord(frame.params.update) ? frame.params.update : undefined;
326 if (update?.sessionUpdate === "task_backgrounded" && UUID.test(String(update.task_id))) {
327 add(update.task_id);
328 }
329 }
330 }
331 return ids;
332}
333
334function normalizeEntries(input: {
335 readonly entries: ReadonlyArray<ProviderReplayEntry>;
336 readonly workspace: string;
337 readonly home: string;
338 readonly user: string;
339 readonly runtimeInstructions: string;
340}): Array<ProviderReplayEntry> {
341 const sessionIds = collectSessionIds(input.entries);
342 const replacements: Array<readonly [string, string]> = [
343 ...sessionIds.map(
344 (id, index) =>
345 [id, `00000000-0000-4000-8000-${String(index + 1).padStart(12, "0")}`] as const,
346 ),
347 [input.workspace, "<workspace>"],
348 [encodeURIComponent(input.workspace), "%3Cworkspace%3E"],
349 [input.home, HOME_PLACEHOLDER],
350 ];
351 const user = /^[a-z_][a-z0-9_-]*$/iu.test(input.user) ? input.user : "";
352 const userPattern = user.length === 0 ? undefined : new RegExp(`\\b${user}\\b`, "gu");
353 const replaceAll = (text: string) => {
354 const replaced = replacements.reduce(
355 (current, [from, to]) => (from.length === 0 ? current : current.replaceAll(from, to)),
356 text,
357 );
358 return userPattern === undefined ? replaced : replaced.replace(userPattern, "grok-replay");
359 };
360 return input.entries.map((entry) => {
361 if (entry.type === "runtime_exit" || !isRecord(entry.frame)) return entry;
362 const frame =
363 entry.type === "expect_outbound"
364 ? normalizeOutboundFrame(entry.frame, input.runtimeInstructions)
365 : normalizeInboundFrame(entry.frame);
366 return {
367 ...entry,
368 ...(entry.label === undefined ? {} : { label: replaceAll(entry.label) }),
369 frame: mapStrings(frame, replaceAll),
370 };
371 });
372}
373
374const DEFAULT_GROK_SETTINGS = Schema.decodeSync(GrokSettings)({});
375
376const RecorderEnv = Config.all({
377 T3_GROK_BIN: Config.String("T3_GROK_BIN").pipe(Config.option, Config.map(Option.getOrUndefined)),
378 HOME: Config.String("HOME").pipe(Config.option, Config.map(Option.getOrUndefined)),
379 USER: Config.String("USER").pipe(Config.option, Config.map(Option.getOrUndefined)),
380});
381
382const recordScenario = Effect.fn("recordGrokScenario")(function* (fixtureName: string) {
383 const recorderEnv = yield* RecorderEnv;
384 const fixture = ORCHESTRATOR_REPLAY_FIXTURES.find((candidate) => candidate.name === fixtureName);
385 const variant = fixture?.providers.find((provider) => provider.driver === GROK_PROVIDER);
386 if (fixture === undefined || variant === undefined) {
387 return yield* Effect.die(new Error(`No Grok replay fixture named '${fixtureName}'.`));
388 }
389 const fs = yield* FileSystem.FileSystem;
390 const path = yield* Path.Path;
391 const fixtureInput = fixture.buildInput();
392 const workspace = yield* checkpointWorkspace(fixtureName, fixtureInput.workspaceFiles);
393 const realWorkspace = yield* fs.realPath(workspace);
394 const materialized = yield* materializeFixtureInput({
395 scenario: fixtureName,
396 fixtureInput,
397 driver: GROK_PROVIDER,
398 modelSelection: variant.modelSelection,
399 }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime);
400 const scenario = {
401 name: `${fixtureName}/grok-record`,
402 commands: materialized.commands,
403 // Live Grok needs no replay gates or clock receipts; record dispatches and wait for idle.
404 steps: materialized.steps.some(
405 (step) =>
406 step.type === "release_replay_gate" || step.type === "release_replay_gate_after_waiting",
407 )
408 ? materialized.steps.filter((step) => step.type === "dispatch")
409 : materialized.steps.map((step) =>
410 step.type === "finish_held_run" ? { ...step, type: "await_run_status" as const } : step,
411 ),
412 projectionThreadIds: materialized.projectionThreadIds,
413 runtimePolicyOverride: { ...variant.runtimePolicyOverride, cwd: realWorkspace },
414 };
415
416 const tee = makeWireTee();
417 const settings = { ...DEFAULT_GROK_SETTINGS, binaryPath: recorderEnv.T3_GROK_BIN ?? "grok" };
418 const layerRegistry = ProviderAdapterRegistry.layerFromAdaptersEffect(
419 Effect.gen(function* () {
420 const environment = yield* HostProcess.Environment;
421 const adapter = yield* makeGrokAdapterV2({
422 instanceId: GROK_DEFAULT_INSTANCE_ID,
423 settings,
424 environment,
425 hostPlatform: yield* HostProcess.Platform,
426 selfInvocation: yield* resolveSelfInvocation(),
427 continuationRequests: yield* ProviderContinuationRequests.ProviderContinuationRequests,
428 makeRuntime: ({ runtimePolicy, ...input }) =>
429 makeGrokAcpRuntime({
430 ...input,
431 protocolLogging: tee.attachRuntime(),
432 interruptPromptOnCancel: input.interruptPromptOnCancel ?? false,
433 grokSettings: settings,
434 environment,
435 runtimeMode: grokLaunchRuntimeMode(runtimePolicy),
436 }),
437 });
438 // Order dispatches on TestClock as in replay, but run Grok's adapter timers on wall time.
439 const onWallClock = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
440 Effect.provideService(effect, Clock.Clock, wallClock);
441 return [
442 {
443 ...adapter,
444 openSession: (input) =>
445 adapter.openSession(input).pipe(
446 Effect.map((session): ProviderAdapter.ProviderAdapterV2SessionRuntime => ({
447 ...session,
448 startTurn: (turnInput) => onWallClock(session.startTurn(turnInput)),
449 steerTurn: (turnInput) => onWallClock(session.steerTurn(turnInput)),
450 interruptTurn: (turnInput) => onWallClock(session.interruptTurn(turnInput)),
451 respondToRuntimeRequest: (turnInput) =>
452 onWallClock(session.respondToRuntimeRequest(turnInput)),
453 })),
454 onWallClock,
455 ),
456 },
457 ];
458 }),
459 ).pipe(
460 Layer.provide(
461 Layer.mergeAll(
462 TestProviderHost.layer().pipe(
463 Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
464 ),
465 NodeServices.layer,
466 NodeCrypto.layer,
467 IdAllocator.layer,
468 ),
469 ),
470 );
471
472 // Wake turns arrive after their triggering frames; require a few quiet seconds too.
473 let quietPolls = 0;
474 let seenFrames = -1;
475 const waitForGrokIdle = Effect.sleep("1 second").pipe(
476 Effect.andThen(
477 Effect.sync(() => {
478 quietPolls = seenFrames === tee.wire.length ? quietPolls + 1 : 0;
479 seenFrames = tee.wire.length;
480 return quietPolls >= 5 && !tee.grokHasPendingWork();
481 }),
482 ),
483 Effect.repeat({ until: (idle) => idle }),
484 Effect.timeout("5 minutes"),
485 Effect.orDie,
486 Effect.asVoid,
487 Effect.provideService(Clock.Clock, wallClock),
488 );
489 const result = yield* runOrchestratorV2Scenario(scenario, { afterSteps: waitForGrokIdle }).pipe(
490 Effect.provide(
491 ProviderReplayHarness.layerWithRegistry(
492 scenario,
493 layerRegistry,
494 variant.runContinuationWorker === true ? { runContinuationWorker: true } : {},
495 ),
496 ),
497 provideDeterministicTestRuntime,
498 Effect.scoped,
499 );
500
501 const { entries, droppedFrames } = wireToEntries(tee.wire);
502 const closedCleanly = entries.some(
503 (entry) =>
504 entry.type === "expect_outbound" &&
505 isRecord(entry.frame) &&
506 entry.frame.method === "session/close",
507 );
508 const initialize = entries.find(
509 (entry) =>
510 entry.type === "emit_inbound" && isRecord(entry.frame) && entry.frame.method === "initialize",
511 );
512 const initializeMeta =
513 initialize?.type === "emit_inbound" &&
514 isRecord(initialize.frame) &&
515 isRecord(initialize.frame.result) &&
516 isRecord(initialize.frame.result._meta)
517 ? initialize.frame.result._meta
518 : {};
519 const transcript = {
520 provider: GROK_PROVIDER,
521 protocol: ACP_PROTOCOL,
522 version: "1",
523 scenario: fixtureName,
524 metadata: {
525 generatedBy: "live-grok-recorder",
526 grokVersion: initializeMeta.agentVersion ?? "unknown",
527 normalization:
528 "Session ids are fixed UUIDs, the workspace is <workspace>, HOME is /home/grok-replay and the recording user is grok-replay. T3-owned prompt text, MCP servers and initialize params other than clientCapabilities and _meta are <any>. Personal skills, machine identity, account settings and announcement broadcasts are removed, as are responses to Grok-internal request ids that T3's protocol drops. Timestamps are kept as recorded.",
529 droppedFrames,
530 },
531 entries: [
532 ...normalizeEntries({
533 entries,
534 workspace: realWorkspace,
535 home: recorderEnv.HOME ?? "",
536 user: recorderEnv.USER ?? "",
537 runtimeInstructions: buildRuntimeInstructions({
538 harness: "Grok",
539 model: variant.modelSelection.model,
540 }),
541 }),
542 { type: "runtime_exit", status: closedCleanly ? "success" : "cancelled" } as const,
543 ],
544 };
545
546 const outputPath = readArgValues("--out")[0] ?? (yield* path.fromFileUrl(variant.transcriptFile));
547 const { entries: transcriptEntries, ...header } = transcript;
548 yield* fs.writeFileString(
549 outputPath,
550 [
551 encodeJson({ type: "transcript_start", ...header }),
552 ...transcriptEntries.map((entry) => encodeJson(entry)),
553 "",
554 ].join("\n"),
555 );
556 yield* Console.log(`Wrote ${transcriptEntries.length} Grok ACP replay entries to ${outputPath}`);
557
558 const liveFailure = Result.try(() => variant.assertOutput(result, transcript));
559 if (Result.isFailure(liveFailure)) {
560 yield* Console.log(`Live orchestration failed ${fixtureName} assertions:`, liveFailure.failure);
561 }
562});
563
564const scenarios = readArgValues("--scenario").flatMap((value) => value.split(","));
565if (scenarios.length === 0) {
566 throw new Error("Pass --scenario <fixture name>[,<fixture name>...]");
567}
568
569await Effect.runPromise(
570 Effect.forEach(scenarios, (name) => Effect.scoped(recordScenario(name)), {
571 discard: true,
572 }).pipe(Effect.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))),
573);