feat(broker): 常开控制面 broker (WSS+PSK+扇出) + abg broker start (PR3)

.dockerignore

@@ -0,0 +1,10 @@
+node_modules
+.git
+dist
+.agent
+*.log
+.DS_Store
+collab.db
+collab.db-wal
+collab.db-shm
+auth-token

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

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` 语义齐备后再扩。

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})`);

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: {}

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}`);

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);
+}

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)

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 };

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();

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();