diff --git a/plugins/platforms/buzz/adapter.py b/plugins/platforms/buzz/adapter.py index b07830ae9b6e4..21d3bd2f38b32 100644 --- a/plugins/platforms/buzz/adapter.py +++ b/plugins/platforms/buzz/adapter.py @@ -60,27 +60,29 @@ from gateway.config import Platform -# Buzz chat messages are Nostr kind 9 events. ``buzz messages get`` also -# returns housekeeping kinds (joins, canvas updates, …) — only kind 9 is -# dispatched to the agent. +# Buzz stream messages are Nostr kind 9 events. Forum roots/comments use +# separate thread semantics and are deliberately outside this adapter. _CHAT_KIND = 9 +_FORUM_EVENT_KINDS = frozenset({45001, 45003}) # How many events to request per poll / seed call. _FETCH_LIMIT = 50 # Bound on the per-channel de-dupe set (events, not bytes). _SEEN_CAP = 500 -# Re-run DM discovery (``dms list`` plus the channels-list fallback) every -# N poll sweeps to pick up conversations opened mid-run. -_DM_DISCOVERY_EVERY = 5 +# Re-run joined-channel and DM discovery every N poll sweeps so the polling +# fallback adopts conversations opened mid-run. +_CHANNEL_DISCOVERY_EVERY = 5 _DEFAULT_POLL_INTERVAL = 4.0 _MIN_POLL_INTERVAL = 1.0 _CLI_TIMEOUT = 30.0 # WebSocket transport (NIP-42 authenticated Nostr subscription). -# kind 44100 is Buzz's channel-membership event — used for live DM discovery. +# Buzz emits relay-signed membership notifications for joins and removals. _WS_AUTH_TIMEOUT = 20.0 _WS_MAX_MESSAGE_BYTES = 2_000_000 _WS_MEMBERSHIP_KIND = 44100 +_WS_MEMBERSHIP_REMOVED_KIND = 44101 +_WS_MEMBERSHIP_KINDS = frozenset({_WS_MEMBERSHIP_KIND, _WS_MEMBERSHIP_REMOVED_KIND}) _WS_MEMBERSHIP_SUB_ID = "hermes-buzz-membership" # Where to look for a credentials JSON (keys: nsec / private_key_hex) when @@ -320,6 +322,17 @@ def _parse_json_list(stdout: str) -> List[dict]: return [item for item in data if isinstance(item, dict)] +def _parse_json_list_strict(stdout: str) -> Optional[List[dict]]: + """Parse a JSON object array, preserving malformed-vs-empty distinction.""" + try: + data = json.loads(stdout) + except (TypeError, ValueError): + return None + if not isinstance(data, list) or any(not isinstance(item, dict) for item in data): + return None + return data + + # --------------------------------------------------------------------------- # Buzz Adapter # --------------------------------------------------------------------------- @@ -486,14 +499,25 @@ async def connect(self, *, is_reconnect: bool = False) -> bool: except ImportError: self._lock_key = None # status module not available (e.g. tests) + # Start the membership cursor before taking the joined-channel + # snapshot. A join racing with startup is then present either in the + # snapshot or in the membership subscription's inclusive overlap. + if self.transport in ("auto", "websocket"): + self._membership_since = int(time.time()) + # Map channel ids to names and pick the watch set. - code, out, err = await self._run_cli(["channels", "list"]) + code, out, err = await self._run_cli(["channels", "list", "--member"]) if code != 0: message = _cli_error_message(err, code) logger.error("Buzz: failed to list channels — %s", message) self._set_fatal_error("connect_failed", message, retryable=code == 2) return False - listed = _parse_json_list(out) + listed = _parse_json_list_strict(out) + if listed is None: + message = "buzz channels list returned malformed JSON" + logger.error("Buzz: %s", message) + self._set_fatal_error("connect_failed", message, retryable=True) + return False self._channel_names = { str(ch.get("channel_id")): str(ch.get("name") or ch.get("channel_id")) for ch in listed @@ -504,9 +528,9 @@ async def connect(self, *, is_reconnect: bool = False) -> bool: self._channel_meta[str(ch["channel_id"])] = ch watch = self.channels or list(self._channel_names) if not watch: - logger.error("Buzz: no channels to watch (configure BUZZ_CHANNELS or join a channel)") - self._set_fatal_error("config_missing", "no Buzz channels to watch", retryable=False) - return False + logger.info( + "Buzz: no joined channels yet; waiting for dynamic discovery" + ) # Seed high-water marks from the newest events so a (re)start never # replays channel history into the agent. @@ -719,7 +743,8 @@ async def _start_websocket(self) -> bool: logger.info("Buzz: WebSocket transport unavailable (%s); falling back to polling", e) return False self._ws_ready = asyncio.Event() - self._membership_since = int(time.time()) + if not self._membership_since: + self._membership_since = int(time.time()) self._ws_task = asyncio.create_task(self._websocket_loop()) try: await asyncio.wait_for(self._ws_ready.wait(), timeout=_WS_AUTH_TIMEOUT + 5) @@ -766,9 +791,21 @@ async def _authenticate_websocket(self, websocket) -> None: detail = response[-1] if len(response) > 1 else "authentication failed" raise ConnectionError(f"Buzz WebSocket AUTH failed: {detail}") - async def _send_channel_subscription(self, websocket, subscription_id: str, channel_id: str) -> None: + async def _send_channel_subscription( + self, + websocket, + subscription_id: str, + channel_id: str, + *, + since: Optional[int] = None, + ) -> None: state = self._channel_state.get(channel_id) or {} - since = max(int(state.get("last_ts") or time.time()) - 1, 0) + if since is None: + last_ts = int(state.get("last_ts") or time.time()) + subscription_floor = int(state.get("subscription_floor") or 0) + since = max(last_ts - 1, subscription_floor, 0) + else: + since = max(int(since), 0) request = [ "REQ", subscription_id, @@ -777,8 +814,8 @@ async def _send_channel_subscription(self, websocket, subscription_id: str, chan await websocket.send(json.dumps(request, separators=(",", ":"))) async def _subscribe_websocket(self, websocket) -> Dict[str, Optional[str]]: - """Subscribe to every watched conversation plus membership events - (kind 44100 p-tagged to us) for live DM discovery.""" + """Subscribe to every watched conversation plus relay-signed + membership changes for live regular-channel and DM discovery.""" subscriptions: Dict[str, Optional[str]] = {} for index, channel_id in enumerate(list(self._channel_state)): subscription_id = f"hermes-buzz-{index}" @@ -789,7 +826,7 @@ async def _subscribe_websocket(self, websocket) -> Dict[str, Optional[str]]: "REQ", _WS_MEMBERSHIP_SUB_ID, { - "kinds": [_WS_MEMBERSHIP_KIND], + "kinds": sorted(_WS_MEMBERSHIP_KINDS), "#p": [self._self_pubkey], "since": max(self._membership_since - 1, 0), }, @@ -799,18 +836,81 @@ async def _subscribe_websocket(self, websocket) -> Dict[str, Optional[str]]: return subscriptions async def _handle_membership_event(self, websocket, subscriptions: Dict[str, Optional[str]], event: dict) -> None: - """A membership event p-tagged to us: rediscover conversations and - subscribe to any new ones (fresh DMs dispatch from their beginning).""" - self._membership_since = max(self._membership_since, int(event.get("created_at") or 0)) + """Reconcile subscriptions after a relay-signed membership change.""" + event_since = max(int(event.get("created_at") or 0), 0) + kind = int(event.get("kind") or 0) + if event_since < self._membership_since: + return + tags = event.get("tags") + channel_id = "" + if isinstance(tags, list): + channel_id = next( + ( + str(tag[1]) + for tag in tags + if isinstance(tag, (list, tuple)) + and len(tag) > 1 + and tag[0] == "h" + ), + "", + ) + + if kind == _WS_MEMBERSHIP_REMOVED_KIND: + self._membership_since = max(self._membership_since, event_since) + if channel_id: + self._channel_state.pop(channel_id, None) + self._channel_names.pop(channel_id, None) + self._channel_meta.pop(channel_id, None) + for subscription_id, subscribed_channel in list(subscriptions.items()): + if subscribed_channel != channel_id: + continue + await websocket.send( + json.dumps(["CLOSE", subscription_id], separators=(",", ":")) + ) + subscriptions.pop(subscription_id, None) + logger.info("Buzz: unsubscribed from removed conversation %s", channel_id) + return + if kind != _WS_MEMBERSHIP_KIND: + return + before = set(self._channel_state) + if not await self._discover_joined_channels( + since=event_since, seed=True, target_channel_id=channel_id + ): + raise ConnectionError("Buzz joined-channel discovery failed") + self._membership_since = max(self._membership_since, event_since) + + async def subscribe_discovered() -> None: + for discovered_channel_id in self._channel_state: + if discovered_channel_id in before: + continue + subscription_index = len(subscriptions) + subscription_id = f"hermes-buzz-dm-{subscription_index}" + while subscription_id in subscriptions: + subscription_index += 1 + subscription_id = f"hermes-buzz-dm-{subscription_index}" + subscriptions[subscription_id] = discovered_channel_id + subscription_since = ( + event_since + if channel_id and discovered_channel_id == channel_id + else None + ) + await self._send_channel_subscription( + websocket, + subscription_id, + discovered_channel_id, + since=subscription_since, + ) + before.add(discovered_channel_id) + logger.info( + "Buzz: subscribed to new conversation %s", discovered_channel_id + ) + + # The membership-target channel is time-sensitive. Subscribe before + # unrelated DM discovery can spend up to two CLI timeouts. + await subscribe_discovered() await self._discover_dms(seed=False) - for channel_id in self._channel_state: - if channel_id in before: - continue - subscription_id = f"hermes-buzz-dm-{len(subscriptions)}" - subscriptions[subscription_id] = channel_id - await self._send_channel_subscription(websocket, subscription_id, channel_id) - logger.info("Buzz: subscribed to new conversation %s", channel_id) + await subscribe_discovered() async def _websocket_loop(self) -> None: """Persistent authenticated subscription with bounded reconnect @@ -833,6 +933,13 @@ async def _websocket_loop(self) -> None: max_size=_WS_MAX_MESSAGE_BYTES, ) as websocket: await self._authenticate_websocket(websocket) + if not await self._discover_joined_channels( + seed=True, reconcile=True + ): + raise ConnectionError( + "Buzz joined-channel reconciliation failed" + ) + await self._discover_dms(seed=False) subscriptions = await self._subscribe_websocket(websocket) self._ws_active = True if self._ws_ready is not None: @@ -883,7 +990,10 @@ async def _poll_loop(self) -> None: await asyncio.sleep(self.poll_interval) self._poll_count += 1 try: - if self._poll_count % _DM_DISCOVERY_EVERY == 0: + if self._poll_count % _CHANNEL_DISCOVERY_EVERY == 0: + await self._discover_joined_channels( + seed=True, reconcile=True + ) await self._discover_dms(seed=False) for channel_id in list(self._channel_state): await self._poll_channel(channel_id) @@ -894,20 +1004,36 @@ async def _poll_loop(self) -> None: except asyncio.CancelledError: raise - async def _seed_channel(self, channel_id: str, chat_type: str) -> None: - """Initialize a channel's high-water mark from its newest events.""" + async def _seed_channel( + self, + channel_id: str, + chat_type: str, + *, + before: int = 0, + safe_floor: bool = False, + ) -> None: + """Initialize a channel's high-water mark from its newest history.""" state = {"chat_type": chat_type, "last_ts": 0, "seen": OrderedDict()} self._channel_state[channel_id] = state - code, out, err = await self._run_cli( - ["messages", "get", "--channel", channel_id, "--limit", str(_FETCH_LIMIT)] - ) + safe_boundary = int(time.time()) if safe_floor else 0 + history_before = before or max(safe_boundary - 1, 0) + args = [ + "messages", "get", "--channel", channel_id, + "--limit", str(_FETCH_LIMIT), + ] + if history_before: + args.extend(["--before", str(history_before)]) + code, out, err = await self._run_cli(args) if code != 0: logger.warning( "Buzz: could not seed channel %s — %s", channel_id, _cli_error_message(err, code) ) - # Fall back to "now" so a transiently unreadable channel does not - # replay its whole history once it becomes readable. - state["last_ts"] = int(time.time()) + # Fall back to a safe current boundary so a transiently unreadable + # channel does not replay history once it becomes readable. + floor = safe_boundary if safe_floor else int(time.time()) + state["last_ts"] = floor + if safe_floor: + state["subscription_floor"] = floor return for event in _parse_json_list(out): event_id = event.get("id") @@ -920,6 +1046,89 @@ async def _seed_channel(self, channel_id: str, chat_type: str) -> None: # so it bypasses the mention gate from the very first poll. self._maybe_latch_dm(channel_id, state, event) self._trim_seen(state) + if safe_floor: + floor = max(safe_boundary, int(state["last_ts"]) + 1) + state["last_ts"] = floor + state["subscription_floor"] = floor + + async def _discover_joined_channels( + self, + *, + since: int = 0, + seed: bool = False, + target_channel_id: str = "", + reconcile: bool = False, + ) -> bool: + """Watch newly joined stream channels when no explicit allowlist is set.""" + if self.channels: + return True + code, out, err = await self._run_cli(["channels", "list", "--member"]) + if code != 0: + logger.warning( + "Buzz: failed to rediscover joined channels — %s", + _cli_error_message(err, code), + ) + return False + channels = _parse_json_list_strict(out) + if channels is None: + logger.warning("Buzz: joined-channel discovery returned malformed JSON") + return False + joined_ids = { + str(channel.get("channel_id") or "") for channel in channels + } + if target_channel_id and target_channel_id not in joined_ids: + logger.warning( + "Buzz: membership channel %s missing from joined-channel snapshot", + target_channel_id, + ) + return False + if reconcile: + for watched_channel_id, state in list(self._channel_state.items()): + if ( + state.get("chat_type") == "group" + and watched_channel_id not in joined_ids + ): + self._channel_state.pop(watched_channel_id, None) + self._channel_names.pop(watched_channel_id, None) + self._channel_meta.pop(watched_channel_id, None) + logger.info( + "Buzz: stopped watching channel no longer joined: %s", + watched_channel_id, + ) + for channel in channels: + channel_id = str(channel.get("channel_id") or "") + if not channel_id: + continue + self._channel_meta[channel_id] = channel + self._channel_names[channel_id] = str(channel.get("name") or channel_id) + if channel_id in self._channel_state: + continue + channel_since = ( + int(since) + if not target_channel_id or channel_id == target_channel_id + else 0 + ) + if seed: + # Buzz maps --before to Nostr's inclusive `until`; stop at the + # prior second so events sharing the join timestamp stay live. + history_before = max(channel_since - 1, 0) if channel_since else 0 + await self._seed_channel( + channel_id, + chat_type="group", + before=history_before, + safe_floor=not bool(channel_since), + ) + state = self._channel_state[channel_id] + state["last_ts"] = max(state["last_ts"], channel_since) + if channel_since: + state["subscription_floor"] = channel_since + else: + self._channel_state[channel_id] = { + "chat_type": "group", + "last_ts": max(channel_since, 0), + "seen": OrderedDict(), + } + return True async def _discover_dms(self, *, seed: bool) -> None: """Watch DM conversations. New ones found mid-run dispatch from their @@ -990,7 +1199,10 @@ async def _handle_event(self, channel_id: str, state: dict, event: dict) -> None state["seen"][event_id] = None state["last_ts"] = max(state["last_ts"], created_at) - if int(event.get("kind") or 0) != _CHAT_KIND: + kind = int(event.get("kind") or 0) + if kind in _FORUM_EVENT_KINDS: + return + if kind != _CHAT_KIND: return pubkey = str(event.get("pubkey") or "").lower() content = event.get("content") diff --git a/tests/gateway/test_buzz_adapter.py b/tests/gateway/test_buzz_adapter.py index d612021bfe39f..7c884a1a63bf4 100644 --- a/tests/gateway/test_buzz_adapter.py +++ b/tests/gateway/test_buzz_adapter.py @@ -381,6 +381,499 @@ async def test_dm_shaped_channel_discovered_when_dms_list_empty(self): assert a._may_reclassify_as_dm(CHANNEL) is False +# ── Dynamic joined-channel discovery ───────────────────────────────────── + + +class TestChannelDiscovery: + + @pytest.mark.asyncio + async def test_connect_lists_only_joined_channels(self, monkeypatch): + import gateway.status as gateway_status + + monkeypatch.setattr( + gateway_status, "acquire_scoped_lock", lambda platform, key: True + ) + monkeypatch.setattr(_buzz_mod, "_resolve_private_key", lambda extra=None: "nsec1test") + monkeypatch.setattr(_buzz_mod.time, "time", lambda: 1000) + adapter = _make_adapter() + adapter.cli_path = "/fake/buzz" + adapter._start_websocket = AsyncMock(return_value=False) + unjoined_channel = "12c81eb7-3a12-47c1-b8af-c66f1c74ca8b" + cli = _ScriptedCli() + cli.script( + "users", "get", + [{"pubkey": SELF_PUBKEY, "display_name": "Chip"}], + ) + cli.script("messages", "get", []) + cli.script("dms", "list", []) + membership_cursors = [] + + async def run_cli(args, *, input_text=None): + if args == ["channels", "list", "--member"]: + membership_cursors.append(adapter._membership_since) + return 0, json.dumps([ + {"channel_id": CHANNEL, "name": "general"}, + ]), "" + if args == ["channels", "list"]: + return 0, json.dumps([ + {"channel_id": CHANNEL, "name": "general", "description": "General"}, + {"channel_id": unjoined_channel, "name": "other", "description": "Other"}, + ]), "" + return await cli(args, input_text=input_text) + + adapter._run_cli = run_cli + + try: + assert await adapter.connect() is True + finally: + await adapter.disconnect() + + assert membership_cursors == [1000] + assert unjoined_channel not in adapter._channel_state + + @pytest.mark.asyncio + async def test_connect_with_no_joined_channels_waits_for_dynamic_discovery( + self, monkeypatch + ): + import gateway.status as gateway_status + + monkeypatch.setattr( + gateway_status, "acquire_scoped_lock", lambda platform, key: True + ) + monkeypatch.setattr( + gateway_status, "release_scoped_lock", lambda platform, key: None + ) + monkeypatch.setattr(_buzz_mod, "_resolve_private_key", lambda extra=None: "nsec1test") + monkeypatch.setattr(_buzz_mod.time, "time", lambda: 1000) + adapter = _make_adapter() + adapter.cli_path = "/fake/buzz" + adapter._start_websocket = AsyncMock(return_value=True) + cli = _ScriptedCli() + cli.script( + "users", "get", + [{"pubkey": SELF_PUBKEY, "display_name": "Chip"}], + ) + cli.script("channels", "list", []) + cli.script("dms", "list", []) + adapter._run_cli = cli + + try: + assert await adapter.connect() is True + assert adapter._channel_state == {} + adapter._start_websocket.assert_awaited_once() + assert adapter._membership_since == 1000 + finally: + await adapter.disconnect() + + @pytest.mark.asyncio + async def test_websocket_membership_subscription_covers_add_and_remove(self): + adapter = _make_adapter() + adapter._self_pubkey = SELF_PUBKEY + adapter._membership_since = 1000 + websocket = AsyncMock() + + subscriptions = await adapter._subscribe_websocket(websocket) + + assert subscriptions == {_buzz_mod._WS_MEMBERSHIP_SUB_ID: None} + request = json.loads(websocket.send.await_args.args[0]) + assert request[2]["kinds"] == sorted(_buzz_mod._WS_MEMBERSHIP_KINDS) + assert request[2]["#p"] == [SELF_PUBKEY] + + @pytest.mark.asyncio + async def test_membership_event_subscribes_to_new_joined_channel(self): + adapter = _make_adapter() + adapter._channel_state[CHANNEL] = { + "chat_type": "group", "last_ts": 100, "seen": {}, + } + new_channel = "4764ae67-7cd8-4f3e-967d-7dd93986b11a" + cli = _ScriptedCli() + cli.script("channels", "list", [ + {"channel_id": CHANNEL, "name": "general", "description": "General"}, + {"channel_id": new_channel, "name": "project", "description": "Project"}, + ]) + cli.script("dms", "list", []) + adapter._run_cli = cli + websocket = AsyncMock() + + async def assert_target_already_subscribed(*, seed): + assert seed is False + assert websocket.send.await_count == 1 + + adapter._discover_dms = AsyncMock(side_effect=assert_target_already_subscribed) + subscriptions = { + "hermes-buzz-0": CHANNEL, + _buzz_mod._WS_MEMBERSHIP_SUB_ID: None, + "hermes-buzz-dm-3": CHANNEL, + } + + await adapter._handle_membership_event( + websocket, + subscriptions, + { + "created_at": 1234, + "kind": _buzz_mod._WS_MEMBERSHIP_KIND, + "tags": [["p", SELF_PUBKEY], ["h", new_channel]], + }, + ) + + assert new_channel in adapter._channel_state + assert adapter._channel_state[new_channel]["chat_type"] == "group" + assert adapter._channel_state[new_channel]["last_ts"] == 1234 + assert new_channel in subscriptions.values() + assert subscriptions["hermes-buzz-dm-3"] == CHANNEL + assert (["channels", "list", "--member"], None) in cli.calls + request = json.loads(websocket.send.await_args.args[0]) + assert request[2]["#h"] == [new_channel] + assert request[2]["kinds"] == [_buzz_mod._CHAT_KIND] + assert not (_buzz_mod._FORUM_EVENT_KINDS & set(request[2]["kinds"])) + assert request[2]["since"] == 1234 + + @pytest.mark.asyncio + async def test_membership_event_seeds_existing_history_without_dispatch(self): + adapter = _make_adapter() + adapter._channel_state[CHANNEL] = { + "chat_type": "group", "last_ts": 100, "seen": {}, + } + adapter._dispatched = [] + + async def capture(**kwargs): + adapter._dispatched.append(kwargs) + + adapter._dispatch_message = capture + adapter._message_handler = AsyncMock() + new_channel = "4764ae67-7cd8-4f3e-967d-7dd93986b11a" + old_event = _event("old", content="@Chip old history", created_at=900) + old_event["tags"] = [["h", new_channel]] + cli = _ScriptedCli() + cli.script("channels", "list", [ + {"channel_id": CHANNEL, "name": "general"}, + {"channel_id": new_channel, "name": "project"}, + ]) + cli.script("messages", "get", [old_event]) + cli.script("dms", "list", []) + adapter._run_cli = cli + websocket = AsyncMock() + + await adapter._handle_membership_event( + websocket, + {"hermes-buzz-0": CHANNEL}, + {"created_at": 1234, "kind": _buzz_mod._WS_MEMBERSHIP_KIND}, + ) + + state = adapter._channel_state[new_channel] + assert state["last_ts"] == 1234 + assert set(state["seen"]) == {"old"} + assert adapter._dispatched == [] + message_call = next(args for args, _ in cli.calls if args[:2] == ["messages", "get"]) + assert message_call[message_call.index("--before") + 1] == "1233" + + live_event = _event("live", content="@Chip after join", created_at=1234) + live_event["tags"] = [["h", new_channel]] + await adapter._handle_event(new_channel, state, live_event) + assert [item["message_id"] for item in adapter._dispatched] == ["live"] + + @pytest.mark.asyncio + async def test_membership_removal_unsubscribes_channel(self): + adapter = _make_adapter() + adapter._membership_since = 100 + adapter._channel_state[CHANNEL] = { + "chat_type": "group", "last_ts": 100, "seen": {}, + } + websocket = AsyncMock() + subscriptions = {"hermes-buzz-0": CHANNEL} + + await adapter._handle_membership_event( + websocket, + subscriptions, + { + "created_at": 1234, + "kind": _buzz_mod._WS_MEMBERSHIP_REMOVED_KIND, + "tags": [["p", SELF_PUBKEY], ["h", CHANNEL]], + }, + ) + + assert CHANNEL not in adapter._channel_state + assert CHANNEL not in subscriptions.values() + assert adapter._membership_since == 1234 + websocket.send.assert_awaited_once_with( + json.dumps(["CLOSE", "hermes-buzz-0"], separators=(",", ":")) + ) + + @pytest.mark.asyncio + async def test_stale_membership_event_does_not_rewind_discovery(self): + adapter = _make_adapter() + adapter._membership_since = 200 + adapter._run_cli = AsyncMock() + websocket = AsyncMock() + + await adapter._handle_membership_event( + websocket, + {}, + { + "created_at": 100, + "kind": _buzz_mod._WS_MEMBERSHIP_KIND, + "tags": [["p", SELF_PUBKEY], ["h", CHANNEL]], + }, + ) + + assert adapter._membership_since == 200 + adapter._run_cli.assert_not_awaited() + websocket.send.assert_not_awaited() + + @pytest.mark.asyncio + async def test_websocket_reconnect_reconciles_before_resubscribing(self, monkeypatch): + import websockets + + adapter = _make_adapter() + stale_channel = "4764ae67-7cd8-4f3e-967d-7dd93986b11a" + adapter._self_pubkey = SELF_PUBKEY + adapter._channel_state[stale_channel] = { + "chat_type": "group", "last_ts": 100, "seen": {}, + } + cli = _ScriptedCli() + cli.script("channels", "list", []) + cli.script("channels", "list", []) + cli.script("dms", "list", []) + adapter._run_cli = cli + adapter._authenticate_websocket = AsyncMock() + + class OneConnection: + def __init__(self): + self.sent = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return False + + async def send(self, raw): + self.sent.append(json.loads(raw)) + + def __aiter__(self): + return self + + async def __anext__(self): + raise asyncio.CancelledError + + websocket = OneConnection() + monkeypatch.setattr(websockets, "connect", lambda *_args, **_kwargs: websocket) + + with pytest.raises(asyncio.CancelledError): + await adapter._websocket_loop() + + channel_requests = [ + request + for request in websocket.sent + if len(request) > 2 + and isinstance(request[2], dict) + and "#h" in request[2] + ] + assert stale_channel not in adapter._channel_state + assert channel_requests == [] + assert any( + request[1] == _buzz_mod._WS_MEMBERSHIP_SUB_ID + for request in websocket.sent + ) + + @pytest.mark.asyncio + async def test_membership_event_respects_explicit_channel_allowlist(self): + adapter = _make_adapter({"channels": [CHANNEL]}) + adapter._channel_state[CHANNEL] = { + "chat_type": "group", "last_ts": 100, "seen": {}, + } + new_channel = "4764ae67-7cd8-4f3e-967d-7dd93986b11a" + cli = _ScriptedCli() + cli.script("channels", "list", [ + {"channel_id": new_channel, "name": "project", "description": "Project"}, + ]) + cli.script("dms", "list", []) + adapter._run_cli = cli + websocket = AsyncMock() + subscriptions = {"hermes-buzz-0": CHANNEL} + + await adapter._handle_membership_event( + websocket, + subscriptions, + {"created_at": 1234, "kind": _buzz_mod._WS_MEMBERSHIP_KIND}, + ) + + assert new_channel not in adapter._channel_state + assert new_channel not in subscriptions.values() + websocket.send.assert_not_awaited() + + @pytest.mark.asyncio + async def test_membership_event_retries_after_joined_channel_discovery_failure(self): + adapter = _make_adapter() + adapter._membership_since = 100 + cli = _ScriptedCli() + cli.script("channels", "list", [], code=2, stderr="temporary failure") + adapter._run_cli = cli + websocket = AsyncMock() + + with pytest.raises(ConnectionError, match="joined-channel discovery failed"): + await adapter._handle_membership_event( + websocket, + {"hermes-buzz-0": CHANNEL}, + {"created_at": 1234, "kind": _buzz_mod._WS_MEMBERSHIP_KIND}, + ) + + assert adapter._membership_since == 100 + websocket.send.assert_not_awaited() + + @pytest.mark.asyncio + async def test_joined_channel_discovery_rejects_malformed_success_output(self): + adapter = _make_adapter() + adapter._run_cli = AsyncMock(return_value=(0, "not-json", "")) + + assert await adapter._discover_joined_channels(seed=True) is False + assert adapter._channel_state == {} + + @pytest.mark.asyncio + async def test_polling_fallback_rediscovers_joined_channels_on_discovery_cadence( + self, monkeypatch + ): + adapter = _make_adapter({"transport": "poll"}) + adapter._poll_count = _buzz_mod._CHANNEL_DISCOVERY_EVERY - 1 + adapter._discover_joined_channels = AsyncMock(return_value=True) + adapter._discover_dms = AsyncMock() + adapter._poll_channel = AsyncMock() + sleep_calls = 0 + + async def one_sweep_then_cancel(_delay): + nonlocal sleep_calls + sleep_calls += 1 + if sleep_calls > 1: + raise asyncio.CancelledError + + monkeypatch.setattr(_buzz_mod.asyncio, "sleep", one_sweep_then_cancel) + + with pytest.raises(asyncio.CancelledError): + await adapter._poll_loop() + + adapter._discover_joined_channels.assert_awaited_once_with( + seed=True, reconcile=True + ) + adapter._discover_dms.assert_awaited_once_with(seed=False) + + @pytest.mark.asyncio + async def test_polling_discovery_seeds_new_channel_without_replaying_history( + self, monkeypatch + ): + monkeypatch.setattr(_buzz_mod.time, "time", lambda: 200) + adapter = _make_adapter({"transport": "poll"}) + adapter._dispatched = [] + + async def capture(**kwargs): + adapter._dispatched.append(kwargs) + + adapter._dispatch_message = capture + adapter._message_handler = AsyncMock() + new_channel = "4764ae67-7cd8-4f3e-967d-7dd93986b11a" + old_event = _event("old", content="@Chip old history", created_at=100) + old_event["tags"] = [["h", new_channel]] + cli = _ScriptedCli() + cli.script("channels", "list", [ + {"channel_id": new_channel, "name": "project", "description": "Project"}, + ]) + cli.script("messages", "get", [old_event]) + adapter._run_cli = cli + + assert await adapter._discover_joined_channels(seed=True) is True + + state = adapter._channel_state[new_channel] + assert state["last_ts"] == 200 + assert state["subscription_floor"] == 200 + assert set(state["seen"]) == {"old"} + assert adapter._dispatched == [] + websocket = AsyncMock() + await adapter._send_channel_subscription( + websocket, "hermes-buzz-dynamic", new_channel + ) + request = json.loads(websocket.send.await_args.args[0]) + assert request[2]["since"] == 200 + message_call = next(args for args, _ in cli.calls if args[:2] == ["messages", "get"]) + assert message_call[message_call.index("--before") + 1] == "199" + + @pytest.mark.asyncio + async def test_discovery_reconciliation_prunes_left_stream_but_preserves_dm(self): + adapter = _make_adapter() + stale_channel = "4764ae67-7cd8-4f3e-967d-7dd93986b11a" + adapter._channel_state[stale_channel] = { + "chat_type": "group", "last_ts": 100, "seen": {}, + } + adapter._channel_state[DM_CHANNEL] = { + "chat_type": "dm", "last_ts": 100, "seen": {}, + } + adapter._channel_names[stale_channel] = "left" + adapter._channel_meta[stale_channel] = {"channel_id": stale_channel} + cli = _ScriptedCli() + cli.script("channels", "list", []) + adapter._run_cli = cli + + discovered = await adapter._discover_joined_channels( + seed=True, reconcile=True + ) + + assert discovered is True + assert stale_channel not in adapter._channel_state + assert stale_channel not in adapter._channel_names + assert stale_channel not in adapter._channel_meta + assert DM_CHANNEL in adapter._channel_state + + @pytest.mark.asyncio + async def test_forum_event_kinds_never_enter_stream_dispatch(self): + adapter = _make_adapter() + stream_channel = "4764ae67-7cd8-4f3e-967d-7dd93986b11a" + stream_state = { + "chat_type": "group", "last_ts": 1234, "seen": {}, + } + adapter._dispatched = [] + + async def capture(**kwargs): + adapter._dispatched.append(kwargs) + + adapter._dispatch_message = capture + adapter._message_handler = AsyncMock() + assert _buzz_mod._FORUM_EVENT_KINDS == frozenset({45001, 45003}) + for kind in _buzz_mod._FORUM_EVENT_KINDS: + event = _event(f"forum-{kind}", content="@Chip forum traffic", kind=kind) + event["tags"] = [["h", stream_channel]] + await adapter._handle_event(stream_channel, stream_state, event) + + assert adapter._dispatched == [] + assert set(stream_state["seen"]) == {"forum-45001", "forum-45003"} + + @pytest.mark.asyncio + async def test_dynamic_channel_uses_existing_mention_allowlist_and_dedupe_pipeline(self): + adapter = _make_adapter() + adapter._allowed_pubkeys = {OTHER_PUBKEY} + adapter._dispatched = [] + + async def capture(**kwargs): + adapter._dispatched.append(kwargs) + + adapter._dispatch_message = capture + adapter._message_handler = AsyncMock() + new_channel = "4764ae67-7cd8-4f3e-967d-7dd93986b11a" + cli = _ScriptedCli() + cli.script("channels", "list", [ + {"channel_id": new_channel, "name": "project", "description": "Project"}, + ]) + adapter._run_cli = cli + assert await adapter._discover_joined_channels(since=100, seed=False) is True + state = adapter._channel_state[new_channel] + + unmentioned = _event("quiet", content="background chatter", created_at=101) + mentioned = _event("ping", content="@Chip canary", created_at=102) + for event in (unmentioned, mentioned, mentioned): + event["tags"] = [["h", new_channel]] + await adapter._handle_event(new_channel, state, event) + + assert [item["message_id"] for item in adapter._dispatched] == ["ping"] + assert state["chat_type"] == "group" + + # ── Sending ───────────────────────────────────────────────────────────────