fix: enforce Buzz room membership

extensions/buzz/src/buzz-bus.lifecycle.test.ts

@@ -1,4 +1,4 @@
-import { finalizeEvent, type Event } from "nostr-tools";
+import { finalizeEvent, getPublicKey, type Event, type Filter } from "nostr-tools";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
 
@@ -8,8 +8,15 @@ const relayMocks = vi.hoisted(() => ({
   publish: vi.fn<(event: Event) => Promise<string>>(),
   subscriptionClose: vi.fn(),
   close: vi.fn(),
-  onevent: undefined as ((event: Event) => void) | undefined,
-  closeHandler: undefined as ((reason: string) => void) | undefined,
+  membershipEvents: [] as Event[],
+  subscriptions: [] as Array<{
+    filter: Filter;
+    handlers: {
+      onevent: (event: Event) => void;
+      oneose?: () => void;
+      onclose: (reason: string) => void;
+    };
+  }>,
 }));
 
 vi.mock("nostr-tools", async (importOriginal) => {
@@ -24,11 +31,23 @@ vi.mock("nostr-tools", async (importOriginal) => {
       close = relayMocks.close;
 
       subscribe(
-        _filters: unknown,
-        handlers: { onevent: (event: Event) => void; onclose: (reason: string) => void },
+        filters: Filter[],
+        handlers: {
+          onevent: (event: Event) => void;
+          oneose?: () => void;
+          onclose: (reason: string) => void;
+        },
       ) {
-        relayMocks.onevent = handlers.onevent;
-        relayMocks.closeHandler = handlers.onclose;
+        const filter = filters[0] ?? {};
+        relayMocks.subscriptions.push({ filter, handlers });
+        if (filter.kinds?.includes(39002)) {
+          for (const event of relayMocks.membershipEvents) {
+            handlers.onevent(event);
+          }
+          handlers.oneose?.();
+        } else if (filter.kinds?.includes(40099)) {
+          handlers.oneose?.();
+        }
         return { close: relayMocks.subscriptionClose };
       }
     },
@@ -40,6 +59,9 @@ import { sendBuzzTextOneShot, startBuzzBus } from "./buzz-bus.js";
 const PRIVATE_KEY = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
 const SENDER_PRIVATE_KEY = "0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20";
 const ACCOUNT_ID = "default";
+const CHANNEL_ID = "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c";
+const BOT_PUBLIC_KEY = getPublicKey(Uint8Array.from(Buffer.from(PRIVATE_KEY, "hex")));
+const SENDER_PUBLIC_KEY = getPublicKey(Uint8Array.from(Buffer.from(SENDER_PRIVATE_KEY, "hex")));
 const tempDirs = useAutoCleanupTempDirTracker(afterEach);
 let previousStateDir: string | undefined;
 let stateDir: string;
@@ -50,8 +72,22 @@ describe("Buzz bus lifecycle", () => {
     stateDir = tempDirs.make("openclaw-buzz-dedupe-");
     process.env.OPENCLAW_STATE_DIR = stateDir;
     vi.clearAllMocks();
-    relayMocks.onevent = undefined;
-    relayMocks.closeHandler = undefined;
+    relayMocks.subscriptions.length = 0;
+    relayMocks.membershipEvents = [
+      {
+        id: "membership-1",
+        kind: 39002,
+        pubkey: "f".repeat(64),
+        created_at: 1_700_000_000,
+        content: "",
+        sig: "e".repeat(128),
+        tags: [
+          ["d", CHANNEL_ID],
+          ["p", BOT_PUBLIC_KEY, "", "bot"],
+          ["p", SENDER_PUBLIC_KEY, "", "member"],
+        ],
+      },
+    ];
     relayMocks.connect.mockResolvedValue();
     relayMocks.auth.mockRejectedValue(new Error("auth rejected"));
     relayMocks.publish.mockResolvedValue("");
@@ -71,7 +107,7 @@ describe("Buzz bus lifecycle", () => {
         accountId: ACCOUNT_ID,
         relayUrl: "wss://buzz.example.com",
         privateKey: PRIVATE_KEY,
-        channelIds: ["7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"],
+        channelIds: [CHANNEL_ID],
         onMessage: async () => {},
       }),
     ).rejects.toThrow("auth rejected");
@@ -86,7 +122,7 @@ describe("Buzz bus lifecycle", () => {
     const messageId = await sendBuzzTextOneShot({
       relayUrl: "wss://buzz.example.com",
       privateKey: PRIVATE_KEY,
-      channelId: "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c",
+      channelId: CHANNEL_ID,
       text: "hello",
       threadId: "root-id",
       replyToId: "parent-id",
@@ -98,7 +134,7 @@ describe("Buzz bus lifecycle", () => {
       kind: 9,
       content: "hello",
       tags: [
-        ["h", "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"],
+        ["h", CHANNEL_ID],
         ["e", "root-id", "", "root"],
         ["e", "parent-id", "", "reply"],
       ],
@@ -114,7 +150,7 @@ describe("Buzz bus lifecycle", () => {
       sendBuzzTextOneShot({
         relayUrl: "wss://buzz.example.com",
         privateKey: PRIVATE_KEY,
-        channelId: "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c",
+        channelId: CHANNEL_ID,
         text: "hello",
       }),
     ).rejects.toThrow("rejected");
@@ -129,7 +165,7 @@ describe("Buzz bus lifecycle", () => {
       accountId: ACCOUNT_ID,
       relayUrl: "wss://buzz.example.com",
       privateKey: PRIVATE_KEY,
-      channelIds: ["7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"],
+      channelIds: [CHANNEL_ID],
       onMessage,
     });
     const event = finalizeEvent(
@@ -137,13 +173,16 @@ describe("Buzz bus lifecycle", () => {
         kind: 9,
         created_at: 1_700_000_000,
         content: "hello",
-        tags: [["h", "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"]],
+        tags: [["h", CHANNEL_ID]],
       },
       Uint8Array.from(Buffer.from(SENDER_PRIVATE_KEY, "hex")),
     );
 
-    relayMocks.onevent?.(event);
-    relayMocks.onevent?.(event);
+    const messageSubscription = relayMocks.subscriptions.find((entry) =>
+      entry.filter.kinds?.includes(9),
+    );
+    messageSubscription?.handlers.onevent(event);
+    messageSubscription?.handlers.onevent(event);
 
     await vi.waitFor(() => expect(onMessage).toHaveBeenCalledOnce());
     await bus.close();
@@ -157,7 +196,7 @@ describe("Buzz bus lifecycle", () => {
       accountId: ACCOUNT_ID,
       relayUrl: "wss://buzz.example.com",
       privateKey: PRIVATE_KEY,
-      channelIds: ["7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"],
+      channelIds: [CHANNEL_ID],
       onMessage: async () => {
         throw new Error("dispatch failed");
       },
@@ -169,12 +208,14 @@ describe("Buzz bus lifecycle", () => {
         kind: 9,
         created_at: 1_700_000_000,
         content: "hello",
-        tags: [["h", "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"]],
+        tags: [["h", CHANNEL_ID]],
       },
       Uint8Array.from(Buffer.from(SENDER_PRIVATE_KEY, "hex")),
     );
 
-    relayMocks.onevent?.(event);
+    relayMocks.subscriptions
+      .find((entry) => entry.filter.kinds?.includes(9))
+      ?.handlers.onevent(event);
 
     await vi.waitFor(() => expect(onMessageError).toHaveBeenCalledWith(expect.any(Error)));
     expect(onFatalError).not.toHaveBeenCalled();
@@ -188,7 +229,7 @@ describe("Buzz bus lifecycle", () => {
         kind: 9,
         created_at: Math.floor(Date.now() / 1000),
         content: "hello",
-        tags: [["h", "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"]],
+        tags: [["h", CHANNEL_ID]],
       },
       Uint8Array.from(Buffer.from(SENDER_PRIVATE_KEY, "hex")),
     );
@@ -197,10 +238,12 @@ describe("Buzz bus lifecycle", () => {
       accountId: ACCOUNT_ID,
       relayUrl: "wss://buzz.example.com",
       privateKey: PRIVATE_KEY,
-      channelIds: ["7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"],
+      channelIds: [CHANNEL_ID],
       onMessage: firstOnMessage,
     });
-    relayMocks.onevent?.(event);
+    relayMocks.subscriptions
+      .find((entry) => entry.filter.kinds?.includes(9))
+      ?.handlers.onevent(event);
     await vi.waitFor(() => expect(firstOnMessage).toHaveBeenCalledOnce());
     await firstBus.close();
 
@@ -209,10 +252,12 @@ describe("Buzz bus lifecycle", () => {
       accountId: ACCOUNT_ID,
       relayUrl: "wss://buzz.example.com",
       privateKey: PRIVATE_KEY,
-      channelIds: ["7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"],
+      channelIds: [CHANNEL_ID],
       onMessage: secondOnMessage,
     });
-    relayMocks.onevent?.(event);
+    relayMocks.subscriptions
+      .findLast((entry) => entry.filter.kinds?.includes(9))
+      ?.handlers.onevent(event);
     await new Promise<void>((resolve) => {
       setTimeout(resolve, 100);
     });

extensions/buzz/src/buzz-bus.ts

@@ -7,6 +7,14 @@ import {
 } from "./message-event.js";
 import { syncBuzzProfile } from "./profile.js";
 import { authenticateBuzzRelay, createBuzzAuthSigner, parseBuzzAuthTag } from "./relay-auth.js";
+import {
+  BUZZ_ROOM_MEMBERSHIP_KIND,
+  BUZZ_ROOM_SYSTEM_KIND,
+  isNewerBuzzRoomMembership,
+  parseBuzzRoomMembershipChangeEvent,
+  parseBuzzRoomMembershipEvent,
+  type BuzzRoomMembership,
+} from "./room-membership.js";
 import { decodeBuzzPrivateKey, resolveBuzzPublicKey } from "./types.js";
 
 const MESSAGE_KIND = 9;
@@ -16,6 +24,9 @@ const REPLAY_TTL_MS = 30 * 24 * 60 * 60 * 1000;
 const REPLAY_MAX_ENTRIES = 10_000;
 const REPLAY_STATE_MAX_ENTRIES = 50_000;
 const REPLAY_NAMESPACE_PREFIX = "buzz.inbound-dedupe";
+const MEMBERSHIP_READY_TIMEOUT_MS = 10_000;
+const MEMBERSHIP_REFRESH_DELAYS_MS = [100, 500, 1_500, 3_000] as const;
+const MEMBERSHIP_EVENT_CACHE_MAX_ENTRIES = 10_000;
 
 export interface BuzzBus {
   publicKey: string;
@@ -123,6 +134,381 @@ async function connectAuthenticatedBuzzRelay(params: {
   }
 }
 
+async function sleepWithSignal(delayMs: number, signal?: AbortSignal): Promise<void> {
+  signal?.throwIfAborted();
+  await new Promise<void>((resolve, reject) => {
+    let settled = false;
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    const finish = (error?: unknown) => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      if (timer) {
+        clearTimeout(timer);
+      }
+      signal?.removeEventListener("abort", onAbort);
+      if (error === undefined) {
+        resolve();
+      } else {
+        reject(error);
+      }
+    };
+    const onAbort = () =>
+      finish(signal?.reason ?? new Error("Buzz room membership refresh aborted"));
+    timer = setTimeout(() => finish(), delayMs);
+    signal?.addEventListener("abort", onAbort, { once: true });
+    if (signal?.aborted) {
+      onAbort();
+    }
+  });
+}
+
+async function queryBuzzRoomMemberships(params: {
+  relay: Relay;
+  channelIds: string[];
+  timeoutMs?: number;
+  signal?: AbortSignal;
+}): Promise<Map<string, BuzzRoomMembership>> {
+  const configuredRooms = new Set(params.channelIds);
+  const memberships = new Map<string, BuzzRoomMembership>();
+  return await new Promise<Map<string, BuzzRoomMembership>>((resolve, reject) => {
+    let settled = false;
+    let subscription: ReturnType<Relay["subscribe"]> | undefined;
+    let timeout: ReturnType<typeof setTimeout> | undefined;
+    const finish = (error?: unknown) => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      if (timeout) {
+        clearTimeout(timeout);
+      }
+      params.signal?.removeEventListener("abort", onAbort);
+      subscription?.close("membership snapshot loaded");
+      if (error === undefined) {
+        resolve(memberships);
+      } else {
+        reject(
+          error instanceof Error
+            ? error
+            : new Error("Buzz room membership query failed", { cause: error }),
+        );
+      }
+    };
+    const onAbort = () =>
+      finish(params.signal?.reason ?? new Error("Buzz room membership query aborted"));
+    timeout = setTimeout(
+      () => finish(new Error("Timed out loading Buzz room membership")),
+      params.timeoutMs ?? MEMBERSHIP_READY_TIMEOUT_MS,
+    );
+    params.signal?.addEventListener("abort", onAbort, { once: true });
+    subscription = params.relay.subscribe(
+      [
+        {
+          kinds: [BUZZ_ROOM_MEMBERSHIP_KIND],
+          "#d": params.channelIds,
+          limit: params.channelIds.length,
+        },
+      ],
+      {
+        onevent: (event) => {
+          const membership = parseBuzzRoomMembershipEvent(event);
+          if (
+            !membership ||
+            !configuredRooms.has(membership.roomId) ||
+            !isNewerBuzzRoomMembership(membership, memberships.get(membership.roomId))
+          ) {
+            return;
+          }
+          memberships.set(membership.roomId, membership);
+        },
+        oneose: () => finish(),
+        onclose: (reason) => {
+          if (reason !== "membership snapshot loaded") {
+            finish(new Error(`Buzz room membership query closed: ${reason}`));
+          }
+        },
+      },
+    );
+    if (settled) {
+      subscription.close("membership snapshot loaded");
+    }
+    if (params.signal?.aborted) {
+      onAbort();
+    }
+  });
+}
+
+async function createBuzzRoomMembershipTracker(params: {
+  relay: Relay;
+  channelIds: string[];
+  botPublicKey: string;
+  since: number;
+  onFatalError?: (error: Error) => void;
+  signal?: AbortSignal;
+}): Promise<{
+  isMember: (channelId: string, publicKey: string) => boolean;
+  subscriptions: Array<ReturnType<Relay["subscribe"]>>;
+}> {
+  type BufferedSystemEvent = { event: Event; historical: boolean };
+  type ExpectedMembership = "present" | "absent";
+  type RefreshState = {
+    generation: number;
+    lastAttemptedGeneration: number;
+    promise: Promise<void>;
+  };
+
+  let initialized = false;
+  const historicalRooms = new Set<string>();
+  const bufferedEvents: BufferedSystemEvent[] = [];
+  const seenEventIds = new Map<string, true>();
+  const blockedRooms = new Set<string>();
+  const deniedMembers = new Map<string, Set<string>>();
+  const pendingMemberships = new Map<string, Map<string, ExpectedMembership>>();
+  const refreshes = new Map<string, RefreshState>();
+  let memberships = new Map<string, BuzzRoomMembership>();
+
+  const markSystemEventSeen = (eventId: string): boolean => {
+    if (seenEventIds.has(eventId)) {
+      return false;
+    }
+    seenEventIds.set(eventId, true);
+    if (seenEventIds.size > MEMBERSHIP_EVENT_CACHE_MAX_ENTRIES) {
+      const oldestEventId = seenEventIds.keys().next().value;
+      if (oldestEventId) {
+        seenEventIds.delete(oldestEventId);
+      }
+    }
+    return true;
+  };
+  const reportSystemEventError = (error: unknown) => {
+    if (params.signal?.aborted) {
+      return;
+    }
+    params.onFatalError?.(error instanceof Error ? error : new Error(String(error)));
+    params.relay.close();
+  };
+
+  const refreshMembership = async (channelId: string, state: RefreshState): Promise<void> => {
+    const baseline = memberships.get(channelId);
+    if (!baseline) {
+      throw new Error(`Missing Buzz room membership for ${channelId}`);
+    }
+    for (const delayMs of MEMBERSHIP_REFRESH_DELAYS_MS) {
+      const generation = state.generation;
+      state.lastAttemptedGeneration = generation;
+      await sleepWithSignal(delayMs, params.signal);
+      if (state.generation !== generation) {
+        continue;
+      }
+      let refreshed: BuzzRoomMembership | undefined;
+      try {
+        refreshed = (
+          await queryBuzzRoomMemberships({
+            relay: params.relay,
+            channelIds: [channelId],
+            timeoutMs: 3_000,
+            signal: params.signal,
+          })
+        ).get(channelId);
+      } catch (error) {
+        if (params.signal?.aborted) {
+          throw error;
+        }
+        continue;
+      }
+      if (state.generation !== generation || !refreshed) {
+        continue;
+      }
+      const pending = pendingMemberships.get(channelId);
+      const pendingMatches =
+        !pending ||
+        [...pending].every(
+          ([publicKey, expected]) => refreshed.members.has(publicKey) === (expected === "present"),
+        );
+      const botMembershipChanged = pending?.has(params.botPublicKey) === true;
+      if (
+        !pendingMatches ||
+        (botMembershipChanged && !isNewerBuzzRoomMembership(refreshed, baseline))
+      ) {
+        continue;
+      }
+      if (
+        refreshed.roles.get(params.botPublicKey) !== "bot" ||
+        !refreshed.members.has(params.botPublicKey)
+      ) {
+        blockedRooms.add(channelId);
+        throw new Error(`Buzz bot no longer has the Bot role in room ${channelId}`);
+      }
+      memberships.set(channelId, refreshed);
+      pendingMemberships.delete(channelId);
+      deniedMembers.delete(channelId);
+      blockedRooms.delete(channelId);
+      return;
+    }
+    if (state.generation !== state.lastAttemptedGeneration) {
+      return;
+    }
+    blockedRooms.add(channelId);
+    throw new Error(`Could not refresh Buzz room membership for ${channelId}`);
+  };
+
+  const refreshMembershipOnce = (channelId: string): Promise<void> => {
+    const current = refreshes.get(channelId);
+    if (current) {
+      current.generation += 1;
+      return current.promise;
+    }
+    const state = {
+      generation: 1,
+      lastAttemptedGeneration: 0,
+      promise: Promise.resolve(),
+    } satisfies RefreshState;
+    state.promise = refreshMembership(channelId, state).finally(() => {
+      if (refreshes.get(channelId) === state) {
+        refreshes.delete(channelId);
+      }
+      if (
+        state.generation !== state.lastAttemptedGeneration &&
+        pendingMemberships.has(channelId) &&
+        !params.signal?.aborted
+      ) {
+        void refreshMembershipOnce(channelId).catch(reportSystemEventError);
+      }
+    });
+    refreshes.set(channelId, state);
+    return state.promise;
+  };
+
+  const handleSystemEvent = (event: Event): Promise<void> | undefined => {
+    if (!markSystemEventSeen(event.id)) {
+      return;
+    }
+    const channelId = event.tags
+      .find((tag) => tag[0] === "h")?.[1]
+      ?.trim()
+      .toLowerCase();
+    if (!channelId) {
+      return;
+    }
+    const membership = memberships.get(channelId);
+    if (!membership) {
+      return;
+    }
+    const change = parseBuzzRoomMembershipChangeEvent(event, membership);
+    if (!change) {
+      return;
+    }
+    // System events invalidate membership; the relay-signed roster decides the
+    // final state. Removals deny immediately, while joins wait for confirmation.
+    const expected = change.type === "member_joined" ? "present" : "absent";
+    const pending = pendingMemberships.get(channelId) ?? new Map<string, ExpectedMembership>();
+    pending.set(change.targetPublicKey, expected);
+    pendingMemberships.set(channelId, pending);
+    if (expected === "absent") {
+      const denied = deniedMembers.get(channelId) ?? new Set<string>();
+      denied.add(change.targetPublicKey);
+      deniedMembers.set(channelId, denied);
+    }
+    if (change.targetPublicKey === params.botPublicKey) {
+      blockedRooms.add(channelId);
+    }
+    return refreshMembershipOnce(channelId);
+  };
+
+  let resolveHistorical: (() => void) | undefined;
+  let rejectHistorical: ((error: Error) => void) | undefined;
+  const historicalReady = new Promise<void>((resolve, reject) => {
+    resolveHistorical = resolve;
+    rejectHistorical = reject;
+  });
+  const historicalTimeout = setTimeout(() => {
+    rejectHistorical?.(new Error("Timed out loading Buzz room membership changes"));
+  }, MEMBERSHIP_READY_TIMEOUT_MS);
+  const subscriptions = params.channelIds.map((channelId) =>
+    params.relay.subscribe(
+      [
+        {
+          kinds: [BUZZ_ROOM_SYSTEM_KIND],
+          "#h": [channelId],
+          since: params.since,
+        },
+      ],
+      {
+        onevent: (event) => {
+          if (!initialized) {
+            bufferedEvents.push({ event, historical: !historicalRooms.has(channelId) });
+            return;
+          }
+          void handleSystemEvent(event)?.catch(reportSystemEventError);
+        },
+        oneose: () => {
+          historicalRooms.add(channelId);
+          if (historicalRooms.size === params.channelIds.length) {
+            resolveHistorical?.();
+          }
+        },
+        onclose: (reason) => {
+          if (!historicalRooms.has(channelId)) {
+            rejectHistorical?.(
+              new Error(`Buzz membership subscription closed for ${channelId}: ${reason}`),
+            );
+          } else if (
+            reason !== "shutdown" &&
+            reason !== "relay connection closed by us" &&
+            !params.signal?.aborted
+          ) {
+            params.onFatalError?.(
+              new Error(`Buzz membership subscription closed for ${channelId}: ${reason}`),
+            );
+          }
+        },
+      },
+    ),
+  );
+
+  try {
+    await historicalReady;
+    memberships = await queryBuzzRoomMemberships(params);
+  } catch (error) {
+    for (const subscription of subscriptions) {
+      subscription.close("membership tracker setup failed");
+    }
+    throw error;
+  } finally {
+    clearTimeout(historicalTimeout);
+  }
+
+  for (const channelId of params.channelIds) {
+    if (memberships.get(channelId)?.roles.get(params.botPublicKey) !== "bot") {
+      for (const subscription of subscriptions) {
+        subscription.close("membership tracker setup failed");
+      }
+      throw new Error(`Buzz bot does not have the Bot role in configured room ${channelId}`);
+    }
+  }
+
+  // Each room subscription reaches EOSE before the snapshot query starts, so
+  // the snapshot owns historical state. Only events received after that room's
+  // EOSE can be newer than the loaded snapshot and need an in-memory overlay.
+  const liveEvents = bufferedEvents
+    .filter((entry) => !entry.historical)
+    .map((entry) => entry.event);
+  for (const event of liveEvents) {
+    void handleSystemEvent(event)?.catch(reportSystemEventError);
+  }
+  initialized = true;
+
+  return {
+    isMember: (channelId, publicKey) =>
+      !blockedRooms.has(channelId) &&
+      !deniedMembers.get(channelId)?.has(publicKey.trim().toLowerCase()) &&
+      memberships.get(channelId)?.members.has(publicKey.trim().toLowerCase()) === true,
+    subscriptions,
+  };
+}
+
 export async function sendBuzzTextOneShot(params: {
   relayUrl: string;
   privateKey: string;
@@ -168,6 +554,10 @@ export async function startBuzzBus(options: {
   const publicKey = resolveBuzzPublicKey(options.privateKey);
   const authTag = parseBuzzAuthTag(options.authTag ?? "");
   const sessionStartedAt = Math.floor(Date.now() / 1000);
+  const lifecycleAbort = new AbortController();
+  const signal = options.signal
+    ? AbortSignal.any([options.signal, lifecycleAbort.signal])
+    : lifecycleAbort.signal;
   const replayGuard = createChannelReplayGuard<Event>({
     dedupe: {
       pluginId: "buzz",
@@ -186,7 +576,7 @@ export async function startBuzzBus(options: {
     relayUrl: options.relayUrl,
     secretKey,
     authTag,
-    signal: options.signal,
+    signal,
   });
   let subscriptions: Array<ReturnType<Relay["subscribe"]>> = [];
   let stopPresenceHeartbeat = () => {};
@@ -198,6 +588,7 @@ export async function startBuzzBus(options: {
       return event.id;
     },
     close: async () => {
+      lifecycleAbort.abort(new Error("Buzz bus closed"));
       stopPresenceHeartbeat();
       for (const subscription of subscriptions) {
         subscription.close("shutdown");
@@ -208,40 +599,57 @@ export async function startBuzzBus(options: {
   };
 
   try {
-    subscriptions = options.channelIds.map((channelId) =>
-      relay.subscribe(
-        [
+    const membershipTracker = await createBuzzRoomMembershipTracker({
+      relay,
+      channelIds: options.channelIds,
+      botPublicKey: publicKey,
+      since: sessionStartedAt,
+      onFatalError: options.onFatalError,
+      signal,
+    });
+    subscriptions.push(...membershipTracker.subscriptions);
+
+    subscriptions.push(
+      ...options.channelIds.map((channelId) =>
+        relay.subscribe(
+          [
+            {
+              kinds: [MESSAGE_KIND],
+              "#h": [channelId],
+              since: options.since ?? sessionStartedAt,
+            },
+          ],
           {
-            kinds: [MESSAGE_KIND],
-            "#h": [channelId],
-            since: options.since ?? sessionStartedAt,
+            onevent: (event) => {
+              if (event.pubkey === publicKey) {
+                return;
+              }
+              if (!membershipTracker.isMember(channelId, event.pubkey)) {
+                return;
+              }
+              const message = parseBuzzMessageEvent(event);
+              if (!message || message.channelId !== channelId) {
+                return;
+              }
+              // Relay reconnects can replay signed events. Only admitted room
+              // members reach the persistent dedupe store or agent pipeline.
+              void replayGuard
+                .processGuarded(event, async () => {
+                  await options.onMessage(message, bus);
+                })
+                .catch((error: unknown) => {
+                  options.onMessageError?.(
+                    error instanceof Error ? error : new Error(String(error)),
+                  );
+                });
+            },
+            onclose: (reason) => {
+              if (reason !== "shutdown" && reason !== "relay connection closed by us") {
+                options.onFatalError?.(new Error(`Buzz subscription closed: ${reason}`));
+              }
+            },
           },
-        ],
-        {
-          onevent: (event) => {
-            // Relay reconnects can replay signed events. Guard by immutable event id
-            // before any authorization, command, or agent work can run twice.
-            void replayGuard
-              .processGuarded(event, async () => {
-                if (event.pubkey === publicKey) {
-                  return;
-                }
-                const message = parseBuzzMessageEvent(event);
-                if (!message) {
-                  return;
-                }
-                await options.onMessage(message, bus);
-              })
-              .catch((error: unknown) => {
-                options.onMessageError?.(error instanceof Error ? error : new Error(String(error)));
-              });
-          },
-          onclose: (reason) => {
-            if (reason !== "shutdown" && reason !== "relay connection closed by us") {
-              options.onFatalError?.(new Error(`Buzz subscription closed: ${reason}`));
-            }
-          },
-        },
+        ),
       ),
     );
     // Buzz presence is a separate ephemeral protocol, not a property of the
@@ -260,7 +668,7 @@ export async function startBuzzBus(options: {
         publicKey,
         displayName: options.profileName,
         authTag,
-        signal: options.signal,
+        signal,
       })
         .then((result) => {
           if (result.status === "published") {
@@ -268,7 +676,7 @@ export async function startBuzzBus(options: {
           }
         })
         .catch((error: unknown) => {
-          if (options.signal?.aborted) {
+          if (signal.aborted) {
             return;
           }
           options.onProfileError?.(
@@ -283,6 +691,7 @@ export async function startBuzzBus(options: {
   } catch (error) {
     // Every failed startup must release the socket before ownership returns to
     // the gateway-level reconnect loop.
+    lifecycleAbort.abort(error);
     relay.close();
     throw error;
   }

extensions/buzz/src/room-membership.ts

@@ -0,0 +1,110 @@
+import type { Event } from "nostr-tools";
+import { BUZZ_CHANNEL_ID_PATTERN } from "./target.js";
+
+export const BUZZ_ROOM_MEMBERSHIP_KIND = 39002;
+export const BUZZ_ROOM_SYSTEM_KIND = 40099;
+
+const HEX_PUBLIC_KEY_PATTERN = /^[0-9a-f]{64}$/u;
+const MEMBERSHIP_CHANGE_TYPES = new Set(["member_joined", "member_left", "member_removed"]);
+
+export type BuzzRoomMembershipChange = {
+  type: "member_joined" | "member_left" | "member_removed";
+  targetPublicKey: string;
+};
+
+export type BuzzRoomMembership = {
+  roomId: string;
+  createdAt: number;
+  eventId: string;
+  publisherPublicKey: string;
+  members: ReadonlySet<string>;
+  roles: ReadonlyMap<string, string>;
+};
+
+export function parseBuzzRoomMembershipEvent(event: Event): BuzzRoomMembership | undefined {
+  if (event.kind !== BUZZ_ROOM_MEMBERSHIP_KIND) {
+    return undefined;
+  }
+  const roomId = event.tags
+    .find((tag) => tag[0] === "d")?.[1]
+    ?.trim()
+    .toLowerCase();
+  if (!roomId || !BUZZ_CHANNEL_ID_PATTERN.test(roomId)) {
+    return undefined;
+  }
+
+  const members = new Set<string>();
+  const roles = new Map<string, string>();
+  for (const tag of event.tags) {
+    if (tag[0] !== "p") {
+      continue;
+    }
+    const publicKey = tag[1]?.trim().toLowerCase();
+    if (!publicKey || !HEX_PUBLIC_KEY_PATTERN.test(publicKey)) {
+      continue;
+    }
+    members.add(publicKey);
+    const role = tag[3]?.trim().toLowerCase();
+    if (role) {
+      roles.set(publicKey, role);
+    }
+  }
+
+  return {
+    roomId,
+    createdAt: event.created_at,
+    eventId: event.id,
+    publisherPublicKey: event.pubkey.toLowerCase(),
+    members,
+    roles,
+  };
+}
+
+export function isNewerBuzzRoomMembership(
+  candidate: BuzzRoomMembership,
+  current: BuzzRoomMembership | undefined,
+): boolean {
+  return (
+    !current ||
+    candidate.createdAt > current.createdAt ||
+    (candidate.createdAt === current.createdAt && candidate.eventId < current.eventId)
+  );
+}
+
+export function parseBuzzRoomMembershipChangeEvent(
+  event: Event,
+  membership: BuzzRoomMembership,
+): BuzzRoomMembershipChange | undefined {
+  if (
+    event.kind !== BUZZ_ROOM_SYSTEM_KIND ||
+    event.pubkey.toLowerCase() !== membership.publisherPublicKey ||
+    !event.tags.some((tag) => tag[0] === "h" && tag[1]?.toLowerCase() === membership.roomId)
+  ) {
+    return undefined;
+  }
+  try {
+    const content = JSON.parse(event.content) as {
+      type?: unknown;
+      target?: unknown;
+      actor?: unknown;
+    };
+    if (typeof content.type !== "string" || !MEMBERSHIP_CHANGE_TYPES.has(content.type)) {
+      return undefined;
+    }
+    const target =
+      typeof content.target === "string"
+        ? content.target.trim().toLowerCase()
+        : content.type === "member_left" && typeof content.actor === "string"
+          ? content.actor.trim().toLowerCase()
+          : "";
+    if (!HEX_PUBLIC_KEY_PATTERN.test(target)) {
+      return undefined;
+    }
+    return {
+      type: content.type as BuzzRoomMembershipChange["type"],
+      targetPublicKey: target,
+    };
+  } catch {
+    return undefined;
+  }
+}