1import * as Context from "effect/Context";
2import * as Deferred from "effect/Deferred";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Schema from "effect/Schema";
6import * as Scope from "effect/Scope";
7import * as Stdio from "effect/Stdio";
8import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
9
10import * as CodexRpc from "./_generated/meta.gen.ts";
11import * as CodexSchema from "./_generated/schema.gen.ts";
12import * as CodexError from "./errors.ts";
13import * as CodexProtocol from "./protocol.ts";
14import {
15 decodeNotificationPayload,
16 decodeOptionalPayload,
17 encodeOptionalPayload,
18 runHandler,
19} from "./_internal/shared.ts";
20import { makeChildStdio, makeStderrTail, makeTerminationError } from "./_internal/stdio.ts";
21
22export interface CodexAppServerClientOptions {
23 readonly logIncoming?: boolean;
24 readonly logOutgoing?: boolean;
25 readonly logger?: (
26 event: CodexProtocol.CodexAppServerProtocolLogEvent,
27 ) => Effect.Effect<void, never>;
28}
29
30interface CodexAppServerClientRaw {
31 readonly notifications: CodexProtocol.CodexAppServerPatchedProtocol["incomingNotifications"];
32 readonly requests: CodexProtocol.CodexAppServerPatchedProtocol["incomingRequests"];
33 readonly request: CodexProtocol.CodexAppServerPatchedProtocol["request"];
34 readonly notify: CodexProtocol.CodexAppServerPatchedProtocol["notify"];
35 readonly respond: CodexProtocol.CodexAppServerPatchedProtocol["respond"];
36 readonly respondError: CodexProtocol.CodexAppServerPatchedProtocol["respondError"];
37}
38
39export class CodexAppServerClient extends Context.Service<
40 CodexAppServerClient,
41 {
42 readonly raw: CodexAppServerClientRaw;
43 readonly awaitTermination: Effect.Effect<never, CodexError.CodexAppServerError>;
44 readonly request: <M extends CodexRpc.ClientRequestMethod>(
45 method: M,
46 payload: CodexRpc.ClientRequestParamsByMethod[M],
47 ) => Effect.Effect<CodexRpc.ClientRequestResponsesByMethod[M], CodexError.CodexAppServerError>;
48 readonly notify: <M extends CodexRpc.ClientNotificationMethod>(
49 method: M,
50 payload: CodexRpc.ClientNotificationParamsByMethod[M],
51 ) => Effect.Effect<void, CodexError.CodexAppServerError>;
52 readonly handleServerRequest: <M extends CodexRpc.ServerRequestMethod>(
53 method: M,
54 handler: (
55 payload: CodexRpc.ServerRequestParamsByMethod[M],
56 ) => Effect.Effect<
57 CodexRpc.ServerRequestResponsesByMethod[M],
58 CodexError.CodexAppServerError
59 >,
60 ) => Effect.Effect<void>;
61 readonly handleServerNotification: <M extends CodexRpc.ServerNotificationMethod>(
62 method: M,
63 handler: (
64 payload: CodexRpc.ServerNotificationParamsByMethod[M],
65 ) => Effect.Effect<void, CodexError.CodexAppServerError>,
66 ) => Effect.Effect<void>;
67 readonly handleUnknownServerRequest: (
68 handler: (
69 method: string,
70 params: unknown,
71 ) => Effect.Effect<unknown, CodexError.CodexAppServerError>,
72 ) => Effect.Effect<void>;
73 readonly handleUnknownServerNotification: (
74 handler: (
75 method: string,
76 params: unknown,
77 ) => Effect.Effect<void, CodexError.CodexAppServerError>,
78 ) => Effect.Effect<void>;
79 }
80>()("effect-codex-app-server/client/CodexAppServerClient") {}
81
82type ServerRequestHandler = (
83 payload: unknown,
84) => Effect.Effect<unknown, CodexError.CodexAppServerError>;
85type ServerNotificationHandler = (
86 payload: unknown,
87) => Effect.Effect<void, CodexError.CodexAppServerError>;
88
89/** The generated `turn/start` schema omits these fields; this is what the wire accepts. */
90export const V2TurnStartParams = CodexSchema.V2TurnStartParams.pipe(
91 Schema.fieldsAssign({
92 collaborationMode: Schema.optionalKey(CodexSchema.ClientRequest__CollaborationMode),
93 additionalContext: Schema.optionalKey(
94 Schema.Record(Schema.String, CodexSchema.V2TurnStartParams__AdditionalContextEntry),
95 ),
96 cyberAccessProgram: Schema.optionalKey(CodexSchema.V2TurnStartParams__CyberAccessProgram),
97 }),
98);
99
100export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make")(function* (
101 stdio: Stdio.Stdio,
102 options: CodexAppServerClientOptions = {},
103 terminationError?: Effect.Effect<CodexError.CodexAppServerError>,
104): Effect.fn.Return<CodexAppServerClient["Service"], never, Scope.Scope> {
105 const requestHandlers = new Map<string, ServerRequestHandler>();
106 const termination = yield* Deferred.make<never, CodexError.CodexAppServerError>();
107 const notificationHandlers = new Map<string, Array<ServerNotificationHandler>>();
108 let unknownRequestHandler:
109 | ((method: string, params: unknown) => Effect.Effect<unknown, CodexError.CodexAppServerError>)
110 | undefined;
111 let unknownNotificationHandler:
112 | ((method: string, params: unknown) => Effect.Effect<void, CodexError.CodexAppServerError>)
113 | undefined;
114
115 const getServerRequestParamSchema = <M extends CodexRpc.ServerRequestMethod>(
116 method: M,
117 ):
118 | Schema.Codec<CodexRpc.ServerRequestParamsByMethod[M], CodexRpc.ServerRequestParamsByMethod[M]>
119 | undefined => CodexRpc.SERVER_REQUEST_PARAMS[method] as never;
120
121 const getServerRequestResponseSchema = <M extends CodexRpc.ServerRequestMethod>(
122 method: M,
123 ):
124 | Schema.Codec<
125 CodexRpc.ServerRequestResponsesByMethod[M],
126 CodexRpc.ServerRequestResponsesByMethod[M]
127 >
128 | undefined => CodexRpc.SERVER_REQUEST_RESPONSES[method] as never;
129
130 const getClientRequestParamSchema = <M extends CodexRpc.ClientRequestMethod>(
131 method: M,
132 ):
133 | Schema.Codec<CodexRpc.ClientRequestParamsByMethod[M], CodexRpc.ClientRequestParamsByMethod[M]>
134 | undefined =>
135 method === "turn/start"
136 ? (V2TurnStartParams as never)
137 : (CodexRpc.CLIENT_REQUEST_PARAMS[method] as never);
138
139 const getClientRequestResponseSchema = <M extends CodexRpc.ClientRequestMethod>(
140 method: M,
141 ):
142 | Schema.Codec<
143 CodexRpc.ClientRequestResponsesByMethod[M],
144 CodexRpc.ClientRequestResponsesByMethod[M]
145 >
146 | undefined => CodexRpc.CLIENT_REQUEST_RESPONSES[method] as never;
147
148 const getClientNotificationParamSchema = <M extends CodexRpc.ClientNotificationMethod>(
149 method: M,
150 ):
151 | Schema.Codec<
152 CodexRpc.ClientNotificationParamsByMethod[M],
153 CodexRpc.ClientNotificationParamsByMethod[M]
154 >
155 | undefined => CodexRpc.CLIENT_NOTIFICATION_PARAMS[method] as never;
156
157 // A notification the schema rejects has no reply to fail, so it is dropped.
158 // A Codex build that drifted from the generated schema repeats the bad shape
159 // on every frame, so each distinct failure is warned about once per client.
160 const warnedDecodeFailures = new Set<string>();
161 const warnUndecodableNotification = (error: CodexError.CodexAppServerProtocolParseError) => {
162 const key = `${error.method}\n${error.issueKinds?.join(",")}`;
163 if (warnedDecodeFailures.has(key)) {
164 return Effect.void;
165 }
166 warnedDecodeFailures.add(key);
167 return Effect.logWarning("Dropped a Codex App Server notification that failed to decode").pipe(
168 Effect.annotateLogs({
169 operation: error.operation,
170 ...(error.method === undefined ? {} : { method: error.method }),
171 ...(error.issueCount === undefined ? {} : { issueCount: error.issueCount }),
172 ...(error.issueKinds === undefined ? {} : { issueKinds: error.issueKinds }),
173 ...(error.maximumPathDepth === undefined
174 ? {}
175 : { maximumPathDepth: error.maximumPathDepth }),
176 }),
177 );
178 };
179
180 const dispatchNotification = (
181 notification: CodexProtocol.CodexAppServerIncomingNotification,
182 ): Effect.Effect<void, never> => {
183 const schema =
184 notification.method in CodexRpc.SERVER_NOTIFICATION_PARAMS
185 ? CodexRpc.SERVER_NOTIFICATION_PARAMS[
186 notification.method as CodexRpc.ServerNotificationMethod
187 ]
188 : undefined;
189 const handlers = notificationHandlers.get(notification.method) ?? [];
190
191 if (schema) {
192 return decodeNotificationPayload(notification.method, schema, notification.params).pipe(
193 Effect.tapError(warnUndecodableNotification),
194 Effect.flatMap((decoded) =>
195 Effect.forEach(handlers, (handler) => handler(decoded), { discard: true }),
196 ),
197 Effect.ignore,
198 );
199 }
200
201 return unknownNotificationHandler
202 ? unknownNotificationHandler(notification.method, notification.params).pipe(Effect.ignore)
203 : Effect.void;
204 };
205
206 const dispatchRequest = (
207 request: CodexProtocol.CodexAppServerIncomingRequest,
208 ): Effect.Effect<unknown, CodexError.CodexAppServerError> => {
209 if (request.method in CodexRpc.SERVER_REQUEST_PARAMS) {
210 const method = request.method as CodexRpc.ServerRequestMethod;
211 const payloadSchema = getServerRequestParamSchema(method);
212 const responseSchema = getServerRequestResponseSchema(method);
213 const handler = requestHandlers.get(method);
214
215 return decodeOptionalPayload(method, payloadSchema, request.params).pipe(
216 Effect.flatMap((decoded) => runHandler(handler, decoded, method)),
217 Effect.flatMap((result) => encodeOptionalPayload(method, responseSchema, result)),
218 );
219 }
220
221 return unknownRequestHandler
222 ? unknownRequestHandler(request.method, request.params)
223 : Effect.fail(CodexError.CodexAppServerRequestError.methodNotFound(request.method));
224 };
225
226 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
227 stdio,
228 ...(terminationError ? { terminationError } : {}),
229 ...(options.logIncoming !== undefined ? { logIncoming: options.logIncoming } : {}),
230 ...(options.logOutgoing !== undefined ? { logOutgoing: options.logOutgoing } : {}),
231 ...(options.logger ? { logger: options.logger } : {}),
232 onNotification: dispatchNotification,
233 onRequest: dispatchRequest,
234 onTermination: (error) => Deferred.fail(termination, error).pipe(Effect.asVoid),
235 });
236
237 const request = <M extends CodexRpc.ClientRequestMethod>(
238 method: M,
239 payload: CodexRpc.ClientRequestParamsByMethod[M],
240 ): Effect.Effect<CodexRpc.ClientRequestResponsesByMethod[M], CodexError.CodexAppServerError> =>
241 encodeOptionalPayload(method, getClientRequestParamSchema(method), payload).pipe(
242 Effect.flatMap((encoded) => transport.request(method, encoded)),
243 Effect.flatMap(
244 (
245 raw,
246 ): Effect.Effect<
247 CodexRpc.ClientRequestResponsesByMethod[M],
248 CodexError.CodexAppServerError
249 > => decodeOptionalPayload(method, getClientRequestResponseSchema(method), raw),
250 ),
251 );
252
253 const notify = <M extends CodexRpc.ClientNotificationMethod>(
254 method: M,
255 payload: CodexRpc.ClientNotificationParamsByMethod[M],
256 ) =>
257 encodeOptionalPayload(method, getClientNotificationParamSchema(method), payload).pipe(
258 Effect.flatMap((encoded) => transport.notify(method, encoded)),
259 );
260
261 return CodexAppServerClient.of({
262 awaitTermination: Deferred.await(termination),
263 raw: {
264 notifications: transport.incomingNotifications,
265 requests: transport.incomingRequests,
266 request: transport.request,
267 notify: transport.notify,
268 respond: transport.respond,
269 respondError: transport.respondError,
270 },
271 request,
272 notify,
273 handleServerRequest: Effect.fn("CodexAppServerClient.handleServerRequest")(
274 function (method, handler) {
275 return Effect.sync(() => {
276 requestHandlers.set(method, handler as ServerRequestHandler);
277 });
278 },
279 ),
280 handleServerNotification: Effect.fn("CodexAppServerClient.handleServerNotification")(
281 function (method, handler) {
282 return Effect.sync(() => {
283 const current = notificationHandlers.get(method) ?? [];
284 current.push(handler as ServerNotificationHandler);
285 notificationHandlers.set(method, current);
286 });
287 },
288 ),
289 handleUnknownServerRequest: Effect.fn("CodexAppServerClient.handleUnknownServerRequest")(
290 function (handler) {
291 return Effect.sync(() => {
292 unknownRequestHandler = handler;
293 });
294 },
295 ),
296 handleUnknownServerNotification: Effect.fn(
297 "CodexAppServerClient.handleUnknownServerNotification",
298 )(function (handler) {
299 return Effect.sync(() => {
300 unknownNotificationHandler = handler;
301 });
302 }),
303 });
304});
305
306export const layerChildProcess = (
307 handle: ChildProcessSpawner.ChildProcessHandle,
308 options: CodexAppServerClientOptions = {},
309): Layer.Layer<CodexAppServerClient> =>
310 Layer.effect(CodexAppServerClient, makeChildProcessClient(handle, options));
311
312const makeChildProcessClient = Effect.fn(
313 "effect-codex-app-server/CodexAppServerClient.makeChildProcessClient",
314)(function* (handle: ChildProcessSpawner.ChildProcessHandle, options: CodexAppServerClientOptions) {
315 const stderrTail = yield* makeStderrTail(handle.stderr);
316 return yield* make(
317 makeChildStdio(handle, stderrTail.awaitDrained),
318 options,
319 makeTerminationError(handle, stderrTail.read),
320 );
321});