1import * as Context from "effect/Context";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Schema from "effect/Schema";
5import * as Scope from "effect/Scope";
6import * as Stdio from "effect/Stdio";
7import * as Stream from "effect/Stream";
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, 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 request: <M extends CodexRpc.ClientRequestMethod>(
44 method: M,
45 payload: CodexRpc.ClientRequestParamsByMethod[M],
46 ) => Effect.Effect<CodexRpc.ClientRequestResponsesByMethod[M], CodexError.CodexAppServerError>;
47 readonly notify: <M extends CodexRpc.ClientNotificationMethod>(
48 method: M,
49 payload: CodexRpc.ClientNotificationParamsByMethod[M],
50 ) => Effect.Effect<void, CodexError.CodexAppServerError>;
51 readonly handleServerRequest: <M extends CodexRpc.ServerRequestMethod>(
52 method: M,
53 handler: (
54 payload: CodexRpc.ServerRequestParamsByMethod[M],
55 ) => Effect.Effect<
56 CodexRpc.ServerRequestResponsesByMethod[M],
57 CodexError.CodexAppServerError
58 >,
59 ) => Effect.Effect<void>;
60 readonly handleServerNotification: <M extends CodexRpc.ServerNotificationMethod>(
61 method: M,
62 handler: (
63 payload: CodexRpc.ServerNotificationParamsByMethod[M],
64 ) => Effect.Effect<void, CodexError.CodexAppServerError>,
65 ) => Effect.Effect<void>;
66 readonly handleUnknownServerRequest: (
67 handler: (
68 method: string,
69 params: unknown,
70 ) => Effect.Effect<unknown, CodexError.CodexAppServerError>,
71 ) => Effect.Effect<void>;
72 readonly handleUnknownServerNotification: (
73 handler: (
74 method: string,
75 params: unknown,
76 ) => Effect.Effect<void, CodexError.CodexAppServerError>,
77 ) => Effect.Effect<void>;
78 }
79>()("effect-codex-app-server/client/CodexAppServerClient") {}
80
81type ServerRequestHandler = (
82 payload: unknown,
83) => Effect.Effect<unknown, CodexError.CodexAppServerError>;
84type ServerNotificationHandler = (
85 payload: unknown,
86) => Effect.Effect<void, CodexError.CodexAppServerError>;
87
88const V2TurnStartParamsWithCollaborationMode = CodexSchema.V2TurnStartParams.pipe(
89 Schema.fieldsAssign({
90 collaborationMode: Schema.optionalKey(CodexSchema.ClientRequest__CollaborationMode),
91 additionalContext: Schema.optionalKey(
92 Schema.Record(Schema.String, CodexSchema.V2TurnStartParams__AdditionalContextEntry),
93 ),
94 }),
95);
96
97export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make")(function* (
98 stdio: Stdio.Stdio,
99 options: CodexAppServerClientOptions = {},
100 terminationError?: Effect.Effect<CodexError.CodexAppServerError>,
101): Effect.fn.Return<CodexAppServerClient["Service"], never, Scope.Scope> {
102 const requestHandlers = new Map<string, ServerRequestHandler>();
103 const notificationHandlers = new Map<string, Array<ServerNotificationHandler>>();
104 let unknownRequestHandler:
105 | ((method: string, params: unknown) => Effect.Effect<unknown, CodexError.CodexAppServerError>)
106 | undefined;
107 let unknownNotificationHandler:
108 | ((method: string, params: unknown) => Effect.Effect<void, CodexError.CodexAppServerError>)
109 | undefined;
110
111 const getServerRequestParamSchema = <M extends CodexRpc.ServerRequestMethod>(
112 method: M,
113 ):
114 | Schema.Codec<CodexRpc.ServerRequestParamsByMethod[M], CodexRpc.ServerRequestParamsByMethod[M]>
115 | undefined => CodexRpc.SERVER_REQUEST_PARAMS[method] as never;
116
117 const getServerRequestResponseSchema = <M extends CodexRpc.ServerRequestMethod>(
118 method: M,
119 ):
120 | Schema.Codec<
121 CodexRpc.ServerRequestResponsesByMethod[M],
122 CodexRpc.ServerRequestResponsesByMethod[M]
123 >
124 | undefined => CodexRpc.SERVER_REQUEST_RESPONSES[method] as never;
125
126 const getClientRequestParamSchema = <M extends CodexRpc.ClientRequestMethod>(
127 method: M,
128 ):
129 | Schema.Codec<CodexRpc.ClientRequestParamsByMethod[M], CodexRpc.ClientRequestParamsByMethod[M]>
130 | undefined =>
131 method === "turn/start"
132 ? (V2TurnStartParamsWithCollaborationMode as never)
133 : (CodexRpc.CLIENT_REQUEST_PARAMS[method] as never);
134
135 const getClientRequestResponseSchema = <M extends CodexRpc.ClientRequestMethod>(
136 method: M,
137 ):
138 | Schema.Codec<
139 CodexRpc.ClientRequestResponsesByMethod[M],
140 CodexRpc.ClientRequestResponsesByMethod[M]
141 >
142 | undefined => CodexRpc.CLIENT_REQUEST_RESPONSES[method] as never;
143
144 const getClientNotificationParamSchema = <M extends CodexRpc.ClientNotificationMethod>(
145 method: M,
146 ):
147 | Schema.Codec<
148 CodexRpc.ClientNotificationParamsByMethod[M],
149 CodexRpc.ClientNotificationParamsByMethod[M]
150 >
151 | undefined => CodexRpc.CLIENT_NOTIFICATION_PARAMS[method] as never;
152
153 const dispatchNotification = (
154 notification: CodexProtocol.CodexAppServerIncomingNotification,
155 ): Effect.Effect<void, never> => {
156 const schema =
157 notification.method in CodexRpc.SERVER_NOTIFICATION_PARAMS
158 ? CodexRpc.SERVER_NOTIFICATION_PARAMS[
159 notification.method as CodexRpc.ServerNotificationMethod
160 ]
161 : undefined;
162 const handlers = notificationHandlers.get(notification.method) ?? [];
163
164 if (schema) {
165 return decodeNotificationPayload(notification.method, schema, notification.params).pipe(
166 Effect.flatMap((decoded) =>
167 Effect.forEach(handlers, (handler) => handler(decoded), { discard: true }),
168 ),
169 Effect.ignore,
170 );
171 }
172
173 return unknownNotificationHandler
174 ? unknownNotificationHandler(notification.method, notification.params).pipe(Effect.ignore)
175 : Effect.void;
176 };
177
178 const dispatchRequest = (
179 request: CodexProtocol.CodexAppServerIncomingRequest,
180 ): Effect.Effect<unknown, CodexError.CodexAppServerError> => {
181 if (request.method in CodexRpc.SERVER_REQUEST_PARAMS) {
182 const method = request.method as CodexRpc.ServerRequestMethod;
183 const payloadSchema = getServerRequestParamSchema(method);
184 const responseSchema = getServerRequestResponseSchema(method);
185 const handler = requestHandlers.get(method);
186
187 return decodeOptionalPayload(method, payloadSchema, request.params).pipe(
188 Effect.flatMap((decoded) => runHandler(handler, decoded, method)),
189 Effect.flatMap((result) => encodeOptionalPayload(method, responseSchema, result)),
190 );
191 }
192
193 return unknownRequestHandler
194 ? unknownRequestHandler(request.method, request.params)
195 : Effect.fail(CodexError.CodexAppServerRequestError.methodNotFound(request.method));
196 };
197
198 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
199 stdio,
200 ...(terminationError ? { terminationError } : {}),
201 ...(options.logIncoming !== undefined ? { logIncoming: options.logIncoming } : {}),
202 ...(options.logOutgoing !== undefined ? { logOutgoing: options.logOutgoing } : {}),
203 ...(options.logger ? { logger: options.logger } : {}),
204 onNotification: dispatchNotification,
205 onRequest: dispatchRequest,
206 });
207
208 const request = <M extends CodexRpc.ClientRequestMethod>(
209 method: M,
210 payload: CodexRpc.ClientRequestParamsByMethod[M],
211 ): Effect.Effect<CodexRpc.ClientRequestResponsesByMethod[M], CodexError.CodexAppServerError> =>
212 encodeOptionalPayload(method, getClientRequestParamSchema(method), payload).pipe(
213 Effect.flatMap((encoded) => transport.request(method, encoded)),
214 Effect.flatMap(
215 (
216 raw,
217 ): Effect.Effect<
218 CodexRpc.ClientRequestResponsesByMethod[M],
219 CodexError.CodexAppServerError
220 > => decodeOptionalPayload(method, getClientRequestResponseSchema(method), raw),
221 ),
222 );
223
224 const notify = <M extends CodexRpc.ClientNotificationMethod>(
225 method: M,
226 payload: CodexRpc.ClientNotificationParamsByMethod[M],
227 ) =>
228 encodeOptionalPayload(method, getClientNotificationParamSchema(method), payload).pipe(
229 Effect.flatMap((encoded) => transport.notify(method, encoded)),
230 );
231
232 return CodexAppServerClient.of({
233 raw: {
234 notifications: transport.incomingNotifications,
235 requests: transport.incomingRequests,
236 request: transport.request,
237 notify: transport.notify,
238 respond: transport.respond,
239 respondError: transport.respondError,
240 },
241 request,
242 notify,
243 handleServerRequest: (method, handler) =>
244 Effect.sync(() => {
245 requestHandlers.set(method, handler as ServerRequestHandler);
246 }),
247 handleServerNotification: (method, handler) =>
248 Effect.sync(() => {
249 const current = notificationHandlers.get(method) ?? [];
250 current.push(handler as ServerNotificationHandler);
251 notificationHandlers.set(method, current);
252 }),
253 handleUnknownServerRequest: (handler) =>
254 Effect.sync(() => {
255 unknownRequestHandler = handler;
256 }),
257 handleUnknownServerNotification: (handler) =>
258 Effect.sync(() => {
259 unknownNotificationHandler = handler;
260 }),
261 });
262});
263
264export const layerChildProcess = (
265 handle: ChildProcessSpawner.ChildProcessHandle,
266 options: CodexAppServerClientOptions = {},
267): Layer.Layer<CodexAppServerClient> =>
268 Layer.effect(CodexAppServerClient, makeChildProcessClient(handle, options));
269
270const makeChildProcessClient = Effect.fn(
271 "effect-codex-app-server/CodexAppServerClient.makeChildProcessClient",
272)(function* (handle: ChildProcessSpawner.ChildProcessHandle, options: CodexAppServerClientOptions) {
273 yield* Stream.runDrain(handle.stderr).pipe(Effect.ignore, Effect.forkScoped);
274 return yield* make(makeChildStdio(handle), options, makeTerminationError(handle));
275});