fix(codex): log undecodable server notifications instead of dropping them silently

packages/effect-codex-app-server/src/client.test.ts

@@ -1,13 +1,17 @@
 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";
 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<string>();
+      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<void>();
+
+      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({

packages/effect-codex-app-server/src/client.ts

@@ -101,13 +101,30 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make
 ): Effect.fn.Return<CodexAppServerClient["Service"], never, Scope.Scope> {
   const requestHandlers = new Map<string, ServerRequestHandler>();
   const notificationHandlers = new Map<string, Array<ServerNotificationHandler>>();
+  const warnedDecodeFailures = new Set<string>();
   let unknownRequestHandler:
     | ((method: string, params: unknown) => Effect.Effect<unknown, CodexError.CodexAppServerError>)
     | undefined;
   let unknownNotificationHandler:
     | ((method: string, params: unknown) => Effect.Effect<void, CodexError.CodexAppServerError>)
     | 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 = <M extends CodexRpc.ServerRequestMethod>(
     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,
       );
     }

packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts

@@ -68,6 +68,19 @@ const handleMethod = (message: Record<string, unknown>) => {
           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": {