1import * as Deferred from "effect/Deferred";
2import * as Effect from "effect/Effect";
3import * as Fiber from "effect/Fiber";
4import * as PlatformError from "effect/PlatformError";
5import * as Queue from "effect/Queue";
6import * as Schema from "effect/Schema";
7import * as Sink from "effect/Sink";
8import * as Stdio from "effect/Stdio";
9import * as Stream from "effect/Stream";
10import * as Tracer from "effect/Tracer";
11import * as TestClock from "effect/testing/TestClock";
12
13import * as NodeServices from "@effect/platform-node/NodeServices";
14import { assert, it } from "@effect/vitest";
15
16import * as CodexError from "./errors.ts";
17import * as CodexProtocol from "./protocol.ts";
18import { makeInMemoryStdio } from "./_internal/stdio.ts";
19const encodeUnknownJsonString = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
20
21const encoder = new TextEncoder();
22
23const encodeJsonl = (value: unknown) => encoder.encode(`${encodeUnknownJsonString(value)}\n`);
24
25const decodeJson = Schema.decodeEffect(Schema.fromJsonString(Schema.Unknown));
26
27it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {
28 it.effect(
29 "encodes requests without a jsonrpc field and routes inbound requests and notifications",
30 () =>
31 Effect.gen(function* () {
32 const { stdio, input, output } = yield* makeInMemoryStdio();
33 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ stdio });
34
35 const notificationDeferred =
36 yield* Deferred.make<ReadonlyArray<CodexProtocol.CodexAppServerIncomingNotification>>();
37 const requestDeferred =
38 yield* Deferred.make<ReadonlyArray<CodexProtocol.CodexAppServerIncomingRequest>>();
39
40 yield* transport.incomingNotifications.pipe(
41 Stream.take(1),
42 Stream.runCollect,
43 Effect.flatMap((notifications) => Deferred.succeed(notificationDeferred, notifications)),
44 Effect.forkScoped,
45 );
46
47 yield* transport.incomingRequests.pipe(
48 Stream.take(1),
49 Stream.runCollect,
50 Effect.flatMap((requests) => Deferred.succeed(requestDeferred, requests)),
51 Effect.forkScoped,
52 );
53
54 yield* transport.notify("initialized");
55 assert.equal(yield* Queue.take(output), '{"method":"initialized"}\n');
56
57 const initializeParams = {
58 clientInfo: {
59 name: "effect-codex-app-server-test",
60 title: "Effect Codex App Server Test",
61 version: "0.0.0",
62 },
63 capabilities: {
64 experimentalApi: true,
65 optOutNotificationMethods: null,
66 },
67 };
68
69 const pendingInitialize = yield* transport
70 .request("initialize", initializeParams)
71 .pipe(Effect.forkScoped);
72 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
73 id: 1,
74 method: "initialize",
75 params: initializeParams,
76 });
77
78 yield* Queue.offer(
79 input,
80 encodeJsonl({
81 method: "item/agentMessage/delta",
82 params: {
83 delta: "Hello from the mock peer.",
84 itemId: "item-1",
85 threadId: "thread-1",
86 turnId: "turn-1",
87 },
88 }),
89 );
90 yield* Queue.offer(
91 input,
92 encodeJsonl({
93 id: 77,
94 method: "item/tool/requestUserInput",
95 params: {
96 itemId: "item-approval-1",
97 threadId: "thread-1",
98 turnId: "turn-1",
99 questions: [
100 {
101 id: "approved",
102 header: "Approve",
103 question: "Continue?",
104 },
105 ],
106 },
107 }),
108 );
109 yield* Queue.offer(
110 input,
111 encodeJsonl({
112 id: 1,
113 result: {
114 userAgent: "mock-codex-app-server",
115 codexHome: "/tmp/codex-home",
116 platformFamily: "unix",
117 platformOs: "macos",
118 },
119 }),
120 );
121
122 assert.deepEqual(yield* Fiber.join(pendingInitialize), {
123 userAgent: "mock-codex-app-server",
124 codexHome: "/tmp/codex-home",
125 platformFamily: "unix",
126 platformOs: "macos",
127 });
128 assert.deepEqual(yield* Deferred.await(notificationDeferred), [
129 {
130 method: "item/agentMessage/delta",
131 params: {
132 delta: "Hello from the mock peer.",
133 itemId: "item-1",
134 threadId: "thread-1",
135 turnId: "turn-1",
136 },
137 },
138 ]);
139 assert.deepEqual(yield* Deferred.await(requestDeferred), [
140 {
141 id: 77,
142 method: "item/tool/requestUserInput",
143 params: {
144 itemId: "item-approval-1",
145 threadId: "thread-1",
146 turnId: "turn-1",
147 questions: [
148 {
149 id: "approved",
150 header: "Approve",
151 question: "Continue?",
152 },
153 ],
154 },
155 },
156 ]);
157
158 yield* transport.respond(77, {
159 answers: {
160 approved: {
161 answers: ["yes"],
162 },
163 },
164 });
165 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
166 id: 77,
167 result: {
168 answers: {
169 approved: {
170 answers: ["yes"],
171 },
172 },
173 },
174 });
175
176 yield* transport.respondError(
177 78,
178 CodexError.CodexAppServerRequestError.methodNotFound("x/test"),
179 );
180 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
181 id: 78,
182 error: {
183 code: -32601,
184 message: "Method not found: x/test",
185 },
186 });
187 }),
188 );
189
190 it.effect(
191 "correlates concurrent request spans through responses, failures and interruption",
192 () => {
193 const spans: Array<Tracer.NativeSpan> = [];
194 const tracer = Tracer.make({
195 span: (options) => {
196 const span = new Tracer.NativeSpan(options);
197 spans.push(span);
198 return span;
199 },
200 });
201 return Effect.gen(function* () {
202 const { stdio, input, output } = yield* makeInMemoryStdio();
203 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ stdio });
204 const methods = ["model/list", "account/read", "skills/list"];
205 const requests = [];
206 for (const method of methods) {
207 requests.push(
208 yield* transport.request(method, { secret: "private-payload" }).pipe(Effect.forkScoped),
209 );
210 yield* Queue.take(output);
211 }
212 yield* Queue.offer(
213 input,
214 encodeJsonl({ id: 2, error: { code: -32602, message: "Invalid params" } }),
215 );
216 yield* Fiber.await(requests[1]!);
217 yield* Queue.offer(input, encodeJsonl({ id: 1, result: { secret: "private-result" } }));
218 yield* Fiber.join(requests[0]!);
219 yield* Fiber.interrupt(requests[2]!);
220 // A late response for an interrupted request must not affect the next request.
221 yield* Queue.offer(input, encodeJsonl({ id: 3, result: {} }));
222 const next = yield* transport.request("config/read").pipe(Effect.forkScoped);
223 yield* Queue.take(output);
224 yield* Queue.offer(input, encodeJsonl({ id: 4, result: {} }));
225 yield* Fiber.join(next);
226 for (const [index, method] of [...methods, "config/read"].entries()) {
227 const request = spans.find(
228 (span) =>
229 span.name === "CodexAppServerProtocol.request" &&
230 span.attributes.get("rpc.request_id") === index + 1,
231 );
232 assert.isDefined(request);
233 assert.equal(request!.attributes.get("rpc.method"), method);
234 assert.equal(request!.attributes.get("rpc.system"), "jsonrpc");
235 assert.equal(request!.status._tag, "Ended");
236 if (request!.status._tag === "Ended") {
237 assert.equal(
238 request!.status.exit._tag,
239 index === 1 || index === 2 ? "Failure" : "Success",
240 );
241 }
242 for (const phase of ["enqueue", "awaitResponse"]) {
243 const child: Tracer.NativeSpan | undefined = spans.find(
244 (span) =>
245 span.name === `CodexAppServerProtocol.request.${phase}` &&
246 span.parent?._tag === "Some" &&
247 span.parent.value.spanId === request!.spanId,
248 );
249 assert.isDefined(child);
250 assert.equal(child!.traceId, request!.traceId);
251 assert.equal(child!.attributes.get("rpc.method"), method);
252 assert.equal(child!.attributes.get("rpc.request_id"), index + 1);
253 assert.equal(child!.status._tag, "Ended");
254 }
255 }
256 const failedSend = yield* transport
257 .request("thread/start", { value: 1n })
258 .pipe(Effect.exit);
259 assert.equal(failedSend._tag, "Failure");
260 const sendSpans = spans.filter((span) => span.attributes.get("rpc.request_id") === 5);
261 assert.deepEqual(
262 sendSpans.map((span) => span.name),
263 ["CodexAppServerProtocol.request", "CodexAppServerProtocol.request.enqueue"],
264 );
265 for (const span of sendSpans) {
266 assert.equal(span.attributes.get("rpc.method"), "thread/start");
267 assert.equal(span.status._tag, "Ended");
268 if (span.status._tag === "Ended") assert.equal(span.status.exit._tag, "Failure");
269 }
270 for (const span of spans) {
271 const attrs = JSON.stringify([...span.attributes]);
272 assert.notInclude(attrs, "private-payload");
273 assert.notInclude(attrs, "private-result");
274 }
275 }).pipe(Effect.withTracer(tracer));
276 },
277 );
278
279 it.effect.each([1, 7, 1024])(
280 "preserves JSONL framing and UTF-8 across %i-byte input chunks",
281 (chunkSize) =>
282 Effect.gen(function* () {
283 const { stdio, input } = yield* makeInMemoryStdio();
284 const notifications: Array<CodexProtocol.CodexAppServerIncomingNotification> = [];
285 const rawLines: Array<unknown> = [];
286 const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
287 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
288 stdio,
289 logIncoming: true,
290 logger: (event) =>
291 Effect.sync(() => {
292 if (event.stage === "raw") {
293 rawLines.push(event.payload);
294 }
295 }),
296 onNotification: (notification) =>
297 Effect.sync(() => {
298 notifications.push(notification);
299 }),
300 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
301 });
302
303 const firstLine = '{"method":"x/first",\r"params":{"text":"hé🙂"}}';
304 const secondLine = '{"method":"x/second","params":{"value":2}}';
305 const finalLine = '{"method":"x/final","params":{"text":"最後"}}\r';
306 const bytes = encoder.encode(`\n \t\r\n${firstLine}\r\n\n${secondLine}\n${finalLine}`);
307 for (let offset = 0; offset < bytes.length; offset += chunkSize) {
308 yield* Queue.offer(input, bytes.subarray(offset, offset + chunkSize));
309 }
310 yield* Queue.end(input);
311
312 assert.instanceOf(
313 yield* Deferred.await(termination),
314 CodexError.CodexAppServerInputStreamEndedError,
315 );
316 assert.deepEqual(notifications, [
317 { method: "x/first", params: { text: "hé🙂" } },
318 { method: "x/second", params: { value: 2 } },
319 { method: "x/final", params: { text: "最後" } },
320 ]);
321 assert.deepEqual(rawLines, [firstLine, secondLine, finalLine]);
322 }),
323 );
324
325 it.effect.each(["newline-terminated", "unfinished"])(
326 "rejects an oversized %s message using UTF-8 bytes before logging or parsing",
327 (framing) =>
328 Effect.gen(function* () {
329 const { stdio, input, output } = yield* makeInMemoryStdio();
330 const notifications: Array<CodexProtocol.CodexAppServerIncomingNotification> = [];
331 const logged: Array<unknown> = [];
332 const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
333 let reachedEnd = false;
334 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
335 stdio: Stdio.make({
336 ...stdio,
337 stdin: stdio.stdin.pipe(
338 Stream.concat(
339 Stream.suspend(() => {
340 reachedEnd = true;
341 return Stream.empty;
342 }),
343 ),
344 ),
345 }),
346 maxIncomingMessageBytes: 16,
347 logIncoming: true,
348 logger: (event) =>
349 Effect.sync(() => {
350 logged.push(event.payload);
351 }),
352 onNotification: (notification) =>
353 Effect.sync(() => {
354 notifications.push(notification);
355 }),
356 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
357 });
358 const pending = yield* Effect.forEach(["thread/read", "thread/list"], (method) =>
359 transport.request(method, {}).pipe(Effect.forkScoped),
360 );
361 yield* Effect.forEach(pending, () => Queue.take(output));
362
363 // Two 16-byte lines must survive; the 17-byte line is only 15 UTF-16 code units.
364 const firstLine = '{"method":"xé"}';
365 const secondLine = '{"method":"x/a"}';
366 const bytes = encoder.encode(
367 `${firstLine}\n${secondLine}\n{"method":"éé"}${framing === "unfinished" ? "" : "\n"}`,
368 );
369 // Split a UTF-8 sequence on both sides of the byte limit.
370 const firstSplit = bytes.indexOf(0xc3) + 1;
371 yield* Queue.offer(input, bytes.subarray(0, firstSplit));
372 if (framing === "unfinished") {
373 const lastSplit = bytes.lastIndexOf(0xc3) + 1;
374 yield* Queue.offer(input, bytes.subarray(firstSplit, lastSplit));
375 yield* Queue.offer(input, bytes.subarray(lastSplit));
376 } else {
377 yield* Queue.offer(input, bytes.subarray(firstSplit));
378 }
379 yield* Queue.end(input);
380
381 const error = yield* Deferred.await(termination);
382 assert.instanceOf(error, CodexError.CodexAppServerTransportError);
383 assert.equal(error.operation, "read-input-stream");
384 assert.isFalse(reachedEnd, "Must reject the unfinished line before reading EOF");
385 for (const response of pending) {
386 const failure = yield* Fiber.join(response).pipe(
387 Effect.match({
388 onFailure: (error) => error,
389 onSuccess: () => assert.fail("Expected the oversized message to fail the request"),
390 }),
391 );
392 assert.strictEqual(failure, error);
393 }
394 assert.deepEqual(notifications, [{ method: "xé" }, { method: "x/a" }]);
395 assert.deepEqual(logged, [firstLine, { method: "xé" }, secondLine, { method: "x/a" }]);
396 }),
397 );
398
399 it.effect("reports a malformed fragmented final line before input stream termination", () =>
400 Effect.gen(function* () {
401 const { stdio, input, output } = yield* makeInMemoryStdio();
402 const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
403 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
404 stdio,
405 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
406 });
407 const response = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
408 yield* Queue.take(output);
409
410 yield* Queue.offer(input, encoder.encode('{"id":1,'));
411 yield* Queue.offer(input, encoder.encode('"result":'));
412 yield* Queue.end(input);
413
414 const error = yield* Deferred.await(termination);
415 assert.instanceOf(error, CodexError.CodexAppServerProtocolParseError);
416 assert.equal(error.operation, "decode-wire-message");
417 const responseError = yield* Fiber.join(response).pipe(
418 Effect.match({
419 onFailure: (failure) => failure,
420 onSuccess: () => assert.fail("Expected the malformed response to fail the request"),
421 }),
422 );
423 assert.strictEqual(responseError, error);
424 }),
425 );
426
427 it.effect("keeps processing protocol messages while an approval is pending", () =>
428 Effect.gen(function* () {
429 const { stdio, input, output } = yield* makeInMemoryStdio();
430 const approvalStarted = yield* Deferred.make<void>();
431 const approvalDecision = yield* Deferred.make<{ readonly decision: string }>();
432 const notificationReceived = yield* Deferred.make<void>();
433 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
434 stdio,
435 onRequest: () =>
436 Deferred.succeed(approvalStarted, undefined).pipe(
437 Effect.andThen(Deferred.await(approvalDecision)),
438 ),
439 onNotification: () => Deferred.succeed(notificationReceived, undefined).pipe(Effect.asVoid),
440 });
441
442 const pendingRequest = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
443 yield* Queue.take(output);
444 yield* Queue.offer(
445 input,
446 encoder.encode(
447 `${[
448 encodeUnknownJsonString({ id: 7, method: "item/tool/requestUserInput", params: {} }),
449 encodeUnknownJsonString({ method: "item/agentMessage/delta", params: { delta: "ok" } }),
450 encodeUnknownJsonString({ id: 1, result: { threadId: "thread-1" } }),
451 ].join("\n")}\n`,
452 ),
453 );
454
455 yield* Deferred.await(approvalStarted);
456 yield* Deferred.await(notificationReceived);
457 assert.deepEqual(yield* Fiber.join(pendingRequest), { threadId: "thread-1" });
458
459 yield* Deferred.succeed(approvalDecision, { decision: "accept" });
460 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
461 id: 7,
462 result: { decision: "accept" },
463 });
464 }),
465 );
466
467 it.effect("rejects incoming requests after the active handler limit is reached", () =>
468 Effect.gen(function* () {
469 const { stdio, input, output } = yield* makeInMemoryStdio();
470 const handlersStarted = yield* Deferred.make<void>();
471 const releaseHandlers = yield* Deferred.make<void>();
472 let activeHandlers = 0;
473 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
474 stdio,
475 onRequest: () =>
476 Effect.sync(() => ++activeHandlers).pipe(
477 Effect.flatMap((count) =>
478 count === 32
479 ? Deferred.succeed(handlersStarted, undefined).pipe(Effect.asVoid)
480 : Effect.void,
481 ),
482 Effect.andThen(Deferred.await(releaseHandlers)),
483 Effect.as({ decision: "accept" }),
484 ),
485 });
486
487 const requests = Array.from({ length: 33 }, (_, index) =>
488 encodeUnknownJsonString({
489 id: index + 1,
490 method: "item/tool/requestUserInput",
491 params: {},
492 }),
493 );
494 yield* Queue.offer(input, encoder.encode(`${requests.join("\n")}\n`));
495 yield* Deferred.await(handlersStarted);
496
497 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
498 id: 33,
499 error: {
500 code: -32001,
501 message: "Too many Codex requests are already active.",
502 },
503 });
504 assert.equal(activeHandlers, 32);
505
506 yield* Deferred.succeed(releaseHandlers, undefined);
507 yield* Effect.forEach(Array.from({ length: 32 }), () => Queue.take(output), {
508 discard: true,
509 });
510 }),
511 );
512
513 it.effect("releases a waiting request handler when Codex resolves its request", () =>
514 Effect.gen(function* () {
515 const { stdio, input, output } = yield* makeInMemoryStdio();
516 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
517 stdio,
518 onRequest: (request) =>
519 request.id === 33 ? Effect.succeed({ decision: "accept" }) : Effect.never,
520 });
521
522 const messages = Array.from({ length: 32 }, (_, index) =>
523 encodeUnknownJsonString({
524 id: index + 1,
525 method: "item/commandExecution/requestApproval",
526 params: {},
527 }),
528 );
529 for (let requestId = 1; requestId <= 32; requestId++) {
530 messages.push(
531 encodeUnknownJsonString({
532 method: "serverRequest/resolved",
533 params: { threadId: "thread-1", requestId },
534 }),
535 );
536 }
537 messages.push(
538 encodeUnknownJsonString({
539 id: 33,
540 method: "item/commandExecution/requestApproval",
541 params: {},
542 }),
543 );
544 yield* Queue.offer(input, encoder.encode(`${messages.join("\n")}\n`));
545
546 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
547 id: 33,
548 result: { decision: "accept" },
549 });
550 }),
551 );
552
553 it.effect("keeps numeric and string request ids apart when Codex resolves one", () =>
554 Effect.gen(function* () {
555 const { stdio, input, output } = yield* makeInMemoryStdio();
556 const interrupted: Array<string | number> = [];
557 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
558 stdio,
559 onRequest: (request) =>
560 request.id === 2
561 ? Effect.sync(() => ({ interrupted }))
562 : Effect.never.pipe(
563 Effect.onInterrupt(() => Effect.sync(() => interrupted.push(request.id))),
564 ),
565 });
566
567 const messages = [
568 { id: 1, method: "item/commandExecution/requestApproval", params: {} },
569 { id: "1", method: "item/commandExecution/requestApproval", params: {} },
570 { method: "serverRequest/resolved", params: { threadId: "thread-1", requestId: "1" } },
571 { id: 2, method: "item/commandExecution/requestApproval", params: {} },
572 ].map((message) => encodeUnknownJsonString(message));
573 yield* Queue.offer(input, encoder.encode(`${messages.join("\n")}\n`));
574
575 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
576 id: 2,
577 result: { interrupted: ["1"] },
578 });
579 }),
580 );
581
582 it.effect("interrupts pending request handlers when the protocol terminates", () =>
583 Effect.gen(function* () {
584 const { stdio, input } = yield* makeInMemoryStdio();
585 const approvalStarted = yield* Deferred.make<void>();
586 const approvalInterrupted = yield* Deferred.make<void>();
587 const terminated = yield* Deferred.make<void>();
588 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
589 stdio,
590 onRequest: () =>
591 Deferred.succeed(approvalStarted, undefined).pipe(
592 Effect.andThen(Effect.never),
593 Effect.onInterrupt(() =>
594 Deferred.succeed(approvalInterrupted, undefined).pipe(Effect.asVoid),
595 ),
596 ),
597 onTermination: () => Deferred.succeed(terminated, undefined).pipe(Effect.asVoid),
598 });
599
600 yield* Queue.offer(
601 input,
602 encodeJsonl({ id: 7, method: "item/tool/requestUserInput", params: {} }),
603 );
604 yield* Deferred.await(approvalStarted);
605 yield* Queue.end(input);
606
607 yield* Deferred.await(approvalInterrupted);
608 yield* Deferred.await(terminated);
609 }),
610 );
611
612 it.effect("rejects outgoing messages after an approval response cannot be encoded", () =>
613 Effect.gen(function* () {
614 const { stdio: baseStdio, input } = yield* makeInMemoryStdio();
615 const terminated = yield* Deferred.make<CodexError.CodexAppServerError>();
616 const readerStopped = yield* Deferred.make<void>();
617 let notificationCount = 0;
618 let requestCount = 0;
619 const stdio = Stdio.make({
620 args: baseStdio.args,
621 stdin: baseStdio.stdin.pipe(
622 Stream.ensuring(Deferred.succeed(readerStopped, undefined).pipe(Effect.asVoid)),
623 ),
624 stdout: baseStdio.stdout,
625 stderr: baseStdio.stderr,
626 });
627 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
628 stdio,
629 onRequest: () =>
630 Effect.sync(() => ++requestCount).pipe(
631 Effect.map((count) => (count === 1 ? { invalid: 1n } : { ok: true })),
632 ),
633 onNotification: () => Effect.sync(() => notificationCount++).pipe(Effect.asVoid),
634 onTermination: (error) => Deferred.succeed(terminated, error).pipe(Effect.asVoid),
635 });
636
637 yield* Queue.offer(
638 input,
639 encodeJsonl({ id: 7, method: "item/tool/requestUserInput", params: {} }),
640 );
641
642 const failure = yield* Deferred.await(terminated);
643 assert.instanceOf(failure, CodexError.CodexAppServerProtocolParseError);
644 const requestFailure = yield* transport.request("thread/read", {}).pipe(
645 Effect.match({
646 onFailure: (error) => error,
647 onSuccess: () => assert.fail("Expected a terminated protocol request to fail"),
648 }),
649 );
650 const notificationFailure = yield* transport.notify("initialized").pipe(Effect.flip);
651 assert.strictEqual(requestFailure, failure);
652 assert.strictEqual(notificationFailure, failure);
653 yield* Deferred.await(readerStopped);
654
655 yield* Queue.offer(
656 input,
657 encoder.encode(
658 `${[
659 encodeUnknownJsonString({ method: "x/late-notification" }),
660 encodeUnknownJsonString({ id: 8, method: "x/late-request" }),
661 ].join("\n")}\n`,
662 ),
663 );
664
665 assert.equal(notificationCount, 0);
666 assert.equal(requestCount, 1);
667 assert.equal(yield* Queue.size(input), 1);
668 }),
669 );
670
671 it.effect("fails pending requests before interrupted handler cleanup completes", () =>
672 Effect.gen(function* () {
673 const { stdio, input, output } = yield* makeInMemoryStdio();
674 const handlerStarted = yield* Deferred.make<void>();
675 const finalizerStarted = yield* Deferred.make<void>();
676 const releaseFinalizer = yield* Deferred.make<void>();
677 const terminated = yield* Deferred.make<CodexError.CodexAppServerError>();
678 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
679 stdio,
680 onRequest: () =>
681 Deferred.succeed(handlerStarted, undefined).pipe(
682 Effect.andThen(Effect.never),
683 Effect.onInterrupt(() =>
684 Deferred.succeed(finalizerStarted, undefined).pipe(
685 Effect.andThen(Deferred.await(releaseFinalizer)),
686 ),
687 ),
688 ),
689 onTermination: (error) => Deferred.succeed(terminated, error).pipe(Effect.asVoid),
690 });
691 const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
692 yield* Queue.take(output);
693 yield* Queue.offer(input, encodeJsonl({ id: 7, method: "x/approval" }));
694 yield* Deferred.await(handlerStarted);
695 yield* Queue.end(input);
696
697 const failure = yield* Deferred.await(terminated);
698 yield* Deferred.await(finalizerStarted);
699 const pendingFailure = yield* Fiber.join(pending).pipe(
700 Effect.match({
701 onFailure: (error) => error,
702 onSuccess: () => assert.fail("Expected the pending request to fail"),
703 }),
704 );
705 assert.strictEqual(pendingFailure, failure);
706
707 yield* Deferred.succeed(releaseFinalizer, undefined);
708 }),
709 );
710
711 it.effect("correlates response errors with the originating request", () =>
712 Effect.gen(function* () {
713 const { stdio, input, output } = yield* makeInMemoryStdio();
714 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ stdio });
715
716 const response = yield* transport.request("thread/start", {}).pipe(Effect.forkScoped);
717 yield* Queue.take(output);
718 yield* Queue.offer(
719 input,
720 encodeJsonl({
721 id: 1,
722 error: {
723 code: -32602,
724 message: "Invalid params",
725 data: { field: "cwd" },
726 },
727 }),
728 );
729
730 const error = yield* Fiber.join(response).pipe(
731 Effect.match({
732 onFailure: (error) => error,
733 onSuccess: () => assert.fail("Expected Codex App Server request to fail"),
734 }),
735 );
736 assert.instanceOf(error, CodexError.CodexAppServerRequestError);
737 assert.deepInclude(error, {
738 code: -32602,
739 errorMessage: "Invalid params",
740 method: "thread/start",
741 requestId: "1",
742 operation: "receive-response",
743 });
744 }),
745 );
746
747 it.effect("logs decode failures without copying the cause or wire payload", () =>
748 Effect.gen(function* () {
749 const secret = "codex-wire-secret-sentinel";
750 const { stdio, input } = yield* makeInMemoryStdio();
751 const events: Array<CodexProtocol.CodexAppServerProtocolLogEvent> = [];
752 const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
753 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
754 stdio,
755 logIncoming: true,
756 logger: (event) =>
757 Effect.sync(() => {
758 events.push(event);
759 }),
760 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
761 });
762
763 yield* Queue.offer(input, encoder.encode(`{"secret":"${secret}"\n`));
764 yield* Deferred.await(termination);
765
766 const event = events.find(({ stage }) => stage === "decode_failed");
767 assert.exists(event);
768 assert.equal(event.direction, "incoming");
769 const payload = event.payload as Record<string, unknown>;
770 assert.equal(payload.operation, "decode-wire-message");
771 assert.isNumber(payload.issueCount);
772 assert.isArray(payload.issueKinds);
773 assert.isNumber(payload.maximumPathDepth);
774 assert.equal("cause" in payload, false);
775 assert.equal("detail" in payload, false);
776 assert.notInclude(encodeUnknownJsonString(event), secret);
777 }),
778 );
779
780 it.effect("describes unroutable messages with safe structural diagnostics", () =>
781 Effect.gen(function* () {
782 const secret = "codex-unroutable-secret-sentinel";
783 const { stdio, input } = yield* makeInMemoryStdio();
784 const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
785 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
786 stdio,
787 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
788 });
789
790 yield* Queue.offer(
791 input,
792 encodeJsonl({ id: true, method: "thread/start", params: { token: secret } }),
793 );
794
795 const error = yield* Deferred.await(termination);
796 assert.instanceOf(error, CodexError.CodexAppServerProtocolParseError);
797 assert.deepInclude(error, {
798 operation: "route-wire-message",
799 method: "thread/start",
800 payloadKind: "object",
801 presentFields: ["id", "method", "params"],
802 });
803 assert.isUndefined(error.requestId);
804 assert.notProperty(error, "detail");
805 assert.notProperty(error, "cause");
806 assert.notInclude(error.message, secret);
807 }),
808 );
809
810 it.effect("fails pending requests when input ends while the process is still running", () =>
811 Effect.gen(function* () {
812 const { stdio, input, output } = yield* makeInMemoryStdio();
813 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
814 stdio,
815 terminationError: Effect.never,
816 });
817 const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
818 yield* Queue.take(output);
819
820 yield* Queue.end(input);
821 yield* TestClock.adjust("249 millis");
822 assert.isUndefined(pending.pollUnsafe());
823 yield* TestClock.adjust("1 millis");
824
825 const failure = yield* Fiber.join(pending).pipe(
826 Effect.match({
827 onFailure: (error) => error,
828 onSuccess: () => assert.fail("Expected the pending request to fail"),
829 }),
830 );
831 assert.instanceOf(failure, CodexError.CodexAppServerInputStreamEndedError);
832 }),
833 );
834
835 it.effect("fails pending requests when a write fails while input stays open", () =>
836 Effect.gen(function* () {
837 const { stdio: baseStdio } = yield* makeInMemoryStdio();
838 const writeFailure = PlatformError.systemError({
839 _tag: "Unknown",
840 module: "Stdio",
841 method: "stdout",
842 cause: new Error("EPIPE"),
843 });
844 const terminated = yield* Deferred.make<CodexError.CodexAppServerError>();
845 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
846 stdio: Stdio.make({
847 args: baseStdio.args,
848 stdin: baseStdio.stdin,
849 stdout: () => Sink.forEach(() => Effect.fail(writeFailure)),
850 stderr: baseStdio.stderr,
851 }),
852 onTermination: (error) => Deferred.succeed(terminated, error).pipe(Effect.asVoid),
853 });
854 const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
855
856 const failure = yield* Deferred.await(terminated);
857 assert.instanceOf(failure, CodexError.CodexAppServerTransportError);
858 assert.equal(failure.operation, "write-output-stream");
859 assert.strictEqual(failure.cause, writeFailure);
860 for (const request of [Fiber.join(pending), transport.request("thread/read", {})]) {
861 const requestFailure = yield* request.pipe(
862 Effect.match({
863 onFailure: (error) => error,
864 onSuccess: () => assert.fail("Expected the broken transport to reject the request"),
865 }),
866 );
867 assert.strictEqual(requestFailure, failure);
868 }
869 }),
870 );
871
872 it.effect("reports a process exit that follows input end", () =>
873 Effect.gen(function* () {
874 const { stdio, input } = yield* makeInMemoryStdio();
875 const exitError = new CodexError.CodexAppServerProcessExitedError({ code: 1, pid: 123 });
876 const exited = yield* Deferred.make<CodexError.CodexAppServerError>();
877 const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
878 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
879 stdio,
880 terminationError: Deferred.await(exited),
881 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
882 });
883
884 yield* Queue.end(input);
885 yield* TestClock.adjust("100 millis");
886 yield* Deferred.succeed(exited, exitError);
887
888 assert.strictEqual(yield* Deferred.await(termination), exitError);
889 }),
890 );
891});