diff --git a/packages/effect-codex-app-server/src/client.test.ts b/packages/effect-codex-app-server/src/client.test.ts index 842f15c0e912c..c5fb8fe999247 100644 --- a/packages/effect-codex-app-server/src/client.test.ts +++ b/packages/effect-codex-app-server/src/client.test.ts @@ -1,15 +1,20 @@ +import * as Deferred from "effect/Deferred"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; +import * as Logger from "effect/Logger"; import * as Path from "effect/Path"; import * as Effect from "effect/Effect"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as References from "effect/References"; import * as Scope from "effect/Scope"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; import * as NodeServices from "@effect/platform-node/NodeServices"; -import { assert, it } from "@effect/vitest"; +import { assert, expect, it } from "@effect/vitest"; import * as CodexClient from "./client.ts"; +import { makeInMemoryStdio } from "./_internal/stdio.ts"; const mockPeerPath = Effect.map(Effect.service(Path.Path), (path) => path.join(import.meta.dirname, "../test/fixtures/codex-app-server-mock-peer.ts"), @@ -155,4 +160,52 @@ it.layer(NodeServices.layer)("effect-codex-app-server client", (it) => { assert.equal(initialized.userAgent, "mock-codex-app-server"); }), ); + + it.effect("warns once about a notification that fails to decode and keeps dispatching", () => + Effect.gen(function* () { + const secret = "codex-notification-secret-sentinel"; + const { stdio, input } = yield* makeInMemoryStdio(); + const warnings: Array = []; + const client = yield* CodexClient.make(stdio).pipe( + Effect.provide( + Logger.layer([ + Logger.make(({ logLevel, fiber }) => { + if (logLevel === "Warn") + warnings.push(fiber.getRef(References.CurrentLogAnnotations)); + }), + ]), + ), + ); + const delivered = yield* Deferred.make(); + yield* client.handleServerNotification("item/agentMessage/delta", (payload) => + Deferred.succeed(delivered, payload).pipe(Effect.asVoid), + ); + + const send = (params: unknown) => + Queue.offer( + input, + new TextEncoder().encode( + `${JSON.stringify({ method: "item/agentMessage/delta", params })}\n`, + ), + ); + const malformed = { itemId: "item-1", threadId: "thread-1", turnId: secret }; + yield* send(malformed); + yield* send(malformed); + yield* send({ delta: "ok", itemId: "item-1", threadId: "thread-1", turnId: "turn-1" }); + + assert.deepEqual(yield* Deferred.await(delivered), { + delta: "ok", + itemId: "item-1", + threadId: "thread-1", + turnId: "turn-1", + }); + assert.lengthOf(warnings, 1); + expect(warnings[0]).toMatchObject({ + operation: "decode-notification-payload", + method: "item/agentMessage/delta", + issueKinds: expect.arrayContaining(["MissingKey"]), + }); + assert.notInclude(JSON.stringify(warnings), secret); + }), + ); }); diff --git a/packages/effect-codex-app-server/src/client.ts b/packages/effect-codex-app-server/src/client.ts index e7c5f50b57045..8ad690bde946d 100644 --- a/packages/effect-codex-app-server/src/client.ts +++ b/packages/effect-codex-app-server/src/client.ts @@ -150,6 +150,29 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make > | undefined => CodexRpc.CLIENT_NOTIFICATION_PARAMS[method] as never; + // A notification the schema rejects has no reply to fail, so it is dropped. + // A Codex build that drifted from the generated schema repeats the bad shape + // on every frame, so each distinct failure is warned about once per client. + const warnedDecodeFailures = new Set(); + const warnUndecodableNotification = (error: CodexError.CodexAppServerProtocolParseError) => { + const key = `${error.method}\n${error.issueKinds?.join(",")}`; + if (warnedDecodeFailures.has(key)) { + return Effect.void; + } + warnedDecodeFailures.add(key); + return Effect.logWarning("Dropped a Codex App Server notification that failed to decode").pipe( + Effect.annotateLogs({ + operation: error.operation, + ...(error.method === undefined ? {} : { method: error.method }), + ...(error.issueCount === undefined ? {} : { issueCount: error.issueCount }), + ...(error.issueKinds === undefined ? {} : { issueKinds: error.issueKinds }), + ...(error.maximumPathDepth === undefined + ? {} + : { maximumPathDepth: error.maximumPathDepth }), + }), + ); + }; + const dispatchNotification = ( notification: CodexProtocol.CodexAppServerIncomingNotification, ): Effect.Effect => { @@ -163,6 +186,7 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make if (schema) { return decodeNotificationPayload(notification.method, schema, notification.params).pipe( + Effect.tapError(warnUndecodableNotification), Effect.flatMap((decoded) => Effect.forEach(handlers, (handler) => handler(decoded), { discard: true }), ),