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