From 66fc2e2a92f53a826d836c167fc13bc68c5030c1 Mon Sep 17 00:00:00 2001 From: Rob Zehner Date: Sat, 25 Jul 2026 19:26:57 +0000 Subject: [PATCH] feat(plugin): add Buzz (Block/Nostr) platform adapter Plugin-path adapter (zero core changes) connecting Hermes to a Buzz community relay via the buzz CLI binary (JSON in/out, arg-list exec, key passed via env only). Inbound uses a poll loop with per-channel high-water marks seeded from newest (no history replay), event-id de-dupe, self-echo suppression by pubkey, and mention gating in channels (DMs always dispatch). Registers env_enablement, cron home-channel delivery, and an out-of-process standalone sender, mirroring the IRC plugin. Verified against a live relay: connect -> send -> poll -> MessageEvent round-trip, self-echo suppressed, clean disconnect. Known limitation: polled inbound (default 4s); a websocket transport (buzz-ws-client) is a future optimization. Co-Authored-By: Claude Fable 5 --- plugins/platforms/buzz/__init__.py | 3 + plugins/platforms/buzz/adapter.py | 973 +++++++++++++++++++++ plugins/platforms/buzz/plugin.yaml | 51 ++ tests/gateway/test_buzz_adapter.py | 640 ++++++++++++++ website/docs/user-guide/messaging/buzz.md | 82 ++ website/docs/user-guide/messaging/index.md | 2 + 6 files changed, 1751 insertions(+) create mode 100644 plugins/platforms/buzz/__init__.py create mode 100644 plugins/platforms/buzz/adapter.py create mode 100644 plugins/platforms/buzz/plugin.yaml create mode 100644 tests/gateway/test_buzz_adapter.py create mode 100644 website/docs/user-guide/messaging/buzz.md diff --git a/plugins/platforms/buzz/__init__.py b/plugins/platforms/buzz/__init__.py new file mode 100644 index 000000000000..d4f1d7bf0e3f --- /dev/null +++ b/plugins/platforms/buzz/__init__.py @@ -0,0 +1,3 @@ +from .adapter import register + +__all__ = ["register"] diff --git a/plugins/platforms/buzz/adapter.py b/plugins/platforms/buzz/adapter.py new file mode 100644 index 000000000000..531b2216be5e --- /dev/null +++ b/plugins/platforms/buzz/adapter.py @@ -0,0 +1,973 @@ +""" +Buzz Platform Adapter for Hermes Agent. + +A plugin-based gateway adapter that connects to a Buzz community relay +(Block's open-source human+agent collaboration platform, built on the +Nostr protocol) and relays messages to/from the Hermes agent. + +The adapter does not speak Nostr itself — it shells out to the ``buzz`` +CLI binary ("JSON in, JSON out") via ``asyncio.create_subprocess_exec``. +Inbound delivery uses a poll loop (the CLI is request/response); see the +"Known limitations" note in the platform docs. + +Configuration in config.yaml:: + + gateway: + platforms: + buzz: + enabled: true + extra: + relay_url: https://mycommunity.communities.buzz.xyz + channels: # channel UUIDs to watch (empty = all joined) + - ccc2bc1a-7a82-5a8f-8c4e-57a070cbe7cd + home_channel: ccc2bc1a-7a82-5a8f-8c4e-57a070cbe7cd + poll_interval: 4 # seconds between poll sweeps + cli_path: "" # path to the buzz binary (default: PATH, then ~/bin/buzz) + credentials_file: "" # JSON file holding the nsec (fallback for BUZZ_PRIVATE_KEY) + allowed_users: [] # empty = allow all; entries are hex pubkeys or npubs + +Or via environment variables (overrides config.yaml): + BUZZ_RELAY_URL, BUZZ_CHANNELS, BUZZ_HOME_CHANNEL, BUZZ_POLL_INTERVAL, + BUZZ_CLI_PATH, BUZZ_CREDENTIALS_FILE, BUZZ_ALLOWED_USERS, + BUZZ_ALLOW_ALL_USERS + +The only secret is BUZZ_PRIVATE_KEY (nsec or hex) — it belongs in +``~/.hermes/.env``. It is passed to the CLI via the subprocess +environment and is never logged. +""" + +import asyncio +import json +import logging +import os +import re +import shutil +import time +from collections import OrderedDict +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + +from gateway.platforms.base import ( + BasePlatformAdapter, + SendResult, + MessageEvent, + MessageType, +) +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. +_CHAT_KIND = 9 +# 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 ``buzz dms list`` every N poll sweeps to pick up new conversations. +_DM_DISCOVERY_EVERY = 5 + +_DEFAULT_POLL_INTERVAL = 4.0 +_MIN_POLL_INTERVAL = 1.0 +_CLI_TIMEOUT = 30.0 + +# 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() + + +# --------------------------------------------------------------------------- +# bech32 (BIP-173) helpers — used to convert between npub and hex pubkeys so +# mention detection and allow-lists accept either form. Pure stdlib. +# --------------------------------------------------------------------------- + +_BECH32_CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l" + + +def _bech32_polymod(values: List[int]) -> int: + generator = (0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3) + chk = 1 + for value in values: + top = chk >> 25 + chk = (chk & 0x1FFFFFF) << 5 ^ value + for i in range(5): + chk ^= generator[i] if ((top >> i) & 1) else 0 + return chk + + +def _bech32_hrp_expand(hrp: str) -> List[int]: + return [ord(c) >> 5 for c in hrp] + [0] + [ord(c) & 31 for c in hrp] + + +def _convertbits(data, frombits: int, tobits: int, pad: bool = True) -> Optional[List[int]]: + acc = 0 + bits = 0 + ret = [] + maxv = (1 << tobits) - 1 + for value in data: + if value < 0 or (value >> frombits): + return None + acc = (acc << frombits) | value + bits += frombits + while bits >= tobits: + bits -= tobits + ret.append((acc >> bits) & maxv) + if pad: + if bits: + ret.append((acc << (tobits - bits)) & maxv) + elif bits >= frombits or ((acc << (tobits - bits)) & maxv): + return None + return ret + + +def hex_to_npub(pubkey_hex: str) -> Optional[str]: + """Encode a 64-char hex pubkey as an ``npub1…`` bech32 string.""" + try: + raw = bytes.fromhex(pubkey_hex) + except ValueError: + return None + if len(raw) != 32: + return None + data = _convertbits(raw, 8, 5) + if data is None: + return None + values = _bech32_hrp_expand("npub") + data + polymod = _bech32_polymod(values + [0, 0, 0, 0, 0, 0]) ^ 1 + checksum = [(polymod >> 5 * (5 - i)) & 31 for i in range(6)] + return "npub1" + "".join(_BECH32_CHARSET[d] for d in data + checksum) + + +def npub_to_hex(npub: str) -> Optional[str]: + """Decode an ``npub1…`` bech32 string to a 64-char hex pubkey.""" + npub = npub.strip().lower() + if not npub.startswith("npub1"): + return None + data_part = npub[len("npub1"):] + try: + data = [_BECH32_CHARSET.index(c) for c in data_part] + except ValueError: + return None + if _bech32_polymod(_bech32_hrp_expand("npub") + data) != 1: + return None + decoded = _convertbits(data[:-6], 5, 8, pad=False) + if decoded is None or len(decoded) != 32: + return None + return bytes(decoded).hex() + + +def _normalize_user_ref(ref: str) -> Optional[str]: + """Normalize a user reference (hex pubkey or npub) to lowercase hex.""" + ref = (ref or "").strip().lower() + if not ref: + return None + if ref.startswith("npub1"): + return npub_to_hex(ref) + if re.fullmatch(r"[0-9a-f]{64}", ref): + return ref + return None + + +# --------------------------------------------------------------------------- +# buzz-cli invocation helpers +# --------------------------------------------------------------------------- + +def _resolve_cli_path(configured: str = "") -> str: + """Resolve the buzz CLI binary path portably. + + Order: explicit config value → ``buzz`` on PATH → ``~/bin/buzz``. + Returns "" when nothing is found so callers can raise a config error. + """ + if configured: + p = Path(configured).expanduser() + return str(p) if p.is_file() else "" + found = shutil.which("buzz") + if found: + return found + fallback = Path.home() / "bin" / "buzz" + return str(fallback) if fallback.is_file() else "" + + +def _resolve_private_key(extra: Optional[dict] = None) -> str: + """Resolve the Nostr private key: env first, then a credentials JSON. + + NEVER log the return value. + """ + key = os.getenv("BUZZ_PRIVATE_KEY", "").strip() + if key: + return key + configured = os.getenv("BUZZ_CREDENTIALS_FILE", "").strip() or (extra or {}).get("credentials_file", "") + if configured: + candidates = [Path(configured).expanduser()] + else: + try: + candidates = sorted(_DEFAULT_CREDENTIALS_DIR.glob("*credentials*.json")) + except OSError: + candidates = [] + for path in candidates: + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + if not isinstance(data, dict): + continue + for field in ("nsec", "private_key_hex", "private_key"): + value = data.get(field) + if isinstance(value, str) and value.strip(): + return value.strip() + return "" + + +async def _exec_buzz( + cli_path: str, + args: List[str], + *, + relay_url: str, + private_key: str, + input_text: Optional[str] = None, + timeout: float = _CLI_TIMEOUT, +) -> Tuple[int, str, str]: + """Run the buzz CLI with an argument list (never a shell) and return + ``(returncode, stdout, stderr)``. + + The private key travels via the subprocess environment only — it never + appears in argv, so process listings and error logs stay clean. + """ + env = os.environ.copy() + env["BUZZ_RELAY_URL"] = relay_url + env["BUZZ_PRIVATE_KEY"] = private_key + proc = await asyncio.create_subprocess_exec( + cli_path, + *args, + stdin=asyncio.subprocess.PIPE if input_text is not None else asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=env, + ) + try: + stdout, stderr = await asyncio.wait_for( + proc.communicate(input_text.encode("utf-8") if input_text is not None else None), + timeout=timeout, + ) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + return 124, "", json.dumps({"error": "timeout", "message": f"buzz {args[0] if args else ''} timed out after {timeout}s"}) + return ( + proc.returncode if proc.returncode is not None else 4, + stdout.decode("utf-8", errors="replace"), + stderr.decode("utf-8", errors="replace"), + ) + + +def _cli_error_message(stderr: str, returncode: int) -> str: + """Extract the human-readable message from the CLI's JSON error contract. + + stderr is ``{"error": "", "message": ""}`` on failure; + fall back to the raw (stripped) stderr when it isn't JSON. + """ + text = (stderr or "").strip() + try: + data = json.loads(text) + if isinstance(data, dict) and data.get("message"): + return f"{data.get('error', 'error')}: {data['message']} (exit {returncode})" + except ValueError: + pass + return text or f"buzz CLI failed with exit code {returncode}" + + +def _parse_json_list(stdout: str) -> List[dict]: + """Parse CLI stdout expected to be a JSON array of objects.""" + try: + data = json.loads(stdout or "[]") + except ValueError: + return [] + if not isinstance(data, list): + return [] + return [item for item in data if isinstance(item, dict)] + + +# --------------------------------------------------------------------------- +# Buzz Adapter +# --------------------------------------------------------------------------- + +class BuzzAdapter(BasePlatformAdapter): + """Poll-based Buzz adapter implementing the BasePlatformAdapter interface. + + Instantiated by the adapter_factory passed to register_platform(). + """ + + def __init__(self, config, **kwargs): + platform = Platform("buzz") + super().__init__(config=config, platform=platform) + + extra = getattr(config, "extra", {}) or {} + self._extra = extra + + # Connection settings (env vars override config.yaml) + self.relay_url = (os.getenv("BUZZ_RELAY_URL") or extra.get("relay_url", "")).strip() + self.cli_path = _resolve_cli_path( + os.getenv("BUZZ_CLI_PATH", "").strip() or str(extra.get("cli_path", "") or "") + ) + + # Channels to watch: env csv > extra list/csv; empty = all joined channels + raw_channels = os.getenv("BUZZ_CHANNELS") or extra.get("channels", []) + if isinstance(raw_channels, str): + raw_channels = raw_channels.split(",") + self.channels: List[str] = [c.strip() for c in raw_channels if isinstance(c, str) and c.strip()] + + self.home_channel = (os.getenv("BUZZ_HOME_CHANNEL") or str(extra.get("home_channel", "") or "")).strip() + + try: + interval = float(os.getenv("BUZZ_POLL_INTERVAL") or extra.get("poll_interval", _DEFAULT_POLL_INTERVAL)) + except (TypeError, ValueError): + interval = _DEFAULT_POLL_INTERVAL + self.poll_interval = max(_MIN_POLL_INTERVAL, interval) + + # 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): + raw_allowed = raw_allowed.split(",") + self._allowed_pubkeys: set = { + normalized + for entry in raw_allowed + if isinstance(entry, str) and (normalized := _normalize_user_ref(entry)) + } + + # Secret — resolved lazily (never at import/registration time and + # never logged). connect() re-resolves it to fail fast with a clear + # error when it is missing. + self._private_key: str = "" + + # Identity — filled in by connect() from ``buzz users get`` + self._self_pubkey: str = "" + self._self_npub: str = "" + self._display_name: str = "" + + # Runtime state + self._poll_task: Optional[asyncio.Task] = None + # channel_id -> {"chat_type", "last_ts", "seen": OrderedDict[event_id, None]} + self._channel_state: Dict[str, dict] = {} + self._channel_names: Dict[str, str] = {} + self._user_names: Dict[str, str] = {} + self._poll_count = 0 + + @property + def name(self) -> str: + return "Buzz" + + # ── buzz-cli plumbing ───────────────────────────────────────────────── + + async def _run_cli(self, args: List[str], *, input_text: Optional[str] = None) -> Tuple[int, str, str]: + if not self._private_key: + self._private_key = _resolve_private_key(self._extra) + return await _exec_buzz( + self.cli_path, + args, + relay_url=self.relay_url, + private_key=self._private_key, + input_text=input_text, + ) + + # ── Connection lifecycle ────────────────────────────────────────────── + + async def connect(self, *, is_reconnect: bool = False) -> bool: + """Verify relay credentials, seed high-water marks, start polling.""" + if not self.relay_url: + logger.error("Buzz: relay URL must be configured") + self._set_fatal_error("config_missing", "BUZZ_RELAY_URL must be set", retryable=False) + return False + if not self.cli_path: + logger.error("Buzz: buzz CLI binary not found (set BUZZ_CLI_PATH or put 'buzz' on PATH)") + self._set_fatal_error("cli_missing", "buzz CLI binary not found", retryable=False) + return False + self._private_key = _resolve_private_key(self._extra) + if not self._private_key: + logger.error("Buzz: no private key (set BUZZ_PRIVATE_KEY or a credentials file)") + self._set_fatal_error("config_missing", "BUZZ_PRIVATE_KEY must be set", retryable=False) + return False + + # Learn our own identity: pubkey drives self-echo suppression and + # display name drives channel mention gating. + code, out, err = await self._run_cli(["users", "get"]) + if code != 0: + message = _cli_error_message(err, code) + logger.error("Buzz: failed to fetch own profile from %s — %s", self.relay_url, message) + self._set_fatal_error("connect_failed", message, retryable=code == 2) + return False + profiles = _parse_json_list(out) + if not profiles or not profiles[0].get("pubkey"): + logger.error("Buzz: 'users get' returned no profile — is the key a member of this community?") + self._set_fatal_error("connect_failed", "buzz users get returned no profile", retryable=True) + return False + self._self_pubkey = str(profiles[0]["pubkey"]).lower() + self._display_name = str(profiles[0].get("display_name") or "").strip() + self._self_npub = hex_to_npub(self._self_pubkey) or "" + + # Map channel ids to names and pick the watch set. + code, out, err = await self._run_cli(["channels", "list"]) + 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) + self._channel_names = { + str(ch.get("channel_id")): str(ch.get("name") or ch.get("channel_id")) + for ch in listed + if ch.get("channel_id") + } + 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 + + # Seed high-water marks from the newest events so a (re)start never + # replays channel history into the agent. + for channel_id in watch: + await self._seed_channel(channel_id, chat_type="group") + await self._discover_dms(seed=True) + + 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", + self.relay_url, + self._display_name or self._self_npub[:16], + len(self._channel_state), + self.poll_interval, + ) + return True + + async def disconnect(self) -> None: + """Stop the poll loop and drop runtime state.""" + self._mark_disconnected() + if self._poll_task and not self._poll_task.done(): + self._poll_task.cancel() + try: + await self._poll_task + except asyncio.CancelledError: + pass + self._poll_task = None + self._channel_state = {} + self._poll_count = 0 + + # ── Sending ─────────────────────────────────────────────────────────── + + async def send( + self, + chat_id: str, + content: str, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + if not content: + return SendResult(success=False, error="Empty message") + args = ["messages", "send", "--channel", str(chat_id), "--content", "-"] + reply_target = reply_to or (metadata or {}).get("thread_id") + if reply_target: + args += ["--reply-to", str(reply_target)] + code, out, err = await self._run_cli(args, input_text=content) + if code != 0: + return SendResult( + success=False, + error=_cli_error_message(err, code), + retryable=code == 2, + ) + try: + data = json.loads(out or "{}") + except ValueError: + data = {} + event_id = data.get("event_id") + if event_id: + # Belt-and-braces echo suppression: the poll loop already skips + # our own pubkey, but marking the id seen makes de-dupe explicit. + self._mark_seen(str(chat_id), str(event_id)) + return SendResult( + success=bool(data.get("accepted", True)), + message_id=str(event_id) if event_id else None, + raw_response=data, + ) + + async def send_typing(self, chat_id: str, metadata=None) -> None: + """Buzz has no typing indicator API — no-op.""" + pass + + async def send_image( + self, + chat_id: str, + image_url: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + """Send an image: local files upload via --file, URLs go as a link.""" + local = Path(image_url).expanduser() if not image_url.startswith(("http://", "https://")) else None + if local is not None and local.is_file(): + args = [ + "messages", "send", + "--channel", str(chat_id), + "--file", str(local), + "--content", "-", + ] + if reply_to: + args += ["--reply-to", str(reply_to)] + code, out, err = await self._run_cli(args, input_text=caption or "") + if code != 0: + return SendResult(success=False, error=_cli_error_message(err, code), retryable=code == 2) + try: + data = json.loads(out or "{}") + except ValueError: + data = {} + event_id = data.get("event_id") + if event_id: + self._mark_seen(str(chat_id), str(event_id)) + return SendResult( + success=bool(data.get("accepted", True)), + message_id=str(event_id) if event_id else None, + raw_response=data, + ) + # Markdown renders in Buzz, so a URL arrives as a clickable image link. + text = f"{caption}\n{image_url}" if caption else image_url + return await self.send(chat_id, text, reply_to=reply_to, metadata=metadata) + + async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: + chat_id = str(chat_id) + state = self._channel_state.get(chat_id) + chat_type = state["chat_type"] if state else "group" + name = self._channel_names.get(chat_id) + if name is None and self.cli_path: + code, out, _err = await self._run_cli(["channels", "get", "--channel", chat_id]) + if code == 0: + try: + data = json.loads(out or "{}") + if isinstance(data, dict) and data.get("name"): + name = str(data["name"]) + self._channel_names[chat_id] = name + except ValueError: + pass + return {"name": name or chat_id, "type": chat_type, "chat_id": chat_id} + + # ── Inbound polling ─────────────────────────────────────────────────── + + async def _poll_loop(self) -> None: + """Poll every watched channel for new events until cancelled.""" + try: + while True: + await asyncio.sleep(self.poll_interval) + self._poll_count += 1 + try: + if self._poll_count % _DM_DISCOVERY_EVERY == 0: + await self._discover_dms(seed=False) + for channel_id in list(self._channel_state): + await self._poll_channel(channel_id) + except asyncio.CancelledError: + raise + except Exception: + logger.warning("Buzz: poll sweep failed", exc_info=True) + 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.""" + 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)] + ) + 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()) + return + for event in _parse_json_list(out): + event_id = event.get("id") + created_at = int(event.get("created_at") or 0) + if event_id: + state["seen"][str(event_id)] = None + state["last_ts"] = max(state["last_ts"], created_at) + self._trim_seen(state) + + async def _discover_dms(self, *, seed: bool) -> None: + """Watch DM conversations. New ones found mid-run dispatch from their + beginning (a fresh conversation has no history worth suppressing); + ones present at startup are seeded like channels.""" + code, out, _err = await self._run_cli(["dms", "list"]) + if code != 0: + return + for dm in _parse_json_list(out): + dm_id = str(dm.get("dm_id") or "") + if not dm_id or dm_id in self._channel_state: + continue + if seed: + await self._seed_channel(dm_id, chat_type="dm") + else: + self._channel_state[dm_id] = {"chat_type": "dm", "last_ts": 0, "seen": OrderedDict()} + self._channel_names.setdefault(dm_id, "DM") + + async def _poll_channel(self, channel_id: str) -> None: + state = self._channel_state.get(channel_id) + if state is None: + return + args = ["messages", "get", "--channel", channel_id, "--limit", str(_FETCH_LIMIT)] + if state["last_ts"]: + # Nostr `since` is inclusive: same-second events are re-fetched + # and de-duped by id below. + args += ["--since", str(state["last_ts"])] + code, out, err = await self._run_cli(args) + if code != 0: + logger.debug( + "Buzz: poll of channel %s failed — %s", channel_id, _cli_error_message(err, code) + ) + return + for event in _parse_json_list(out): + await self._handle_event(channel_id, state, event) + self._trim_seen(state) + + async def _handle_event(self, channel_id: str, state: dict, event: dict) -> None: + """De-dupe, filter, and dispatch a single ``messages get`` event.""" + event_id = str(event.get("id") or "") + created_at = int(event.get("created_at") or 0) + if not event_id or event_id in state["seen"]: + return + state["seen"][event_id] = None + state["last_ts"] = max(state["last_ts"], created_at) + + if int(event.get("kind") or 0) != _CHAT_KIND: + return + pubkey = str(event.get("pubkey") or "").lower() + content = event.get("content") + if not pubkey or not isinstance(content, str) or not content.strip(): + return + + # Suppress self-echo: never dispatch our own messages back to the agent. + if pubkey == self._self_pubkey: + return + + is_dm = state["chat_type"] == "dm" + # In shared channels only respond when addressed; DMs always dispatch. + if not is_dm and not self._is_mentioned(content): + return + + # Adapter-level allow-list (the gateway applies BUZZ_ALLOWED_USERS / + # BUZZ_ALLOW_ALL_USERS centrally as well; empty list = no filter here). + if self._allowed_pubkeys and pubkey not in self._allowed_pubkeys: + logger.debug("Buzz: ignoring message from unauthorized pubkey %s…", pubkey[:8]) + return + + await self._dispatch_message( + text=content, + chat_id=channel_id, + chat_type="dm" if is_dm else "group", + user_id=pubkey, + user_name=await self._resolve_user_name(pubkey), + message_id=event_id, + created_at=created_at, + ) + + def _is_mentioned(self, content: str) -> bool: + """True when the message addresses this agent (npub, hex, or name).""" + lowered = content.lower() + if self._self_pubkey and self._self_pubkey in lowered: + return True + if self._self_npub and self._self_npub in lowered: + return True + if self._display_name: + pattern = rf"(? str: + """Resolve a pubkey to a display name (cached; falls back to npub prefix).""" + cached = self._user_names.get(pubkey) + if cached: + return cached + name = "" + code, out, _err = await self._run_cli(["users", "get", "--pubkey", pubkey]) + if code == 0: + profiles = _parse_json_list(out) + if profiles: + name = str(profiles[0].get("display_name") or "").strip() + if not name: + name = (hex_to_npub(pubkey) or pubkey)[:16] + self._user_names[pubkey] = name + return name + + @staticmethod + def _trim_seen(state: dict) -> None: + seen = state["seen"] + while len(seen) > _SEEN_CAP: + seen.popitem(last=False) + + def _mark_seen(self, channel_id: str, event_id: str) -> None: + state = self._channel_state.get(channel_id) + if state is not None: + state["seen"][event_id] = None + self._trim_seen(state) + + async def _dispatch_message( + self, + text: str, + chat_id: str, + chat_type: str, + user_id: str, + user_name: str, + message_id: str, + created_at: int, + ) -> None: + """Build a MessageEvent and hand it to the base class handler.""" + if not self._message_handler: + return + + source = self.build_source( + chat_id=chat_id, + chat_name=self._channel_names.get(chat_id, chat_id), + chat_type=chat_type, + user_id=user_id, + user_name=user_name, + ) + + event = MessageEvent( + text=text, + message_type=MessageType.TEXT, + source=source, + message_id=message_id, + timestamp=datetime.fromtimestamp(created_at) if created_at else datetime.now(), + ) + + await self.handle_message(event) + + +# --------------------------------------------------------------------------- +# Plugin registration +# --------------------------------------------------------------------------- + +def check_requirements() -> bool: + """Check if Buzz is configured: a relay URL plus a resolvable key.""" + if not os.getenv("BUZZ_RELAY_URL", "").strip(): + return False + return bool(_resolve_private_key()) + + +def validate_config(config) -> bool: + """Validate that the platform config has enough info to connect.""" + extra = getattr(config, "extra", {}) or {} + relay = os.getenv("BUZZ_RELAY_URL") or extra.get("relay_url", "") + return bool(relay and _resolve_private_key(extra)) + + +def is_connected(config) -> bool: + """Check whether Buzz is configured (env or config.yaml).""" + return validate_config(config) + + +def _env_enablement() -> Optional[dict]: + """Seed ``PlatformConfig.extra`` from env vars during gateway config load. + + Called BEFORE adapter construction so env-only setups show up in + ``hermes gateway status`` and ``get_connected_platforms()``. Returns + ``None`` when Buzz isn't minimally configured. + + The special ``home_channel`` key is handled by the core hook — it becomes + a proper ``HomeChannel`` on the ``PlatformConfig``. + """ + relay = os.getenv("BUZZ_RELAY_URL", "").strip() + if not relay or not _resolve_private_key(): + return None + seed: dict = {"relay_url": relay} + channels = os.getenv("BUZZ_CHANNELS", "").strip() + if channels: + seed["channels"] = [c.strip() for c in channels.split(",") if c.strip()] + interval = os.getenv("BUZZ_POLL_INTERVAL", "").strip() + if interval: + try: + seed["poll_interval"] = float(interval) + except ValueError: + pass + cli_path = os.getenv("BUZZ_CLI_PATH", "").strip() + if cli_path: + seed["cli_path"] = cli_path + # Home channel for deliver=buzz cron jobs; defaults to the first watched + # channel so env-only setups get a sensible target without extra config. + home = os.getenv("BUZZ_HOME_CHANNEL", "").strip() or (seed.get("channels") or [""])[0] + if home: + seed["home_channel"] = { + "chat_id": home, + "name": os.getenv("BUZZ_HOME_CHANNEL_NAME", home), + } + return seed + + +async def _standalone_send( + pconfig, + chat_id: str, + message: str, + *, + thread_id: Optional[str] = None, + media_files: Optional[List[str]] = None, + force_document: bool = False, +) -> Dict[str, Any]: + """One-shot send without a live adapter (out-of-process cron delivery). + + Used by ``tools/send_message_tool`` when ``hermes cron`` runs separately + from the gateway process. Without this hook, ``deliver=buzz`` cron jobs + fail with ``No live adapter for platform 'buzz'``. + """ + extra = getattr(pconfig, "extra", {}) or {} + relay = (os.getenv("BUZZ_RELAY_URL") or extra.get("relay_url", "")).strip() + private_key = _resolve_private_key(extra) + cli_path = _resolve_cli_path( + os.getenv("BUZZ_CLI_PATH", "").strip() or str(extra.get("cli_path", "") or "") + ) + if not relay or not private_key: + return {"error": "Buzz standalone send: BUZZ_RELAY_URL and BUZZ_PRIVATE_KEY must be configured"} + if not cli_path: + return {"error": "Buzz standalone send: buzz CLI binary not found"} + target = (chat_id or "").strip() or (os.getenv("BUZZ_HOME_CHANNEL") or str(extra.get("home_channel", "") or "")).strip() + if not target: + return {"error": "Buzz standalone send: no target channel (set BUZZ_HOME_CHANNEL)"} + + args = ["messages", "send", "--channel", target, "--content", "-"] + if thread_id: + args += ["--reply-to", str(thread_id)] + for path in media_files or []: + args += ["--file", str(path)] + try: + code, out, err = await _exec_buzz( + cli_path, args, relay_url=relay, private_key=private_key, input_text=message + ) + except asyncio.CancelledError: + raise + except OSError as e: + return {"error": f"Buzz standalone send failed to launch CLI: {e}"} + if code != 0: + return {"error": f"Buzz standalone send failed: {_cli_error_message(err, code)}"} + try: + data = json.loads(out or "{}") + except ValueError: + data = {} + return {"success": True, "message_id": str(data.get("event_id") or "")} + + +def interactive_setup() -> None: + """Interactive ``hermes gateway setup`` flow for the Buzz platform. + + Lazy-imports ``hermes_cli.setup`` helpers so the plugin stays importable + in non-CLI contexts (gateway runtime, tests). + """ + from hermes_cli.setup import ( + prompt, + prompt_yes_no, + save_env_value, + get_env_value, + print_header, + print_info, + print_warning, + print_success, + ) + + print_header("Buzz") + existing_relay = get_env_value("BUZZ_RELAY_URL") + if existing_relay: + print_info(f"Buzz: already configured (relay: {existing_relay})") + if not prompt_yes_no("Reconfigure Buzz?", False): + return + + print_info("Connect Hermes to a Buzz community (Block's Nostr-based human+agent platform).") + print_info(" Requires the buzz CLI binary and a Nostr key that is a community member.") + print() + + relay = prompt( + "Relay URL (e.g. https://mycommunity.communities.buzz.xyz)", + default=existing_relay or "", + ) + if not relay: + print_warning("Relay URL is required — skipping Buzz setup") + return + save_env_value("BUZZ_RELAY_URL", relay.strip()) + + key = prompt("Nostr private key (nsec or hex; leave blank to keep current)", password=True) + if key: + save_env_value("BUZZ_PRIVATE_KEY", key.strip()) + elif not _resolve_private_key(): + print_warning("No private key configured — set BUZZ_PRIVATE_KEY before starting the gateway") + + channels = prompt( + "Channel UUIDs to watch (comma-separated, empty = all joined channels)", + default=get_env_value("BUZZ_CHANNELS") or "", + ) + if channels: + save_env_value("BUZZ_CHANNELS", channels.replace(" ", "")) + + home = prompt( + "Home channel UUID for cron/notification delivery (optional)", + default=get_env_value("BUZZ_HOME_CHANNEL") or "", + ) + if home: + save_env_value("BUZZ_HOME_CHANNEL", home.strip()) + + print() + print_info("🔒 Access control: restrict who can talk to the agent") + allow_all = prompt_yes_no("Allow all community members to talk to the agent?", False) + if allow_all: + save_env_value("BUZZ_ALLOW_ALL_USERS", "true") + save_env_value("BUZZ_ALLOWED_USERS", "") + print_warning("⚠️ Open access — anyone in the community can command the agent.") + else: + save_env_value("BUZZ_ALLOW_ALL_USERS", "false") + allowed = prompt( + "Allowed users (comma-separated npubs or hex pubkeys, empty to deny everyone)", + default=get_env_value("BUZZ_ALLOWED_USERS") or "", + ) + save_env_value("BUZZ_ALLOWED_USERS", allowed.replace(" ", "") if allowed else "") + + print() + print_success("Buzz configuration saved to ~/.hermes/.env") + print_info("Restart the gateway for changes to take effect: hermes gateway restart") + + +def register(ctx): + """Plugin entry point: called by the Hermes plugin system.""" + ctx.register_platform( + name="buzz", + label="Buzz", + adapter_factory=lambda cfg: BuzzAdapter(cfg), + check_fn=check_requirements, + validate_config=validate_config, + is_connected=is_connected, + required_env=["BUZZ_RELAY_URL", "BUZZ_PRIVATE_KEY"], + install_hint="Requires the buzz CLI binary (https://github.com/block/buzz) on PATH or at BUZZ_CLI_PATH", + setup_fn=interactive_setup, + # Env-driven auto-configuration: seeds PlatformConfig.extra with + # relay/channels/poll interval + home_channel so env-only setups show + # up in gateway status without instantiating the adapter. + env_enablement_fn=_env_enablement, + # Cron home-channel delivery support (deliver=buzz). + cron_deliver_env_var="BUZZ_HOME_CHANNEL", + # Out-of-process cron delivery. Without this hook, deliver=buzz + # cron jobs fail with "No live adapter" when cron runs separately + # from the gateway. + standalone_sender_fn=_standalone_send, + # Auth env vars for _is_user_authorized() integration + allowed_users_env="BUZZ_ALLOWED_USERS", + allow_all_env="BUZZ_ALLOW_ALL_USERS", + # Display + emoji="🐝", + # Buzz identities are pubkeys, not phone numbers + pii_safe=False, + allow_update_command=True, + # LLM guidance + platform_hint=( + "You are collaborating in a Buzz workspace (Block's Nostr-based " + "human+agent platform). Markdown IS supported. Users address you " + "by @-mentioning your name or npub in channels; direct messages " + "reach you without a mention. Keep responses conversational." + ), + ) diff --git a/plugins/platforms/buzz/plugin.yaml b/plugins/platforms/buzz/plugin.yaml new file mode 100644 index 000000000000..a988d6ee1c08 --- /dev/null +++ b/plugins/platforms/buzz/plugin.yaml @@ -0,0 +1,51 @@ +name: buzz-platform +label: Buzz +kind: platform +version: 1.0.0 +description: > + Buzz gateway adapter for Hermes Agent. + Connects to a Buzz community relay (Block's open-source human+agent + collaboration platform built on Nostr) and relays messages between + channels/DMs and the Hermes agent. Pure stdlib — shells out to the + buzz CLI binary (JSON in/out); no Python packages required. +author: Nous Research +# ``requires_env`` entries are surfaced in ``hermes config`` UI via the +# platform-plugin env var injector in ``hermes_cli/config.py``. +requires_env: + - name: BUZZ_RELAY_URL + description: "Base URL of the Buzz community relay (e.g. https://mycommunity.communities.buzz.xyz)" + prompt: "Buzz relay URL" + password: false + - name: BUZZ_PRIVATE_KEY + description: "Nostr private key for the agent's Buzz identity (nsec or hex) — the only Buzz secret" + prompt: "Nostr private key (nsec or hex)" + password: true +optional_env: + - name: BUZZ_CHANNELS + description: "Comma-separated channel UUIDs to watch (default: all joined channels)" + prompt: "Channel UUIDs (comma-separated)" + password: false + - name: BUZZ_HOME_CHANNEL + description: "Channel UUID for cron / notification delivery (defaults to the first watched channel)" + prompt: "Home channel UUID (or empty)" + password: false + - name: BUZZ_ALLOWED_USERS + description: "Comma-separated npubs or hex pubkeys allowed to talk to the agent" + prompt: "Allowed users (comma-separated)" + password: false + - name: BUZZ_ALLOW_ALL_USERS + description: "Allow any community member to talk to the agent (true/false)" + prompt: "Allow all users? (true/false)" + password: false + - name: BUZZ_POLL_INTERVAL + description: "Seconds between inbound poll sweeps (default: 4)" + prompt: "Poll interval seconds" + password: false + - name: BUZZ_CLI_PATH + description: "Path to the buzz CLI binary (default: 'buzz' on PATH, then ~/bin/buzz)" + prompt: "buzz CLI path (or empty)" + password: false + - name: BUZZ_CREDENTIALS_FILE + description: "JSON credentials file holding the nsec (fallback when BUZZ_PRIVATE_KEY is unset)" + prompt: "Credentials file path (or empty)" + password: false diff --git a/tests/gateway/test_buzz_adapter.py b/tests/gateway/test_buzz_adapter.py new file mode 100644 index 000000000000..db1468cffa03 --- /dev/null +++ b/tests/gateway/test_buzz_adapter.py @@ -0,0 +1,640 @@ +"""Tests for the Buzz platform adapter plugin.""" + +import asyncio +import json + +import pytest +from unittest.mock import AsyncMock, MagicMock + +from tests.gateway._plugin_adapter_loader import load_plugin_adapter + +# Load plugins/platforms/buzz/adapter.py under a unique module name +# (plugin_adapter_buzz) so it cannot collide with other plugin adapters +# loaded by sibling tests in the same xdist worker. +_buzz_mod = load_plugin_adapter("buzz") + +BuzzAdapter = _buzz_mod.BuzzAdapter +hex_to_npub = _buzz_mod.hex_to_npub +npub_to_hex = _buzz_mod.npub_to_hex +_normalize_user_ref = _buzz_mod._normalize_user_ref +_cli_error_message = _buzz_mod._cli_error_message +_resolve_private_key = _buzz_mod._resolve_private_key +check_requirements = _buzz_mod.check_requirements +validate_config = _buzz_mod.validate_config +register = _buzz_mod.register +_env_enablement = _buzz_mod._env_enablement +_standalone_send = _buzz_mod._standalone_send + +# Real key pair (Chip's public identity — public information, not a secret) +SELF_PUBKEY = "9fd5c7ba6d3ef224da78f541e0fcb9c50f72cc63edb19aae76ac6a0474dfa860" +SELF_NPUB = "npub1nl2u0wnd8mezfknc74q7pl9ec58h9nrrakce4tnk434qgaxl4psqe5twr6" +OTHER_PUBKEY = "a" * 64 +CHANNEL = "ccc2bc1a-7a82-5a8f-8c4e-57a070cbe7cd" + +_ENV_VARS = ( + "BUZZ_RELAY_URL", + "BUZZ_PRIVATE_KEY", + "BUZZ_CHANNELS", + "BUZZ_HOME_CHANNEL", + "BUZZ_ALLOWED_USERS", + "BUZZ_ALLOW_ALL_USERS", + "BUZZ_POLL_INTERVAL", + "BUZZ_CLI_PATH", + "BUZZ_CREDENTIALS_FILE", +) + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch, tmp_path): + """Keep tests hermetic: no ambient Buzz env vars or real credentials.""" + for var in _ENV_VARS: + monkeypatch.delenv(var, raising=False) + monkeypatch.setattr(_buzz_mod, "_DEFAULT_CREDENTIALS_DIR", tmp_path / "no-creds") + yield + + +def _event(event_id, pubkey=OTHER_PUBKEY, content="hello", created_at=1000, kind=9): + return { + "id": event_id, + "pubkey": pubkey, + "content": content, + "created_at": created_at, + "kind": kind, + "tags": [["h", CHANNEL]], + } + + +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._self_npub = SELF_NPUB + adapter._display_name = "Chip" + adapter._private_key = "nsec1test" + return adapter + + +class _ScriptedCli: + """Fake ``_run_cli`` that routes on the buzz subcommand and records calls.""" + + def __init__(self): + self.responses = {} # (group, cmd) -> list of (code, stdout, stderr) + self.calls = [] + + def script(self, group, cmd, payload, code=0, stderr=""): + stdout = payload if isinstance(payload, str) else json.dumps(payload) + self.responses.setdefault((group, cmd), []).append((code, stdout, stderr)) + + async def __call__(self, args, *, input_text=None): + self.calls.append((list(args), input_text)) + queue = self.responses.get((args[0], args[1]), []) + if len(queue) > 1: + return queue.pop(0) + if queue: + return queue[0] + return 0, "[]", "" + + +# ── bech32 / identity helpers ───────────────────────────────────────────── + + +class TestBech32Helpers: + + def test_hex_to_npub_known_pair(self): + assert hex_to_npub(SELF_PUBKEY) == SELF_NPUB + + def test_npub_to_hex_known_pair(self): + assert npub_to_hex(SELF_NPUB) == SELF_PUBKEY + + def test_round_trip(self): + npub = hex_to_npub(OTHER_PUBKEY) + assert npub is not None and npub.startswith("npub1") + assert npub_to_hex(npub) == OTHER_PUBKEY + + def test_invalid_inputs(self): + assert hex_to_npub("not-hex") is None + assert hex_to_npub("ab") is None + assert npub_to_hex("npub1invalidchecksum") is None + assert npub_to_hex("nsec1notanpub") is None + + def test_normalize_user_ref(self): + assert _normalize_user_ref(SELF_NPUB) == SELF_PUBKEY + assert _normalize_user_ref(SELF_PUBKEY.upper()) == SELF_PUBKEY + assert _normalize_user_ref("") is None + assert _normalize_user_ref("bogus") is None + + +# ── Adapter init / config precedence ────────────────────────────────────── + + +class TestBuzzAdapterInit: + + def test_init_from_env(self, monkeypatch): + monkeypatch.setenv("BUZZ_RELAY_URL", "https://env.relay") + monkeypatch.setenv("BUZZ_CHANNELS", "aaa, bbb ,") + monkeypatch.setenv("BUZZ_POLL_INTERVAL", "7.5") + + from gateway.config import PlatformConfig + adapter = BuzzAdapter(PlatformConfig(enabled=True)) + + assert adapter.relay_url == "https://env.relay" + assert adapter.channels == ["aaa", "bbb"] + assert adapter.poll_interval == 7.5 + + def test_init_from_config_extra(self): + from gateway.config import PlatformConfig + cfg = PlatformConfig( + enabled=True, + extra={ + "relay_url": "https://cfg.relay", + "channels": ["ccc"], + "poll_interval": 2, + "home_channel": "ccc", + }, + ) + adapter = BuzzAdapter(cfg) + assert adapter.relay_url == "https://cfg.relay" + assert adapter.channels == ["ccc"] + assert adapter.poll_interval == 2.0 + assert adapter.home_channel == "ccc" + + def test_env_overrides_config(self, monkeypatch): + monkeypatch.setenv("BUZZ_RELAY_URL", "https://env.relay") + from gateway.config import PlatformConfig + adapter = BuzzAdapter(PlatformConfig(enabled=True, extra={"relay_url": "https://cfg.relay"})) + assert adapter.relay_url == "https://env.relay" + + def test_poll_interval_clamped_and_defaulted(self): + adapter = _make_adapter({"poll_interval": 0.01}) + assert adapter.poll_interval == _buzz_mod._MIN_POLL_INTERVAL + adapter = _make_adapter({"poll_interval": "garbage"}) + assert adapter.poll_interval == _buzz_mod._DEFAULT_POLL_INTERVAL + + def test_allowed_users_normalized_from_mixed_forms(self): + adapter = _make_adapter({"allowed_users": [SELF_NPUB, OTHER_PUBKEY.upper(), "junk"]}) + assert adapter._allowed_pubkeys == {SELF_PUBKEY, OTHER_PUBKEY} + + +# ── CLI error contract ──────────────────────────────────────────────────── + + +class TestCliErrorContract: + + def test_parses_json_error(self): + msg = _cli_error_message('{"error":"relay_error","message":"boom","retryable":false}', 2) + assert "relay_error" in msg and "boom" in msg and "exit 2" in msg + + def test_falls_back_to_raw_stderr(self): + assert "garbage" in _cli_error_message("garbage output", 4) + + def test_empty_stderr(self): + assert "exit code 3" in _cli_error_message("", 3) + + +# ── Seeding / high-water mark / de-dupe ─────────────────────────────────── + + +class TestPollingDedupe: + + @pytest.fixture + def adapter(self): + a = _make_adapter() + a._dispatched = [] + + async def capture(**kwargs): + a._dispatched.append(kwargs) + + a._dispatch_message = capture + a._message_handler = AsyncMock() + return a + + @pytest.mark.asyncio + async def test_seed_sets_high_water_mark_without_dispatch(self, adapter): + cli = _ScriptedCli() + cli.script("messages", "get", [ + _event("e1", content="@Chip old history", created_at=100), + _event("e2", content="@Chip newer history", created_at=200), + ]) + adapter._run_cli = cli + await adapter._seed_channel(CHANNEL, chat_type="group") + + state = adapter._channel_state[CHANNEL] + assert state["last_ts"] == 200 + assert set(state["seen"]) == {"e1", "e2"} + # Seeding must never replay history into the agent + assert adapter._dispatched == [] + + @pytest.mark.asyncio + async def test_new_event_dispatched_once(self, adapter): + cli = _ScriptedCli() + cli.script("messages", "get", [_event("e1", content="@Chip hi", created_at=100)]) + adapter._run_cli = cli + await adapter._seed_channel(CHANNEL, chat_type="group") + + # Poll 1: seeded event + a genuinely new mention + cli.responses.clear() + cli.script("messages", "get", [ + _event("e1", content="@Chip hi", created_at=100), + _event("e2", content="hey @Chip, ping", created_at=150), + ]) + await adapter._poll_channel(CHANNEL) + assert [d["message_id"] for d in adapter._dispatched] == ["e2"] + assert adapter._dispatched[0]["text"] == "hey @Chip, ping" + assert adapter._channel_state[CHANNEL]["last_ts"] == 150 + + # Poll 2: identical response — the seen-id set must de-dupe + await adapter._poll_channel(CHANNEL) + assert len(adapter._dispatched) == 1 + + @pytest.mark.asyncio + async def test_poll_uses_since_high_water_mark(self, adapter): + adapter._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 4242, "seen": {}} + cli = _ScriptedCli() + adapter._run_cli = cli + await adapter._poll_channel(CHANNEL) + args, _stdin = cli.calls[0] + assert "--since" in args + assert args[args.index("--since") + 1] == "4242" + + @pytest.mark.asyncio + async def test_self_echo_suppressed(self, adapter): + adapter._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 0, "seen": {}} + cli = _ScriptedCli() + cli.script("messages", "get", [ + _event("own", pubkey=SELF_PUBKEY, content="@Chip I mention myself", created_at=300), + ]) + adapter._run_cli = cli + await adapter._poll_channel(CHANNEL) + assert adapter._dispatched == [] + # …but the event still advances the high-water mark + assert adapter._channel_state[CHANNEL]["last_ts"] == 300 + + @pytest.mark.asyncio + async def test_non_chat_kinds_ignored(self, adapter): + adapter._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 0, "seen": {}} + cli = _ScriptedCli() + cli.script("messages", "get", [ + _event("j1", content="@Chip join event", created_at=300, kind=40002), + ]) + adapter._run_cli = cli + await adapter._poll_channel(CHANNEL) + assert adapter._dispatched == [] + + @pytest.mark.asyncio + async def test_seen_set_is_bounded(self, adapter): + adapter._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 0, "seen": {}} + from collections import OrderedDict + adapter._channel_state[CHANNEL]["seen"] = OrderedDict() + cli = _ScriptedCli() + cli.script("messages", "get", [ + _event(f"e{i}", content="unrelated chatter", created_at=i) + for i in range(_buzz_mod._SEEN_CAP + 100) + ]) + adapter._run_cli = cli + await adapter._poll_channel(CHANNEL) + assert len(adapter._channel_state[CHANNEL]["seen"]) <= _buzz_mod._SEEN_CAP + + +# ── Mention gating / DMs / authorization ────────────────────────────────── + + +class TestMentionGating: + + @pytest.fixture + def adapter(self): + a = _make_adapter() + a._dispatched = [] + + async def capture(**kwargs): + a._dispatched.append(kwargs) + + a._dispatch_message = capture + a._message_handler = AsyncMock() + a._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 0, "seen": {}} + return a + + async def _poll_with(self, adapter, *events): + cli = _ScriptedCli() + cli.script("messages", "get", list(events)) + adapter._run_cli = cli + await adapter._poll_channel(CHANNEL) + + @pytest.mark.asyncio + async def test_unaddressed_channel_message_ignored(self, adapter): + await self._poll_with(adapter, _event("e1", content="just chatting", created_at=10)) + assert adapter._dispatched == [] + + @pytest.mark.asyncio + async def test_name_mention_dispatched(self, adapter): + await self._poll_with(adapter, _event("e1", content="hey @Chip can you help?", created_at=10)) + assert len(adapter._dispatched) == 1 + + @pytest.mark.asyncio + async def test_bare_name_mention_dispatched_case_insensitive(self, adapter): + await self._poll_with(adapter, _event("e1", content="chip: what do you think?", created_at=10)) + assert len(adapter._dispatched) == 1 + + @pytest.mark.asyncio + async def test_name_inside_word_not_a_mention(self, adapter): + await self._poll_with(adapter, _event("e1", content="I love potato chips", created_at=10)) + assert adapter._dispatched == [] + + @pytest.mark.asyncio + async def test_npub_mention_dispatched(self, adapter): + await self._poll_with(adapter, _event("e1", content=f"nostr:{SELF_NPUB} hello", created_at=10)) + assert len(adapter._dispatched) == 1 + + @pytest.mark.asyncio + async def test_dm_always_dispatched(self, adapter): + adapter._channel_state[CHANNEL]["chat_type"] = "dm" + await self._poll_with(adapter, _event("e1", content="no mention here", created_at=10)) + assert len(adapter._dispatched) == 1 + assert adapter._dispatched[0]["chat_type"] == "dm" + + @pytest.mark.asyncio + async def test_allowlist_blocks_unauthorized(self, adapter): + adapter._allowed_pubkeys = {"b" * 64} + await self._poll_with(adapter, _event("e1", content="@Chip hello", created_at=10)) + assert adapter._dispatched == [] + + @pytest.mark.asyncio + async def test_allowlist_admits_npub_entry(self): + a = _make_adapter({"allowed_users": [hex_to_npub(OTHER_PUBKEY)]}) + a._dispatched = [] + + async def capture(**kwargs): + a._dispatched.append(kwargs) + + a._dispatch_message = capture + a._message_handler = AsyncMock() + a._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 0, "seen": {}} + cli = _ScriptedCli() + cli.script("messages", "get", [_event("e1", content="@Chip hello", created_at=10)]) + a._run_cli = cli + await a._poll_channel(CHANNEL) + assert len(a._dispatched) == 1 + + +# ── Sending ─────────────────────────────────────────────────────────────── + + +class TestBuzzAdapterSend: + + @pytest.mark.asyncio + async def test_send_success_via_stdin(self): + adapter = _make_adapter() + adapter._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 0, "seen": {}} + cli = _ScriptedCli() + cli.script("messages", "send", {"accepted": True, "event_id": "evt123", "message": ""}) + adapter._run_cli = cli + + result = await adapter.send(CHANNEL, "hello **markdown**") + assert result.success is True + assert result.message_id == "evt123" + + args, stdin_text = cli.calls[0] + assert args[:2] == ["messages", "send"] + assert args[args.index("--channel") + 1] == CHANNEL + # Content travels via stdin (--content -), never argv + assert args[args.index("--content") + 1] == "-" + assert stdin_text == "hello **markdown**" + # Our own event id is marked seen for echo suppression + assert "evt123" in adapter._channel_state[CHANNEL]["seen"] + + @pytest.mark.asyncio + async def test_send_reply_to(self): + adapter = _make_adapter() + cli = _ScriptedCli() + cli.script("messages", "send", {"accepted": True, "event_id": "evt124", "message": ""}) + adapter._run_cli = cli + await adapter.send(CHANNEL, "threaded", reply_to="root-event") + args, _stdin = cli.calls[0] + assert args[args.index("--reply-to") + 1] == "root-event" + + @pytest.mark.asyncio + async def test_send_failure_maps_error_contract(self): + adapter = _make_adapter() + cli = _ScriptedCli() + cli.script("messages", "send", "", code=2, + stderr='{"error":"relay_error","message":"relay unreachable"}') + adapter._run_cli = cli + result = await adapter.send(CHANNEL, "hi") + assert result.success is False + assert "relay unreachable" in result.error + assert result.retryable is True + + @pytest.mark.asyncio + async def test_send_empty_message_rejected(self): + adapter = _make_adapter() + result = await adapter.send(CHANNEL, "") + assert result.success is False + + @pytest.mark.asyncio + async def test_send_image_url_falls_back_to_link(self): + adapter = _make_adapter() + cli = _ScriptedCli() + cli.script("messages", "send", {"accepted": True, "event_id": "evt125", "message": ""}) + adapter._run_cli = cli + result = await adapter.send_image(CHANNEL, "https://example.com/cat.png", caption="a cat") + assert result.success is True + _args, stdin_text = cli.calls[0] + assert "https://example.com/cat.png" in stdin_text + assert "a cat" in stdin_text + + @pytest.mark.asyncio + async def test_send_image_local_file_uses_file_flag(self, tmp_path): + img = tmp_path / "shot.png" + img.write_bytes(b"\x89PNG fake") + adapter = _make_adapter() + cli = _ScriptedCli() + cli.script("messages", "send", {"accepted": True, "event_id": "evt126", "message": ""}) + adapter._run_cli = cli + result = await adapter.send_image(CHANNEL, str(img), caption="screenshot") + assert result.success is True + args, _stdin = cli.calls[0] + assert args[args.index("--file") + 1] == str(img) + + +# ── Lifecycle ───────────────────────────────────────────────────────────── + + +class TestBuzzAdapterLifecycle: + + @pytest.mark.asyncio + async def test_connect_fails_without_relay_url(self): + from gateway.config import PlatformConfig + adapter = BuzzAdapter(PlatformConfig(enabled=True, extra={})) + assert await adapter.connect() is False + + @pytest.mark.asyncio + async def test_disconnect_cancels_poll_task(self): + adapter = _make_adapter() + + async def forever(): + while True: + await asyncio.sleep(3600) + + adapter._poll_task = asyncio.create_task(forever()) + task = adapter._poll_task + await adapter.disconnect() + assert task.cancelled() + assert adapter._poll_task is None + assert adapter.is_connected is False + + @pytest.mark.asyncio + async def test_get_chat_info_uses_cached_names(self): + adapter = _make_adapter() + adapter._channel_names[CHANNEL] = "general" + adapter._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 0, "seen": {}} + info = await adapter.get_chat_info(CHANNEL) + assert info == {"name": "general", "type": "group", "chat_id": CHANNEL} + + @pytest.mark.asyncio + async def test_dm_discovery_registers_conversation(self): + adapter = _make_adapter() + cli = _ScriptedCli() + cli.script("dms", "list", [{"dm_id": "dm-1", "participants": [OTHER_PUBKEY], "created_at": 5}]) + adapter._run_cli = cli + await adapter._discover_dms(seed=False) + assert adapter._channel_state["dm-1"]["chat_type"] == "dm" + + +# ── Credentials / requirements ──────────────────────────────────────────── + + +class TestCredentialResolution: + + def test_env_key_wins(self, monkeypatch): + monkeypatch.setenv("BUZZ_PRIVATE_KEY", "nsec1fromenv") + assert _resolve_private_key() == "nsec1fromenv" + + def test_credentials_file_fallback(self, monkeypatch, tmp_path): + creds = tmp_path / "agent_credentials.json" + creds.write_text(json.dumps({"nsec": "nsec1fromfile", "npub": "npub1x"}), encoding="utf-8") + monkeypatch.setenv("BUZZ_CREDENTIALS_FILE", str(creds)) + assert _resolve_private_key() == "nsec1fromfile" + + def test_default_dir_glob(self, monkeypatch, tmp_path): + (tmp_path / "chip_nostr_credentials.json").write_text( + json.dumps({"private_key_hex": "ab" * 32}), encoding="utf-8" + ) + monkeypatch.setattr(_buzz_mod, "_DEFAULT_CREDENTIALS_DIR", tmp_path) + assert _resolve_private_key() == "ab" * 32 + + def test_missing_everywhere(self): + assert _resolve_private_key() == "" + + def test_check_requirements(self, monkeypatch): + assert check_requirements() is False + monkeypatch.setenv("BUZZ_RELAY_URL", "https://r") + assert check_requirements() is False # still no key + monkeypatch.setenv("BUZZ_PRIVATE_KEY", "nsec1x") + assert check_requirements() is True + + def test_validate_config_from_extra(self, monkeypatch): + monkeypatch.setenv("BUZZ_PRIVATE_KEY", "nsec1x") + from gateway.config import PlatformConfig + assert validate_config(PlatformConfig(extra={"relay_url": "https://r"})) is True + assert validate_config(PlatformConfig(extra={})) is False + + +# ── Env enablement / registration / standalone send ────────────────────── + + +class TestEnvEnablement: + + def test_returns_none_when_unconfigured(self): + assert _env_enablement() is None + + def test_seeds_extra_and_home_channel(self, monkeypatch): + monkeypatch.setenv("BUZZ_RELAY_URL", "https://r") + monkeypatch.setenv("BUZZ_PRIVATE_KEY", "nsec1x") + monkeypatch.setenv("BUZZ_CHANNELS", "chan-a,chan-b") + seed = _env_enablement() + assert seed["relay_url"] == "https://r" + assert seed["channels"] == ["chan-a", "chan-b"] + # Home channel defaults to the first watched channel + assert seed["home_channel"]["chat_id"] == "chan-a" + + def test_explicit_home_channel(self, monkeypatch): + monkeypatch.setenv("BUZZ_RELAY_URL", "https://r") + monkeypatch.setenv("BUZZ_PRIVATE_KEY", "nsec1x") + monkeypatch.setenv("BUZZ_HOME_CHANNEL", "home-chan") + assert _env_enablement()["home_channel"]["chat_id"] == "home-chan" + + +class TestBuzzPluginRegistration: + + def test_register_platform_contract(self): + from gateway.platform_registry import platform_registry + + platform_registry.unregister("buzz") + ctx = MagicMock() + register(ctx) + ctx.register_platform.assert_called_once() + kwargs = ctx.register_platform.call_args.kwargs + assert kwargs["name"] == "buzz" + assert kwargs["cron_deliver_env_var"] == "BUZZ_HOME_CHANNEL" + assert kwargs["allowed_users_env"] == "BUZZ_ALLOWED_USERS" + assert kwargs["allow_all_env"] == "BUZZ_ALLOW_ALL_USERS" + assert callable(kwargs["standalone_sender_fn"]) + assert callable(kwargs["env_enablement_fn"]) + assert set(kwargs["required_env"]) == {"BUZZ_RELAY_URL", "BUZZ_PRIVATE_KEY"} + + +class TestStandaloneSend: + + @pytest.mark.asyncio + async def test_standalone_send_success(self, monkeypatch, tmp_path): + from gateway.config import PlatformConfig + + fake_cli = tmp_path / "buzz" + fake_cli.write_text("#!/bin/sh\n", encoding="utf-8") + monkeypatch.setenv("BUZZ_RELAY_URL", "https://r") + monkeypatch.setenv("BUZZ_PRIVATE_KEY", "nsec1x") + monkeypatch.setenv("BUZZ_CLI_PATH", str(fake_cli)) + + captured = {} + + async def fake_exec(cli_path, args, *, relay_url, private_key, input_text=None, timeout=30.0): + captured.update(cli_path=cli_path, args=args, relay_url=relay_url, input_text=input_text) + return 0, json.dumps({"accepted": True, "event_id": "evt-cron", "message": ""}), "" + + monkeypatch.setattr(_buzz_mod, "_exec_buzz", fake_exec) + + result = await _standalone_send(PlatformConfig(enabled=True, extra={}), CHANNEL, "cron says hi") + assert result == {"success": True, "message_id": "evt-cron"} + assert captured["args"][:2] == ["messages", "send"] + assert captured["input_text"] == "cron says hi" + # The private key must never be part of argv + assert all("nsec1x" not in str(a) for a in captured["args"]) + + @pytest.mark.asyncio + async def test_standalone_send_unconfigured(self): + from gateway.config import PlatformConfig + result = await _standalone_send(PlatformConfig(enabled=True, extra={}), CHANNEL, "hi") + assert "error" in result + + @pytest.mark.asyncio + async def test_standalone_send_falls_back_to_home_channel(self, monkeypatch, tmp_path): + from gateway.config import PlatformConfig + + fake_cli = tmp_path / "buzz" + fake_cli.write_text("#!/bin/sh\n", encoding="utf-8") + monkeypatch.setenv("BUZZ_RELAY_URL", "https://r") + monkeypatch.setenv("BUZZ_PRIVATE_KEY", "nsec1x") + monkeypatch.setenv("BUZZ_CLI_PATH", str(fake_cli)) + monkeypatch.setenv("BUZZ_HOME_CHANNEL", "home-chan") + + captured = {} + + async def fake_exec(cli_path, args, *, relay_url, private_key, input_text=None, timeout=30.0): + captured["args"] = args + return 0, json.dumps({"accepted": True, "event_id": "e"}), "" + + monkeypatch.setattr(_buzz_mod, "_exec_buzz", fake_exec) + result = await _standalone_send(PlatformConfig(enabled=True, extra={}), "", "hi") + assert result["success"] is True + assert captured["args"][captured["args"].index("--channel") + 1] == "home-chan" diff --git a/website/docs/user-guide/messaging/buzz.md b/website/docs/user-guide/messaging/buzz.md new file mode 100644 index 000000000000..5de82d71bbaf --- /dev/null +++ b/website/docs/user-guide/messaging/buzz.md @@ -0,0 +1,82 @@ +# 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. + +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. + +> Run `hermes gateway setup` and pick **Buzz** for a guided walk-through. + +## Prerequisites + +- The `buzz` CLI binary on your `PATH` (or point `BUZZ_CLI_PATH` at it) — build it from the [Buzz repo](https://github.com/block/buzz) with `cargo build --release -p buzz-cli` +- A Buzz community relay URL (e.g. `https://mycommunity.communities.buzz.xyz`) +- A Nostr private key (nsec or hex) whose identity is already a **member** of that community + +## Configure Hermes + +You can configure Buzz two ways — the `gateway` block in `config.yaml` (canonical) or environment variables (which override it). The private key is a **secret** and always belongs in `~/.hermes/.env`. + +### Option A — config.yaml + +```yaml +gateway: + platforms: + buzz: + enabled: true + extra: + relay_url: https://mycommunity.communities.buzz.xyz + channels: # channel UUIDs to watch (empty = all joined) + - ccc2bc1a-7a82-5a8f-8c4e-57a070cbe7cd + home_channel: ccc2bc1a-7a82-5a8f-8c4e-57a070cbe7cd + poll_interval: 4 # seconds between inbound poll sweeps + cli_path: "" # buzz binary (default: PATH, then ~/bin/buzz) + credentials_file: "" # JSON file with the nsec (BUZZ_PRIVATE_KEY fallback) + allowed_users: [] # empty = allow all; hex pubkeys or npubs +``` + +Plus, in `~/.hermes/.env`: + +``` +BUZZ_PRIVATE_KEY=nsec1... +``` + +### Option B — environment variables + +| Variable | Required | Description | +|----------|:--------:|-------------| +| `BUZZ_RELAY_URL` | ✅ | Base URL of the community relay | +| `BUZZ_PRIVATE_KEY` | ✅ | Nostr private key (nsec or hex) — the only secret | +| `BUZZ_CHANNELS` | — | Comma-separated channel UUIDs to watch (default: all joined channels) | +| `BUZZ_HOME_CHANNEL` | — | Channel UUID for cron / notification delivery (defaults to the first watched channel) | +| `BUZZ_ALLOWED_USERS` | — | Comma-separated npubs or hex pubkeys allowed to talk to the agent | +| `BUZZ_ALLOW_ALL_USERS` | — | Allow any community member to talk to the agent | +| `BUZZ_POLL_INTERVAL` | — | Seconds between inbound poll sweeps (default: 4) | +| `BUZZ_CLI_PATH` | — | Path to the `buzz` binary (default: `buzz` on PATH, then `~/bin/buzz`) | +| `BUZZ_CREDENTIALS_FILE` | — | JSON credentials file holding the nsec, used when `BUZZ_PRIVATE_KEY` is unset | + +## Mentions, channels, and DMs + +- In shared channels the agent only responds when **addressed** — by `@name`, its npub, or its hex pubkey. Everything else is ignored. +- Direct messages always reach the agent, no mention needed. +- The agent's own messages are never dispatched back to it (self-echo suppression by pubkey), and every event is de-duplicated by event id against a per-channel high-water mark. + +## Access control + +By default the allow-list is empty, which means every community member who mentions the agent gets a response only if `BUZZ_ALLOW_ALL_USERS=true`; otherwise restrict access by listing npubs or hex pubkeys in `BUZZ_ALLOWED_USERS` (or `allowed_users` in config.yaml). Community membership itself is enforced by the relay — only members can post. + +Cron jobs and notifications (`deliver=buzz`) are delivered to the **home channel** — `BUZZ_HOME_CHANNEL` if set, otherwise the first watched channel — and work even when cron runs outside the gateway process. + +## Run the gateway + +```bash +hermes gateway start +``` + +Check status with `hermes gateway status` — Buzz connection state is reported there, including for env-only setups. + +## Notes and limitations + +- **Inbound is polled, not streamed.** The `buzz` CLI is request/response, so the adapter polls `buzz messages get` per watched channel every `poll_interval` seconds (default 4). Expect up to one interval of latency on inbound messages. A future optimization is a websocket transport (the Buzz repo ships `buzz-ws-client` for true streaming). +- On (re)connect the adapter seeds its high-water mark from the newest events, so channel history is never replayed into the agent. +- New DM conversations are discovered automatically (every few poll sweeps). +- The private key is passed to the CLI via the subprocess environment — it never appears in argv or logs. diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index da107c4ae613..a9cfcc85ba87 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -42,6 +42,7 @@ Bots need both a model provider and tool providers (TTS, web). A [Nous Portal](/ | ntfy | — | — | — | — | — | — | — | | Raft | — | — | — | — | — | — | — | | IRC | — | — | — | — | — | — | — | +| Buzz | — | ✅ | — | ✅ | — | — | — | **Voice** = TTS audio replies and/or voice message transcription. **Images** = send/receive images. **Files** = send/receive file attachments. **Threads** = threaded conversations. **Reactions** = emoji reactions on messages. **Typing** = typing indicator while processing. **Streaming** = progressive message updates via editing. @@ -727,4 +728,5 @@ Defaults to `false`. Only platforms whose adapter implements `delete_message` ho - [Open WebUI + API Server](open-webui.md) - [Raft Setup](raft.md) - [IRC Setup](irc.md) +- [Buzz Setup](buzz.md) - [Webhooks](webhooks.md)