1import * as Cause from "effect/Cause";
2import * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Queue from "effect/Queue";
6import * as Ref from "effect/Ref";
7import * as Schema from "effect/Schema";
8import * as Sink from "effect/Sink";
9import * as Stdio from "effect/Stdio";
10import * as Stream from "effect/Stream";
11
12import * as CodexClient from "./client.ts";
13import * as CodexError from "./errors.ts";
14
15export const CodexAppServerReplayEntry = Schema.Union([
16 Schema.Struct({
17 type: Schema.Literal("expect_outbound"),
18 label: Schema.optional(Schema.String),
19 frame: Schema.Unknown,
20 }),
21 Schema.Struct({
22 type: Schema.Literal("emit_inbound"),
23 label: Schema.optional(Schema.String),
24 frame: Schema.Unknown,
25 afterMs: Schema.optional(Schema.Number),
26 }),
27 Schema.Struct({
28 type: Schema.Literal("runtime_exit"),
29 status: Schema.Literals(["success", "error", "cancelled"]),
30 error: Schema.optional(Schema.Unknown),
31 }),
32]);
33export type CodexAppServerReplayEntry = typeof CodexAppServerReplayEntry.Type;
34
35export const CodexAppServerReplayTranscript = Schema.Struct({
36 provider: Schema.Literal("codex"),
37 protocol: Schema.Literal("codex.app-server"),
38 version: Schema.String,
39 scenario: Schema.String,
40 metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
41 entries: Schema.Array(CodexAppServerReplayEntry),
42});
43export type CodexAppServerReplayTranscript = typeof CodexAppServerReplayTranscript.Type;
44const decodeOutboundJsonFrame = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
45
46export class CodexAppServerReplayJsonParseError extends Schema.TaggedError<CodexAppServerReplayJsonParseError>()(
47 "CodexAppServerReplayJsonParseError",
48 {
49 scenario: Schema.String,
50 line: Schema.String,
51 cause: Schema.Defect(),
52 },
53) {
54 override get message(): string {
55 return `Failed to parse outbound Codex app-server replay frame for scenario ${this.scenario}.`;
56 }
57}
58
59export class CodexAppServerReplayExhaustedError extends Schema.TaggedError<CodexAppServerReplayExhaustedError>()(
60 "CodexAppServerReplayExhaustedError",
61 {
62 scenario: Schema.String,
63 cursor: Schema.Number,
64 actual: Schema.Unknown,
65 },
66) {
67 override get message(): string {
68 return `Codex app-server replay transcript exhausted before outbound frame ${this.cursor} in scenario ${this.scenario}.`;
69 }
70}
71
72export class CodexAppServerReplayUnexpectedOutboundError extends Schema.TaggedError<CodexAppServerReplayUnexpectedOutboundError>()(
73 "CodexAppServerReplayUnexpectedOutboundError",
74 {
75 scenario: Schema.String,
76 cursor: Schema.Number,
77 expectedType: Schema.String,
78 actual: Schema.Unknown,
79 },
80) {
81 override get message(): string {
82 return `Unexpected outbound Codex app-server frame at replay cursor ${this.cursor} in scenario ${this.scenario}.`;
83 }
84}
85
86export class CodexAppServerReplayFrameMismatchError extends Schema.TaggedError<CodexAppServerReplayFrameMismatchError>()(
87 "CodexAppServerReplayFrameMismatchError",
88 {
89 scenario: Schema.String,
90 cursor: Schema.Number,
91 label: Schema.optional(Schema.String),
92 expected: Schema.Unknown,
93 actual: Schema.Unknown,
94 },
95) {
96 override get message(): string {
97 return `Outbound Codex app-server frame did not match replay cursor ${this.cursor} in scenario ${this.scenario}. Expected ${JSON.stringify(this.expected)}, received ${JSON.stringify(this.actual)}.`;
98 }
99}
100
101export class CodexAppServerReplayRuntimeExitError extends Schema.TaggedError<CodexAppServerReplayRuntimeExitError>()(
102 "CodexAppServerReplayRuntimeExitError",
103 {
104 scenario: Schema.String,
105 cursor: Schema.Number,
106 status: Schema.Literals(["error", "cancelled"]),
107 error: Schema.optional(Schema.Unknown),
108 },
109) {
110 override get message(): string {
111 return `Codex app-server replay exited with status ${this.status} at cursor ${this.cursor} in scenario ${this.scenario}.`;
112 }
113}
114
115export class CodexAppServerReplayIncompleteError extends Schema.TaggedError<CodexAppServerReplayIncompleteError>()(
116 "CodexAppServerReplayIncompleteError",
117 {
118 scenario: Schema.String,
119 cursor: Schema.Number,
120 remaining: Schema.Number,
121 },
122) {
123 override get message(): string {
124 return `Codex app-server replay ended with ${this.remaining} unconsumed entries in scenario ${this.scenario}.`;
125 }
126}
127
128export const CodexAppServerReplayError = Schema.Union([
129 CodexAppServerReplayJsonParseError,
130 CodexAppServerReplayExhaustedError,
131 CodexAppServerReplayUnexpectedOutboundError,
132 CodexAppServerReplayFrameMismatchError,
133 CodexAppServerReplayRuntimeExitError,
134 CodexAppServerReplayIncompleteError,
135]);
136export type CodexAppServerReplayError = typeof CodexAppServerReplayError.Type;
137
138export interface CodexAppServerReplayState {
139 readonly cursor: number;
140 readonly failure: CodexAppServerReplayError | null;
141}
142
143export interface CodexAppServerReplayDriver {
144 readonly transcript: CodexAppServerReplayTranscript;
145 readonly state: Ref.Ref<CodexAppServerReplayState>;
146 readonly beforeEmitInbound?: (
147 entry: Extract<CodexAppServerReplayEntry, { readonly type: "emit_inbound" }>,
148 ) => Effect.Effect<void>;
149}
150
151const encoder = new TextEncoder();
152
153function stableStringify(value: unknown): string {
154 if (Array.isArray(value)) {
155 return `[${value.map(stableStringify).join(",")}]`;
156 }
157 if (typeof value === "object" && value !== null) {
158 const record = value as Record<string, unknown>;
159 return `{${Object.keys(record)
160 .toSorted()
161 .map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`)
162 .join(",")}}`;
163 }
164 return JSON.stringify(value);
165}
166
167function normalizeContextHandoffText(value: string): string {
168 if (!value.startsWith("Context handoff (")) {
169 return value;
170 }
171 const userMessageMarker = "\n\nUser message:\n";
172 const userMessageIndex = value.indexOf(userMessageMarker);
173 const headerEndIndex = value.indexOf(":\n");
174 if (headerEndIndex === -1 || userMessageIndex === -1 || headerEndIndex >= userMessageIndex) {
175 return value;
176 }
177 return `${value.slice(0, headerEndIndex + 2)}<dynamic-summary>${value.slice(userMessageIndex)}`;
178}
179
180function normalizeReplayFrame(value: unknown): unknown {
181 if (typeof value === "string") {
182 return normalizeContextHandoffText(value);
183 }
184 if (typeof value !== "object" || value === null) {
185 return value;
186 }
187 if (Array.isArray(value)) {
188 return value.map(normalizeReplayFrame);
189 }
190
191 const record = value as Record<string, unknown>;
192 const normalized = Object.fromEntries(
193 Object.entries(record).map(([key, entry]) => [key, normalizeReplayFrame(entry)]),
194 );
195
196 if (
197 normalized.method === "initialize" &&
198 typeof normalized.params === "object" &&
199 normalized.params !== null
200 ) {
201 const params = normalized.params as Record<string, unknown>;
202 if (typeof params.clientInfo === "object" && params.clientInfo !== null) {
203 normalized.params = {
204 ...params,
205 clientInfo: {
206 ...(params.clientInfo as Record<string, unknown>),
207 version: "<ignored>",
208 },
209 };
210 }
211 }
212
213 if (
214 normalized.method === "turn/start" &&
215 typeof normalized.params === "object" &&
216 normalized.params !== null
217 ) {
218 const params = { ...(normalized.params as Record<string, unknown>) };
219 if (params.approvalPolicy === "never") {
220 delete params.approvalPolicy;
221 }
222 if (
223 typeof params.sandboxPolicy === "object" &&
224 params.sandboxPolicy !== null &&
225 (params.sandboxPolicy as Record<string, unknown>).type === "dangerFullAccess"
226 ) {
227 delete params.sandboxPolicy;
228 }
229 if (typeof params.collaborationMode === "object" && params.collaborationMode !== null) {
230 const collaborationMode = {
231 ...(params.collaborationMode as Record<string, unknown>),
232 };
233 if (typeof collaborationMode.settings === "object" && collaborationMode.settings !== null) {
234 collaborationMode.settings = {
235 ...(collaborationMode.settings as Record<string, unknown>),
236 developer_instructions: "<ignored>",
237 };
238 }
239 params.collaborationMode = collaborationMode;
240 }
241 normalized.params = params;
242 }
243
244 return normalized;
245}
246
247/**
248 * `config` keys a recorder sent on thread requests that the client under test
249 * never sends, such as `agents.max_depth`. Recorders list them in the
250 * transcript header as `metadata.recorderThreadConfigKeys`.
251 */
252function recorderThreadConfigKeys(transcript: CodexAppServerReplayTranscript): ReadonlySet<string> {
253 const keys = transcript.metadata?.recorderThreadConfigKeys;
254 return new Set(Array.isArray(keys) ? keys.filter((key) => typeof key === "string") : []);
255}
256
257/**
258 * Thread requests must match on the `config` the client owns. `cwd`, `model`
259 * and host-supplied MCP servers (machine-local URLs, short-lived authorization
260 * headers) vary per machine, and recorder-only keys are not the client's.
261 */
262function normalizeThreadRequest(frame: unknown, ignoredConfigKeys: ReadonlySet<string>): unknown {
263 if (typeof frame !== "object" || frame === null) return frame;
264 const record = frame as Record<string, unknown>;
265 if (
266 (record.method !== "thread/start" &&
267 record.method !== "thread/resume" &&
268 record.method !== "thread/fork") ||
269 typeof record.params !== "object" ||
270 record.params === null
271 ) {
272 return frame;
273 }
274 const { cwd: _cwd, model: _model, config, ...params } = record.params as Record<string, unknown>;
275 if (typeof config !== "object" || config === null) return { ...record, params };
276 const { mcp_servers: _mcpServers, ...owned } = config as Record<string, unknown>;
277 for (const key of ignoredConfigKeys) delete owned[key];
278 return { ...record, params: { ...params, config: owned } };
279}
280
281function sameFrame(
282 left: unknown,
283 right: unknown,
284 ignoredConfigKeys: ReadonlySet<string> = new Set(),
285): boolean {
286 const normalize = (frame: unknown) =>
287 stableStringify(normalizeThreadRequest(normalizeReplayFrame(frame), ignoredConfigKeys));
288 return normalize(left) === normalize(right);
289}
290
291function normalizeLegacyInboundFrame(value: unknown): unknown {
292 if (Array.isArray(value)) {
293 return value.map(normalizeLegacyInboundFrame);
294 }
295 if (typeof value !== "object" || value === null) {
296 return value;
297 }
298 const record = value as Record<string, unknown>;
299 const normalized = Object.fromEntries(
300 Object.entries(record).map(([key, entry]) => [key, normalizeLegacyInboundFrame(entry)]),
301 );
302 if (
303 typeof normalized.id === "string" &&
304 "modelProvider" in normalized &&
305 "status" in normalized
306 ) {
307 if (normalized.sessionId === undefined) normalized.sessionId = normalized.id;
308 if (normalized.projectId === undefined) normalized.projectId = null;
309 }
310 if (
311 (normalized.method === "item/started" || normalized.method === "item/completed") &&
312 typeof normalized.params === "object" &&
313 normalized.params !== null
314 ) {
315 const params = { ...(normalized.params as Record<string, unknown>) };
316 if (normalized.method === "item/started" && params.startedAtMs === undefined) {
317 params.startedAtMs = 0;
318 }
319 if (normalized.method === "item/completed" && params.completedAtMs === undefined) {
320 params.completedAtMs = 0;
321 }
322 normalized.params = params;
323 }
324 if (
325 typeof normalized.method === "string" &&
326 normalized.method.endsWith("/requestApproval") &&
327 typeof normalized.params === "object" &&
328 normalized.params !== null
329 ) {
330 const params = { ...(normalized.params as Record<string, unknown>) };
331 if (params.startedAtMs === undefined) {
332 params.startedAtMs = 0;
333 }
334 normalized.params = params;
335 }
336 if (
337 normalized.method === "item/tool/requestUserInput" &&
338 typeof normalized.params === "object" &&
339 normalized.params !== null
340 ) {
341 const params = { ...(normalized.params as Record<string, unknown>) };
342 // Codex treats a legacy request without `isBlocking` as blocking.
343 if (params.isBlocking === undefined) {
344 params.isBlocking = true;
345 }
346 normalized.params = params;
347 }
348 return normalized;
349}
350
351function encodeInboundFrame(frame: unknown): Uint8Array {
352 return encoder.encode(`${JSON.stringify(normalizeLegacyInboundFrame(frame))}\n`);
353}
354
355function replayTransportError(
356 error: CodexAppServerReplayError,
357): CodexError.CodexAppServerTransportError {
358 return new CodexError.CodexAppServerTransportError({
359 operation: "read-input-stream",
360 cause: error,
361 });
362}
363
364export function layerReplay(
365 transcript: CodexAppServerReplayTranscript,
366): Layer.Layer<CodexClient.CodexAppServerClient, CodexAppServerReplayError> {
367 return Layer.effect(CodexClient.CodexAppServerClient, makeReplayClient(transcript));
368}
369
370export const makeReplayDriver = Effect.fn("effect-codex-app-server/replay.makeReplayDriver")(
371 function* (
372 transcript: CodexAppServerReplayTranscript,
373 options: Pick<CodexAppServerReplayDriver, "beforeEmitInbound"> = {},
374 ) {
375 return {
376 transcript,
377 state: yield* Ref.make<CodexAppServerReplayState>({ cursor: 0, failure: null }),
378 ...options,
379 } satisfies CodexAppServerReplayDriver;
380 },
381);
382
383export function layerReplayWithDriver(
384 driver: CodexAppServerReplayDriver,
385): Layer.Layer<CodexClient.CodexAppServerClient, CodexAppServerReplayError> {
386 return Layer.effect(CodexClient.CodexAppServerClient, makeReplayClientWithState(driver));
387}
388
389const makeReplayClient = Effect.fn("effect-codex-app-server/replay.makeReplayClient")(function* (
390 transcript: CodexAppServerReplayTranscript,
391 options: CodexClient.CodexAppServerClientOptions = {},
392) {
393 const driver = yield* makeReplayDriver(transcript);
394 return yield* makeReplayClientWithState(driver, options);
395});
396
397const makeReplayClientWithState = Effect.fn(
398 "effect-codex-app-server/replay.makeReplayClientWithState",
399)(function* (
400 driver: CodexAppServerReplayDriver,
401 options: CodexClient.CodexAppServerClientOptions = {},
402) {
403 const transcript = driver.transcript;
404 const ignoredThreadConfigKeys = recorderThreadConfigKeys(transcript);
405 const input = yield* Queue.unbounded<Uint8Array, Cause.Done<void>>();
406 const state = driver.state;
407 const outboundRemainder = yield* Ref.make("");
408 const decoder = new TextDecoder();
409
410 const failReplay = (error: CodexAppServerReplayError) =>
411 Ref.update(state, (current) => ({
412 ...current,
413 failure: current.failure ?? error,
414 })).pipe(Effect.andThen(Queue.end(input)));
415
416 const drainInbound = Effect.fn("effect-codex-app-server/replay.drainInbound")(function* () {
417 while (true) {
418 const current = yield* Ref.get(state);
419 if (current.failure) {
420 return;
421 }
422
423 const entry = transcript.entries[current.cursor];
424 if (!entry) {
425 return;
426 }
427
428 if (entry.type === "emit_inbound") {
429 if (driver.beforeEmitInbound !== undefined) {
430 yield* driver.beforeEmitInbound(entry);
431 }
432 if (entry.afterMs !== undefined && entry.afterMs > 0) {
433 yield* Effect.sleep(Duration.millis(entry.afterMs));
434 }
435 yield* Queue.offer(input, encodeInboundFrame(entry.frame));
436 yield* Ref.update(state, (latest) => ({ ...latest, cursor: latest.cursor + 1 }));
437 continue;
438 }
439
440 if (entry.type === "runtime_exit") {
441 yield* Ref.update(state, (latest) => ({ ...latest, cursor: latest.cursor + 1 }));
442 if (entry.status === "success") {
443 yield* Queue.end(input);
444 return;
445 }
446 yield* failReplay(
447 new CodexAppServerReplayRuntimeExitError({
448 scenario: transcript.scenario,
449 cursor: current.cursor,
450 status: entry.status,
451 ...(entry.error === undefined ? {} : { error: entry.error }),
452 }),
453 );
454 return;
455 }
456
457 return;
458 }
459 });
460
461 const processOutboundFrame = (actual: unknown) =>
462 drainInbound().pipe(
463 Effect.andThen(
464 Effect.gen(function* () {
465 const current = yield* Ref.get(state);
466 if (current.failure) {
467 return;
468 }
469
470 const entry = transcript.entries[current.cursor];
471 if (!entry) {
472 yield* failReplay(
473 new CodexAppServerReplayExhaustedError({
474 scenario: transcript.scenario,
475 cursor: current.cursor,
476 actual,
477 }),
478 );
479 return;
480 }
481
482 if (entry.type !== "expect_outbound") {
483 yield* failReplay(
484 new CodexAppServerReplayUnexpectedOutboundError({
485 scenario: transcript.scenario,
486 cursor: current.cursor,
487 expectedType: entry.type,
488 actual,
489 }),
490 );
491 return;
492 }
493
494 if (!sameFrame(entry.frame, actual, ignoredThreadConfigKeys)) {
495 yield* failReplay(
496 new CodexAppServerReplayFrameMismatchError({
497 scenario: transcript.scenario,
498 cursor: current.cursor,
499 ...(entry.label === undefined ? {} : { label: entry.label }),
500 expected: entry.frame,
501 actual,
502 }),
503 );
504 return;
505 }
506
507 yield* Ref.update(state, (latest) => ({ ...latest, cursor: latest.cursor + 1 }));
508 yield* drainInbound();
509 }),
510 ),
511 );
512
513 const processOutboundLine = (line: string) => {
514 const trimmed = line.trim();
515 if (trimmed.length === 0) {
516 return Effect.void;
517 }
518 return Effect.try({
519 try: () => decodeOutboundJsonFrame(trimmed),
520 catch: (cause) =>
521 new CodexAppServerReplayJsonParseError({
522 scenario: transcript.scenario,
523 line: trimmed,
524 cause,
525 }),
526 }).pipe(
527 Effect.matchEffect({
528 onFailure: failReplay,
529 onSuccess: processOutboundFrame,
530 }),
531 );
532 };
533
534 const processOutboundChunk = (chunk: string | Uint8Array) =>
535 Ref.modify(outboundRemainder, (current) => {
536 const text = typeof chunk === "string" ? chunk : decoder.decode(chunk, { stream: true });
537 const combined = current + text;
538 const lines = combined.split("\n");
539 const nextRemainder = lines.pop() ?? "";
540 return [lines.map((line) => line.replace(/\r$/, "")), nextRemainder] as const;
541 }).pipe(
542 Effect.flatMap((lines) => Effect.forEach(lines, processOutboundLine, { discard: true })),
543 );
544
545 const terminationError: Effect.Effect<CodexError.CodexAppServerError> = Ref.get(state).pipe(
546 Effect.map((current) =>
547 current.failure
548 ? replayTransportError(current.failure)
549 : new CodexError.CodexAppServerProcessExitedError({ code: 0 }),
550 ),
551 );
552
553 const stdio = Stdio.make({
554 args: Effect.succeed([]),
555 stdin: Stream.fromQueue(input),
556 stdout: () => Sink.forEach(processOutboundChunk),
557 stderr: () => Sink.drain,
558 });
559
560 yield* drainInbound();
561
562 return yield* CodexClient.make(stdio, options, terminationError);
563});