feat(presence): member_joined / member_left + 预留 host/capabilities/budgetHint (PR8)

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("bdfea8e", "source"),
+  commit: defineString("a5a6005", "source"),
   bundle: defineBundle("plugin"),
   contractVersion: defineNumber(1, CONTRACT_VERSION),
-  codeHash: defineString("ac631ba2670f", "source")
+  codeHash: defineString("5f8ef4c63fe4", "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("bdfea8e", "source"),
+  commit: defineString("a5a6005", "source"),
   bundle: defineBundle("plugin"),
   contractVersion: defineNumber(1, CONTRACT_VERSION),
-  codeHash: defineString("ac631ba2670f", "source")
+  codeHash: defineString("5f8ef4c63fe4", "source")
 });
 function daemonStatusBuildInfo() {
   return { ...BUILD_INFO };

src/broker-client.ts

@@ -1,9 +1,12 @@
 import type { Envelope } from "./backbone/envelope";
 import type { Identity } from "./backbone/identity";
+import type { PresenceMeta } from "./presence";
 
 export interface BrokerClientOptions {
   url: string;
   token: string;
+  /** Reserved presence metadata (host/capabilities/...) declared at hello (§11.1 bullet 9). */
+  presence?: PresenceMeta;
   log?: (msg: string) => void;
   /** Initial reconnect backoff (default 250ms), doubled up to {@link reconnectMaxMs}. */
   reconnectBaseMs?: number;
@@ -136,7 +139,7 @@ export class BrokerClient {
     this.ws = ws;
 
     ws.onopen = () => {
-      this.sendRaw({ type: "hello", token: this.opts.token });
+      this.sendRaw({ type: "hello", token: this.opts.token, presence: this.opts.presence });
     };
     ws.onmessage = (ev) => {
       let msg: any;

src/broker.ts

@@ -4,19 +4,37 @@ 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";
+import { buildPresenceEnvelope, type PresenceMeta } from "./presence";
 
 export const DEFAULT_BROKER_PORT = 4700; // outside the multi-pair 4500/4501/4502+stride range
 const CLOSE_AUTH_FAILED = 4401;
 
+/** Validate the optional reserved presence blob from hello — best-effort, drop anything malformed. Exported for boundary tests. */
+export function sanitizePresence(raw: unknown): PresenceMeta | undefined {
+  if (typeof raw !== "object" || raw === null || Array.isArray(raw)) return undefined;
+  const r = raw as Record<string, unknown>;
+  const out: PresenceMeta = {};
+  if (typeof r.agentType === "string") out.agentType = r.agentType;
+  if (typeof r.host === "string") out.host = r.host;
+  if (Array.isArray(r.capabilities)) {
+    const caps = r.capabilities.filter((c): c is string => typeof c === "string");
+    if (caps.length > 0) out.capabilities = caps;
+  }
+  if (typeof r.budgetHint === "string") out.budgetHint = r.budgetHint;
+  return Object.keys(out).length > 0 ? out : undefined;
+}
+
 interface BrokerSocketData {
   connId: number;
   identity?: Identity;
+  /** Reserved presence metadata declared at hello (§11.1 bullet 9); echoed in member_joined. */
+  presence?: PresenceMeta;
   /** topic → unsubscribe handle for this connection's subscriptions. */
   subs: Map<string, () => void>;
 }
 
 type ClientMessage =
-  | { type: "hello"; token: string }
+  | { type: "hello"; token: string; presence?: unknown }
   | { type: "subscribe"; topic: string }
   | { type: "unsubscribe"; topic: string }
   | { type: "publish"; topic: string; envelope: Envelope };
@@ -94,8 +112,16 @@ 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);
+          const identity = ws.data.identity;
+          if (identity) {
+            // A crash-disconnect still yields member_left here (presence tracks real
+            // connectivity). Fire-and-forget: close() is sync, emit can't be awaited.
+            for (const topic of ws.data.subs.keys()) {
+              if (self.removeTopicMember(topic, identity.id)) {
+                void self.emitPresence(topic, "member_left", identity, ws.data.presence);
+              }
+            }
+          }
           for (const unsub of ws.data.subs.values()) unsub();
           ws.data.subs.clear();
           self.log(`conn #${ws.data.connId} closed`);
@@ -156,6 +182,7 @@ export class Broker {
         return;
       }
       ws.data.identity = identity;
+      ws.data.presence = sanitizePresence(msg.presence); // reserved meta, best-effort
       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
@@ -187,8 +214,16 @@ export class Broker {
           if (this.shouldDeliver(me, envelope)) this.send(ws, { type: "event", topic, envelope });
         });
         ws.data.subs.set(topic, unsub);
-        this.addTopicMember(topic, me);
+        const becamePresent = this.addTopicMember(topic, me);
         this.send(ws, { type: "subscribed", topic });
+        // Presence (§11.1 bullet 9): announce only on the 0→1 transition, so a
+        // second connection for the same identity doesn't re-announce a join.
+        // Emit BEFORE draining this subscriber's own backlog: join notification
+        // doesn't depend on the joiner's pending queue, and keeping it ahead of the
+        // 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);
         // 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.
@@ -200,7 +235,10 @@ export class Broker {
         if (unsub) {
           unsub();
           ws.data.subs.delete(msg.topic);
-          this.removeTopicMember(msg.topic, me);
+          // member_left only on the →0 transition (last connection for this identity left the topic).
+          if (this.removeTopicMember(msg.topic, me)) {
+            await this.emitPresence(msg.topic, "member_left", ws.data.identity, ws.data.presence);
+          }
         }
         return;
       }
@@ -260,22 +298,55 @@ export class Broker {
     return true; // broadcast / @mention (highlight is client-side via mentions[])
   }
 
-  private addTopicMember(topic: string, id: string): void {
+  /** Add a live subscription for `id` on `topic`. Returns true iff this is a 0→1 transition (newly present). */
+  private addTopicMember(topic: string, id: string): boolean {
     let m = this.topicMembers.get(topic);
     if (!m) {
       m = new Map();
       this.topicMembers.set(topic, m);
     }
-    m.set(id, (m.get(id) ?? 0) + 1);
+    const prev = m.get(id) ?? 0;
+    m.set(id, prev + 1);
+    return prev === 0;
   }
 
-  private removeTopicMember(topic: string, id: string): void {
+  /** Drop a live subscription for `id` on `topic`. Returns true iff this is a →0 transition (now absent). */
+  private removeTopicMember(topic: string, id: string): boolean {
     const m = this.topicMembers.get(topic);
-    if (!m) return;
-    const n = (m.get(id) ?? 0) - 1;
+    if (!m) return false;
+    const had = m.get(id) ?? 0;
+    if (had === 0) return false;
+    const n = had - 1;
     if (n <= 0) m.delete(id);
     else m.set(id, n);
     if (m.size === 0) this.topicMembers.delete(topic);
+    return n <= 0;
+  }
+
+  /**
+   * Synthesize a presence event (§11.1 bullet 9) on a membership transition and
+   * fan it out to the topic. Broker-authored (not client-published) so it tracks
+   * ACTUAL connectivity — a crash-disconnect still yields member_left via close().
+   * `online_only` (never stored); shouldDeliver skips the subject themselves.
+   */
+  private async emitPresence(
+    topic: string,
+    kind: "member_joined" | "member_left",
+    identity: Identity,
+    presence?: PresenceMeta,
+  ): Promise<void> {
+    const env = buildPresenceEnvelope({
+      kind,
+      roomId: topic,
+      agentId: identity.id,
+      displayName: identity.displayName,
+      meta: presence,
+    });
+    try {
+      await this.transport.publish(topic, env);
+    } catch (e) {
+      this.log(`presence ${kind} publish failed for ${identity.id}@${topic}: ${String(e)}`);
+    }
   }
 
   private isReachable(topic: string, id: string): boolean {

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

@@ -0,0 +1,110 @@
+import { describe, test, expect, afterEach } from "bun:test";
+import { Broker } from "../broker";
+import { BrokerClient } from "../broker-client";
+import { InMemoryStore } from "../backbone/store/memory-store";
+import { IdentityService } from "../backbone/identity-service";
+import { StorePskIdentityProvider } from "../backbone/identity/store-psk-identity-provider";
+import type { Envelope } from "../backbone/envelope";
+
+const ROOM = "checkout";
+
+async function delay(ms: number): Promise<void> {
+  await new Promise((r) => setTimeout(r, ms));
+}
+async function waitFor(cond: () => boolean, timeoutMs = 2000): Promise<void> {
+  const start = performance.now();
+  while (!cond()) {
+    if (performance.now() - start > timeoutMs) throw new Error("waitFor timed out");
+    await delay(10);
+  }
+}
+
+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 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, tokenA, tokenB, url: `ws://127.0.0.1:${port}/ws` };
+}
+
+/** A subscribed BrokerClient that records every event it receives. */
+async function subscriber(url: string, token: string, presence?: Record<string, unknown>) {
+  const client = new BrokerClient({ url, token, presence: presence as never });
+  const events: Envelope[] = [];
+  client.onEvent((_topic, env) => events.push(env));
+  await client.connect();
+  client.subscribe(ROOM);
+  await delay(60); // let the subscribe register at the broker
+  return { client, events };
+}
+
+describe("Broker presence — member_joined / member_left (§11.1 bullet 9)", () => {
+  let cleanup: Array<() => void> = [];
+  afterEach(() => {
+    for (const fn of cleanup) fn();
+    cleanup = [];
+  });
+
+  test("an existing subscriber sees member_joined (with reserved meta); the joiner does not see its own", async () => {
+    const { broker, tokenA, tokenB, url } = await startBroker();
+    cleanup.push(() => broker.stop());
+
+    const bob = await subscriber(url, tokenB);
+    const alice = await subscriber(url, tokenA, { agentType: "claude", host: "tailnet-1", capabilities: ["review"] });
+    cleanup.push(() => bob.client.close(), () => alice.client.close());
+
+    await waitFor(() => bob.events.some((e) => e.kind === "member_joined" && e.from.agentId === "alice@x.com"));
+    const joined = bob.events.find((e) => e.kind === "member_joined" && e.from.agentId === "alice@x.com")!;
+    expect(joined.deliveryMode).toBe("online_only");
+    expect(joined.from.agentType).toBe("claude");
+    expect(joined.payload).toMatchObject({ displayName: "Alice", host: "tailnet-1", capabilities: ["review"] });
+
+    // self-skip: alice never receives her own join
+    expect(alice.events.some((e) => e.kind === "member_joined" && e.from.agentId === "alice@x.com")).toBe(false);
+  });
+
+  test("member_left fires when a member disconnects", async () => {
+    const { broker, tokenA, tokenB, url } = await startBroker();
+    cleanup.push(() => broker.stop());
+    const bob = await subscriber(url, tokenB);
+    const alice = await subscriber(url, tokenA);
+    cleanup.push(() => bob.client.close());
+
+    await waitFor(() => bob.events.some((e) => e.kind === "member_joined" && e.from.agentId === "alice@x.com"));
+    alice.client.close(); // disconnect
+
+    await waitFor(() => bob.events.some((e) => e.kind === "member_left" && e.from.agentId === "alice@x.com"));
+  });
+
+  test("a second connection for the same identity does not re-announce; member_left only on the last leave", async () => {
+    const { broker, tokenA, tokenB, url } = await startBroker();
+    cleanup.push(() => broker.stop());
+    const bob = await subscriber(url, tokenB);
+    cleanup.push(() => bob.client.close());
+
+    const a1 = await subscriber(url, tokenA);
+    await waitFor(() => bob.events.filter((e) => e.kind === "member_joined" && e.from.agentId === "alice@x.com").length === 1);
+
+    const a2 = await subscriber(url, tokenA); // same identity, second connection
+    await delay(120);
+    // still exactly one member_joined — alice was already present
+    expect(bob.events.filter((e) => e.kind === "member_joined" && e.from.agentId === "alice@x.com").length).toBe(1);
+
+    a1.client.close(); // one of two connections leaves
+    await delay(120);
+    expect(bob.events.some((e) => e.kind === "member_left" && e.from.agentId === "alice@x.com")).toBe(false);
+
+    a2.client.close(); // last connection leaves
+    await waitFor(() => bob.events.some((e) => e.kind === "member_left" && e.from.agentId === "alice@x.com"));
+  });
+});

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

@@ -18,6 +18,9 @@ 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).
+      if (m?.type === "event" && (m.envelope?.kind === "member_joined" || m.envelope?.kind === "member_left")) return;
       const w = c.waiters.shift();
       if (w) w(m);
       else c.q.push(m);

src/presence.ts

@@ -0,0 +1,59 @@
+import { randomUUID } from "node:crypto";
+import type { Envelope } from "./backbone/envelope";
+
+export type PresenceKind = "member_joined" | "member_left";
+
+/**
+ * Reserved presence metadata a client may declare at `hello` (§5.2 / §11.1
+ * bullet 9). `host`/`capabilities` describe the agent; `budgetHint` is reserved
+ * for the budget-aware B-class and is IGNORED by the A-class MVP. All optional —
+ * presence works with none of them.
+ */
+export interface PresenceMeta {
+  agentType?: string;
+  host?: string;
+  capabilities?: string[];
+  /** Reserved (§11.1): budget coordination is B-class; A-class never reads this. */
+  budgetHint?: string;
+}
+
+export interface BuildPresenceInput {
+  kind: PresenceKind;
+  roomId: string;
+  agentId: string;
+  /** Server-authoritative display name (from the resolved identity), for UI only. */
+  displayName?: string;
+  meta?: PresenceMeta;
+  /** Clock injection for tests. */
+  now?: () => number;
+}
+
+/**
+ * Build a presence Envelope (member_joined / member_left, §11.1 bullet 9).
+ *
+ * Broadcast to the room (no `to`) and `online_only` — presence is EPHEMERAL, so
+ * it is never persisted for offline replay (a member who was absent doesn't need
+ * a backlog of stale join/leave churn; they get the live roster on reconnect).
+ * The reserved `host`/`capabilities`/`budgetHint` ride in the payload for the
+ * receiving adapter to render; routing never uses them.
+ */
+export function buildPresenceEnvelope(input: BuildPresenceInput): Envelope {
+  const payload: Record<string, unknown> = {};
+  if (input.displayName) payload.displayName = input.displayName;
+  if (input.meta?.host) payload.host = input.meta.host;
+  if (input.meta?.capabilities && input.meta.capabilities.length > 0) {
+    payload.capabilities = input.meta.capabilities;
+  }
+  if (input.meta?.budgetHint) payload.budgetHint = input.meta.budgetHint;
+  return {
+    roomId: input.roomId,
+    messageId: randomUUID(),
+    traceId: randomUUID(),
+    idempotencyKey: randomUUID(),
+    from: { agentId: input.agentId, agentType: input.meta?.agentType ?? "unknown" },
+    kind: input.kind,
+    payload,
+    timestamp: (input.now ?? Date.now)(),
+    deliveryMode: "online_only",
+  };
+}