import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Fiber from "effect/Fiber";import * as Queue from "effect/Queue";import * as Schema from "effect/Schema";import * as Stdio from "effect/Stdio";import * as Stream from "effect/Stream"; import * as NodeServices from "@effect/platform-node/NodeServices";import { assert, it } from "@effect/vitest"; import * as CodexError from "./errors.ts";import * as CodexProtocol from "./protocol.ts";import * as CodexRpc from "./rpc.ts";import * as CodexSchema from "./schema.ts";import { makeInMemoryStdio } from "./_internal/stdio.ts";const encodeUnknownJsonString = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); const encoder = new TextEncoder(); const encodeJsonl = (value: unknown) => encoder.encode(`${encodeUnknownJsonString(value)}\n`); const decodeJson = Schema.decodeEffect(Schema.fromJsonString(Schema.Unknown));const decodeAccountTokenUsageResponse = Schema.decodeUnknownEffect(  CodexRpc.CLIENT_REQUEST_RESPONSES["account/usage/read"],);const decodeAccountRateLimitsResponse = Schema.decodeUnknownEffect(  CodexRpc.CLIENT_REQUEST_RESPONSES["account/rateLimits/read"],);const decodeConsumeRateLimitResetCreditParams = Schema.decodeUnknownEffect(  CodexRpc.CLIENT_REQUEST_PARAMS["account/rateLimitResetCredit/consume"],);const decodeConsumeRateLimitResetCreditResponse = Schema.decodeUnknownEffect(  CodexRpc.CLIENT_REQUEST_RESPONSES["account/rateLimitResetCredit/consume"],); it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {  it.effect("maps account usage responses to the upstream token usage schema", () =>    Effect.gen(function* () {      assert.strictEqual(        CodexRpc.CLIENT_REQUEST_RESPONSES["account/usage/read"],        CodexSchema.V2GetAccountTokenUsageResponse,      );      const decoded = yield* decodeAccountTokenUsageResponse({        dailyUsageBuckets: [{ startDate: "2026-06-10", tokens: 42 }],        summary: { lifetimeTokens: 42 },      });      assert.deepEqual(decoded, {        dailyUsageBuckets: [{ startDate: "2026-06-10", tokens: 42 }],        summary: { lifetimeTokens: 42 },      });    }),  );   it.effect("maps earned rate-limit reset credits from account rate-limit snapshots", () =>    Effect.gen(function* () {      assert.strictEqual(        CodexRpc.CLIENT_REQUEST_RESPONSES["account/rateLimits/read"],        CodexSchema.V2GetAccountRateLimitsResponse,      );       const response = {        rateLimits: {},        rateLimitResetCredits: {          availableCount: 2,          credits: [            {              id: "RateLimitResetCredit_1",              resetType: "codexRateLimits",              status: "available",              grantedAt: 1_781_654_400,              expiresAt: 1_784_246_400,              title: "Full reset",              description: "Ready to redeem",            },            {              id: "RateLimitResetCredit_2",              resetType: "unknown",              status: "unknown",              grantedAt: 1_781_654_401,              expiresAt: null,            },          ],        },      } as const;       assert.deepEqual(yield* decodeAccountRateLimitsResponse(response), response);      assert.deepEqual(        yield* decodeAccountRateLimitsResponse({          rateLimits: {},          rateLimitResetCredits: { availableCount: 2, credits: null },        }),        {          rateLimits: {},          rateLimitResetCredits: { availableCount: 2, credits: null },        },      );    }),  );   it.effect("maps the earned rate-limit reset consume request and response", () =>    Effect.gen(function* () {      assert.equal(        CodexRpc.CLIENT_REQUEST_METHODS["account/rateLimitResetCredit/consume"],        "account/rateLimitResetCredit/consume",      );      assert.strictEqual(        CodexRpc.CLIENT_REQUEST_PARAMS["account/rateLimitResetCredit/consume"],        CodexSchema.V2ConsumeAccountRateLimitResetCreditParams,      );      assert.strictEqual(        CodexRpc.CLIENT_REQUEST_RESPONSES["account/rateLimitResetCredit/consume"],        CodexSchema.V2ConsumeAccountRateLimitResetCreditResponse,      );       assert.deepEqual(        yield* decodeConsumeRateLimitResetCreditParams({          idempotencyKey: "8ae96ff3-3425-4f4c-8772-b6fd61502868",          creditId: "RateLimitResetCredit_1",        }),        {          idempotencyKey: "8ae96ff3-3425-4f4c-8772-b6fd61502868",          creditId: "RateLimitResetCredit_1",        },      );      assert.deepEqual(yield* decodeConsumeRateLimitResetCreditResponse({ outcome: "reset" }), {        outcome: "reset",      });    }),  );   it.effect(    "encodes requests without a jsonrpc field and routes inbound requests and notifications",    () =>      Effect.gen(function* () {        const { stdio, input, output } = yield* makeInMemoryStdio();        const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ stdio });         const notificationDeferred =          yield* Deferred.make<ReadonlyArray<CodexProtocol.CodexAppServerIncomingNotification>>();        const requestDeferred =          yield* Deferred.make<ReadonlyArray<CodexProtocol.CodexAppServerIncomingRequest>>();         yield* transport.incomingNotifications.pipe(          Stream.take(1),          Stream.runCollect,          Effect.flatMap((notifications) => Deferred.succeed(notificationDeferred, notifications)),          Effect.forkScoped,        );         yield* transport.incomingRequests.pipe(          Stream.take(1),          Stream.runCollect,          Effect.flatMap((requests) => Deferred.succeed(requestDeferred, requests)),          Effect.forkScoped,        );         yield* transport.notify("initialized");        assert.equal(yield* Queue.take(output), '{"method":"initialized"}\n');         const initializeParams = {          clientInfo: {            name: "effect-codex-app-server-test",            title: "Effect Codex App Server Test",            version: "0.0.0",          },          capabilities: {            experimentalApi: true,            optOutNotificationMethods: null,          },        };         const pendingInitialize = yield* transport          .request("initialize", initializeParams)          .pipe(Effect.forkScoped);        assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {          id: 1,          method: "initialize",          params: initializeParams,        });         yield* Queue.offer(          input,          encodeJsonl({            method: "item/agentMessage/delta",            params: {              delta: "Hello from the mock peer.",              itemId: "item-1",              threadId: "thread-1",              turnId: "turn-1",            },          }),        );        yield* Queue.offer(          input,          encodeJsonl({            id: 77,            method: "item/tool/requestUserInput",            params: {              itemId: "item-approval-1",              threadId: "thread-1",              turnId: "turn-1",              questions: [                {                  id: "approved",                  header: "Approve",                  question: "Continue?",                },              ],            },          }),        );        yield* Queue.offer(          input,          encodeJsonl({            id: 1,            result: {              userAgent: "mock-codex-app-server",              codexHome: "/tmp/codex-home",              platformFamily: "unix",              platformOs: "macos",            },          }),        );         assert.deepEqual(yield* Fiber.join(pendingInitialize), {          userAgent: "mock-codex-app-server",          codexHome: "/tmp/codex-home",          platformFamily: "unix",          platformOs: "macos",        });        assert.deepEqual(yield* Deferred.await(notificationDeferred), [          {            method: "item/agentMessage/delta",            params: {              delta: "Hello from the mock peer.",              itemId: "item-1",              threadId: "thread-1",              turnId: "turn-1",            },          },        ]);        assert.deepEqual(yield* Deferred.await(requestDeferred), [          {            id: 77,            method: "item/tool/requestUserInput",            params: {              itemId: "item-approval-1",              threadId: "thread-1",              turnId: "turn-1",              questions: [                {                  id: "approved",                  header: "Approve",                  question: "Continue?",                },              ],            },          },        ]);         yield* transport.respond(77, {          answers: {            approved: {              answers: ["yes"],            },          },        });        assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {          id: 77,          result: {            answers: {              approved: {                answers: ["yes"],              },            },          },        });         yield* transport.respondError(          78,          CodexError.CodexAppServerRequestError.methodNotFound("x/test"),        );        assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {          id: 78,          error: {            code: -32601,            message: "Method not found: x/test",          },        });      }),  );   it.effect("routes a large notification fragmented across thousands of input chunks", () =>    Effect.gen(function* () {      const { stdio, input, output } = yield* makeInMemoryStdio();      const notifications: Array<CodexProtocol.CodexAppServerIncomingNotification> = [];      const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({        stdio,        onNotification: (notification) =>          Effect.sync(() => {            notifications.push(notification);          }),      });      const response = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);      yield* Queue.take(output);       const notification = {        method: "turn/diff/updated",        params: {          threadId: "thread-1",          turnId: "turn-1",          diff: "x".repeat(4 * 1024 * 1024),        },      };      const bytes = encoder.encode(        `${encodeUnknownJsonString(notification)}\n${encodeUnknownJsonString({ id: 1, result: { ok: true } })}\n`,      );      for (let offset = 0; offset < bytes.length; offset += 1024) {        yield* Queue.offer(input, bytes.subarray(offset, offset + 1024));      }       assert.deepEqual(yield* Fiber.join(response), { ok: true });      assert.deepEqual(notifications, [notification]);    }),  );   it.effect.each([1, 7, 1024])(    "preserves JSONL framing and UTF-8 across %i-byte input chunks",    (chunkSize) =>      Effect.gen(function* () {        const { stdio, input } = yield* makeInMemoryStdio();        const notifications: Array<CodexProtocol.CodexAppServerIncomingNotification> = [];        const rawLines: Array<unknown> = [];        const termination = yield* Deferred.make<CodexError.CodexAppServerError>();        yield* CodexProtocol.makeCodexAppServerPatchedProtocol({          stdio,          logIncoming: true,          logger: (event) =>            Effect.sync(() => {              if (event.stage === "raw") {                rawLines.push(event.payload);              }            }),          onNotification: (notification) =>            Effect.sync(() => {              notifications.push(notification);            }),          onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),        });         const firstLine = '{"method":"x/first",\r"params":{"text":"hé🙂"}}';        const secondLine = '{"method":"x/second","params":{"value":2}}';        const finalLine = '{"method":"x/final","params":{"text":"最後"}}\r';        const bytes = encoder.encode(`\n \t\r\n${firstLine}\r\n\n${secondLine}\n${finalLine}`);        for (let offset = 0; offset < bytes.length; offset += chunkSize) {          yield* Queue.offer(input, bytes.subarray(offset, offset + chunkSize));        }        yield* Queue.end(input);         assert.instanceOf(          yield* Deferred.await(termination),          CodexError.CodexAppServerInputStreamEndedError,        );        assert.deepEqual(notifications, [          { method: "x/first", params: { text: "hé🙂" } },          { method: "x/second", params: { value: 2 } },          { method: "x/final", params: { text: "最後" } },        ]);        assert.deepEqual(rawLines, [firstLine, secondLine, finalLine]);      }),  );   it.effect("reports a malformed fragmented final line before input stream termination", () =>    Effect.gen(function* () {      const { stdio, input, output } = yield* makeInMemoryStdio();      const termination = yield* Deferred.make<CodexError.CodexAppServerError>();      const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({        stdio,        onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),      });      const response = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);      yield* Queue.take(output);       yield* Queue.offer(input, encoder.encode('{"id":1,'));      yield* Queue.offer(input, encoder.encode('"result":'));      yield* Queue.end(input);       const error = yield* Deferred.await(termination);      assert.instanceOf(error, CodexError.CodexAppServerProtocolParseError);      assert.equal(error.operation, "decode-wire-message");      const responseError = yield* Fiber.join(response).pipe(        Effect.match({          onFailure: (failure) => failure,          onSuccess: () => assert.fail("Expected the malformed response to fail the request"),        }),      );      assert.strictEqual(responseError, error);    }),  );   it.effect("keeps only recent raw notifications after their callbacks run", () =>    Effect.gen(function* () {      const { stdio, input } = yield* makeInMemoryStdio();      const handled = yield* Deferred.make<void>();      let handledCount = 0;      const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({        stdio,        onNotification: () =>          Effect.sync(() => ++handledCount).pipe(            Effect.flatMap((count) =>              count === 64 ? Deferred.succeed(handled, undefined).pipe(Effect.asVoid) : Effect.void,            ),          ),      });       const messages = Array.from({ length: 64 }, (_, index) =>        encodeUnknownJsonString({          method: "item/agentMessage/delta",          params: { index },        }),      );      yield* Queue.offer(input, encoder.encode(`${messages.join("\n")}\n`));      yield* Deferred.await(handled);       const retained = yield* transport.incomingNotifications.pipe(        Stream.take(32),        Stream.runCollect,      );       assert.equal(handledCount, 64);      assert.equal(retained.length, 32);      assert.deepEqual(retained[0]?.params, { index: 32 });      assert.deepEqual(retained[31]?.params, { index: 63 });    }),  );   it.effect("keeps processing protocol messages while an approval is pending", () =>    Effect.gen(function* () {      const { stdio, input, output } = yield* makeInMemoryStdio();      const approvalStarted = yield* Deferred.make<void>();      const approvalDecision = yield* Deferred.make<{ readonly decision: string }>();      const notificationReceived = yield* Deferred.make<void>();      const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({        stdio,        onRequest: () =>          Deferred.succeed(approvalStarted, undefined).pipe(            Effect.andThen(Deferred.await(approvalDecision)),          ),        onNotification: () => Deferred.succeed(notificationReceived, undefined).pipe(Effect.asVoid),      });       const pendingRequest = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);      yield* Queue.take(output);      yield* Queue.offer(        input,        encoder.encode(          `${[            encodeUnknownJsonString({ id: 7, method: "item/tool/requestUserInput", params: {} }),            encodeUnknownJsonString({ method: "item/agentMessage/delta", params: { delta: "ok" } }),            encodeUnknownJsonString({ id: 1, result: { threadId: "thread-1" } }),          ].join("\n")}\n`,        ),      );       yield* Deferred.await(approvalStarted);      yield* Deferred.await(notificationReceived);      assert.deepEqual(yield* Fiber.join(pendingRequest), { threadId: "thread-1" });       yield* Deferred.succeed(approvalDecision, { decision: "accept" });      assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {        id: 7,        result: { decision: "accept" },      });    }),  );   it.effect("rejects incoming requests after the active handler limit is reached", () =>    Effect.gen(function* () {      const { stdio, input, output } = yield* makeInMemoryStdio();      const handlersStarted = yield* Deferred.make<void>();      const releaseHandlers = yield* Deferred.make<void>();      let activeHandlers = 0;      yield* CodexProtocol.makeCodexAppServerPatchedProtocol({        stdio,        onRequest: () =>          Effect.sync(() => ++activeHandlers).pipe(            Effect.flatMap((count) =>              count === 32                ? Deferred.succeed(handlersStarted, undefined).pipe(Effect.asVoid)                : Effect.void,            ),            Effect.andThen(Deferred.await(releaseHandlers)),            Effect.as({ decision: "accept" }),          ),      });       const requests = Array.from({ length: 33 }, (_, index) =>        encodeUnknownJsonString({          id: index + 1,

Lines 1–500 of 807. Load more