feat(routing): Envelope 三档路由 + traceId/hop 防环 + 离线补投 (PR6)

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("3b71186", "source"),
+  commit: defineString("cdb0dcc", "source"),
   bundle: defineBundle("plugin"),
   contractVersion: defineNumber(1, CONTRACT_VERSION),
-  codeHash: defineString("d4792cf81ed0", "source")
+  codeHash: defineString("3908e9b66df5", "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("3b71186", "source"),
+  commit: defineString("cdb0dcc", "source"),
   bundle: defineBundle("plugin"),
   contractVersion: defineNumber(1, CONTRACT_VERSION),
-  codeHash: defineString("d4792cf81ed0", "source")
+  codeHash: defineString("3908e9b66df5", "source")
 });
 function daemonStatusBuildInfo() {
   return { ...BUILD_INFO };

src/broker.ts

@@ -43,12 +43,19 @@ export interface BrokerOptions {
  * 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.
+ * Routing (§3.2): three-tier — broadcast (no `to` → all room subscribers),
+ * @mention (delivered to all, highlight is client-side via `mentions[]`), and DM
+ * (`to: [agentId]` → only those identities). Loop prevention: never echo to the
+ * sender (`from.agentId`) + a `hop<=0` drop (the multi-hop generalisation of the
+ * v1 binary-source guard). Offline replay: a `store_if_offline` envelope is
+ * persisted (Store.pending) for any intended recipient with no live subscription,
+ * and drained to them on their next (re)connect.
  */
 export class Broker {
   private server: ReturnType<typeof Bun.serve> | null = null;
   private nextConnId = 0;
+  /** topic → (identityId → live-subscription count) — who is reachable per topic. */
+  private readonly topicMembers = new Map<string, Map<string, number>>();
   private readonly transport: MessageTransport;
   private readonly log: (msg: string) => void;
 
@@ -87,6 +94,8 @@ export class Broker {
           });
         },
         close(ws) {
+          const me = ws.data.identity?.id;
+          if (me) for (const topic of ws.data.subs.keys()) self.removeTopicMember(topic, me);
           for (const unsub of ws.data.subs.values()) unsub();
           ws.data.subs.clear();
           self.log(`conn #${ws.data.connId} closed`);
@@ -129,15 +138,33 @@ export class Broker {
     const msg = parsed as ClientMessage;
 
     if (msg.type === "hello") {
+      // Reject a second hello on an already-authenticated socket: re-binding the
+      // identity would strand the old identity's subscriptions + topicMembers
+      // counts (close removes under the NEW identity), leaving the old one
+      // permanently "reachable" and silently dropping its store_if_offline.
+      if (ws.data.identity) {
+        this.send(ws, { type: "error", reason: "already authenticated" });
+        return;
+      }
+      let identity: Identity;
       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}`);
+        identity = await this.opts.identityProvider.authenticate(msg.token);
       } 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;
+      }
+      ws.data.identity = identity;
+      this.send(ws, { type: "welcome", identity });
+      this.log(`conn #${ws.data.connId} authenticated as ${identity.id}`);
+      // Reconnect replay (§3.2) — OUTSIDE the auth try/catch: a transient store
+      // error during drain must NOT be misreported as an auth failure (which
+      // would close a connection that was already welcomed + resolved client-side).
+      try {
+        await this.drainPendingTo(ws, identity.id);
+      } catch (e) {
+        this.log(`pending drain failed for ${identity.id}: ${String(e)}`);
       }
       return;
     }
@@ -147,35 +174,130 @@ export class Broker {
       this.send(ws, { type: "error", reason: "not authenticated (send hello first)" });
       return;
     }
+    const me = ws.data.identity.id;
 
     switch (msg.type) {
       case "subscribe": {
-        if (ws.data.subs.has(msg.topic)) return; // idempotent
         const topic = msg.topic;
+        if (ws.data.subs.has(topic)) {
+          this.send(ws, { type: "subscribed", topic }); // re-ack so a re-subscribe never hangs
+          return;
+        }
         const unsub = this.transport.subscribe(topic, (envelope) => {
-          this.send(ws, { type: "event", topic, envelope });
+          if (this.shouldDeliver(me, 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
+        this.addTopicMember(topic, me);
+        this.send(ws, { type: "subscribed", topic });
+        // 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.
+        await this.drainPendingTo(ws, me);
         return;
       }
       case "unsubscribe": {
         const unsub = ws.data.subs.get(msg.topic);
         if (unsub) {
           unsub();
           ws.data.subs.delete(msg.topic);
+          this.removeTopicMember(msg.topic, me);
         }
         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);
+        // no repo access — code sync is git's job (§2.6). The broker is a trust
+        // boundary (§7.3), so validate the envelope SHAPE at runtime — TS types
+        // are compile-time only; a client can send anything over the wire.
+        const env = msg.envelope as Envelope | null | undefined;
+        if (!env || typeof env !== "object" || Array.isArray(env)) {
+          this.send(ws, { type: "error", reason: "malformed envelope" });
+          return;
+        }
+        if (!env.from || typeof env.from !== "object" || Array.isArray(env.from)) {
+          this.send(ws, { type: "error", reason: "envelope.from must be an object" });
+          return;
+        }
+        if (env.to !== undefined && !Array.isArray(env.to)) {
+          // A string `to` would make `to.includes(me)` a SUBSTRING match (DM leak).
+          this.send(ws, { type: "error", reason: "envelope.to must be an array" });
+          return;
+        }
+        // Load-bearing for offline replay: a missing idempotencyKey hits the
+        // NOT NULL + OR IGNORE in SqliteStore → silently dropped (≠ InMemoryStore,
+        // a §6.4 divergence); a missing roomId drains as `topic: undefined`.
+        if (typeof env.idempotencyKey !== "string" || env.idempotencyKey === "") {
+          this.send(ws, { type: "error", reason: "envelope.idempotencyKey must be a non-empty string" });
+          return;
+        }
+        if (typeof env.roomId !== "string" || env.roomId === "") {
+          this.send(ws, { type: "error", reason: "envelope.roomId must be a non-empty string" });
+          return;
+        }
+        // Anti-spoof + reliable loop prevention: stamp the authenticated sender
+        // unconditionally (from is now guaranteed a plain object).
+        env.from.agentId = me;
+        // hop<=0 → drop (multi-hop loop guard, §3.2).
+        if (typeof env.hop === "number" && env.hop <= 0) return;
+        // Offline replay: persist for intended recipients with no live subscription.
+        if (env.deliveryMode === "store_if_offline") {
+          await this.storeForOfflineRecipients(msg.topic, env, me);
+        }
+        // Live fan-out; each subscriber's handler applies shouldDeliver (DM / from-skip).
+        await this.transport.publish(msg.topic, env);
         return;
       }
       default: {
         this.send(ws, { type: "error", reason: "unknown message type" });
       }
     }
   }
+
+  /** Should `me` receive `env`? Loop prevention (skip sender) + DM filter. */
+  private shouldDeliver(me: string, env: Envelope): boolean {
+    if (env.from?.agentId === me) return false; // never echo to the sender
+    if (Array.isArray(env.to)) return env.to.includes(me); // DM: present `to` ⇒ only named targets ([] ⇒ nobody, never a broadcast)
+    return true; // broadcast / @mention (highlight is client-side via mentions[])
+  }
+
+  private addTopicMember(topic: string, id: string): void {
+    let m = this.topicMembers.get(topic);
+    if (!m) {
+      m = new Map();
+      this.topicMembers.set(topic, m);
+    }
+    m.set(id, (m.get(id) ?? 0) + 1);
+  }
+
+  private removeTopicMember(topic: string, id: string): void {
+    const m = this.topicMembers.get(topic);
+    if (!m) return;
+    const n = (m.get(id) ?? 0) - 1;
+    if (n <= 0) m.delete(id);
+    else m.set(id, n);
+    if (m.size === 0) this.topicMembers.delete(topic);
+  }
+
+  private isReachable(topic: string, id: string): boolean {
+    return (this.topicMembers.get(topic)?.get(id) ?? 0) > 0;
+  }
+
+  /** 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);
+    for (const id of intended) {
+      if (id === from) continue; // never store for the sender
+      if (!this.isReachable(topic, id)) {
+        await this.opts.store.enqueuePending(id, env);
+      }
+    }
+  }
+
+  /** Reconnect replay (§3.2): drain + deliver everything queued for this identity. */
+  private async drainPendingTo(ws: ServerWebSocket<BrokerSocketData>, id: string): Promise<void> {
+    const pending = await this.opts.store.drainPending(id);
+    for (const env of pending) {
+      this.send(ws, { type: "event", topic: env.roomId, envelope: env });
+    }
+  }
 }

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

@@ -12,7 +12,9 @@ async function startBroker() {
   const store = new InMemoryStore();
   const svc = new IdentityService(store);
   await svc.registerIdentity("alice@x.com", "Alice");
+  await svc.registerIdentity("bob@x.com", "Bob");
   const token = await svc.issueToken("alice@x.com");
+  const tokenB = await svc.issueToken("bob@x.com");
   const broker = new Broker({
     store,
     identityProvider: new StorePskIdentityProvider(store),
@@ -21,11 +23,11 @@ async function startBroker() {
     log: () => {},
   });
   const { port } = broker.start();
-  return { broker, store, token, url: `ws://127.0.0.1:${port}/ws` };
+  return { broker, store, token, tokenB, url: `ws://127.0.0.1:${port}/ws` };
 }
 
 describe("BrokerClient ↔ real Broker", () => {
-  test("connects, authenticates, and round-trips an event", async () => {
+  test("connects + authenticates; does NOT echo a client's own broadcast (loop prevention)", async () => {
     const { broker, store, token, url } = await startBroker();
     const c = new BrokerClient({ url, token, log: () => {} });
     try {
@@ -36,18 +38,18 @@ describe("BrokerClient ↔ real Broker", () => {
       await sleep(60); // let subscribe + ack land
       c.publish("room-1", makeEnvelope({ messageId: "e1" }));
       await sleep(60);
-      expect(got).toEqual(["e1"]);
+      expect(got).toEqual([]); // the sender never receives its own broadcast
     } finally {
       c.close();
       broker.stop();
       await store.close();
     }
   });
 
-  test("fans out across clients: A publishes, B receives", async () => {
-    const { broker, store, token, url } = await startBroker();
-    const a = new BrokerClient({ url, token, log: () => {} });
-    const b = new BrokerClient({ url, token, log: () => {} });
+  test("fans out across clients: A publishes, B (different identity) receives", async () => {
+    const { broker, store, token, tokenB, url } = await startBroker();
+    const a = new BrokerClient({ url, token, log: () => {} }); // alice
+    const b = new BrokerClient({ url, token: tokenB, log: () => {} }); // bob
     try {
       await a.connect();
       await b.connect();