From 8284e8a30c6af08542f48dd34866ba76ef65a00a Mon Sep 17 00:00:00 2001 From: "rayson951005@gmail.com" Date: Fri, 26 Jun 2026 11:50:35 +0800 Subject: [PATCH] =?UTF-8?q?feat(presence):=20member=5Fjoined=20/=20member?= =?UTF-8?q?=5Fleft=20+=20=E9=A2=84=E7=95=99=20host/capabilities/budgetHint?= =?UTF-8?q?=20(PR8)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit v3 §11.1 单团队 MVP 的 PR8。broker 在房间成员到场/离场时合成 presence 事件 (§11.1 bullet 9),让成员实时看到谁在房间里。 PR8 of the v3 §11.1 single-team MVP. The broker synthesizes presence events as members arrive/leave a room, so members see who is live. - src/presence.ts: buildPresenceEnvelope — broadcast, online_only (EPHEMERAL, never persisted for offline replay), payload {displayName, host, capabilities, budgetHint}. budgetHint is reserved for the B-class; the A-class MVP ignores it. - src/broker.ts: addTopicMember/removeTopicMember now return the membership transition. member_joined on the 0→1 transition (a 2nd connection for the same identity does NOT re-announce); member_left on →0 (last connection leaving), emitted from BOTH unsubscribe AND close() so a crash-disconnect still leaves. from.agentId is the broker-authenticated identity (unspoofable). hello may carry a reserved presence blob, validated by sanitizePresence (drops malformed input — non-string fields, nested objects, __proto__). - src/broker-client.ts: optional presence meta in the hello frame. member_joined is emitted BEFORE draining the joiner's own backlog so a future truly-async Store can't reorder it after a close()-emitted member_left. Tests: presence (3) + sanitize-presence boundary (4, incl. __proto__ + malformed) + broker-presence integration (3: existing-subscriber sees join with reserved meta + self-skip, member_left on disconnect, multi-connection refcount = join once / leave only on last). broker-routing WsClient filters presence churn (those tests assert routing of published envelopes, not presence). Cross-review: 4 rounds (foreground), fixed 1 latent ordering (join-after-drain) + 1 HIGH test gap (sanitizePresence had zero adversarial-input coverage); converged on 2 consecutive clean rounds. bun run check green: 1786 pass. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --- plugins/agentbridge/server/bridge-server.js | 4 +- plugins/agentbridge/server/daemon.js | 4 +- src/broker-client.ts | 5 +- src/broker.ts | 91 +++++++++++++-- src/integration-test/broker-presence.test.ts | 110 +++++++++++++++++++ src/integration-test/broker-routing.test.ts | 3 + src/presence.ts | 59 ++++++++++ src/unit-test/presence.test.ts | 46 ++++++++ src/unit-test/sanitize-presence.test.ts | 40 +++++++ 9 files changed, 347 insertions(+), 15 deletions(-) create mode 100644 src/integration-test/broker-presence.test.ts create mode 100644 src/presence.ts create mode 100644 src/unit-test/presence.test.ts create mode 100644 src/unit-test/sanitize-presence.test.ts diff --git a/plugins/agentbridge/server/bridge-server.js b/plugins/agentbridge/server/bridge-server.js index ff7e7a0..9e711f2 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("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) diff --git a/plugins/agentbridge/server/daemon.js b/plugins/agentbridge/server/daemon.js index 83f0f08..0848857 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("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 }; diff --git a/src/broker-client.ts b/src/broker-client.ts index 10a7097..2940bcb 100644 --- a/src/broker-client.ts +++ b/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; diff --git a/src/broker.ts b/src/broker.ts index b1797e2..a58dc71 100644 --- a/src/broker.ts +++ b/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; + 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 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 { + 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 { diff --git a/src/integration-test/broker-presence.test.ts b/src/integration-test/broker-presence.test.ts new file mode 100644 index 0000000..2cad7cf --- /dev/null +++ b/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 { + await new Promise((r) => setTimeout(r, ms)); +} +async function waitFor(cond: () => boolean, timeoutMs = 2000): Promise { + 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) { + 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")); + }); +}); diff --git a/src/integration-test/broker-routing.test.ts b/src/integration-test/broker-routing.test.ts index b8fcef0..c9dbb71 100644 --- a/src/integration-test/broker-routing.test.ts +++ b/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); diff --git a/src/presence.ts b/src/presence.ts new file mode 100644 index 0000000..e18b1a3 --- /dev/null +++ b/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 = {}; + 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", + }; +} diff --git a/src/unit-test/presence.test.ts b/src/unit-test/presence.test.ts new file mode 100644 index 0000000..4a81d50 --- /dev/null +++ b/src/unit-test/presence.test.ts @@ -0,0 +1,46 @@ +import { describe, test, expect } from "bun:test"; +import { buildPresenceEnvelope } from "../presence"; + +describe("buildPresenceEnvelope (§11.1 bullet 9)", () => { + test("member_joined: broadcast, online_only, echoes displayName + reserved meta", () => { + const env = buildPresenceEnvelope({ + kind: "member_joined", + roomId: "checkout", + agentId: "alice@x.com", + displayName: "Alice", + meta: { agentType: "claude", host: "tailnet-1", capabilities: ["review", "plan"], budgetHint: "low" }, + now: () => 7, + }); + expect(env.kind).toBe("member_joined"); + expect(env.roomId).toBe("checkout"); + expect(env.deliveryMode).toBe("online_only"); // ephemeral — never stored for offline replay + expect(env.to).toBeUndefined(); // broadcast + expect(env.timestamp).toBe(7); + expect(env.from).toEqual({ agentId: "alice@x.com", agentType: "claude" }); + expect(env.payload).toEqual({ + displayName: "Alice", + host: "tailnet-1", + capabilities: ["review", "plan"], + budgetHint: "low", + }); + }); + + test("member_left with no meta: empty payload, agentType defaults to unknown", () => { + const env = buildPresenceEnvelope({ kind: "member_left", roomId: "r1", agentId: "bob@x.com" }); + expect(env.kind).toBe("member_left"); + expect(env.from).toEqual({ agentId: "bob@x.com", agentType: "unknown" }); + expect(env.payload).toEqual({}); + expect(env.deliveryMode).toBe("online_only"); + }); + + test("omits empty capabilities and absent reserved fields", () => { + const env = buildPresenceEnvelope({ + kind: "member_joined", + roomId: "r1", + agentId: "a", + displayName: "A", + meta: { capabilities: [] }, + }); + expect(env.payload).toEqual({ displayName: "A" }); + }); +}); diff --git a/src/unit-test/sanitize-presence.test.ts b/src/unit-test/sanitize-presence.test.ts new file mode 100644 index 0000000..1696cd4 --- /dev/null +++ b/src/unit-test/sanitize-presence.test.ts @@ -0,0 +1,40 @@ +import { describe, test, expect } from "bun:test"; +import { sanitizePresence } from "../broker"; + +// sanitizePresence is the trust boundary for the untrusted `presence` blob a +// client sends at hello (§7.3). These assert malformed input is dropped, never +// echoed into a member_joined that fans out to the room. +describe("sanitizePresence — hello presence trust boundary", () => { + test("non-objects ⇒ undefined", () => { + for (const bad of [null, undefined, 42, "str", true, []]) { + expect(sanitizePresence(bad)).toBeUndefined(); + } + }); + + test("malformed fields are dropped; an all-malformed blob ⇒ undefined", () => { + expect(sanitizePresence({ agentType: 123 })).toBeUndefined(); // non-string + expect(sanitizePresence({ host: { nested: true } })).toBeUndefined(); // object + expect(sanitizePresence({ capabilities: "not-an-array" })).toBeUndefined(); + expect(sanitizePresence({ capabilities: [42, { x: 1 }] })).toBeUndefined(); // filtered empty ⇒ omitted + expect(sanitizePresence({ budgetHint: ["arr"] })).toBeUndefined(); + }); + + test("valid fields survive; non-string capability entries are filtered out", () => { + expect( + sanitizePresence({ agentType: "claude", host: 5, capabilities: ["a", 1, { x: 1 }, "b"], budgetHint: 7 }), + ).toEqual({ agentType: "claude", capabilities: ["a", "b"] }); // host/budgetHint dropped (non-string) + expect(sanitizePresence({ agentType: "codex", host: "tailnet-1", capabilities: ["review"], budgetHint: "low" })).toEqual({ + agentType: "codex", + host: "tailnet-1", + capabilities: ["review"], + budgetHint: "low", + }); + }); + + test("a __proto__ payload neither pollutes Object.prototype nor leaks into the result", () => { + const raw = JSON.parse('{"__proto__":{"polluted":1},"host":"h"}'); + const out = sanitizePresence(raw); + expect(out).toEqual({ host: "h" }); // only the known string field survives + expect(({} as Record).polluted).toBeUndefined(); // no prototype pollution + }); +});