feat(resilience): broker /healthz + 优雅关闭 + pending 上限 + 重连 jitter (§8.2)

plugins/agentbridge/server/bridge-server.js

@@ -14707,10 +14707,10 @@ function defineNumber(value, fallback) {
 }
 var BUILD_INFO = Object.freeze({
   version: defineString("0.1.24", "0.0.0-source"),
-  commit: defineString("8284e8a", "source"),
+  commit: defineString("b202e8f", "source"),
   bundle: defineBundle("plugin"),
   contractVersion: defineNumber(1, CONTRACT_VERSION),
-  codeHash: defineString("0a63b984bf2a", "source")
+  codeHash: defineString("7c98560c206e", "source")
 });
 function sameRuntimeContract(a, b) {
   if (!a || !b)

plugins/agentbridge/server/daemon.js

@@ -30,10 +30,10 @@ function defineNumber(value, fallback) {
 }
 var BUILD_INFO = Object.freeze({
   version: defineString("0.1.24", "0.0.0-source"),
-  commit: defineString("8284e8a", "source"),
+  commit: defineString("b202e8f", "source"),
   bundle: defineBundle("plugin"),
   contractVersion: defineNumber(1, CONTRACT_VERSION),
-  codeHash: defineString("0a63b984bf2a", "source")
+  codeHash: defineString("7c98560c206e", "source")
 });
 function daemonStatusBuildInfo() {
   return { ...BUILD_INFO };
@@ -6811,6 +6811,11 @@ class RoomManager {
 }
 
 // src/broker-client.ts
+function reconnectDelay(baseMs, maxMs, attempt, rand) {
+  const ceiling = Math.min(maxMs, baseMs * 2 ** attempt);
+  return ceiling / 2 + rand * (ceiling / 2);
+}
+
 class BrokerClient {
   opts;
   ws = null;
@@ -6830,13 +6835,15 @@ class BrokerClient {
   baseMs;
   maxMs;
   maxOutbox;
+  rand;
   constructor(opts) {
     this.opts = opts;
     this.log = opts.log ?? (() => {});
     this.mkWs = opts.wsFactory ?? ((url) => new WebSocket(url));
     this.baseMs = opts.reconnectBaseMs ?? 250;
     this.maxMs = opts.reconnectMaxMs ?? 1e4;
     this.maxOutbox = opts.maxOutbox ?? 1000;
+    this.rand = opts.random ?? Math.random;
   }
   get connected() {
     return this.ws !== null && this.ws.readyState === WebSocket.OPEN && this.identity !== null;
@@ -6990,9 +6997,9 @@ class BrokerClient {
   scheduleReconnect() {
     if (this.closed || this.reconnectTimer)
       return;
-    const delay = Math.min(this.maxMs, this.baseMs * 2 ** this.reconnectAttempt);
+    const delay = reconnectDelay(this.baseMs, this.maxMs, this.reconnectAttempt, this.rand());
     this.reconnectAttempt++;
-    this.log(`reconnecting in ${delay}ms (attempt ${this.reconnectAttempt})`);
+    this.log(`reconnecting in ${Math.round(delay)}ms (attempt ${this.reconnectAttempt})`);
     this.reconnectTimer = setTimeout(() => {
       this.reconnectTimer = null;
       if (this.closed)
@@ -7065,6 +7072,10 @@ import { dirname as dirname3, join as join11 } from "path";
 // src/backbone/store/sqlite-store.ts
 import { Database } from "bun:sqlite";
 
+// src/backbone/store.ts
+var MAX_PENDING_PER_TARGET = 1000;
+
+// src/backbone/store/sqlite-store.ts
 class SqliteStore {
   db;
   closed = false;
@@ -7203,6 +7214,9 @@ class SqliteStore {
   }
   async enqueuePending(targetAgentId, envelope) {
     this.db.query("INSERT OR IGNORE INTO pending_deliveries(target_agent_id, idempotency_key, envelope) VALUES(?, ?, ?)").run(targetAgentId, envelope.idempotencyKey, JSON.stringify(envelope));
+    this.db.query(`DELETE FROM pending_deliveries WHERE target_agent_id=? AND seq NOT IN (
+           SELECT seq FROM pending_deliveries WHERE target_agent_id=? ORDER BY seq DESC LIMIT ?
+         )`).run(targetAgentId, targetAgentId, MAX_PENDING_PER_TARGET);
   }
   async drainPending(targetAgentId) {
     const rows = this.db.query("SELECT envelope FROM pending_deliveries WHERE target_agent_id=? ORDER BY seq").all(targetAgentId);

src/backbone/store.ts

@@ -1,5 +1,12 @@
 import type { Envelope } from "./envelope";
 
+/**
+ * Max queued envelopes per offline target before the oldest is dropped (§8.2).
+ * Mirrors the BrokerClient outbox bound: logged, bounded loss beats unbounded
+ * growth for a member that never comes back.
+ */
+export const MAX_PENDING_PER_TARGET = 1000;
+
 /**
  * Store interface (spec §6.1, §12 data model).
  *
@@ -83,6 +90,11 @@ export interface Store {
   saveWhiteboard(roomId: string, whiteboard: WhiteboardRecord): Promise<void>;
 
   // --- pending_deliveries: offline replay queue (§3.2), dedup by idempotencyKey ---
+  /**
+   * Queue an envelope for an offline target. Bounded per target at
+   * {@link MAX_PENDING_PER_TARGET} (drop-oldest) so a member that never reconnects
+   * can't grow the backlog without limit (§8.2 resilience).
+   */
   enqueuePending(targetAgentId: string, envelope: Envelope): Promise<void>;
   /** Remove and return the target's pending envelopes (deduped by idempotencyKey). */
   drainPending(targetAgentId: string): Promise<Envelope[]>;

src/backbone/store/memory-store.ts

@@ -1,4 +1,5 @@
 import type { Envelope } from "../envelope";
+import { MAX_PENDING_PER_TARGET } from "../store";
 import type {
   AgentRecord,
   IdentityRecord,
@@ -140,6 +141,11 @@ export class InMemoryStore implements Store {
     if (!byKey.has(envelope.idempotencyKey)) {
       byKey.set(envelope.idempotencyKey, envelope); // dedup: first wins
     }
+    // Bound the per-target backlog (§8.2): Map preserves insertion order, so drop
+    // from the front (oldest) until within MAX_PENDING_PER_TARGET. Matches SqliteStore.
+    while (byKey.size > MAX_PENDING_PER_TARGET) {
+      byKey.delete(byKey.keys().next().value as string);
+    }
   }
 
   async drainPending(targetAgentId: string): Promise<Envelope[]> {

src/backbone/store/sqlite-store.ts

@@ -1,5 +1,6 @@
 import { Database } from "bun:sqlite";
 import type { Envelope } from "../envelope";
+import { MAX_PENDING_PER_TARGET } from "../store";
 import type {
   AgentRecord,
   IdentityRecord,
@@ -242,6 +243,15 @@ export class SqliteStore implements Store {
         "INSERT OR IGNORE INTO pending_deliveries(target_agent_id, idempotency_key, envelope) VALUES(?, ?, ?)",
       )
       .run(targetAgentId, envelope.idempotencyKey, JSON.stringify(envelope));
+    // Bound the per-target backlog (§8.2): keep the newest MAX_PENDING_PER_TARGET,
+    // drop the oldest beyond it, so a never-reconnecting member can't grow the table.
+    this.db
+      .query(
+        `DELETE FROM pending_deliveries WHERE target_agent_id=? AND seq NOT IN (
+           SELECT seq FROM pending_deliveries WHERE target_agent_id=? ORDER BY seq DESC LIMIT ?
+         )`,
+      )
+      .run(targetAgentId, targetAgentId, MAX_PENDING_PER_TARGET);
   }
 
   async drainPending(targetAgentId: string): Promise<Envelope[]> {

src/broker-client.ts

@@ -16,10 +16,23 @@ export interface BrokerClientOptions {
   maxOutbox?: number;
   /** WebSocket factory — injectable so tests can drive reconnect without a real socket. */
   wsFactory?: (url: string) => WebSocket;
+  /** Randomness source for reconnect jitter [0,1) — injectable so tests are deterministic. */
+  random?: () => number;
 }
 
 type EventHandler = (topic: string, envelope: Envelope) => void;
 
+/**
+ * Reconnect backoff with EQUAL JITTER (§8.2). `ceiling = min(maxMs, baseMs·2^attempt)`;
+ * the delay is `ceiling/2 + rand·ceiling/2`, i.e. uniformly in `[ceiling/2, ceiling]`.
+ * Half-fixed keeps a sane minimum wait; half-random de-synchronises adapters that
+ * dropped together (no thundering herd). Result is always `≤ ceiling ≤ maxMs`.
+ */
+export function reconnectDelay(baseMs: number, maxMs: number, attempt: number, rand: number): number {
+  const ceiling = Math.min(maxMs, baseMs * 2 ** attempt);
+  return ceiling / 2 + rand * (ceiling / 2);
+}
+
 /**
  * Edge-side client to the control-plane broker (§5 adapter transport + §8.2
  * resilience foundation).
@@ -55,13 +68,15 @@ export class BrokerClient {
   private readonly baseMs: number;
   private readonly maxMs: number;
   private readonly maxOutbox: number;
+  private readonly rand: () => number;
 
   constructor(private readonly opts: BrokerClientOptions) {
     this.log = opts.log ?? (() => {});
     this.mkWs = opts.wsFactory ?? ((url) => new WebSocket(url));
     this.baseMs = opts.reconnectBaseMs ?? 250;
     this.maxMs = opts.reconnectMaxMs ?? 10_000;
     this.maxOutbox = opts.maxOutbox ?? 1000;
+    this.rand = opts.random ?? Math.random;
   }
 
   get connected(): boolean {
@@ -237,9 +252,9 @@ export class BrokerClient {
 
   private scheduleReconnect(): void {
     if (this.closed || this.reconnectTimer) return;
-    const delay = Math.min(this.maxMs, this.baseMs * 2 ** this.reconnectAttempt);
+    const delay = reconnectDelay(this.baseMs, this.maxMs, this.reconnectAttempt, this.rand());
     this.reconnectAttempt++;
-    this.log(`reconnecting in ${delay}ms (attempt ${this.reconnectAttempt})`);
+    this.log(`reconnecting in ${Math.round(delay)}ms (attempt ${this.reconnectAttempt})`);
     this.reconnectTimer = setTimeout(() => {
       this.reconnectTimer = null;
       if (this.closed) return;

src/broker.ts

@@ -72,6 +72,10 @@ export interface BrokerOptions {
 export class Broker {
   private server: ReturnType<typeof Bun.serve> | null = null;
   private nextConnId = 0;
+  /** Live WS connections (incremented on upgrade, decremented on close) — for /healthz. */
+  private liveConnections = 0;
+  /** Epoch ms the server started, for /healthz uptime. 0 until start(). */
+  private startedAt = 0;
   /** topic → (identityId → live-subscription count) — who is reachable per topic. */
   private readonly topicMembers = new Map<string, Map<string, number>>();
   private readonly transport: MessageTransport;
@@ -90,14 +94,23 @@ export class Broker {
     // loopback rather than become an all-interfaces bind (`Bun.serve({hostname:""})`).
     const host = this.opts.host || "127.0.0.1";
     const port = this.opts.port ?? DEFAULT_BROKER_PORT;
+    this.startedAt = Date.now();
     // eslint-disable-next-line @typescript-eslint/no-this-alias
     const self = this;
     const server = Bun.serve<BrokerSocketData>({
       hostname: host,
       port,
       fetch(req, server) {
-        if (new URL(req.url).pathname === "/ws") {
+        const pathname = new URL(req.url).pathname;
+        if (pathname === "/healthz") {
+          // Liveness probe for a watchdog/supervisor (§8.2). Minimal, NON-SENSITIVE
+          // body — the broker binds a Tailscale 100.x address reachable by any
+          // tailnet node, so this must never leak tokens/PII/identities.
+          return Response.json(self.healthBody());
+        }
+        if (pathname === "/ws") {
           if (server.upgrade(req, { data: { connId: ++self.nextConnId, subs: new Map() } })) {
+            self.liveConnections++;
             return undefined;
           }
         }
@@ -124,6 +137,7 @@ export class Broker {
           }
           for (const unsub of ws.data.subs.values()) unsub();
           ws.data.subs.clear();
+          if (self.liveConnections > 0) self.liveConnections--;
           self.log(`conn #${ws.data.connId} closed`);
         },
       },
@@ -133,6 +147,16 @@ export class Broker {
     return { host, port: server.port ?? port };
   }
 
+  /** Non-sensitive liveness body for GET /healthz (§8.2 watchdog). No tokens/PII/identities. */
+  private healthBody(): { ok: true; pid: number; uptimeMs: number; connections: number } {
+    return {
+      ok: true,
+      pid: process.pid,
+      uptimeMs: this.startedAt === 0 ? 0 : Date.now() - this.startedAt,
+      connections: this.liveConnections,
+    };
+  }
+
   stop(): void {
     this.server?.stop(true);
     this.server = null;

src/cli/broker.ts

@@ -91,7 +91,22 @@ export async function runBrokerStart(argv: string[]): Promise<void> {
   console.log(`AgentBridge broker 已启动，监听 ${bound.host}:${bound.port}`);
   console.log(`协作数据库：${dbPath}`);
   console.log("用 abg auth login 签发的 token 连接；Ctrl-C 停止。");
-  // Bun.serve keeps the event loop alive — the process stays up until killed.
+
+  // Graceful shutdown (§8.2): on SIGTERM/SIGINT, stop the server then close the
+  // Store — db.close() checkpoints the WAL so pending_deliveries survive the
+  // restart and a reconnecting member can still drain them. `once` so a second
+  // signal during teardown doesn't re-enter; exit 0 even if close() rejects.
+  let stopping = false;
+  const shutdown = (sig: string) => {
+    if (stopping) return;
+    stopping = true;
+    console.error(`[broker] ${sig} 收到，正在优雅关闭…`);
+    broker.stop();
+    store.close().finally(() => process.exit(0));
+  };
+  process.once("SIGTERM", () => shutdown("SIGTERM"));
+  process.once("SIGINT", () => shutdown("SIGINT"));
+  // Bun.serve keeps the event loop alive — the process stays up until a signal.
 }
 
 export async function runBroker(args: string[]): Promise<void> {

src/integration-test/broker-graceful-shutdown.test.ts

@@ -0,0 +1,47 @@
+import { describe, test, expect, afterEach } from "bun:test";
+import { spawn } from "node:child_process";
+import { fileURLToPath } from "node:url";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+const CLI_PATH = fileURLToPath(new URL("../cli.ts", import.meta.url));
+
+describe("abg broker start — graceful shutdown (§8.2)", () => {
+  let dir: string | undefined;
+  afterEach(() => {
+    if (dir) rmSync(dir, { recursive: true, force: true });
+    dir = undefined;
+  });
+
+  test("SIGTERM stops the server, closes the Store, and exits 0", async () => {
+    dir = mkdtempSync(join(tmpdir(), "agentbridge-broker-sig-"));
+    const dbPath = join(dir, "collab.db");
+    const child = spawn(process.execPath, ["run", CLI_PATH, "broker", "start", "--port", "0", "--db", dbPath], {
+      env: { ...process.env, AGENTBRIDGE_COLLAB_DB: dbPath },
+      stdio: ["ignore", "pipe", "pipe"],
+    });
+
+    // Wait for the broker to finish binding before signalling (so the SIGTERM
+    // handler is installed — it's registered right after start()).
+    await new Promise<void>((resolve, reject) => {
+      const to = setTimeout(() => reject(new Error("broker did not start in time")), 8000);
+      child.stdout.on("data", (d: Buffer) => {
+        if (String(d).includes("已启动")) {
+          clearTimeout(to);
+          resolve();
+        }
+      });
+      child.once("exit", (c) => {
+        clearTimeout(to);
+        reject(new Error(`broker exited before startup (code ${c})`));
+      });
+    });
+
+    const exitCode = await new Promise<number | null>((resolve) => {
+      child.once("exit", (code) => resolve(code));
+      child.kill("SIGTERM");
+    });
+    expect(exitCode).toBe(0); // graceful: exit 0, not killed by the signal
+  }, 15000);
+});

src/integration-test/broker-healthz.test.ts

@@ -0,0 +1,54 @@
+import { describe, test, expect, afterEach } from "bun:test";
+import { Broker } from "../broker";
+import { InMemoryStore } from "../backbone/store/memory-store";
+import { IdentityService } from "../backbone/identity-service";
+import { StorePskIdentityProvider } from "../backbone/identity/store-psk-identity-provider";
+
+describe("Broker /healthz — liveness probe (§8.2 watchdog)", () => {
+  let stop: (() => void) | undefined;
+  afterEach(() => {
+    stop?.();
+    stop = undefined;
+  });
+
+  async function start() {
+    const store = new InMemoryStore();
+    const svc = new IdentityService(store);
+    await svc.registerIdentity("alice@x.com", "Alice");
+    const token = await svc.issueToken("alice@x.com");
+    const broker = new Broker({ store, identityProvider: new StorePskIdentityProvider(store), host: "127.0.0.1", port: 0, log: () => {} });
+    const { port } = broker.start();
+    stop = () => broker.stop();
+    return { port, token, base: `http://127.0.0.1:${port}` };
+  }
+
+  test("GET /healthz returns 200 + a minimal non-sensitive JSON body", async () => {
+    const { base, token } = await start();
+    const res = await fetch(`${base}/healthz`);
+    expect(res.status).toBe(200);
+    const body = await res.json();
+    expect(body.ok).toBe(true);
+    expect(typeof body.pid).toBe("number");
+    expect(typeof body.uptimeMs).toBe("number");
+    expect(body.uptimeMs).toBeGreaterThanOrEqual(0);
+    expect(body.connections).toBe(0); // no ws connected yet
+    // must NOT leak secrets/PII
+    const raw = JSON.stringify(body);
+    expect(raw).not.toContain(token);
+    expect(raw).not.toContain("alice@x.com");
+  });
+
+  test("connections reflects a live ws and drops when it closes", async () => {
+    const { base, port, token } = await start();
+    const ws = new WebSocket(`ws://127.0.0.1:${port}/ws`);
+    await new Promise<void>((res, rej) => {
+      ws.onopen = () => res();
+      ws.onerror = () => rej(new Error("ws failed"));
+    });
+    await new Promise((r) => setTimeout(r, 30));
+    expect((await (await fetch(`${base}/healthz`)).json()).connections).toBe(1);
+    ws.close();
+    await new Promise((r) => setTimeout(r, 60));
+    expect((await (await fetch(`${base}/healthz`)).json()).connections).toBe(0);
+  });
+});