feat(room-memory): 房间底账 room_events + 白板机械合并 + 新成员注入 (§4)

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("b202e8f", "source"),
+  commit: defineString("d824fd2", "source"),
   bundle: defineBundle("plugin"),
   contractVersion: defineNumber(1, CONTRACT_VERSION),
-  codeHash: defineString("7c98560c206e", "source")
+  codeHash: defineString("2784fec70261", "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("b202e8f", "source"),
+  commit: defineString("d824fd2", "source"),
   bundle: defineBundle("plugin"),
   contractVersion: defineNumber(1, CONTRACT_VERSION),
-  codeHash: defineString("7c98560c206e", "source")
+  codeHash: defineString("2784fec70261", "source")
 });
 function daemonStatusBuildInfo() {
   return { ...BUILD_INFO };
@@ -6823,6 +6823,7 @@ class BrokerClient {
   subscriptions = new Set;
   outbox = [];
   eventHandlers = [];
+  whiteboardHandlers = [];
   closed = false;
   authFailed = false;
   reconnectAttempt = 0;
@@ -6890,6 +6891,9 @@ class BrokerClient {
   onEvent(handler) {
     this.eventHandlers.push(handler);
   }
+  onWhiteboard(handler) {
+    this.whiteboardHandlers.push(handler);
+  }
   close() {
     this.closed = true;
     this.clearReconnectTimer();
@@ -6947,6 +6951,14 @@ class BrokerClient {
             this.log(`event handler threw: ${String(e)}`);
           }
         }
+      } else if (msg.type === "whiteboard") {
+        for (const h of this.whiteboardHandlers) {
+          try {
+            h(msg.roomId, msg.whiteboard);
+          } catch (e) {
+            this.log(`whiteboard handler threw: ${String(e)}`);
+          }
+        }
       }
     };
     ws.onclose = () => {
@@ -7282,6 +7294,29 @@ function label(env) {
   const dn = env.payload?.displayName;
   return env.from?.name || (typeof dn === "string" ? dn : "") || env.from?.agentId || "\u67D0\u6210\u5458";
 }
+function renderWhiteboard(wb) {
+  if (!wb || typeof wb !== "object")
+    return null;
+  const w = wb;
+  const arr = (x) => Array.isArray(x) ? x : [];
+  const contracts = arr(w.contractsReady);
+  const inProgress = arr(w.inProgress);
+  const blockers = arr(w.blockers);
+  const milestones = arr(w.recentMilestones);
+  if (contracts.length + inProgress.length + blockers.length + milestones.length === 0)
+    return null;
+  const names = (items, key) => items.slice(-3).map((it) => typeof it[key] === "string" ? it[key] : "?").join(key === "summary" ? " / " : ", ");
+  const parts2 = ["\uD83D\uDCCB \u623F\u95F4\u767D\u677F"];
+  if (contracts.length)
+    parts2.push(`\u5DF2\u5C31\u7EEA\u5951\u7EA6 ${contracts.length}\uFF08${names(contracts, "contract")}\uFF09`);
+  if (inProgress.length)
+    parts2.push(`\u8FDB\u884C\u4E2D ${inProgress.length}`);
+  if (blockers.length)
+    parts2.push(`\u963B\u585E ${blockers.length}`);
+  if (milestones.length)
+    parts2.push(`\u6700\u8FD1\uFF1A${names(milestones, "summary")}`);
+  return parts2.join(" \xB7 ");
+}
 function renderRoomEvent(env) {
   const who = label(env);
   switch (env.kind) {
@@ -7344,6 +7379,11 @@ async function startRoomBridge(deps) {
     if (text)
       deps.emit(text);
   });
+  client.onWhiteboard((_roomId, wb) => {
+    const text = renderWhiteboard(wb);
+    if (text)
+      deps.emit(text);
+  });
   client.subscribe(room);
   client.connect().catch((e) => log(`room bridge: connect failed \u2014 ${String(e)}`));
   log(`room bridge: subscribed to room ${room}`);

src/broker-client.ts

@@ -33,6 +33,8 @@ export function reconnectDelay(baseMs: number, maxMs: number, attempt: number, r
   return ceiling / 2 + rand * (ceiling / 2);
 }
 
+type WhiteboardHandler = (roomId: string, whiteboard: unknown) => void;
+
 /**
  * Edge-side client to the control-plane broker (§5 adapter transport + §8.2
  * resilience foundation).
@@ -55,6 +57,7 @@ export class BrokerClient {
   private readonly subscriptions = new Set<string>();
   private readonly outbox: Array<{ topic: string; envelope: Envelope }> = [];
   private readonly eventHandlers: EventHandler[] = [];
+  private readonly whiteboardHandlers: WhiteboardHandler[] = [];
   private closed = false;
   /** Set on auth_error: a bad token must NOT trigger an infinite reconnect loop. */
   private authFailed = false;
@@ -134,6 +137,11 @@ export class BrokerClient {
     this.eventHandlers.push(handler);
   }
 
+  /** Register a handler for the room whiteboard snapshot pushed on join (§4.4). */
+  onWhiteboard(handler: WhiteboardHandler): void {
+    this.whiteboardHandlers.push(handler);
+  }
+
   close(): void {
     this.closed = true;
     this.clearReconnectTimer();
@@ -194,6 +202,15 @@ export class BrokerClient {
             this.log(`event handler threw: ${String(e)}`);
           }
         }
+      } else if (msg.type === "whiteboard") {
+        // §4.4 new-member injection: a room whiteboard snapshot pushed on join.
+        for (const h of this.whiteboardHandlers) {
+          try {
+            h(msg.roomId, msg.whiteboard);
+          } catch (e) {
+            this.log(`whiteboard handler threw: ${String(e)}`);
+          }
+        }
       }
     };
     ws.onclose = () => {

src/broker.ts

@@ -5,6 +5,7 @@ import type { MessageTransport } from "./backbone/transport";
 import type { Envelope } from "./backbone/envelope";
 import { InProcTransport } from "./backbone/transport/inproc-transport";
 import { buildPresenceEnvelope, type PresenceMeta } from "./presence";
+import { mergeWhiteboard } from "./whiteboard";
 
 export const DEFAULT_BROKER_PORT = 4700; // outside the multi-pair 4500/4501/4502+stride range
 const CLOSE_AUTH_FAILED = 4401;
@@ -247,7 +248,19 @@ export class Broker {
         // drain await means a disconnect mid-drain can't reorder it after the
         // close()-emitted member_left (a "left-then-joined" ghost) under a future
         // truly-async Store. Drain is broadcast-irrelevant; ordering vs join is moot.
-        if (becamePresent) await this.emitPresence(topic, "member_joined", ws.data.identity, ws.data.presence);
+        if (becamePresent) {
+          await this.emitPresence(topic, "member_joined", ws.data.identity, ws.data.presence);
+          // New-member injection (§4.4): hand the joiner the room's distilled
+          // whiteboard so it has context immediately, WITHOUT replaying the raw
+          // ledger (that would double-deliver against drainPendingTo). Sent only to
+          // this socket, best-effort.
+          try {
+            const whiteboard = await this.opts.store.getWhiteboard(topic);
+            if (whiteboard) this.send(ws, { type: "whiteboard", roomId: topic, whiteboard });
+          } catch (e) {
+            this.log(`whiteboard inject failed for ${me}@${topic}: ${String(e)}`);
+          }
+        }
         // Drain anything queued during the connected-but-not-yet-subscribed gap
         // (between hello's drain and this subscribe). Safe: drainPending removes,
         // so an already-drained message is never re-delivered.
@@ -307,6 +320,16 @@ export class Broker {
         }
         // Live fan-out; each subscriber's handler applies shouldDeliver (DM / from-skip).
         await this.transport.publish(msg.topic, env);
+        // Room memory (§4): append to the ledger + distil into the whiteboard AFTER
+        // delivery, best-effort — a memory write must never roll back or block the
+        // live fan-out that already happened. Presence is broker-synthesized via
+        // emitPresence (not this path), so it's naturally excluded from the ledger.
+        try {
+          await this.opts.store.appendEvent(env.roomId, env);
+          await this.updateWhiteboard(env);
+        } catch (e) {
+          this.log(`room-memory update failed (${env.idempotencyKey}): ${String(e)}`);
+        }
         return;
       }
       default: {
@@ -377,6 +400,17 @@ export class Broker {
     return (this.topicMembers.get(topic)?.get(id) ?? 0) > 0;
   }
 
+  /**
+   * Distil an event into the room whiteboard (§4.2), zero-LLM. mergeWhiteboard
+   * returns the SAME reference when the kind doesn't touch the board, so an
+   * unmergeable event (a DM, etc.) skips the Store write entirely.
+   */
+  private async updateWhiteboard(env: Envelope): Promise<void> {
+    const prev = await this.opts.store.getWhiteboard(env.roomId);
+    const next = mergeWhiteboard(prev, env);
+    if (next !== prev && next !== null) await this.opts.store.saveWhiteboard(env.roomId, next);
+  }
+
   /** Persist a store_if_offline envelope for intended recipients with no live subscription (§3.2). */
   private async storeForOfflineRecipients(topic: string, env: Envelope, from?: string): Promise<void> {
     const intended = Array.isArray(env.to) ? env.to : await this.opts.store.getMembers(topic);

src/integration-test/broker-room-memory.test.ts

@@ -0,0 +1,133 @@
+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";
+import { buildTaskCompletedEnvelope } from "../task-completed";
+import type { Store } from "../backbone/store";
+
+const ROOM = "checkout";
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+/** Raw buffering WS client so we can observe non-event frames (e.g. `whiteboard`). */
+class WsClient {
+  ws!: WebSocket;
+  frames: any[] = [];
+  static async connect(url: string): Promise<WsClient> {
+    const c = new WsClient();
+    c.ws = new WebSocket(url);
+    c.ws.onmessage = (ev) => c.frames.push(JSON.parse(ev.data as string));
+    await new Promise<void>((res, rej) => {
+      c.ws.onopen = () => res();
+      c.ws.onerror = () => rej(new Error("ws connect failed"));
+    });
+    return c;
+  }
+  send(m: unknown) {
+    this.ws.send(JSON.stringify(m));
+  }
+  close() {
+    this.ws.close();
+  }
+  async helloAndSubscribe(token: string) {
+    this.send({ type: "hello", token });
+    await sleep(30);
+    this.send({ type: "subscribe", topic: ROOM });
+    await sleep(40);
+  }
+}
+
+async function startBroker(store: Store = new InMemoryStore()) {
+  const svc = new IdentityService(store);
+  await svc.registerIdentity("alice@x.com", "Alice");
+  await svc.registerIdentity("bob@x.com", "Bob");
+  const tokenA = await svc.issueToken("alice@x.com");
+  const tokenB = await svc.issueToken("bob@x.com");
+  const broker = new Broker({ store, identityProvider: new StorePskIdentityProvider(store), host: "127.0.0.1", port: 0, log: () => {} });
+  const { port } = broker.start();
+  return { broker, store, tokenA, tokenB, url: `ws://127.0.0.1:${port}/ws` };
+}
+
+function taskEnv(summary: string, contract?: string) {
+  return buildTaskCompletedEnvelope({ roomId: ROOM, from: { agentId: "bob@x.com", agentType: "codex" }, summary, contract });
+}
+
+describe("Broker room memory — ledger + whiteboard (§4)", () => {
+  let cleanup: Array<() => void> = [];
+  afterEach(() => {
+    for (const fn of cleanup) fn();
+    cleanup = [];
+  });
+
+  test("a published task_completed is appended to the ledger; broker-synthesized presence is NOT", async () => {
+    const { broker, store, tokenA, tokenB, url } = await startBroker();
+    cleanup.push(() => broker.stop());
+    const alice = await WsClient.connect(url); // subscribing alice triggers a member_joined (presence)
+    cleanup.push(() => alice.close());
+    await alice.helloAndSubscribe(tokenA);
+
+    const bob = await WsClient.connect(url);
+    cleanup.push(() => bob.close());
+    bob.send({ type: "hello", token: tokenB });
+    await sleep(30);
+    bob.send({ type: "publish", topic: ROOM, envelope: taskEnv("auth done", "auth/v1") });
+    await sleep(60);
+
+    const events = await store.getRecentEvents(ROOM, 10);
+    expect(events).toHaveLength(1); // ONLY the task_completed — presence never hits the ledger
+    expect(events[0]!.kind).toBe("task_completed");
+    const wb = await store.getWhiteboard(ROOM);
+    expect(wb!.recentMilestones).toHaveLength(1);
+    expect(wb!.contractsReady[0]).toMatchObject({ contract: "auth/v1" });
+  });
+
+  test("a new member is handed the whiteboard on join (only on the 0→1 transition)", async () => {
+    const { broker, store, tokenA, tokenB, url } = await startBroker();
+    cleanup.push(() => broker.stop());
+    // bob publishes first so the whiteboard has content.
+    const bob = await WsClient.connect(url);
+    cleanup.push(() => bob.close());
+    bob.send({ type: "hello", token: tokenB });
+    await sleep(30);
+    bob.send({ type: "publish", topic: ROOM, envelope: taskEnv("checkout shipped", "checkout/v1") });
+    await sleep(50);
+    expect((await store.getWhiteboard(ROOM))!.contractsReady).toHaveLength(1);
+
+    // alice joins → receives a whiteboard frame.
+    const alice = await WsClient.connect(url);
+    cleanup.push(() => alice.close());
+    await alice.helloAndSubscribe(tokenA);
+    const wbFrame = alice.frames.find((f) => f.type === "whiteboard");
+    expect(wbFrame).toBeDefined();
+    expect(wbFrame.roomId).toBe(ROOM);
+    expect(wbFrame.whiteboard.contractsReady[0]).toMatchObject({ contract: "checkout/v1" });
+
+    // a second subscribe on the SAME connection must not re-inject (no 0→1).
+    const before = alice.frames.filter((f) => f.type === "whiteboard").length;
+    alice.send({ type: "subscribe", topic: ROOM });
+    await sleep(40);
+    expect(alice.frames.filter((f) => f.type === "whiteboard").length).toBe(before);
+  });
+
+  test("a ledger-write failure does NOT block live delivery (best-effort)", async () => {
+    const store = new InMemoryStore();
+    (store as unknown as { appendEvent: () => Promise<void> }).appendEvent = () => Promise.reject(new Error("ledger down"));
+    const { broker, tokenA, tokenB, url } = await startBroker(store);
+    cleanup.push(() => broker.stop());
+    const alice = await WsClient.connect(url);
+    cleanup.push(() => alice.close());
+    await alice.helloAndSubscribe(tokenA);
+
+    const bob = await WsClient.connect(url);
+    cleanup.push(() => bob.close());
+    bob.send({ type: "hello", token: tokenB });
+    await sleep(30);
+    bob.send({ type: "publish", topic: ROOM, envelope: taskEnv("still delivered") });
+    await sleep(60);
+
+    // alice still received the event despite appendEvent throwing.
+    const evt = alice.frames.find((f) => f.type === "event" && f.envelope?.kind === "task_completed");
+    expect(evt).toBeDefined();
+    expect(evt.envelope.payload.summary).toBe("still delivered");
+  });
+});

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

@@ -18,9 +18,11 @@ class WsClient {
     c.ws = new WebSocket(url);
     c.ws.onmessage = (ev) => {
       const m = JSON.parse(ev.data as string);
-      // Ignore presence churn (§11.1 bullet 9): these tests assert on the routing
-      // of PUBLISHED envelopes, not member_joined/left (covered by broker-presence).
+      // Ignore presence churn (§11.1 bullet 9) + whiteboard injection (§4.4):
+      // these tests assert on the routing of PUBLISHED envelopes, not member_joined/
+      // left (broker-presence) or the on-join whiteboard snapshot (broker-room-memory).
       if (m?.type === "event" && (m.envelope?.kind === "member_joined" || m.envelope?.kind === "member_left")) return;
+      if (m?.type === "whiteboard") return;
       const w = c.waiters.shift();
       if (w) w(m);
       else c.q.push(m);

src/integration-test/room-bridge.test.ts

@@ -60,6 +60,29 @@ describe("startRoomBridge — last-mile broker→session injection (§11.1)", ()
     expect(emitted).toEqual([]);
   });
 
+  test("on join, the room whiteboard is rendered and injected (§4.4 new-member injection)", async () => {
+    const { dir, store, tokenB, broker, url, dbPath } = await setup();
+    cleanup.push(() => broker.stop(), () => rmSync(dir, { recursive: true, force: true }));
+
+    // bob publishes a task_completed first so the room has a whiteboard.
+    const bob = new BrokerClient({ url, token: tokenB });
+    await bob.connect();
+    cleanup.push(() => bob.close());
+    bob.publish(
+      ROOM,
+      buildTaskCompletedEnvelope({ roomId: ROOM, from: { agentId: "bob@x.com", agentType: "codex" }, summary: "auth done", contract: "auth/v1" }),
+    );
+    await delay(120); // let the broker append + distil the whiteboard
+
+    // alice's room bridge starts → subscribes → broker pushes the whiteboard snapshot.
+    const emitted: string[] = [];
+    const handle = await startRoomBridge({ cwd: dir, emit: (t) => emitted.push(t), store, dbPath, brokerUrl: url });
+    cleanup.push(() => handle.stop());
+    await waitFor(() => emitted.some((t) => t.includes("📋 房间白板")));
+    const wbLine = emitted.find((t) => t.includes("📋 房间白板"))!;
+    expect(wbLine).toContain("auth/v1");
+  });
+
   test("inert when cwd is not mapped to a room", async () => {
     const { dir, store, broker, url, dbPath } = await setup({ mapCwd: false });
     cleanup.push(() => broker.stop(), () => rmSync(dir, { recursive: true, force: true }));