fix(gateway): stop broadcast serialization failures from storming gap detectors (#124726)

broadcastInternal consumed each client's seq before building the frame,
and swallowed any error in a bare catch. An unserializable payload
(circular reference, BigInt) threw identically for every client: the
event vanished for all of them, every seq advanced, and the next good
frame fired every client's gap detector simultaneously — a synchronized
full-reconnect storm with zero server-side evidence.

Root cause: seq consumption ordered before frame construction, and
serialization failure conflated with per-client send failure in one
catch. The frame now builds before the seq is consumed; a serialization
failure aborts the broadcast once with a logged error and touches no
seqs, while per-client send failures keep consuming the seq so that
client's gap detector still sees the loss.

Regression: new server-broadcast.serialization.test.ts — a circular
payload consumes no seqs, logs once, and the next broadcast is seq 1
for every client. Fails pre-fix (no log, seqs consumed).

src/gateway/server-broadcast.serialization.test.ts

@@ -0,0 +1,77 @@
+// Covers broadcast frame-serialization failure: an unserializable payload must
+// not consume per-client seqs (which would fire every client's gap detector and
+// cause a synchronized reconnect storm) and must leave a server-side record.
+import { describe, expect, it, vi } from "vitest";
+import { createGatewayBroadcaster } from "./server-broadcast.js";
+import type { GatewayWsClient } from "./server/ws-types.js";
+
+const warnSpy = vi.hoisted(() => vi.fn());
+
+vi.mock("../logging/subsystem.js", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("../logging/subsystem.js")>();
+  return {
+    ...actual,
+    createSubsystemLogger: (subsystem: string) => {
+      const logger = actual.createSubsystemLogger(subsystem);
+      if (subsystem !== "gateway/broadcast") {
+        return logger;
+      }
+      return { ...logger, error: warnSpy };
+    },
+  };
+});
+
+type RecordingSocket = {
+  bufferedAmount: number;
+  close: ReturnType<typeof vi.fn>;
+  send: ReturnType<typeof vi.fn>;
+  frames: Array<{ event: string; seq: number }>;
+};
+
+function makeClient(connId: string): { client: GatewayWsClient; socket: RecordingSocket } {
+  const frames: Array<{ event: string; seq: number }> = [];
+  const socket: RecordingSocket = {
+    bufferedAmount: 0,
+    close: vi.fn(),
+    send: vi.fn((payload: string) => {
+      const frame = JSON.parse(payload) as { event: string; seq: number };
+      frames.push({ event: frame.event, seq: frame.seq });
+    }),
+    frames,
+  };
+  return {
+    client: {
+      socket: socket as unknown as GatewayWsClient["socket"],
+      connect: { role: "operator", scopes: ["operator.read"] } as GatewayWsClient["connect"],
+      connId,
+      usesSharedGatewayAuth: false,
+    },
+    socket,
+  };
+}
+
+describe("broadcast serialization failures", () => {
+  it("drops the event without consuming seqs when the payload cannot serialize", () => {
+    warnSpy.mockClear();
+    const first = makeClient("first");
+    const second = makeClient("second");
+    const clients = new Set([first.client, second.client]);
+    const { broadcast } = createGatewayBroadcaster({ clients });
+
+    const circular: Record<string, unknown> = {};
+    circular.self = circular;
+    broadcast("skills.changed", circular);
+
+    // Neither socket saw the bad frame, and the failure is recorded once.
+    expect(first.socket.send).not.toHaveBeenCalled();
+    expect(second.socket.send).not.toHaveBeenCalled();
+    expect(warnSpy).toHaveBeenCalledTimes(1);
+    expect(String(warnSpy.mock.calls[0]?.[0])).toContain("skills.changed");
+
+    // The next good broadcast starts at seq 1 for every client: the dropped
+    // event consumed no seq, so no gap detector fires.
+    broadcast("skills.changed", { reason: "recovered" });
+    expect(first.socket.frames).toEqual([{ event: "skills.changed", seq: 1 }]);
+    expect(second.socket.frames).toEqual([{ event: "skills.changed", seq: 1 }]);
+  });
+});

src/gateway/server-broadcast.ts

@@ -2,9 +2,11 @@ import {
   GATEWAY_CLIENT_CAPS,
   hasGatewayClientCap,
 } from "../../packages/gateway-protocol/src/client-info.js";
+import { formatErrorMessage } from "../infra/errors.js";
 // Gateway WebSocket broadcaster.
 // Applies event scope guards and slow-consumer handling before sending frames.
 import { logRejectedLargePayload } from "../logging/diagnostic-payload.js";
+import { createSubsystemLogger } from "../logging/subsystem.js";
 import { queuePluginSessionsChanged } from "../plugins/gateway-events.js";
 import { isBrowserCopilotClient } from "../utils/message-channel.js";
 import { GATEWAY_EVENT_NODE_RUNNER_INVENTORY_CHANGED } from "./events.js";
@@ -94,6 +96,8 @@ const EVENT_SCOPE_GUARDS: Record<string, string[]> = {
 
 // Opt-in scoped clients never receive session-bearing broadcasts without an
 // authoritative registry key, including malformed/sessionless agent events.
+const log = createSubsystemLogger("gateway/broadcast");
+
 const SESSION_SUBSCRIPTION_EVENTS = new Set([
   "agent",
   "chat",
@@ -248,6 +252,8 @@ export function createGatewayBroadcaster(params: {
           stateVersionFragment: string;
         }
       | undefined;
+    // Lazy so filtered-out broadcasts (zero eligible clients) never pay
+    // JSON.stringify for the payload.
     const getFrameBase = () => {
       if (!frameBase) {
         frameBase = {
@@ -325,16 +331,27 @@ export function createGatewayBroadcaster(params: {
         }
         continue;
       }
+      // Build the frame before consuming the seq: a serialization failure
+      // (circular/BigInt payload) throws identically for every client, and
+      // advancing seqs for a frame that never existed would fire every gap
+      // detector at once — a synchronized reconnect storm with no evidence.
+      let frame: string;
       try {
-        // Targeted frames ride the same per-client sequence as fanout frames:
-        // an unstamped frame is invisible to the client's gap detector, so a
-        // drop between two targeted sends would go unnoticed forever.
-        clientSeq.set(c, nextSeq);
         const base = getFrameBase();
-        const frame = `{"type":"event","event":${base.eventJSON}${base.payloadFragment},"seq":${nextSeq}${base.stateVersionFragment}}`;
+        frame = `{"type":"event","event":${base.eventJSON}${base.payloadFragment},"seq":${nextSeq}${base.stateVersionFragment}}`;
+      } catch (err) {
+        log.error(`broadcast serialization failed for event ${event}: ${formatErrorMessage(err)}`);
+        return;
+      }
+      // Targeted frames ride the same per-client sequence as fanout frames:
+      // an unstamped frame is invisible to the client's gap detector, so a
+      // drop between two targeted sends would go unnoticed forever.
+      clientSeq.set(c, nextSeq);
+      try {
         c.socket.send(frame);
       } catch {
-        /* ignore */
+        // The consumed seq makes this send failure visible to the client's
+        // gap detector on its next received frame.
       }
     }
   };