1import * as Deferred from "effect/Deferred";
2import * as Effect from "effect/Effect";
3import * as Fiber from "effect/Fiber";
4import * as Queue from "effect/Queue";
5import * as Schema from "effect/Schema";
6import * as Stdio from "effect/Stdio";
7import * as Stream from "effect/Stream";
8
9import * as NodeServices from "@effect/platform-node/NodeServices";
10import { assert, it } from "@effect/vitest";
11
12import * as CodexError from "./errors.ts";
13import * as CodexProtocol from "./protocol.ts";
14import * as CodexRpc from "./rpc.ts";
15import * as CodexSchema from "./schema.ts";
16import { makeInMemoryStdio } from "./_internal/stdio.ts";
17const encodeUnknownJsonString = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
18
19const encoder = new TextEncoder();
20
21const encodeJsonl = (value: unknown) => encoder.encode(`${encodeUnknownJsonString(value)}\n`);
22
23const decodeJson = Schema.decodeEffect(Schema.fromJsonString(Schema.Unknown));
24const decodeAccountTokenUsageResponse = Schema.decodeUnknownEffect(
25 CodexRpc.CLIENT_REQUEST_RESPONSES["account/usage/read"],
26);
27const decodeAccountRateLimitsResponse = Schema.decodeUnknownEffect(
28 CodexRpc.CLIENT_REQUEST_RESPONSES["account/rateLimits/read"],
29);
30const decodeConsumeRateLimitResetCreditParams = Schema.decodeUnknownEffect(
31 CodexRpc.CLIENT_REQUEST_PARAMS["account/rateLimitResetCredit/consume"],
32);
33const decodeConsumeRateLimitResetCreditResponse = Schema.decodeUnknownEffect(
34 CodexRpc.CLIENT_REQUEST_RESPONSES["account/rateLimitResetCredit/consume"],
35);
36
37it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {
38 it.effect("maps account usage responses to the upstream token usage schema", () =>
39 Effect.gen(function* () {
40 assert.strictEqual(
41 CodexRpc.CLIENT_REQUEST_RESPONSES["account/usage/read"],
42 CodexSchema.V2GetAccountTokenUsageResponse,
43 );
44 const decoded = yield* decodeAccountTokenUsageResponse({
45 dailyUsageBuckets: [{ startDate: "2026-06-10", tokens: 42 }],
46 summary: { lifetimeTokens: 42 },
47 });
48 assert.deepEqual(decoded, {
49 dailyUsageBuckets: [{ startDate: "2026-06-10", tokens: 42 }],
50 summary: { lifetimeTokens: 42 },
51 });
52 }),
53 );
54
55 it.effect("maps earned rate-limit reset credits from account rate-limit snapshots", () =>
56 Effect.gen(function* () {
57 assert.strictEqual(
58 CodexRpc.CLIENT_REQUEST_RESPONSES["account/rateLimits/read"],
59 CodexSchema.V2GetAccountRateLimitsResponse,
60 );
61
62 const response = {
63 rateLimits: {},
64 rateLimitResetCredits: {
65 availableCount: 2,
66 credits: [
67 {
68 id: "RateLimitResetCredit_1",
69 resetType: "codexRateLimits",
70 status: "available",
71 grantedAt: 1_781_654_400,
72 expiresAt: 1_784_246_400,
73 title: "Full reset",
74 description: "Ready to redeem",
75 },
76 {
77 id: "RateLimitResetCredit_2",
78 resetType: "unknown",
79 status: "unknown",
80 grantedAt: 1_781_654_401,
81 expiresAt: null,
82 },
83 ],
84 },
85 } as const;
86
87 assert.deepEqual(yield* decodeAccountRateLimitsResponse(response), response);
88 assert.deepEqual(
89 yield* decodeAccountRateLimitsResponse({
90 rateLimits: {},
91 rateLimitResetCredits: { availableCount: 2, credits: null },
92 }),
93 {
94 rateLimits: {},
95 rateLimitResetCredits: { availableCount: 2, credits: null },
96 },
97 );
98 }),
99 );
100
101 it.effect("maps the earned rate-limit reset consume request and response", () =>
102 Effect.gen(function* () {
103 assert.equal(
104 CodexRpc.CLIENT_REQUEST_METHODS["account/rateLimitResetCredit/consume"],
105 "account/rateLimitResetCredit/consume",
106 );
107 assert.strictEqual(
108 CodexRpc.CLIENT_REQUEST_PARAMS["account/rateLimitResetCredit/consume"],
109 CodexSchema.V2ConsumeAccountRateLimitResetCreditParams,
110 );
111 assert.strictEqual(
112 CodexRpc.CLIENT_REQUEST_RESPONSES["account/rateLimitResetCredit/consume"],
113 CodexSchema.V2ConsumeAccountRateLimitResetCreditResponse,
114 );
115
116 assert.deepEqual(
117 yield* decodeConsumeRateLimitResetCreditParams({
118 idempotencyKey: "8ae96ff3-3425-4f4c-8772-b6fd61502868",
119 creditId: "RateLimitResetCredit_1",
120 }),
121 {
122 idempotencyKey: "8ae96ff3-3425-4f4c-8772-b6fd61502868",
123 creditId: "RateLimitResetCredit_1",
124 },
125 );
126 assert.deepEqual(yield* decodeConsumeRateLimitResetCreditResponse({ outcome: "reset" }), {
127 outcome: "reset",
128 });
129 }),
130 );
131
132 it.effect(
133 "encodes requests without a jsonrpc field and routes inbound requests and notifications",
134 () =>
135 Effect.gen(function* () {
136 const { stdio, input, output } = yield* makeInMemoryStdio();
137 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ stdio });
138
139 const notificationDeferred =
140 yield* Deferred.make<ReadonlyArray<CodexProtocol.CodexAppServerIncomingNotification>>();
141 const requestDeferred =
142 yield* Deferred.make<ReadonlyArray<CodexProtocol.CodexAppServerIncomingRequest>>();
143
144 yield* transport.incomingNotifications.pipe(
145 Stream.take(1),
146 Stream.runCollect,
147 Effect.flatMap((notifications) => Deferred.succeed(notificationDeferred, notifications)),
148 Effect.forkScoped,
149 );
150
151 yield* transport.incomingRequests.pipe(
152 Stream.take(1),
153 Stream.runCollect,
154 Effect.flatMap((requests) => Deferred.succeed(requestDeferred, requests)),
155 Effect.forkScoped,
156 );
157
158 yield* transport.notify("initialized");
159 assert.equal(yield* Queue.take(output), '{"method":"initialized"}\n');
160
161 const initializeParams = {
162 clientInfo: {
163 name: "effect-codex-app-server-test",
164 title: "Effect Codex App Server Test",
165 version: "0.0.0",
166 },
167 capabilities: {
168 experimentalApi: true,
169 optOutNotificationMethods: null,
170 },
171 };
172
173 const pendingInitialize = yield* transport
174 .request("initialize", initializeParams)
175 .pipe(Effect.forkScoped);
176 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
177 id: 1,
178 method: "initialize",
179 params: initializeParams,
180 });
181
182 yield* Queue.offer(
183 input,
184 encodeJsonl({
185 method: "item/agentMessage/delta",
186 params: {
187 delta: "Hello from the mock peer.",
188 itemId: "item-1",
189 threadId: "thread-1",
190 turnId: "turn-1",
191 },
192 }),
193 );
194 yield* Queue.offer(
195 input,
196 encodeJsonl({
197 id: 77,
198 method: "item/tool/requestUserInput",
199 params: {
200 itemId: "item-approval-1",
201 threadId: "thread-1",
202 turnId: "turn-1",
203 questions: [
204 {
205 id: "approved",
206 header: "Approve",
207 question: "Continue?",
208 },
209 ],
210 },
211 }),
212 );
213 yield* Queue.offer(
214 input,
215 encodeJsonl({
216 id: 1,
217 result: {
218 userAgent: "mock-codex-app-server",
219 codexHome: "/tmp/codex-home",
220 platformFamily: "unix",
221 platformOs: "macos",
222 },
223 }),
224 );
225
226 assert.deepEqual(yield* Fiber.join(pendingInitialize), {
227 userAgent: "mock-codex-app-server",
228 codexHome: "/tmp/codex-home",
229 platformFamily: "unix",
230 platformOs: "macos",
231 });
232 assert.deepEqual(yield* Deferred.await(notificationDeferred), [
233 {
234 method: "item/agentMessage/delta",
235 params: {
236 delta: "Hello from the mock peer.",
237 itemId: "item-1",
238 threadId: "thread-1",
239 turnId: "turn-1",
240 },
241 },
242 ]);
243 assert.deepEqual(yield* Deferred.await(requestDeferred), [
244 {
245 id: 77,
246 method: "item/tool/requestUserInput",
247 params: {
248 itemId: "item-approval-1",
249 threadId: "thread-1",
250 turnId: "turn-1",
251 questions: [
252 {
253 id: "approved",
254 header: "Approve",
255 question: "Continue?",
256 },
257 ],
258 },
259 },
260 ]);
261
262 yield* transport.respond(77, {
263 answers: {
264 approved: {
265 answers: ["yes"],
266 },
267 },
268 });
269 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
270 id: 77,
271 result: {
272 answers: {
273 approved: {
274 answers: ["yes"],
275 },
276 },
277 },
278 });
279
280 yield* transport.respondError(
281 78,
282 CodexError.CodexAppServerRequestError.methodNotFound("x/test"),
283 );
284 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
285 id: 78,
286 error: {
287 code: -32601,
288 message: "Method not found: x/test",
289 },
290 });
291 }),
292 );
293
294 it.effect("routes a large notification fragmented across thousands of input chunks", () =>
295 Effect.gen(function* () {
296 const { stdio, input, output } = yield* makeInMemoryStdio();
297 const notifications: Array<CodexProtocol.CodexAppServerIncomingNotification> = [];
298 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
299 stdio,
300 onNotification: (notification) =>
301 Effect.sync(() => {
302 notifications.push(notification);
303 }),
304 });
305 const response = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
306 yield* Queue.take(output);
307
308 const notification = {
309 method: "turn/diff/updated",
310 params: {
311 threadId: "thread-1",
312 turnId: "turn-1",
313 diff: "x".repeat(4 * 1024 * 1024),
314 },
315 };
316 const bytes = encoder.encode(
317 `${encodeUnknownJsonString(notification)}\n${encodeUnknownJsonString({ id: 1, result: { ok: true } })}\n`,
318 );
319 for (let offset = 0; offset < bytes.length; offset += 1024) {
320 yield* Queue.offer(input, bytes.subarray(offset, offset + 1024));
321 }
322
323 assert.deepEqual(yield* Fiber.join(response), { ok: true });
324 assert.deepEqual(notifications, [notification]);
325 }),
326 );
327
328 it.effect.each([1, 7, 1024])(
329 "preserves JSONL framing and UTF-8 across %i-byte input chunks",
330 (chunkSize) =>
331 Effect.gen(function* () {
332 const { stdio, input } = yield* makeInMemoryStdio();
333 const notifications: Array<CodexProtocol.CodexAppServerIncomingNotification> = [];
334 const rawLines: Array<unknown> = [];
335 const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
336 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
337 stdio,
338 logIncoming: true,
339 logger: (event) =>
340 Effect.sync(() => {
341 if (event.stage === "raw") {
342 rawLines.push(event.payload);
343 }
344 }),
345 onNotification: (notification) =>
346 Effect.sync(() => {
347 notifications.push(notification);
348 }),
349 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
350 });
351
352 const firstLine = '{"method":"x/first",\r"params":{"text":"héπŸ™‚"}}';
353 const secondLine = '{"method":"x/second","params":{"value":2}}';
354 const finalLine = '{"method":"x/final","params":{"text":"ζœ€εΎŒ"}}\r';
355 const bytes = encoder.encode(`\n \t\r\n${firstLine}\r\n\n${secondLine}\n${finalLine}`);
356 for (let offset = 0; offset < bytes.length; offset += chunkSize) {
357 yield* Queue.offer(input, bytes.subarray(offset, offset + chunkSize));
358 }
359 yield* Queue.end(input);
360
361 assert.instanceOf(
362 yield* Deferred.await(termination),
363 CodexError.CodexAppServerInputStreamEndedError,
364 );
365 assert.deepEqual(notifications, [
366 { method: "x/first", params: { text: "héπŸ™‚" } },
367 { method: "x/second", params: { value: 2 } },
368 { method: "x/final", params: { text: "ζœ€εΎŒ" } },
369 ]);
370 assert.deepEqual(rawLines, [firstLine, secondLine, finalLine]);
371 }),
372 );
373
374 it.effect("reports a malformed fragmented final line before input stream termination", () =>
375 Effect.gen(function* () {
376 const { stdio, input, output } = yield* makeInMemoryStdio();
377 const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
378 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
379 stdio,
380 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
381 });
382 const response = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
383 yield* Queue.take(output);
384
385 yield* Queue.offer(input, encoder.encode('{"id":1,'));
386 yield* Queue.offer(input, encoder.encode('"result":'));
387 yield* Queue.end(input);
388
389 const error = yield* Deferred.await(termination);
390 assert.instanceOf(error, CodexError.CodexAppServerProtocolParseError);
391 assert.equal(error.operation, "decode-wire-message");
392 const responseError = yield* Fiber.join(response).pipe(
393 Effect.match({
394 onFailure: (failure) => failure,
395 onSuccess: () => assert.fail("Expected the malformed response to fail the request"),
396 }),
397 );
398 assert.strictEqual(responseError, error);
399 }),
400 );
401
402 it.effect("keeps only recent raw notifications after their callbacks run", () =>
403 Effect.gen(function* () {
404 const { stdio, input } = yield* makeInMemoryStdio();
405 const handled = yield* Deferred.make<void>();
406 let handledCount = 0;
407 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
408 stdio,
409 onNotification: () =>
410 Effect.sync(() => ++handledCount).pipe(
411 Effect.flatMap((count) =>
412 count === 64 ? Deferred.succeed(handled, undefined).pipe(Effect.asVoid) : Effect.void,
413 ),
414 ),
415 });
416
417 const messages = Array.from({ length: 64 }, (_, index) =>
418 encodeUnknownJsonString({
419 method: "item/agentMessage/delta",
420 params: { index },
421 }),
422 );
423 yield* Queue.offer(input, encoder.encode(`${messages.join("\n")}\n`));
424 yield* Deferred.await(handled);
425
426 const retained = yield* transport.incomingNotifications.pipe(
427 Stream.take(32),
428 Stream.runCollect,
429 );
430
431 assert.equal(handledCount, 64);
432 assert.equal(retained.length, 32);
433 assert.deepEqual(retained[0]?.params, { index: 32 });
434 assert.deepEqual(retained[31]?.params, { index: 63 });
435 }),
436 );
437
438 it.effect("keeps processing protocol messages while an approval is pending", () =>
439 Effect.gen(function* () {
440 const { stdio, input, output } = yield* makeInMemoryStdio();
441 const approvalStarted = yield* Deferred.make<void>();
442 const approvalDecision = yield* Deferred.make<{ readonly decision: string }>();
443 const notificationReceived = yield* Deferred.make<void>();
444 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
445 stdio,
446 onRequest: () =>
447 Deferred.succeed(approvalStarted, undefined).pipe(
448 Effect.andThen(Deferred.await(approvalDecision)),
449 ),
450 onNotification: () => Deferred.succeed(notificationReceived, undefined).pipe(Effect.asVoid),
451 });
452
453 const pendingRequest = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
454 yield* Queue.take(output);
455 yield* Queue.offer(
456 input,
457 encoder.encode(
458 `${[
459 encodeUnknownJsonString({ id: 7, method: "item/tool/requestUserInput", params: {} }),
460 encodeUnknownJsonString({ method: "item/agentMessage/delta", params: { delta: "ok" } }),
461 encodeUnknownJsonString({ id: 1, result: { threadId: "thread-1" } }),
462 ].join("\n")}\n`,
463 ),
464 );
465
466 yield* Deferred.await(approvalStarted);
467 yield* Deferred.await(notificationReceived);
468 assert.deepEqual(yield* Fiber.join(pendingRequest), { threadId: "thread-1" });
469
470 yield* Deferred.succeed(approvalDecision, { decision: "accept" });
471 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
472 id: 7,
473 result: { decision: "accept" },
474 });
475 }),
476 );
477
478 it.effect("rejects incoming requests after the active handler limit is reached", () =>
479 Effect.gen(function* () {
480 const { stdio, input, output } = yield* makeInMemoryStdio();
481 const handlersStarted = yield* Deferred.make<void>();
482 const releaseHandlers = yield* Deferred.make<void>();
483 let activeHandlers = 0;
484 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
485 stdio,
486 onRequest: () =>
487 Effect.sync(() => ++activeHandlers).pipe(
488 Effect.flatMap((count) =>
489 count === 32
490 ? Deferred.succeed(handlersStarted, undefined).pipe(Effect.asVoid)
491 : Effect.void,
492 ),
493 Effect.andThen(Deferred.await(releaseHandlers)),
494 Effect.as({ decision: "accept" }),
495 ),
496 });
497
498 const requests = Array.from({ length: 33 }, (_, index) =>
499 encodeUnknownJsonString({
500 id: index + 1,
501 method: "item/tool/requestUserInput",
502 params: {},
503 }),
504 );
505 yield* Queue.offer(input, encoder.encode(`${requests.join("\n")}\n`));
506 yield* Deferred.await(handlersStarted);
507
508 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
509 id: 33,
510 error: {
511 code: -32001,
512 message: "Too many Codex requests are already active.",
513 },
514 });
515 assert.equal(activeHandlers, 32);
516
517 yield* Deferred.succeed(releaseHandlers, undefined);
518 yield* Effect.forEach(Array.from({ length: 32 }), () => Queue.take(output), {
519 discard: true,
520 });
521 }),
522 );
523
524 it.effect("interrupts pending request handlers when the protocol terminates", () =>
525 Effect.gen(function* () {
526 const { stdio, input } = yield* makeInMemoryStdio();
527 const approvalStarted = yield* Deferred.make<void>();
528 const approvalInterrupted = yield* Deferred.make<void>();
529 const terminated = yield* Deferred.make<void>();
530 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
531 stdio,
532 onRequest: () =>
533 Deferred.succeed(approvalStarted, undefined).pipe(
534 Effect.andThen(Effect.never),
535 Effect.onInterrupt(() =>
536 Deferred.succeed(approvalInterrupted, undefined).pipe(Effect.asVoid),
537 ),
538 ),
539 onTermination: () => Deferred.succeed(terminated, undefined).pipe(Effect.asVoid),
540 });
541
542 yield* Queue.offer(
543 input,
544 encodeJsonl({ id: 7, method: "item/tool/requestUserInput", params: {} }),
545 );
546 yield* Deferred.await(approvalStarted);
547 yield* Queue.end(input);
548
549 yield* Deferred.await(approvalInterrupted);
550 yield* Deferred.await(terminated);
551 }),
552 );
553
554 it.effect("rejects outgoing messages after an approval response cannot be encoded", () =>
555 Effect.gen(function* () {
556 const { stdio: baseStdio, input } = yield* makeInMemoryStdio();
557 const terminated = yield* Deferred.make<CodexError.CodexAppServerError>();
558 const readerStopped = yield* Deferred.make<void>();
559 let notificationCount = 0;
560 let requestCount = 0;
561 const stdio = Stdio.make({
562 args: baseStdio.args,
563 stdin: baseStdio.stdin.pipe(
564 Stream.ensuring(Deferred.succeed(readerStopped, undefined).pipe(Effect.asVoid)),
565 ),
566 stdout: baseStdio.stdout,
567 stderr: baseStdio.stderr,
568 });
569 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
570 stdio,
571 onRequest: () =>
572 Effect.sync(() => ++requestCount).pipe(
573 Effect.map((count) => (count === 1 ? { invalid: 1n } : { ok: true })),
574 ),
575 onNotification: () => Effect.sync(() => notificationCount++).pipe(Effect.asVoid),
576 onTermination: (error) => Deferred.succeed(terminated, error).pipe(Effect.asVoid),
577 });
578
579 yield* Queue.offer(
580 input,
581 encodeJsonl({ id: 7, method: "item/tool/requestUserInput", params: {} }),
582 );
583
584 const failure = yield* Deferred.await(terminated);
585 assert.instanceOf(failure, CodexError.CodexAppServerProtocolParseError);
586 const requestFailure = yield* transport.request("thread/read", {}).pipe(
587 Effect.match({
588 onFailure: (error) => error,
589 onSuccess: () => assert.fail("Expected a terminated protocol request to fail"),
590 }),
591 );
592 const notificationFailure = yield* transport.notify("initialized").pipe(Effect.flip);
593 assert.strictEqual(requestFailure, failure);
594 assert.strictEqual(notificationFailure, failure);
595 yield* Deferred.await(readerStopped);
596
597 yield* Queue.offer(
598 input,
599 encoder.encode(
600 `${[
601 encodeUnknownJsonString({ method: "x/late-notification" }),
602 encodeUnknownJsonString({ id: 8, method: "x/late-request" }),
603 ].join("\n")}\n`,
604 ),
605 );
606
607 assert.equal(notificationCount, 0);
608 assert.equal(requestCount, 1);
609 assert.equal(yield* Queue.size(input), 1);
610 }),
611 );
612
613 it.effect("fails pending requests before interrupted handler cleanup completes", () =>
614 Effect.gen(function* () {
615 const { stdio, input, output } = yield* makeInMemoryStdio();
616 const handlerStarted = yield* Deferred.make<void>();
617 const finalizerStarted = yield* Deferred.make<void>();
618 const releaseFinalizer = yield* Deferred.make<void>();
619 const terminated = yield* Deferred.make<CodexError.CodexAppServerError>();
620 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
621 stdio,
622 onRequest: () =>
623 Deferred.succeed(handlerStarted, undefined).pipe(
624 Effect.andThen(Effect.never),
625 Effect.onInterrupt(() =>
626 Deferred.succeed(finalizerStarted, undefined).pipe(
627 Effect.andThen(Deferred.await(releaseFinalizer)),
628 ),
629 ),
630 ),
631 onTermination: (error) => Deferred.succeed(terminated, error).pipe(Effect.asVoid),
632 });
633 const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
634 yield* Queue.take(output);
635 yield* Queue.offer(input, encodeJsonl({ id: 7, method: "x/approval" }));
636 yield* Deferred.await(handlerStarted);
637 yield* Queue.end(input);
638
639 const failure = yield* Deferred.await(terminated);
640 yield* Deferred.await(finalizerStarted);
641 const pendingFailure = yield* Fiber.join(pending).pipe(
642 Effect.match({
643 onFailure: (error) => error,
644 onSuccess: () => assert.fail("Expected the pending request to fail"),
645 }),
646 );
647 assert.strictEqual(pendingFailure, failure);
648
649 yield* Deferred.succeed(releaseFinalizer, undefined);
650 }),
651 );
652
653 it.effect("surfaces JSON encoding failures as protocol parse errors", () =>
654 Effect.gen(function* () {
655 const { stdio } = yield* makeInMemoryStdio();
656 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ stdio });
657
658 const bigintError = yield* transport.notify("x/test", 1n).pipe(Effect.flip);
659 assert.instanceOf(bigintError, CodexError.CodexAppServerProtocolParseError);
660 assert.equal(bigintError.operation, "encode-wire-message");
661 assert.equal(bigintError.method, "x/test");
662 assert.exists(bigintError.cause);
663 assert.equal(
664 bigintError.message,
665 "Codex App Server protocol operation 'encode-wire-message' failed for method 'x/test'.",
666 );
667
668 const circular: Record<string, unknown> = {};
669 circular.self = circular;
670 const circularError = yield* transport.notify("x/test", circular).pipe(Effect.flip);
671 assert.instanceOf(circularError, CodexError.CodexAppServerProtocolParseError);
672 assert.equal(circularError.operation, "encode-wire-message");
673 assert.equal(circularError.method, "x/test");
674 assert.exists(circularError.cause);
675
676 const requestError = yield* transport.request("x/request", 1n).pipe(
677 Effect.match({
678 onFailure: (error) => error,
679 onSuccess: () => assert.fail("Expected request encoding to fail"),
680 }),
681 );
682 assert.instanceOf(requestError, CodexError.CodexAppServerProtocolParseError);
683 assert.deepInclude(requestError, {
684 operation: "encode-wire-message",
685 method: "x/request",
686 requestId: "1",
687 });
688 }),
689 );
690
691 it.effect("correlates response errors with the originating request", () =>
692 Effect.gen(function* () {
693 const { stdio, input, output } = yield* makeInMemoryStdio();
694 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ stdio });
695
696 const response = yield* transport.request("thread/start", {}).pipe(Effect.forkScoped);
697 yield* Queue.take(output);
698 yield* Queue.offer(
699 input,
700 encodeJsonl({
701 id: 1,
702 error: {
703 code: -32602,
704 message: "Invalid params",
705 data: { field: "cwd" },
706 },
707 }),
708 );
709
710 const error = yield* Fiber.join(response).pipe(
711 Effect.match({
712 onFailure: (error) => error,
713 onSuccess: () => assert.fail("Expected Codex App Server request to fail"),
714 }),
715 );
716 assert.instanceOf(error, CodexError.CodexAppServerRequestError);
717 assert.deepInclude(error, {
718 code: -32602,
719 errorMessage: "Invalid params",
720 method: "thread/start",
721 requestId: "1",
722 operation: "receive-response",
723 });
724 }),
725 );
726
727 it.effect("logs decode failures without copying the cause or wire payload", () =>
728 Effect.gen(function* () {
729 const secret = "codex-wire-secret-sentinel";
730 const { stdio, input } = yield* makeInMemoryStdio();
731 const events: Array<CodexProtocol.CodexAppServerProtocolLogEvent> = [];
732 const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
733 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
734 stdio,
735 logIncoming: true,
736 logger: (event) =>
737 Effect.sync(() => {
738 events.push(event);
739 }),
740 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
741 });
742
743 yield* Queue.offer(input, encoder.encode(`{"secret":"${secret}"\n`));
744 yield* Deferred.await(termination);
745
746 const event = events.find(({ stage }) => stage === "decode_failed");
747 assert.exists(event);
748 assert.equal(event.direction, "incoming");
749 const payload = event.payload as Record<string, unknown>;
750 assert.equal(payload.operation, "decode-wire-message");
751 assert.isNumber(payload.issueCount);
752 assert.isArray(payload.issueKinds);
753 assert.isNumber(payload.maximumPathDepth);
754 assert.equal("cause" in payload, false);
755 assert.equal("detail" in payload, false);
756 assert.notInclude(encodeUnknownJsonString(event), secret);
757 }),
758 );
759
760 it.effect("describes unroutable messages with safe structural diagnostics", () =>
761 Effect.gen(function* () {
762 const secret = "codex-unroutable-secret-sentinel";
763 const { stdio, input } = yield* makeInMemoryStdio();
764 const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
765 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
766 stdio,
767 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
768 });
769
770 yield* Queue.offer(
771 input,
772 encodeJsonl({ id: true, method: "thread/start", params: { token: secret } }),
773 );
774
775 const error = yield* Deferred.await(termination);
776 assert.instanceOf(error, CodexError.CodexAppServerProtocolParseError);
777 assert.deepInclude(error, {
778 operation: "route-wire-message",
779 method: "thread/start",
780 payloadKind: "object",
781 presentFields: ["id", "method", "params"],
782 });
783 assert.isUndefined(error.requestId);
784 assert.notProperty(error, "detail");
785 assert.notProperty(error, "cause");
786 assert.notInclude(error.message, secret);
787 }),
788 );
789
790 it.effect("classifies an input stream ending without inventing a cause", () =>
791 Effect.gen(function* () {
792 const { stdio, input } = yield* makeInMemoryStdio();
793 const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
794 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
795 stdio,
796 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
797 });
798
799 yield* Queue.end(input);
800
801 const error = yield* Deferred.await(termination);
802 assert.instanceOf(error, CodexError.CodexAppServerInputStreamEndedError);
803 assert.equal(error.message, "Codex App Server input stream ended.");
804 assert.equal("cause" in error, false);
805 }),
806 );
807});