diff --git a/packages/effect-codex-app-server/src/client.test.ts b/packages/effect-codex-app-server/src/client.test.ts index 842f15c0e912c..14fca793d5190 100644 --- a/packages/effect-codex-app-server/src/client.test.ts +++ b/packages/effect-codex-app-server/src/client.test.ts @@ -1,6 +1,9 @@ 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 Queue from "effect/Queue"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Ref from "effect/Ref"; import * as Scope from "effect/Scope"; @@ -8,6 +11,7 @@ import { ChildProcess, ChildProcessSpawner } from "effect/process"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; +import * as TestClock from "effect/testing/TestClock"; import * as CodexClient from "./client.ts"; @@ -124,6 +128,57 @@ it.layer(NodeServices.layer)("effect-codex-app-server client", (it) => { ]); }), ); + it.effect("warns once for an undecodable notification and keeps serving requests", () => + Effect.gen(function* () { + const warnings = yield* Queue.unbounded(); + const logger = Logger.make(({ logLevel, message }) => { + if (logLevel === "Warn") Queue.offerUnsafe(warnings, String([message].flat()[0])); + }); + const handle = yield* makeHandle({ + CODEX_APP_SERVER_TEST_INVALID_NOTIFICATION: "1", + }); + const scope = yield* Scope.make(); + const layerClient = CodexClient.layerChildProcess(handle); + // The transport fiber that logs the warning forks while the client layer + // builds, so the logger must wrap the build itself (CurrentLoggers is a + // FiberRef: a merged-in layer branch does not reach the client's fibers). + const context = yield* Layer.buildWithScope(layerClient, scope).pipe( + Effect.provide(Logger.layer([logger])), + ); + const validDelta = yield* Deferred.make(); + + const account = yield* Effect.gen(function* () { + const client = yield* CodexClient.CodexAppServerClient; + yield* client.handleServerNotification("item/agentMessage/delta", () => + Deferred.succeed(validDelta, undefined).pipe(Effect.asVoid), + ); + yield* client.request("initialize", { + clientInfo: { + name: "effect-codex-app-server-test", + title: "Effect Codex App Server Test", + version: "0.0.0", + }, + capabilities: { + experimentalApi: true, + optOutNotificationMethods: null, + }, + }); + yield* client.notify("initialized", undefined); + // The peer answers `initialized` with a valid delta followed by one the + // payload schema rejects; the warning for the rejected one follows the + // accepted one on the same processing fiber. + yield* Deferred.await(validDelta); + const warning = yield* TestClock.withLive( + Queue.take(warnings).pipe(Effect.timeout("5 seconds")), + ); + assert.equal(warning, "Dropped a Codex App Server notification that failed to decode"); + // The dropped notification must not have broken the client. + return yield* client.request("account/read", {}); + }).pipe(Effect.provide(context), Effect.ensuring(Scope.close(scope, Exit.void))); + + assert.equal(account.requiresOpenaiAuth, false); + }), + ); it.effect("drains child stderr so large diagnostics cannot block protocol responses", () => Effect.gen(function* () { const handle = yield* makeHandle({ diff --git a/packages/effect-codex-app-server/src/client.ts b/packages/effect-codex-app-server/src/client.ts index d872b9fa8473e..27a0c34663f89 100644 --- a/packages/effect-codex-app-server/src/client.ts +++ b/packages/effect-codex-app-server/src/client.ts @@ -101,6 +101,7 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make ): Effect.fn.Return { const requestHandlers = new Map(); const notificationHandlers = new Map>(); + const warnedDecodeFailures = new Set(); let unknownRequestHandler: | ((method: string, params: unknown) => Effect.Effect) | undefined; @@ -108,6 +109,22 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make | ((method: string, params: unknown) => Effect.Effect) | undefined; + // Drop undecodable notifications without ending the session; a newer app server may + // send shapes this client's schemas predate. Warn once per distinct failure because + // the server repeats bad payloads. + const dropUndecodableNotification = + (method: string) => (error: CodexError.CodexAppServerError) => { + const key = `${method}\n${error.message}`; + const firstOfItsKind = !warnedDecodeFailures.has(key); + warnedDecodeFailures.add(key); + return firstOfItsKind + ? Effect.logWarning("Dropped a Codex App Server notification that failed to decode").pipe( + Effect.annotateLogs({ method, detail: error.message }), + Effect.asVoid, + ) + : Effect.void; + }; + const getServerRequestParamSchema = ( method: M, ): @@ -166,6 +183,7 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make Effect.flatMap((decoded) => Effect.forEach(handlers, (handler) => handler(decoded), { discard: true }), ), + Effect.catch(dropUndecodableNotification(notification.method)), Effect.ignore, ); } diff --git a/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts b/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts index 59e5eb28faf94..ad8274cc57c8d 100644 --- a/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts +++ b/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts @@ -68,6 +68,19 @@ const handleMethod = (message: Record) => { turnId: "turn-1", }, }); + // `item/agentMessage/delta` expects a string delta; the number fails the + // payload schema, so the client must drop the notification with a warning. + if (process.env.CODEX_APP_SERVER_TEST_INVALID_NOTIFICATION) { + writeMessage({ + method: "item/agentMessage/delta", + params: { + delta: 42, + itemId: "item-invalid", + threadId: "thread-1", + turnId: "turn-1", + }, + }); + } return; } case "account/read": {