From 07e931fcb4d48c4c7aaa4121f426e5eaea7e77fe Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Tue, 28 Jul 2026 18:22:28 -0700 Subject: [PATCH] =?UTF-8?q?feat(buzz):=20WebSocket=20inbound=20transport?= =?UTF-8?q?=20=E2=80=94=20NIP-42=20auth,=20live=20DM=20discovery,=20poll?= =?UTF-8?q?=20fallback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consolidates the native-transport half of PR #73636 by @ScaleLeanChris onto the merged adapter: persistent NIP-42-authenticated Nostr WebSocket subscription as the default inbound path (transport=auto|websocket|poll), kind-44100 membership events for live DM discovery, since-timestamp resume on reconnect with bounded exponential backoff, and automatic fallback to CLI polling when the WS can't be established. Events route through the same _handle_event() pipeline as the poll loop, so de-dupe, mention gating, p-tag DM latching, and allow-lists behave identically on both transports. Outbound stays on the CLI (one-shot sends never race a WS auth handshake — his design). E2E verified against a real in-process websockets relay: NIP-42 challenge -> signed kind-22242 AUTH (event id re-derived server-side) -> REQ subscription -> EVENT dispatch -> clean disconnect. Co-authored-by: ScaleLeanChris --- contributors/emails/chris@scalelean.com | 1 + plugins/platforms/buzz/adapter.py | 257 +++++++++++++++++++++- plugins/platforms/buzz/plugin.yaml | 8 + tests/gateway/test_buzz_websocket.py | 244 ++++++++++++++++++++ website/docs/user-guide/messaging/buzz.md | 4 +- 5 files changed, 509 insertions(+), 5 deletions(-) create mode 100644 contributors/emails/chris@scalelean.com create mode 100644 tests/gateway/test_buzz_websocket.py diff --git a/contributors/emails/chris@scalelean.com b/contributors/emails/chris@scalelean.com new file mode 100644 index 000000000000..a487a63f9365 --- /dev/null +++ b/contributors/emails/chris@scalelean.com @@ -0,0 +1 @@ +ScaleLeanChris diff --git a/plugins/platforms/buzz/adapter.py b/plugins/platforms/buzz/adapter.py index 55e2c50fef4f..b07830ae9b6e 100644 --- a/plugins/platforms/buzz/adapter.py +++ b/plugins/platforms/buzz/adapter.py @@ -47,6 +47,7 @@ from collections import OrderedDict from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, Tuple +from urllib.parse import urlsplit, urlunsplit logger = logging.getLogger(__name__) @@ -75,11 +76,40 @@ _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. +_WS_AUTH_TIMEOUT = 20.0 +_WS_MAX_MESSAGE_BYTES = 2_000_000 +_WS_MEMBERSHIP_KIND = 44100 +_WS_MEMBERSHIP_SUB_ID = "hermes-buzz-membership" + # Where to look for a credentials JSON (keys: nsec / private_key_hex) when # BUZZ_PRIVATE_KEY is not set. Module-level so tests can point it at a tmpdir. _DEFAULT_CREDENTIALS_DIR = Path("~/.config/buzz").expanduser() +def _load_nostr_auth(): + """Import the sibling nostr_auth module in a loader-agnostic way. + + The adapter is imported both as a package module + (``plugins.platforms.buzz.adapter``) and as a bare single-file module by + the test plugin loader, where relative imports have no parent package. + """ + try: + from . import nostr_auth # type: ignore[no-redef] + + return nostr_auth + except ImportError: + import importlib.util + + path = Path(__file__).with_name("nostr_auth.py") + spec = importlib.util.spec_from_file_location("plugin_adapter_buzz_nostr_auth", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + # --------------------------------------------------------------------------- # bech32 (BIP-173) helpers — used to convert between npub and hex pubkeys so # mention detection and allow-lists accept either form. Pure stdlib. @@ -338,6 +368,15 @@ class BuzzAdapter(BasePlatformAdapter): _rm_cfg = _rm_raw self.require_mention = str(_rm_cfg).strip().lower() not in ("false", "0", "no", "off") + # Inbound transport: "auto" (WebSocket with poll fallback, default), + # "websocket" (require WS; fail connect when it can't authenticate), + # or "poll" (CLI polling only). Env (BUZZ_TRANSPORT) overrides + # config.yaml. + _transport = ( + os.getenv("BUZZ_TRANSPORT") or str(extra.get("transport", "auto") or "auto") + ).strip().lower() + self.transport = _transport if _transport in ("auto", "websocket", "poll") else "auto" + # Auth: entries may be hex pubkeys or npubs; normalized to hex raw_allowed = os.getenv("BUZZ_ALLOWED_USERS") or extra.get("allowed_users", []) if isinstance(raw_allowed, str): @@ -360,6 +399,10 @@ class BuzzAdapter(BasePlatformAdapter): # Runtime state self._poll_task: Optional[asyncio.Task] = None + self._ws_task: Optional[asyncio.Task] = None + self._ws_ready: Optional[asyncio.Event] = None + self._ws_active = False # True while the WS loop owns inbound delivery + self._membership_since = 0 self._lock_key: Optional[str] = None # channel_id -> {"chat_type", "last_ts", "seen": OrderedDict[event_id, None]} self._channel_state: Dict[str, dict] = {} @@ -471,19 +514,37 @@ class BuzzAdapter(BasePlatformAdapter): await self._seed_channel(channel_id, chat_type="group") await self._discover_dms(seed=True) - self._poll_task = asyncio.create_task(self._poll_loop()) + # Inbound transport: prefer the NIP-42-authenticated WebSocket + # subscription (push, near-zero latency); fall back to CLI polling + # when the WS can't be established (transport="auto") or when the + # user pinned transport="poll". + transport_used = "poll" + if self.transport in ("auto", "websocket"): + if await self._start_websocket(): + transport_used = "websocket" + elif self.transport == "websocket": + self._set_fatal_error( + "ws_auth_failed", + "Buzz WebSocket transport did not authenticate (transport=websocket)", + retryable=True, + ) + await self.disconnect() + return False + if transport_used == "poll": + self._poll_task = asyncio.create_task(self._poll_loop()) self._mark_connected() logger.info( - "Buzz: connected to %s as %s, watching %d channel(s), poll interval %.1fs", + "Buzz: connected to %s as %s, watching %d channel(s) via %s%s", self.relay_url, self._display_name or self._self_npub[:16], len(self._channel_state), - self.poll_interval, + transport_used, + "" if transport_used == "websocket" else f", poll interval {self.poll_interval:.1f}s", ) return True async def disconnect(self) -> None: - """Stop the poll loop and drop runtime state.""" + """Stop the inbound transport and drop runtime state.""" self._mark_disconnected() lock_key = getattr(self, "_lock_key", None) if lock_key: @@ -494,6 +555,14 @@ class BuzzAdapter(BasePlatformAdapter): except Exception: pass self._lock_key = None + self._ws_active = False + if self._ws_task and not self._ws_task.done(): + self._ws_task.cancel() + try: + await self._ws_task + except asyncio.CancelledError: + pass + self._ws_task = None if self._poll_task and not self._poll_task.done(): self._poll_task.cancel() try: @@ -626,6 +695,185 @@ class BuzzAdapter(BasePlatformAdapter): pass return {"name": name or chat_id, "type": chat_type, "chat_id": chat_id} + # ── Inbound: WebSocket transport (NIP-42 authenticated) ────────────── + # + # Push transport contributed in PR #73636 by @ScaleLeanChris, adapted to + # dispatch through the same _handle_event() machinery as the poll loop so + # de-dupe, mention gating, DM latching, and the allow-list behave + # identically on both transports. + + def _websocket_url(self) -> str: + parsed = urlsplit(self.relay_url.strip()) + scheme = {"http": "ws", "https": "wss"}.get(parsed.scheme, parsed.scheme) + if scheme not in ("ws", "wss") or not parsed.netloc: + raise ValueError("Buzz relay URL must use http(s) or ws(s)") + return urlunsplit((scheme, parsed.netloc, parsed.path or "", parsed.query, "")) + + async def _start_websocket(self) -> bool: + """Start the WS loop; True when it authenticates within the timeout.""" + try: + import websockets # noqa: F401 (availability probe) + + self._websocket_url() + except Exception as e: + 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()) + self._ws_task = asyncio.create_task(self._websocket_loop()) + try: + await asyncio.wait_for(self._ws_ready.wait(), timeout=_WS_AUTH_TIMEOUT + 5) + except (asyncio.TimeoutError, TimeoutError): + logger.warning("Buzz: WebSocket did not authenticate in time") + self._ws_active = False + if self._ws_task and not self._ws_task.done(): + self._ws_task.cancel() + try: + await self._ws_task + except asyncio.CancelledError: + pass + self._ws_task = None + return False + return True + + async def _authenticate_websocket(self, websocket) -> None: + """NIP-42: wait for the relay's AUTH challenge, answer with a signed + kind-22242 event (plus the optional NIP-OA owner-attestation tag from + BUZZ_AUTH_TAG), and wait for the OK acknowledgment.""" + build_auth_event = _load_nostr_auth().build_auth_event + + raw = await asyncio.wait_for(websocket.recv(), timeout=_WS_AUTH_TIMEOUT) + message = json.loads(raw) + if not isinstance(message, list) or len(message) < 2 or message[0] != "AUTH": + raise ConnectionError("Buzz relay did not send a NIP-42 AUTH challenge") + event = build_auth_event( + private_key=self._private_key, + challenge=str(message[1]), + relay_url=self._websocket_url(), + auth_tag_json=os.getenv("BUZZ_AUTH_TAG", ""), + ) + await websocket.send(json.dumps(["AUTH", event], separators=(",", ":"))) + while True: + raw = await asyncio.wait_for(websocket.recv(), timeout=_WS_AUTH_TIMEOUT) + response = json.loads(raw) + if not isinstance(response, list) or not response: + continue + if response[0] == "OK" and len(response) >= 4 and response[1] == event["id"]: + if response[2] is True: + return + raise ConnectionError(f"Buzz WebSocket AUTH rejected: {response[3]}") + if response[0] in ("NOTICE", "CLOSED"): + 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: + state = self._channel_state.get(channel_id) or {} + since = max(int(state.get("last_ts") or time.time()) - 1, 0) + request = [ + "REQ", + subscription_id, + {"kinds": [_CHAT_KIND], "#h": [channel_id], "since": since}, + ] + 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.""" + subscriptions: Dict[str, Optional[str]] = {} + for index, channel_id in enumerate(list(self._channel_state)): + subscription_id = f"hermes-buzz-{index}" + subscriptions[subscription_id] = channel_id + await self._send_channel_subscription(websocket, subscription_id, channel_id) + if self._self_pubkey: + request = [ + "REQ", + _WS_MEMBERSHIP_SUB_ID, + { + "kinds": [_WS_MEMBERSHIP_KIND], + "#p": [self._self_pubkey], + "since": max(self._membership_since - 1, 0), + }, + ] + await websocket.send(json.dumps(request, separators=(",", ":"))) + subscriptions[_WS_MEMBERSHIP_SUB_ID] = None + 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)) + before = set(self._channel_state) + 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) + + async def _websocket_loop(self) -> None: + """Persistent authenticated subscription with bounded reconnect + backoff. Events route through _handle_event() — identical semantics + to the poll loop. On reconnect, per-channel `since` filters resume + from the last observed timestamps (same-second overlap de-duped by + event id).""" + import websockets + + backoff = 1.0 + try: + while True: + try: + async with websockets.connect( + self._websocket_url(), + open_timeout=_WS_AUTH_TIMEOUT, + close_timeout=5, + ping_interval=20, + ping_timeout=20, + max_size=_WS_MAX_MESSAGE_BYTES, + ) as websocket: + await self._authenticate_websocket(websocket) + subscriptions = await self._subscribe_websocket(websocket) + self._ws_active = True + if self._ws_ready is not None: + self._ws_ready.set() + backoff = 1.0 + async for raw in websocket: + try: + message = json.loads(raw) + except (ValueError, TypeError): + logger.warning("Buzz: ignoring malformed WebSocket frame") + continue + if not isinstance(message, list) or not message: + continue + if message[0] == "EVENT" and len(message) >= 3: + subscription_id = str(message[1]) + event = message[2] + if not isinstance(event, dict): + continue + if subscription_id == _WS_MEMBERSHIP_SUB_ID: + await self._handle_membership_event(websocket, subscriptions, event) + continue + channel_id = subscriptions.get(subscription_id) + state = self._channel_state.get(channel_id or "") + if channel_id and state is not None: + await self._handle_event(channel_id, state, event) + self._trim_seen(state) + elif message[0] == "CLOSED": + detail = message[-1] if len(message) > 2 else "subscription closed" + raise ConnectionError(str(detail)) + elif message[0] == "NOTICE": + logger.warning("Buzz: relay notice: %s", message[-1]) + except asyncio.CancelledError: + raise + except Exception as e: + self._ws_active = False + logger.warning("Buzz: WebSocket disconnected; retrying in %.1fs: %s", backoff, e) + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 30.0) + finally: + self._ws_active = False + # ── Inbound polling ─────────────────────────────────────────────────── async def _poll_loop(self) -> None: @@ -1021,6 +1269,7 @@ def _apply_yaml_config(yaml_cfg: dict, buzz_cfg: dict) -> Optional[dict]: "relay_url": "BUZZ_RELAY_URL", "cli_path": "BUZZ_CLI_PATH", "home_channel": "BUZZ_HOME_CHANNEL", + "transport": "BUZZ_TRANSPORT", } for src, env in _str_keys.items(): val = extra.get(src) diff --git a/plugins/platforms/buzz/plugin.yaml b/plugins/platforms/buzz/plugin.yaml index a988d6ee1c08..5bc8bedbf9cd 100644 --- a/plugins/platforms/buzz/plugin.yaml +++ b/plugins/platforms/buzz/plugin.yaml @@ -21,6 +21,14 @@ requires_env: prompt: "Nostr private key (nsec or hex)" password: true optional_env: + - name: BUZZ_TRANSPORT + description: "Inbound transport: auto (WebSocket w/ poll fallback, default), websocket, or poll" + prompt: "Transport (auto/websocket/poll)" + password: false + - name: BUZZ_AUTH_TAG + description: "Optional NIP-OA owner-attestation auth tag JSON for NIP-42 WebSocket auth" + prompt: "NIP-OA auth tag JSON (or empty)" + password: false - name: BUZZ_CHANNELS description: "Comma-separated channel UUIDs to watch (default: all joined channels)" prompt: "Channel UUIDs (comma-separated)" diff --git a/tests/gateway/test_buzz_websocket.py b/tests/gateway/test_buzz_websocket.py new file mode 100644 index 000000000000..8935b74664c2 --- /dev/null +++ b/tests/gateway/test_buzz_websocket.py @@ -0,0 +1,244 @@ +"""Tests for the Buzz WebSocket transport (NIP-42) and Nostr signing module. + +The signing module and WS transport were contributed in PR #73636 by +@ScaleLeanChris and consolidated onto the merged poll-based adapter; these +tests cover the crypto (against the official BIP-340 vector) and the WS +lifecycle as wired into BuzzAdapter. +""" + +import asyncio +import json +import time + +import pytest + +from tests.gateway._plugin_adapter_loader import load_plugin_adapter + +_buzz_mod = load_plugin_adapter("buzz") +BuzzAdapter = _buzz_mod.BuzzAdapter + +import importlib.util as _ilu +from pathlib import Path as _Path + +_auth_path = _Path(_buzz_mod.__file__).with_name("nostr_auth.py") +_spec = _ilu.spec_from_file_location("plugin_adapter_buzz_nostr_auth", _auth_path) +nostr_auth = _ilu.module_from_spec(_spec) +_spec.loader.exec_module(nostr_auth) + +SELF_PUBKEY = "9fd5c7ba6d3ef224da78f541e0fcb9c50f72cc63edb19aae76ac6a0474dfa860" +# BIP-340 test vector 0 private key +TEST_PRIVATE_KEY = "00" * 31 + "03" +CHANNEL = "ccc2bc1a-7a82-5a8f-8c4e-57a070cbe7cd" + + +def _make_adapter(extra=None): + from gateway.config import PlatformConfig + + cfg = PlatformConfig(enabled=True, extra={"relay_url": "https://test.relay", **(extra or {})}) + adapter = BuzzAdapter(cfg) + adapter._self_pubkey = SELF_PUBKEY + adapter._private_key = TEST_PRIVATE_KEY + adapter._display_name = "Chip" + return adapter + + +# ── nostr_auth: BIP-340 / NIP-42 ────────────────────────────────────────── + + +def test_schnorr_sign_matches_official_bip340_vector_zero(): + signature = nostr_auth.schnorr_sign( + bytes(32), TEST_PRIVATE_KEY, auxiliary_randomness=bytes(32) + ) + assert nostr_auth.public_key_hex(TEST_PRIVATE_KEY).upper() == ( + "F9308A019258C31049344F85F89D5229B531C845836F99B08601F113BCE036F9" + ) + assert signature.hex().upper() == ( + "E907831F80848D1069A5371B402410364BDF1C5F8307B0084C55F1CE2DCA8215" + "25F66A4A85EA8B71E482A74F382D2CE5EBEEE8FDB2172F477DF4900D310536C0" + ) + + +def test_decode_private_key_rejects_bad_input(): + with pytest.raises(ValueError): + nostr_auth.decode_private_key("not-a-key") + with pytest.raises(ValueError): + nostr_auth.decode_private_key("00" * 32) # zero — outside range + with pytest.raises(ValueError): + nostr_auth.decode_private_key("nsec1qqqqqqqq") # bad checksum/length + + +def test_build_auth_event_shape_and_owner_tag(): + tag = json.dumps(["auth", "b" * 64, "", "c" * 128]) + event = nostr_auth.build_auth_event( + private_key=TEST_PRIVATE_KEY, + challenge="challenge-1", + relay_url="wss://relay.example", + auth_tag_json=tag, + created_at=1_700_000_000, + auxiliary_randomness=bytes(32), + ) + assert event["kind"] == 22242 + assert ["relay", "wss://relay.example"] in event["tags"] + assert ["challenge", "challenge-1"] in event["tags"] + assert ["auth", "b" * 64, "", "c" * 128] in event["tags"] + assert len(bytes.fromhex(event["sig"])) == 64 + assert event["pubkey"] == nostr_auth.public_key_hex(TEST_PRIVATE_KEY) + + +def test_build_auth_event_rejects_malformed_auth_tag(): + with pytest.raises(ValueError): + nostr_auth.build_auth_event( + private_key=TEST_PRIVATE_KEY, + challenge="c", + relay_url="wss://r", + auth_tag_json='{"not": "a list"}', + ) + + +# ── Adapter WS wiring ───────────────────────────────────────────────────── + + +class _FakeWebSocket: + """Replays a NIP-42 handshake: AUTH challenge, then OK for the reply.""" + + def __init__(self): + self.sent = [] + + async def recv(self): + if self.sent: + auth_event = self.sent[0][1] + return json.dumps(["OK", auth_event["id"], True, "authenticated"]) + return json.dumps(["AUTH", "relay-challenge"]) + + async def send(self, raw): + self.sent.append(json.loads(raw)) + + +def test_transport_config_parsing(): + assert _make_adapter().transport == "auto" + assert _make_adapter({"transport": "poll"}).transport == "poll" + assert _make_adapter({"transport": "websocket"}).transport == "websocket" + assert _make_adapter({"transport": "bogus"}).transport == "auto" + + +def test_websocket_url_converts_rest_schemes(): + adapter = _make_adapter() + adapter.relay_url = "https://community.example/path" + assert adapter._websocket_url() == "wss://community.example/path" + adapter.relay_url = "http://localhost:3000" + assert adapter._websocket_url() == "ws://localhost:3000" + adapter.relay_url = "ftp://nope" + with pytest.raises(ValueError): + adapter._websocket_url() + + +@pytest.mark.asyncio +async def test_websocket_auth_signs_nip42_challenge(monkeypatch): + adapter = _make_adapter() + monkeypatch.setenv("BUZZ_AUTH_TAG", json.dumps(["auth", "b" * 64, "", "c" * 128])) + ws = _FakeWebSocket() + await adapter._authenticate_websocket(ws) + frame = ws.sent[0] + assert frame[0] == "AUTH" + event = frame[1] + assert event["kind"] == 22242 + assert ["challenge", "relay-challenge"] in event["tags"] + assert ["auth", "b" * 64, "", "c" * 128] in event["tags"] + + +@pytest.mark.asyncio +async def test_websocket_auth_raises_on_rejection(): + adapter = _make_adapter() + + class RejectingWs(_FakeWebSocket): + async def recv(self): + if self.sent: + auth_event = self.sent[0][1] + return json.dumps(["OK", auth_event["id"], False, "denied"]) + return json.dumps(["AUTH", "relay-challenge"]) + + with pytest.raises(ConnectionError): + await adapter._authenticate_websocket(RejectingWs()) + + +@pytest.mark.asyncio +async def test_subscriptions_resume_from_channel_high_water_marks(): + adapter = _make_adapter() + adapter._channel_state = { + "general-id": {"chat_type": "group", "last_ts": 100, "seen": {}}, + "dm-one": {"chat_type": "dm", "last_ts": 200, "seen": {}}, + } + adapter._membership_since = 300 + ws = _FakeWebSocket() + subscriptions = await adapter._subscribe_websocket(ws) + assert subscriptions["hermes-buzz-0"] == "general-id" + assert subscriptions["hermes-buzz-1"] == "dm-one" + assert subscriptions[_buzz_mod._WS_MEMBERSHIP_SUB_ID] is None + assert ws.sent[0][2]["#h"] == ["general-id"] + assert ws.sent[0][2]["since"] == 99 + assert ws.sent[1][2]["since"] == 199 + assert ws.sent[2][2]["kinds"] == [_buzz_mod._WS_MEMBERSHIP_KIND] + assert ws.sent[2][2]["since"] == 299 + + +@pytest.mark.asyncio +async def test_ws_events_route_through_handle_event_with_dedup(): + """WS events use the same _handle_event pipeline as the poll loop — + same de-dupe, same mention gate, same DM semantics.""" + adapter = _make_adapter() + adapter._channel_state = { + CHANNEL: {"chat_type": "dm", "last_ts": 0, "seen": __import__("collections").OrderedDict()} + } + received = [] + + async def handle_message(event): + received.append(event) + + adapter.handle_message = handle_message + adapter._message_handler = handle_message # dispatch gate requires a handler + adapter._user_names["b" * 64] = "user" + event = { + "id": "evt-1", + "kind": 9, + "pubkey": "b" * 64, + "content": "hello", + "created_at": 1_700_000_000, + "tags": [], + } + state = adapter._channel_state[CHANNEL] + await adapter._handle_event(CHANNEL, state, event) + await adapter._handle_event(CHANNEL, state, event) + assert len(received) == 1 + assert state["last_ts"] == 1_700_000_000 + + +@pytest.mark.asyncio +async def test_start_websocket_times_out_and_cleans_up(monkeypatch): + adapter = _make_adapter() + monkeypatch.setattr(_buzz_mod, "_WS_AUTH_TIMEOUT", 0.0) + + async def never_ready(): + await asyncio.sleep(3600) + + monkeypatch.setattr(adapter, "_websocket_loop", never_ready) + assert await adapter._start_websocket() is False + assert adapter._ws_task is None + assert adapter._ws_active is False + + +@pytest.mark.asyncio +async def test_membership_event_subscribes_new_conversations(): + adapter = _make_adapter() + adapter._channel_state = {"old-chan": {"chat_type": "group", "last_ts": 5, "seen": {}}} + + async def fake_discover(seed): + adapter._channel_state["new-dm"] = {"chat_type": "dm", "last_ts": 0, "seen": {}} + + adapter._discover_dms = fake_discover + ws = _FakeWebSocket() + subscriptions = {"hermes-buzz-0": "old-chan"} + await adapter._handle_membership_event( + ws, subscriptions, {"created_at": int(time.time()), "tags": []} + ) + assert "new-dm" in subscriptions.values() + assert any(f[2].get("#h") == ["new-dm"] for f in ws.sent) diff --git a/website/docs/user-guide/messaging/buzz.md b/website/docs/user-guide/messaging/buzz.md index 386a760245ae..848e21c0532e 100644 --- a/website/docs/user-guide/messaging/buzz.md +++ b/website/docs/user-guide/messaging/buzz.md @@ -1,9 +1,11 @@ # Buzz -The Buzz adapter connects Hermes to a [Buzz](https://github.com/block/buzz) community — Block's open-source human+agent collaboration platform built on the Nostr protocol — and relays messages between Buzz channels (or DMs) and the agent. The adapter is pure Python stdlib: it shells out to the `buzz` CLI binary ("JSON in, JSON out") instead of speaking Nostr itself, so **no Python packages are required** — just the `buzz` binary. +The Buzz adapter connects Hermes to a [Buzz](https://github.com/block/buzz) community — Block's open-source human+agent collaboration platform built on the Nostr protocol — and relays messages between Buzz channels (or DMs) and the agent. Outbound traffic shells out to the `buzz` CLI binary ("JSON in, JSON out"); inbound uses a native Nostr WebSocket subscription (via the already-bundled `websockets` package) with CLI polling as fallback. **No extra Python packages are required** — just the `buzz` binary. Buzz renders markdown, so agent replies keep their formatting. Images are delivered as uploads (local files) or links (URLs). Replies can thread onto an existing message via its event id. +Inbound messages arrive over a persistent NIP-42-authenticated Nostr WebSocket subscription by default (near-instant delivery), with automatic fallback to CLI polling when the WebSocket can't be established. Outbound messages always go through the `buzz` CLI. Control it with `transport` / `BUZZ_TRANSPORT`: `auto` (default), `websocket` (require WS, fail otherwise), or `poll`. If your relay membership uses NIP-OA owner attestation, set `BUZZ_AUTH_TAG` to the four-string auth tag JSON. + > Run `hermes gateway setup` and pick **Buzz** for a guided walk-through. ## Prerequisites