1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { CodexSettings, type ProviderReplayTranscript } from "@t3tools/contracts";
4import * as CodexClient from "effect-codex-app-server/client";
5import type * as CodexError from "effect-codex-app-server/errors";
6import * as CodexReplay from "effect-codex-app-server/replay";
7import * as Effect from "effect/Effect";
8import * as Layer from "effect/Layer";
9import * as Predicate from "effect/Predicate";
10import * as Schema from "effect/Schema";
11
12import * as ServerConfig from "../../config.ts";
13import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
14import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
15import { ProviderAdapterDriverCreateError } from "@t3tools/provider-core/server/adapterDriver";
16import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts";
17import {
18 makeReplayServerConfig,
19 type OrchestratorV2ProviderReplayHarness,
20} from "../testkit/ProviderReplayHarness.ts";
21import type { ProviderReplayGate } from "@t3tools/provider-testing/replayGate";
22import * as CodexAdapterV2 from "./CodexAdapterV2.ts";
23
24const DEFAULT_CODEX_SETTINGS = Schema.decodeSync(CodexSettings)({});
25
26export class CodexReplayTranscriptDecodeError extends Schema.TaggedError<CodexReplayTranscriptDecodeError>()(
27 "CodexReplayTranscriptDecodeError",
28 {
29 driver: Schema.optional(Schema.String),
30 protocol: Schema.optional(Schema.String),
31 scenario: Schema.optional(Schema.String),
32 cause: Schema.Defect(),
33 },
34) {
35 override get message(): string {
36 return `Failed to decode Codex app-server replay transcript for scenario ${this.scenario ?? "<unknown>"}.`;
37 }
38}
39
40export const CodexOrchestratorReplayHarnessError = Schema.Union([
41 CodexReplayTranscriptDecodeError,
42 CodexReplay.CodexAppServerReplayError,
43 ProviderAdapterDriverCreateError,
44]);
45export type CodexOrchestratorReplayHarnessError = typeof CodexOrchestratorReplayHarnessError.Type;
46
47export function withCodexReplayChildMetadata(
48 client: CodexClient.CodexAppServerClient["Service"],
49 transcript: CodexReplay.CodexAppServerReplayTranscript,
50 readMetadata: (
51 threadId: string,
52 method: "thread/read" | "thread/resume",
53 ) => Effect.Effect<unknown, CodexError.CodexAppServerError> = (threadId) =>
54 Effect.succeed({ thread: { id: threadId }, model: null }),
55): CodexClient.CodexAppServerClient["Service"] {
56 const childThreadIds = new Set(
57 transcript.entries.flatMap((entry) => {
58 if (entry.type !== "emit_inbound" || !Predicate.isObject(entry.frame)) return [];
59 const params = entry.frame.params;
60 if (!Predicate.isObject(params) || !Predicate.isObject(params.item)) return [];
61 const item = params.item;
62 if (item.type === "subAgentActivity" && typeof item.agentThreadId === "string") {
63 return [item.agentThreadId];
64 }
65 return item.type === "collabAgentToolCall" && Array.isArray(item.receiverThreadIds)
66 ? item.receiverThreadIds.filter(Predicate.isString)
67 : [];
68 }),
69 );
70 return {
71 ...client,
72 raw: {
73 ...client.raw,
74 request: (method, params) =>
75 (method === "thread/read" || method === "thread/resume") &&
76 Predicate.isObject(params) &&
77 (method === "thread/read" ? params.includeTurns === false : params.excludeTurns === true) &&
78 typeof params.threadId === "string" &&
79 childThreadIds.has(params.threadId)
80 ? readMetadata(params.threadId, method)
81 : client.raw.request(method, params),
82 },
83 };
84}
85
86function metadataFromTranscript(transcript: ProviderReplayTranscript): {
87 readonly provider?: string;
88 readonly protocol?: string;
89 readonly scenario?: string;
90} {
91 return {
92 provider: transcript.provider,
93 protocol: transcript.protocol,
94 scenario: transcript.scenario,
95 };
96}
97
98export function layer(input: {
99 readonly transcript: CodexReplay.CodexAppServerReplayTranscript;
100 readonly driver?: CodexReplay.CodexAppServerReplayDriver;
101}) {
102 const layerReplay =
103 input.driver === undefined
104 ? CodexReplay.layerReplay(input.transcript)
105 : CodexReplay.layerReplayWithDriver(input.driver);
106 const layerReplayClientFactory = Layer.succeed(CodexAdapterV2.CodexAppServerClientFactory, {
107 open: (openInput) =>
108 Effect.gen(function* () {
109 const context = yield* Layer.build(layerReplay).pipe(
110 Effect.mapError(
111 (cause) =>
112 new ProviderAdapter.ProviderAdapterOpenSessionError({
113 driver: CodexAdapterV2.CODEX_DRIVER_KIND,
114 providerSessionId: openInput.providerSessionId,
115 cause,
116 }),
117 ),
118 );
119 return yield* Effect.service(CodexClient.CodexAppServerClient).pipe(
120 Effect.map((client) => withCodexReplayChildMetadata(client, input.transcript)),
121 Effect.provide(context),
122 );
123 }),
124 });
125 const layerServerConfig = Layer.effect(
126 ServerConfig.ServerConfig,
127 makeReplayServerConfig(input.transcript.scenario).pipe(Effect.orDie),
128 ).pipe(Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)));
129 const layerRegistry = ProviderAdapterRegistry.layerFromAdaptersEffect(
130 CodexAdapterV2.createCodexAdapterV2({
131 instanceId: CodexAdapterV2.CODEX_DEFAULT_INSTANCE_ID,
132 displayName: undefined,
133 environment: [],
134 enabled: true,
135 config: DEFAULT_CODEX_SETTINGS,
136 }).pipe(Effect.map((adapter) => [adapter])),
137 ).pipe(
138 Layer.provide(
139 Layer.mergeAll(
140 layerReplayClientFactory,
141 layerServerConfig,
142 NodeServices.layer,
143 NodeCrypto.layer,
144 IdAllocator.layer,
145 ),
146 ),
147 );
148
149 return layerRegistry;
150}
151
152const decodeCodexAppServerReplayTranscript = Schema.decodeUnknownEffect(
153 CodexReplay.CodexAppServerReplayTranscript,
154);
155
156export const CodexOrchestratorReplayHarness: OrchestratorV2ProviderReplayHarness<
157 CodexReplay.CodexAppServerReplayTranscript,
158 CodexOrchestratorReplayHarnessError
159> = {
160 driver: CodexAdapterV2.CODEX_DRIVER_KIND,
161 decodeTranscript: (transcript) =>
162 decodeCodexAppServerReplayTranscript(transcript).pipe(
163 Effect.mapError(
164 (cause) =>
165 new CodexReplayTranscriptDecodeError({
166 ...metadataFromTranscript(transcript),
167 cause,
168 }),
169 ),
170 ),
171 makeProviderAdapterRegistryLayer: (
172 transcript,
173 options: { readonly replayGate?: ProviderReplayGate } = {},
174 ) => {
175 return Layer.effectContext(
176 Effect.gen(function* () {
177 const replayGate = options.replayGate;
178 if (replayGate !== undefined) {
179 yield* Effect.addFinalizer(() => Effect.sync(() => replayGate.releaseAll()));
180 }
181 const driver = yield* CodexReplay.makeReplayDriver(
182 transcript,
183 replayGate === undefined
184 ? {}
185 : {
186 beforeEmitInbound: (entry) =>
187 Effect.promise((signal) => replayGate.beforeEmit(entry.label, signal)),
188 },
189 );
190 return yield* Layer.build(layer({ transcript, driver }));
191 }),
192 );
193 },
194};