diff --git a/.dockerignore b/.dockerignore
new file mode 100644
index 00000000..8863eca1
--- /dev/null
+++ b/.dockerignore
@@ -0,0 +1,10 @@
+node_modules
+.git
+dist
+.agent
+*.log
+.DS_Store
+collab.db
+collab.db-wal
+collab.db-shm
+auth-token
diff --git a/docker/Dockerfile b/docker/Dockerfile
new file mode 100644
index 00000000..ef43ddbb
--- /dev/null
+++ b/docker/Dockerfile
@@ -0,0 +1,7 @@
+# Multi-machine simulation image (§13). Zero runtime deps — Bun runs the TS
+# directly and the broker/client only import src/backbone + Bun built-ins.
+FROM oven/bun:1.3.11
+WORKDIR /app
+COPY package.json tsconfig.json ./
+COPY src ./src
+COPY docker ./docker
diff --git a/docker/README.md b/docker/README.md
new file mode 100644
index 00000000..4c2092fd
--- /dev/null
+++ b/docker/README.md
@@ -0,0 +1,37 @@
+# 多机协作模拟（Docker）
+
+用多个容器模拟「多台机器、多个 agent」连同一个常开 broker 协作，验证规格 §13 的跨机切片
+（PSK 鉴权可达 + broker 扇出），无需真实多台物理机。
+
+## 拓扑
+
+| 容器 | 角色 |
+|------|------|
+| `provision` | 注册身份、签发 PSK token 写入共享卷（一次性） |
+| `broker` | 常开控制面 broker（一台「服务器机」），WSS + PSK + 扇出 |
+| `subscriber` | 一台「agent 机」：鉴权 → 订阅房间 → 等事件 |
+| `publisher` | 另一台「agent 机」：鉴权 → 发布 `task_completed` 事件 |
+
+## 跑
+
+```bash
+docker compose -f docker/docker-compose.yml up --build --abort-on-container-exit
+```
+
+**通过判据**：`subscriber` 容器打印 `SIM_OK` 并以 0 退出——表示 publisher 容器发布的事件
+经 broker 跨容器扇出，被 subscriber 容器收到（跨「机」鉴权 + 路由全链路通）。
+
+清理：
+
+```bash
+docker compose -f docker/docker-compose.yml down -v
+```
+
+## 说明
+
+- 容器仅为模拟共享 `/data` 卷分发 token + DB；真实部署里 token 走带外分发、每台机器各自持
+  工作副本（**代码同步是 git 的职责，§2.6；broker 永不传文件**）。
+- broker 在容器内绑 `0.0.0.0`（容器网络已隔离）；**跨内网真机请绑 Tailscale 的 `100.x`，
+  绝不绑 `0.0.0.0`（§7.3）**。
+- 这是 broker 层的跨机切片验证。完整的「多人多 agent 完成事件协作」E2E 需要 Edge adapter
+  （后续 PR）与 `task_completed` 语义齐备后再扩。
diff --git a/docker/broker-entry.ts b/docker/broker-entry.ts
new file mode 100644
index 00000000..de904433
--- /dev/null
+++ b/docker/broker-entry.ts
@@ -0,0 +1,27 @@
+/**
+ * Container entrypoint for the broker (multi-machine simulation, §11.1/§13).
+ * Constructs the broker directly (no CLI) so the image needs no extra deps.
+ */
+import { chmodSync, mkdirSync } from "node:fs";
+import { dirname } from "node:path";
+import { Broker } from "../src/broker";
+import { SqliteStore } from "../src/backbone/store/sqlite-store";
+import { StorePskIdentityProvider } from "../src/backbone/identity/store-psk-identity-provider";
+
+const host = process.env.BROKER_HOST ?? "0.0.0.0"; // 0.0.0.0 is fine INSIDE a container
+const port = parseInt(process.env.BROKER_PORT ?? "4700", 10);
+const db = process.env.COLLAB_DB ?? "/data/collab.db";
+
+mkdirSync(dirname(db), { recursive: true, mode: 0o700 });
+chmodSync(dirname(db), 0o700);
+
+const store = new SqliteStore(db);
+const broker = new Broker({
+  store,
+  identityProvider: new StorePskIdentityProvider(store),
+  host,
+  port,
+  log: (m) => console.error(`[broker] ${m}`),
+});
+const bound = broker.start();
+console.log(`[broker] up on ${bound.host}:${bound.port} (db ${db})`);
diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
new file mode 100644
index 00000000..828935b2
--- /dev/null
+++ b/docker/docker-compose.yml
@@ -0,0 +1,68 @@
+# Multi-machine broker simulation (§13 cross-machine slice).
+#
+#   provision → issues a PSK token into the shared collab DB
+#   broker    → the always-on control-plane broker (one "server machine")
+#   subscriber/publisher → two "agent machines" exchanging an event over the broker
+#
+# Run:  docker compose -f docker/docker-compose.yml up --build --abort-on-container-exit
+# Pass: the `subscriber` container prints `SIM_OK` and exits 0.
+#
+# NOTE: the containers share a /data volume only to distribute the token + DB for
+# the sim; in a real deployment the token is distributed out of band and each
+# machine has its own working copy (code sync is git's job, §2.6).
+
+services:
+  provision:
+    build:
+      context: ..
+      dockerfile: docker/Dockerfile
+    volumes: [collab:/data]
+    command: ["bun", "docker/provision.ts"]
+    environment:
+      COLLAB_DB: /data/collab.db
+      TOKEN_FILE: /data/token
+      SIM_ID: sim@example.com
+
+  broker:
+    build:
+      context: ..
+      dockerfile: docker/Dockerfile
+    volumes: [collab:/data]
+    depends_on:
+      provision:
+        condition: service_completed_successfully
+    command: ["bun", "docker/broker-entry.ts"]
+    environment:
+      BROKER_HOST: 0.0.0.0
+      BROKER_PORT: "4700"
+      COLLAB_DB: /data/collab.db
+    ports: ["4700:4700"]
+
+  subscriber:
+    build:
+      context: ..
+      dockerfile: docker/Dockerfile
+    volumes: [collab:/data]
+    depends_on: [broker]
+    command: ["bun", "docker/sim-client.ts"]
+    environment:
+      ROLE: subscriber
+      BROKER_URL: ws://broker:4700/ws
+      TOKEN_FILE: /data/token
+      TOPIC: demo-room
+
+  publisher:
+    build:
+      context: ..
+      dockerfile: docker/Dockerfile
+    volumes: [collab:/data]
+    depends_on: [broker, subscriber]
+    command: ["bun", "docker/sim-client.ts"]
+    environment:
+      ROLE: publisher
+      BROKER_URL: ws://broker:4700/ws
+      TOKEN_FILE: /data/token
+      TOPIC: demo-room
+
+volumes:
+  collab: {}
diff --git a/docker/provision.ts b/docker/provision.ts
new file mode 100644
index 00000000..cd6133d1
--- /dev/null
+++ b/docker/provision.ts
@@ -0,0 +1,28 @@
+/**
+ * Provision step for the multi-machine sim: register an identity, issue a PSK
+ * token, write it to the shared volume for the client containers to read.
+ *
+ * (In a real deployment the token is distributed out of band; the sim shares a
+ * volume purely for convenience.)
+ */
+import { chmodSync, mkdirSync, writeFileSync } from "node:fs";
+import { dirname } from "node:path";
+import { SqliteStore } from "../src/backbone/store/sqlite-store";
+import { IdentityService } from "../src/backbone/identity-service";
+
+const db = process.env.COLLAB_DB ?? "/data/collab.db";
+const id = process.env.SIM_ID ?? "sim@example.com";
+const name = process.env.SIM_NAME ?? "Sim Agent";
+const tokenFile = process.env.TOKEN_FILE ?? "/data/token";
+
+mkdirSync(dirname(db), { recursive: true, mode: 0o700 });
+chmodSync(dirname(db), 0o700);
+
+const store = new SqliteStore(db);
+const svc = new IdentityService(store);
+await svc.registerIdentity(id, name);
+const token = await svc.issueToken(id);
+await store.close();
+
+writeFileSync(tokenFile, token, { mode: 0o600 });
+console.log(`[provision] identity ${id} registered, token written to ${tokenFile}`);
diff --git a/docker/sim-client.ts b/docker/sim-client.ts
new file mode 100644
index 00000000..556502f2
--- /dev/null
+++ b/docker/sim-client.ts
@@ -0,0 +1,79 @@
+/**
+ * Simulated agent client (one per container = one "machine"). Authenticates by
+ * PSK over the broker WS, then either subscribes-and-waits or publishes.
+ *
+ * ROLE=subscriber: subscribe to TOPIC, wait for one event, print SIM_OK, exit 0.
+ * ROLE=publisher : wait for the subscriber, publish a few task_completed events.
+ */
+import { readFileSync } from "node:fs";
+
+const url = process.env.BROKER_URL ?? "ws://broker:4700/ws";
+const role = process.env.ROLE ?? "subscriber";
+const topic = process.env.TOPIC ?? "demo";
+const token = (process.env.TOKEN ?? readFileSync(process.env.TOKEN_FILE ?? "/data/token", "utf8")).trim();
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+const ws = new WebSocket(url);
+const inbox: any[] = [];
+const waiters: ((m: any) => void)[] = [];
+ws.onmessage = (ev) => {
+  const m = JSON.parse(ev.data as string);
+  const w = waiters.shift();
+  if (w) w(m);
+  else inbox.push(m);
+};
+function next(): Promise<any> {
+  const m = inbox.shift();
+  if (m !== undefined) return Promise.resolve(m);
+  return new Promise((r) => waiters.push(r));
+}
+const send = (m: unknown) => ws.send(JSON.stringify(m));
+
+await new Promise<void>((res, rej) => {
+  ws.onopen = () => res();
+  ws.onerror = () => rej(new Error("connect failed"));
+});
+
+send({ type: "hello", token });
+const welcome = await next();
+if (welcome.type !== "welcome") {
+  console.error(`[${role}] AUTH FAILED:`, welcome);
+  process.exit(1);
+}
+console.log(`[${role}] authenticated as ${welcome.identity.id}`);
+
+if (role === "subscriber") {
+  send({ type: "subscribe", topic });
+  await next(); // subscribed ack
+  console.log(`[subscriber] subscribed to "${topic}", awaiting event...`);
+  const ev = await next();
+  console.log(`[subscriber] RECEIVED:`, JSON.stringify(ev));
+  if (ev.type === "event" && ev.envelope?.from?.agentId) {
+    console.log("SIM_OK");
+    process.exit(0);
+  }
+  console.error("[subscriber] unexpected message:", ev);
+  process.exit(1);
+} else {
+  await sleep(5000); // give the subscriber container time to subscribe
+  for (let i = 0; i < 3; i++) {
+    send({
+      type: "publish",
+      topic,
+      envelope: {
+        roomId: topic,
+        messageId: `m${i}`,
+        traceId: "trace-sim",
+        idempotencyKey: `k${i}`,
+        from: { agentId: "sim-publisher", agentType: "claude" },
+        kind: "task_completed",
+        timestamp: Date.now(),
+        deliveryMode: "store_if_offline",
+        payload: { summary: "hello from the publisher container" },
+      },
+    });
+    await sleep(1000);
+  }
+  console.log("[publisher] published 3 events");
+  process.exit(0);
+}
diff --git a/plugins/agentbridge/server/bridge-server.js b/plugins/agentbridge/server/bridge-server.js
index 0f7334d7..86833ff2 100755
--- a/plugins/agentbridge/server/bridge-server.js
+++ b/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("dbefc72", "source"),
+  commit: defineString("851f82a", "source"),
   bundle: defineBundle("plugin"),
   contractVersion: defineNumber(1, CONTRACT_VERSION),
-  codeHash: defineString("fc5f11bd22b6", "source")
+  codeHash: defineString("29db7e466211", "source")
 });
 function sameRuntimeContract(a, b) {
   if (!a || !b)
diff --git a/plugins/agentbridge/server/daemon.js b/plugins/agentbridge/server/daemon.js
index 1cb62d74..d383adc0 100755
--- a/plugins/agentbridge/server/daemon.js
+++ 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("dbefc72", "source"),
+  commit: defineString("851f82a", "source"),
   bundle: defineBundle("plugin"),
   contractVersion: defineNumber(1, CONTRACT_VERSION),
-  codeHash: defineString("fc5f11bd22b6", "source")
+  codeHash: defineString("29db7e466211", "source")
 });
 function daemonStatusBuildInfo() {
   return { ...BUILD_INFO };
diff --git a/src/backbone/transport/in-memory-transport.ts b/src/backbone/transport/in-memory-transport.ts
index 5c179e69..fcf1a4f4 100644
--- a/src/backbone/transport/in-memory-transport.ts
+++ b/src/backbone/transport/in-memory-transport.ts
@@ -27,7 +27,11 @@ export class InMemoryTransport implements MessageTransport {
     const set = this.topics.get(topic);
     if (set) {
       for (const handler of [...set]) {
-        handler(msg);
+        try {
+          handler(msg);
+        } catch {
+          // Isolate a throwing subscriber (parity with InProcTransport, §6.4).
+        }
       }
     }
     return Promise.resolve();
diff --git a/src/backbone/transport/inproc-transport.ts b/src/backbone/transport/inproc-transport.ts
index e8a3adba..1ee04f15 100644
--- a/src/backbone/transport/inproc-transport.ts
+++ b/src/backbone/transport/inproc-transport.ts
@@ -1,14 +1,26 @@
 import type { Envelope } from "../envelope";
 import type { MessageTransport, Unsubscribe } from "../transport";
 
+export interface InProcTransportOptions {
+  /**
+   * Invoked when a subscriber handler throws. A throwing subscriber MUST NOT
+   * block fan-out to its siblings (the broker fans out to many WS clients), so
+   * each handler is isolated; this sink surfaces the error for logging.
+   */
+  onHandlerError?: (err: unknown, topic: string) => void;
+}
+
 /**
  * In-process pub/sub transport (spec §6.1 battery impl). Topic → handler set;
  * publish snapshots the set before fanning out so a handler that (un)subscribes
- * during delivery cannot corrupt the iteration.
+ * during delivery cannot corrupt the iteration, and isolates each handler so one
+ * throwing subscriber cannot starve the others.
  */
 export class InProcTransport implements MessageTransport {
   private readonly topics = new Map<string, Set<(m: Envelope) => void>>();
 
+  constructor(private readonly options: InProcTransportOptions = {}) {}
+
   subscribe(topic: string, handler: (msg: Envelope) => void): Unsubscribe {
     let set = this.topics.get(topic);
     if (!set) {
@@ -25,7 +37,12 @@ export class InProcTransport implements MessageTransport {
     const set = this.topics.get(topic);
     if (set) {
       for (const handler of [...set]) {
-        handler(msg);
+        try {
+          handler(msg);
+        } catch (err) {
+          // Isolate: a throwing subscriber must not block delivery to siblings.
+          this.options.onHandlerError?.(err, topic);
+        }
       }
     }
     return Promise.resolve();
diff --git a/src/broker.ts b/src/broker.ts
new file mode 100644
index 00000000..d87ad434
--- /dev/null
+++ b/src/broker.ts
@@ -0,0 +1,181 @@
+import type { ServerWebSocket } from "bun";
+import type { Store } from "./backbone/store";
+import type { Identity, IdentityProvider } from "./backbone/identity";
+import type { MessageTransport } from "./backbone/transport";
+import type { Envelope } from "./backbone/envelope";
+import { InProcTransport } from "./backbone/transport/inproc-transport";
+
+export const DEFAULT_BROKER_PORT = 4700; // outside the multi-pair 4500/4501/4502+stride range
+const CLOSE_AUTH_FAILED = 4401;
+
+interface BrokerSocketData {
+  connId: number;
+  identity?: Identity;
+  /** topic → unsubscribe handle for this connection's subscriptions. */
+  subs: Map<string, () => void>;
+}
+
+type ClientMessage =
+  | { type: "hello"; token: string }
+  | { type: "subscribe"; topic: string }
+  | { type: "unsubscribe"; topic: string }
+  | { type: "publish"; topic: string; envelope: Envelope };
+
+export interface BrokerOptions {
+  store: Store;
+  identityProvider: IdentityProvider;
+  /** Bind host. Default 127.0.0.1; for Tailscale bind the 100.x address (never 0.0.0.0, §7.3). */
+  host?: string;
+  /** Bind port. Default {@link DEFAULT_BROKER_PORT}; 0 picks a random free port. */
+  port?: number;
+  transport?: MessageTransport;
+  log?: (msg: string) => void;
+}
+
+/**
+ * The always-on, multi-tenant control-plane event broker (§11.1).
+ *
+ * A WSS endpoint that authenticates every connection by PSK (IdentityProvider),
+ * then routes Envelopes between authenticated clients via a MessageTransport
+ * (in-process bus + WSS fan-out, §6.2). **CONTROL PLANE ONLY**: it accepts and
+ * forwards Envelopes (structured signals) and NEVER reads/writes repo files —
+ * code sync is git's job (§2.6). It is a SEPARATE process from the per-pair
+ * daemon (independent failure domain) and binds a CONFIGURABLE host (default
+ * loopback; Tailscale uses the 100.x address, never 0.0.0.0 — §7.3).
+ *
+ * Loop prevention / dedup (traceId/hop) and three-tier routing (broadcast /
+ * @mention / DM) are layered on top in a later PR; this PR is plain topic fan-out.
+ */
+export class Broker {
+  private server: ReturnType<typeof Bun.serve> | null = null;
+  private nextConnId = 0;
+  private readonly transport: MessageTransport;
+  private readonly log: (msg: string) => void;
+
+  constructor(private readonly opts: BrokerOptions) {
+    this.log = opts.log ?? (() => {});
+    this.transport =
+      opts.transport ??
+      new InProcTransport({ onHandlerError: (e) => this.log(`subscriber handler error: ${String(e)}`) });
+  }
+
+  /** Start listening. Returns the bound { host, port } (port resolved if 0 was given). */
+  start(): { host: string; port: number } {
+    // `||` not `??`: a programmatically-passed empty string must fall back to
+    // 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;
+    // 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") {
+          if (server.upgrade(req, { data: { connId: ++self.nextConnId, subs: new Map() } })) {
+            return undefined;
+          }
+        }
+        return new Response("AgentBridge broker");
+      },
+      websocket: {
+        message(ws, raw) {
+          // Catch any unexpected rejection so a bad message can never become an
+          // unhandled promise rejection that takes the process down.
+          self.handleMessage(ws, typeof raw === "string" ? raw : raw.toString()).catch((e) => {
+            self.log(`message handler error (#${ws.data.connId}): ${String(e)}`);
+          });
+        },
+        close(ws) {
+          for (const unsub of ws.data.subs.values()) unsub();
+          ws.data.subs.clear();
+          self.log(`conn #${ws.data.connId} closed`);
+        },
+      },
+    });
+    this.server = server;
+    this.log(`broker listening on ${host}:${server.port}`);
+    return { host, port: server.port ?? port };
+  }
+
+  stop(): void {
+    this.server?.stop(true);
+    this.server = null;
+  }
+
+  private send(ws: ServerWebSocket<BrokerSocketData>, msg: unknown): void {
+    try {
+      ws.send(JSON.stringify(msg));
+    } catch (e) {
+      this.log(`send failed (#${ws.data.connId}): ${String(e)}`);
+    }
+  }
+
+  private async handleMessage(ws: ServerWebSocket<BrokerSocketData>, raw: string): Promise<void> {
+    let parsed: unknown;
+    try {
+      parsed = JSON.parse(raw);
+    } catch {
+      this.send(ws, { type: "error", reason: "invalid JSON" });
+      return;
+    }
+    // Valid JSON that isn't a tagged object (null / number / string / array)
+    // must not reach `.type` access — that would throw out of this async handler
+    // and become an unhandled rejection. Reject it as malformed.
+    if (typeof parsed !== "object" || parsed === null || typeof (parsed as { type?: unknown }).type !== "string") {
+      this.send(ws, { type: "error", reason: "malformed message" });
+      return;
+    }
+    const msg = parsed as ClientMessage;
+
+    if (msg.type === "hello") {
+      try {
+        const identity = await this.opts.identityProvider.authenticate(msg.token);
+        ws.data.identity = identity;
+        this.send(ws, { type: "welcome", identity });
+        this.log(`conn #${ws.data.connId} authenticated as ${identity.id}`);
+      } catch {
+        // Never echo the presented token or the underlying reason.
+        this.send(ws, { type: "auth_error", reason: "invalid token" });
+        ws.close(CLOSE_AUTH_FAILED, "auth failed");
+      }
+      return;
+    }
+
+    // Every non-hello message requires a prior successful hello.
+    if (!ws.data.identity) {
+      this.send(ws, { type: "error", reason: "not authenticated (send hello first)" });
+      return;
+    }
+
+    switch (msg.type) {
+      case "subscribe": {
+        if (ws.data.subs.has(msg.topic)) return; // idempotent
+        const topic = msg.topic;
+        const unsub = this.transport.subscribe(topic, (envelope) => {
+          this.send(ws, { type: "event", topic, envelope });
+        });
+        ws.data.subs.set(topic, unsub);
+        this.send(ws, { type: "subscribed", topic }); // ack: safe to publish now
+        return;
+      }
+      case "unsubscribe": {
+        const unsub = ws.data.subs.get(msg.topic);
+        if (unsub) {
+          unsub();
+          ws.data.subs.delete(msg.topic);
+        }
+        return;
+      }
+      case "publish": {
+        // CONTROL PLANE ONLY: forward the structured Envelope. No filesystem,
+        // no repo access — code sync is git's job (§2.6).
+        await this.transport.publish(msg.topic, msg.envelope);
+        return;
+      }
+      default: {
+        this.send(ws, { type: "error", reason: "unknown message type" });
+      }
+    }
+  }
+}
diff --git a/src/cli.ts b/src/cli.ts
index fe699d9e..af51142e 100755
--- a/src/cli.ts
+++ b/src/cli.ts
@@ -123,6 +123,10 @@ async function main(command: string | undefined, restArgs: string[]) {
       const { runAuth } = await import("./cli/auth");
       await runAuth(restArgs);
       break;
+    case "broker":
+      const { runBroker } = await import("./cli/broker");
+      await runBroker(restArgs);
+      break;
     case "--help":
     case "-h":
     case undefined:
diff --git a/src/cli/broker.ts b/src/cli/broker.ts
new file mode 100644
index 00000000..f8bad45a
--- /dev/null
+++ b/src/cli/broker.ts
@@ -0,0 +1,108 @@
+/**
+ * `abg broker start` — run the always-on control-plane broker (§11.1).
+ *
+ * Opens the local collab Store (shared with `abg auth login`), authenticates
+ * connections by PSK, and routes Envelopes. Binds a configurable host (default
+ * loopback; for Tailscale pass the 100.x address, never 0.0.0.0 — §7.3).
+ */
+
+import { chmodSync, mkdirSync } from "node:fs";
+import { dirname, join } from "node:path";
+import { Broker, DEFAULT_BROKER_PORT } from "../broker";
+import { SqliteStore } from "../backbone/store/sqlite-store";
+import { StorePskIdentityProvider } from "../backbone/identity/store-psk-identity-provider";
+import { StateDirResolver } from "../state-dir";
+
+function resolveDbPath(explicit?: string): string {
+  if (explicit) return explicit;
+  const env = process.env.AGENTBRIDGE_COLLAB_DB;
+  if (env && env.length > 0) return env;
+  return join(new StateDirResolver().dir, "collab.db");
+}
+
+const LOOPBACK_HOSTS = new Set(["127.0.0.1", "::1", "localhost"]);
+
+/**
+ * True iff `host` is in the Tailscale CGNAT range 100.64.0.0/10 (second octet
+ * 64–127). A bare `/^100\./` would wrongly silence the warning for PUBLIC 100.x
+ * addresses (e.g. 100.0.x / 100.200.x) that are NOT Tailscale.
+ */
+function isTailscaleCgnat(host: string): boolean {
+  const m = /^100\.(\d{1,3})\./.exec(host);
+  if (!m) return false;
+  const octet = Number(m[1]);
+  return octet >= 64 && octet <= 127;
+}
+
+/**
+ * Normalise the bind host and decide whether to warn (§7.3). Empty → loopback (a
+ * malformed/unset `--host` must NOT silently bind all interfaces). Warn on any
+ * non-loopback, non-Tailscale address — a WHITELIST, so `0.0.0.0` / `""` / `::` /
+ * a LAN IP / a public 100.x all surface the exposure warning (the prior
+ * `=== "0.0.0.0"` blacklist missed empty string and `::`).
+ */
+export function resolveBindHost(raw: string): { host: string; warning: string | null } {
+  const host = raw === "" ? "127.0.0.1" : raw;
+  if (LOOPBACK_HOSTS.has(host) || isTailscaleCgnat(host)) return { host, warning: null };
+  return {
+    host,
+    warning:
+      "⚠️ 绑定非 loopback 地址会把 broker 暴露给物理 LAN/WiFi。跨网请绑 Tailscale 的 100.x 地址（§7.3）。",
+  };
+}
+
+export async function runBrokerStart(argv: string[]): Promise<void> {
+  let host = "127.0.0.1";
+  let port = DEFAULT_BROKER_PORT;
+  let db: string | undefined;
+  for (let i = 0; i < argv.length; i++) {
+    const a = argv[i]!;
+    if (a === "--host") host = argv[++i] ?? host;
+    else if (a.startsWith("--host=")) host = a.slice("--host=".length);
+    else if (a === "--port") port = parseInt(argv[++i] ?? "", 10);
+    else if (a.startsWith("--port=")) port = parseInt(a.slice("--port=".length), 10);
+    else if (a === "--db") db = argv[++i];
+    else if (a.startsWith("--db=")) db = a.slice("--db=".length);
+  }
+
+  if (!Number.isInteger(port) || port < 0 || port > 65535) {
+    console.error(`无效的 --port：${port}`);
+    process.exit(1);
+    return;
+  }
+  const bind = resolveBindHost(host);
+  if (bind.warning) console.error(bind.warning);
+
+  const dbPath = resolveDbPath(db);
+  const dir = dirname(dbPath);
+  // Same 0700 lockdown as `abg auth login`: the collab DB holds raw PSK tokens + PII.
+  mkdirSync(dir, { recursive: true, mode: 0o700 });
+  chmodSync(dir, 0o700);
+
+  const store = new SqliteStore(dbPath);
+  const broker = new Broker({
+    store,
+    identityProvider: new StorePskIdentityProvider(store),
+    host: bind.host,
+    port,
+    log: (m) => console.error(`[broker] ${m}`),
+  });
+  const bound = broker.start();
+  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.
+}
+
+export async function runBroker(args: string[]): Promise<void> {
+  const sub = args[0];
+  switch (sub) {
+    case "start":
+      await runBrokerStart(args.slice(1));
+      break;
+    default:
+      console.error(`未知的 broker 子命令：${sub ?? "(空)"}`);
+      console.error("用法：abg broker start [--host <ip>] [--port <n>] [--db <path>]");
+      process.exit(1);
+  }
+}
diff --git a/src/integration-test/broker.test.ts b/src/integration-test/broker.test.ts
new file mode 100644
index 00000000..fbac71c6
--- /dev/null
+++ b/src/integration-test/broker.test.ts
@@ -0,0 +1,160 @@
+import { describe, test, expect } 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";
+
+/** Minimal buffering WS client so no inbound message is lost between awaits. */
+class WsClient {
+  ws!: WebSocket;
+  private queue: any[] = [];
+  private waiters: ((m: any) => void)[] = [];
+  private closed?: { code: number; reason: string };
+  private closeWaiters: ((c: { code: number; reason: string }) => void)[] = [];
+
+  static async connect(url: string): Promise<WsClient> {
+    const c = new WsClient();
+    c.ws = new WebSocket(url);
+    c.ws.onmessage = (ev) => {
+      const m = JSON.parse(ev.data as string);
+      const w = c.waiters.shift();
+      if (w) w(m);
+      else c.queue.push(m);
+    };
+    c.ws.onclose = (ev) => {
+      c.closed = { code: ev.code, reason: ev.reason };
+      for (const w of c.closeWaiters) w(c.closed);
+      c.closeWaiters = [];
+    };
+    await new Promise<void>((res, rej) => {
+      c.ws.onopen = () => res();
+      c.ws.onerror = () => rej(new Error("ws connect failed"));
+    });
+    return c;
+  }
+
+  next(): Promise<any> {
+    const m = this.queue.shift();
+    if (m !== undefined) return Promise.resolve(m);
+    return new Promise((res) => this.waiters.push(res));
+  }
+
+  waitClose(): Promise<{ code: number; reason: string }> {
+    if (this.closed) return Promise.resolve(this.closed);
+    return new Promise((res) => this.closeWaiters.push(res));
+  }
+
+  send(m: unknown) {
+    this.ws.send(JSON.stringify(m));
+  }
+  close() {
+    this.ws.close();
+  }
+}
+
+async function startBroker() {
+  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, // random free port
+    log: () => {},
+  });
+  const { port } = broker.start();
+  return { broker, store, token, url: `ws://127.0.0.1:${port}/ws` };
+}
+
+describe("Broker — WSS + PSK auth + transport fan-out", () => {
+  test("a valid PSK token authenticates; an invalid one is rejected and closed", async () => {
+    const { broker, store, token, url } = await startBroker();
+    try {
+      const ok = await WsClient.connect(url);
+      ok.send({ type: "hello", token });
+      expect(await ok.next()).toEqual({
+        type: "welcome",
+        identity: { id: "alice@x.com", displayName: "Alice" },
+      });
+
+      const bad = await WsClient.connect(url);
+      bad.send({ type: "hello", token: "not-a-real-token" });
+      expect(await bad.next()).toMatchObject({ type: "auth_error" });
+      expect((await bad.waitClose()).code).toBe(4401);
+
+      ok.close();
+    } finally {
+      broker.stop();
+      await store.close();
+    }
+  });
+
+  test("publish fans out the envelope to a subscriber (control-plane only)", async () => {
+    const { broker, store, token, url } = await startBroker();
+    try {
+      const pub = await WsClient.connect(url);
+      pub.send({ type: "hello", token });
+      await pub.next(); // welcome
+
+      const sub = await WsClient.connect(url);
+      sub.send({ type: "hello", token });
+      await sub.next(); // welcome
+      sub.send({ type: "subscribe", topic: "room-1" });
+      expect(await sub.next()).toEqual({ type: "subscribed", topic: "room-1" });
+
+      const envelope = {
+        roomId: "room-1",
+        messageId: "e1",
+        traceId: "t1",
+        idempotencyKey: "k1",
+        from: { agentId: "ag-1", agentType: "claude" },
+        kind: "task_completed",
+        timestamp: 1,
+        deliveryMode: "store_if_offline" as const,
+      };
+      pub.send({ type: "publish", topic: "room-1", envelope });
+
+      const ev = await sub.next();
+      expect(ev).toMatchObject({ type: "event", topic: "room-1", envelope: { messageId: "e1" } });
+
+      pub.close();
+      sub.close();
+    } finally {
+      broker.stop();
+      await store.close();
+    }
+  });
+
+  test("a non-hello message before auth is rejected", async () => {
+    const { broker, store, url } = await startBroker();
+    try {
+      const c = await WsClient.connect(url);
+      c.send({ type: "subscribe", topic: "x" });
+      expect(await c.next()).toMatchObject({ type: "error" });
+      c.close();
+    } finally {
+      broker.stop();
+      await store.close();
+    }
+  });
+
+  test("malformed-but-valid JSON (null / number) is rejected, never crashes the handler", async () => {
+    const { broker, store, url } = await startBroker();
+    try {
+      const c = await WsClient.connect(url);
+      c.ws.send("null"); // valid JSON, not a tagged object → would throw on .type access
+      expect(await c.next()).toMatchObject({ type: "error" });
+      c.ws.send("42");
+      expect(await c.next()).toMatchObject({ type: "error" });
+      // still responsive after malformed input (handler did not die)
+      c.send({ type: "subscribe", topic: "x" });
+      expect(await c.next()).toMatchObject({ type: "error" });
+      c.close();
+    } finally {
+      broker.stop();
+      await store.close();
+    }
+  });
+});
diff --git a/src/unit-test/broker-cli.test.ts b/src/unit-test/broker-cli.test.ts
new file mode 100644
index 00000000..e02f0acd
--- /dev/null
+++ b/src/unit-test/broker-cli.test.ts
@@ -0,0 +1,36 @@
+import { describe, test, expect } from "bun:test";
+import { resolveBindHost } from "../cli/broker";
+
+describe("resolveBindHost — §7.3 bind-host guard (loopback whitelist)", () => {
+  test("empty string normalises to loopback, no warning (no silent all-interfaces bind)", () => {
+    expect(resolveBindHost("")).toEqual({ host: "127.0.0.1", warning: null });
+  });
+
+  test("loopback hosts pass without warning", () => {
+    for (const h of ["127.0.0.1", "::1", "localhost"]) {
+      const r = resolveBindHost(h);
+      expect(r.host).toBe(h);
+      expect(r.warning).toBeNull();
+    }
+  });
+
+  test("Tailscale CGNAT 100.64.0.0/10 passes without warning", () => {
+    for (const h of ["100.64.0.1", "100.100.100.100", "100.127.255.254"]) {
+      expect(resolveBindHost(h).warning).toBeNull();
+    }
+  });
+
+  test("public 100.x OUTSIDE the CGNAT range still warns (not real Tailscale)", () => {
+    for (const h of ["100.0.0.1", "100.63.0.1", "100.128.0.1", "100.200.0.1"]) {
+      expect(resolveBindHost(h).warning).toBeTruthy();
+    }
+  });
+
+  test("exposed addresses warn (0.0.0.0, ::, LAN IPs — the blacklist gap)", () => {
+    for (const h of ["0.0.0.0", "::", "192.168.1.5", "10.0.0.2"]) {
+      const r = resolveBindHost(h);
+      expect(r.host).toBe(h);
+      expect(r.warning).toBeTruthy();
+    }
+  });
+});
diff --git a/src/unit-test/transport-contract.ts b/src/unit-test/transport-contract.ts
index 717e3273..be7561f3 100644
--- a/src/unit-test/transport-contract.ts
+++ b/src/unit-test/transport-contract.ts
@@ -39,6 +39,17 @@ export function runTransportContract(label: string, makeTransport: () => Message
       expect(b).toEqual(["z"]);
     });
 
+    test("a throwing subscriber does not block delivery to its siblings", async () => {
+      const t = makeTransport();
+      const got: string[] = [];
+      t.subscribe("r", () => {
+        throw new Error("bad subscriber");
+      });
+      t.subscribe("r", (m) => got.push(m.messageId));
+      await t.publish("r", makeEnvelope({ messageId: "z" }));
+      expect(got).toEqual(["z"]); // sibling still received despite the thrower
+    });
+
     test("one subscriber's unsubscribe does not affect the other", async () => {
       const t = makeTransport();
       const a: string[] = [];