From a6fb23d56f553f856db07dfa12505177bcba3c7b Mon Sep 17 00:00:00 2001 From: Reinhold <310554180+reinhold-ph@users.noreply.github.com> Date: Wed, 16 Sep 2026 08:58:26 +0000 Subject: [PATCH] feat(gateway): publish native remote turn activity Squashed and rebased onto current main (PR #80217). A platform-neutral GatewayTurnObserver reports turn/session/tool lifecycle through a new BasePlatformAdapter.on_turn_lifecycle hook; the Buzz adapter translates those into owner-encrypted NIP-44 observer events (kind 24200) published on the authenticated WebSocket with bounded ACK tracking, terminal-frame retention across reconnects and one delayed retry. Configured with activity_owner_pubkey; requires the websocket or auto transport. --- gateway/platforms/base.py | 5 + gateway/run_turn.py | 41 +- gateway/run_turn_runner.py | 42 +- gateway/turn_observer.py | 293 ++++++++ plugins/platforms/buzz/adapter.py | 475 ++++++++++++- plugins/platforms/buzz/nostr_auth.py | 166 +++++ tests/gateway/test_buzz_activity_bridge.py | 445 +++++++++++++ tests/gateway/test_buzz_websocket.py | 738 ++++++++++++++++++++- tests/gateway/test_turn_observer.py | 95 +++ website/docs/integrations/buzz.md | 3 +- website/docs/user-guide/messaging/buzz.md | 19 + 11 files changed, 2315 insertions(+), 7 deletions(-) create mode 100644 gateway/turn_observer.py create mode 100644 tests/gateway/test_buzz_activity_bridge.py create mode 100644 tests/gateway/test_turn_observer.py diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index a29918f9e0a6..6ad20036c9fb 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -2219,6 +2219,11 @@ def set_message_handler(self, handler: MessageHandler) -> None: """Set the incoming-message handler (MessageEvent -> optional response str).""" self._message_handler = handler + def on_turn_lifecycle(self, event: Any) -> bool: + """Consume a platform-neutral Gateway turn event when supported. Fail-open no-op by default; + an adapter may translate the closed event contract into platform-native activity telemetry.""" + return False + def set_platform_event_handler( self, handler: Optional[Callable[[Dict[str, Any], Any], Awaitable[None]]]) -> None: """Install the gateway-owned normalized platform-event boundary (stable dicts + internal diff --git a/gateway/run_turn.py b/gateway/run_turn.py index 3eac3a283c48..8226e53669f1 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -1957,6 +1957,7 @@ class _PreparedTurn: persist_user_display_kind: Optional[str] persistence_session_id: Optional[str] = None persistence_owner: Optional[str] = None + is_new_session: bool = False async def _hmwa_prepare_turn(self, event, source, session_entry, session_key, _quick_key, run_generation): """Everything between session resolution and the agent run: session open, task-local env, @@ -2047,7 +2048,7 @@ async def _hmwa_prepare_turn(self, event, source, session_entry, session_key, _q if event.message_id else str(uuid.uuid4())) return self._PreparedTurn( history, context_prompt, message_text, persist_user_message, persist_user_timestamp, - persist_user_display_kind, session_entry.session_id, owner, + persist_user_display_kind, session_entry.session_id, owner, _is_new_session, ), _session_env_tokens async def _handle_message_with_agent(self, event, source, _quick_key: str, run_generation: int): @@ -2098,6 +2099,7 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g agent_result = await self._run_agent( message=message_text, context_prompt=prepared.context_prompt, history=history, source=source, session_id=_run_start_session_id, session_key=session_key, + is_new_session=prepared.is_new_session, run_generation=run_generation, event_message_id=self._reply_anchor_for_event(event), inbound_message_id=str(event.message_id) if event.message_id else None, channel_prompt=event.channel_prompt, moa_config=getattr(event, "_moa_config", None), @@ -2780,8 +2782,30 @@ async def _run_agent( ) -> Dict[str, Any]: """Profile-scoping wrapper around ``_run_agent_inner`` (same keyword parameters; pass-through when multiplexing is off).""" + import sys + from gateway.turn_observer import GatewayTurnObserver with self._profile_scope_for_source(source): - return await self._run_agent_inner(message, context_prompt, history, source, session_id, **turn_kwargs) + session_key = turn_kwargs.get("session_key") + generation = turn_kwargs.get("run_generation") + observer = GatewayTurnObserver( + platform=source.platform.value, profile=getattr(source, "profile", None) or "default", + channel_id=source.chat_id, session_id=session_id, + triggering_event_id=turn_kwargs.get("inbound_message_id") or turn_kwargs.get("event_message_id"), + is_new_session=turn_kwargs.pop("is_new_session", False), + route=self._adapter_for_source(source), loop=asyncio.get_running_loop(), + is_current=lambda: generation is None or self._is_session_run_current(session_key, generation), + ) + response = None + observer.start() + observer.session_resolved() + try: + response = await self._run_agent_inner( + message, context_prompt, history, source, session_id, + turn_observer=observer, **turn_kwargs, + ) + return response + finally: + observer.finish(response, exception_type=sys.exc_info()[0]) def _run_agent_display_settings(self, source: SessionSource) -> "GatewayRunner._RunAgentDisplay": """Resolve per-platform display, progress, status and streaming-surface settings for a turn.""" @@ -3443,7 +3467,11 @@ async def _run_agent_await_turn_worker( ).start() break await self._run_agent_backup_interrupt_check(turn_ctx, _interrupt_detected, interrupt_monitor) - return self._run_agent_timeout_result(worker, turn_ctx) + result = self._run_agent_timeout_result(worker, turn_ctx) + observer = getattr(turn_ctx, "turn_observer", None) + if observer is not None: + observer.finish(result, timed_out=True) + return result def _run_agent_evict_on_fallback(self, turn_ctx: TurnContext) -> None: """Evict the cached agent when a fallback model activated on a SUCCESSFUL run (so /model shows @@ -3729,6 +3757,7 @@ async def _run_agent_queued_followup( message=next_message, context_prompt=turn_ctx.context_prompt, history=updated_history, source=next_source, session_id=session_id, session_key=next_session_key, run_generation=run_generation, _interrupt_depth=_interrupt_depth + 1, + is_new_session=False, event_message_id=next_message_id, inbound_message_id=next_inbound_id, channel_prompt=next_channel_prompt, message_type=next_message_type, persist_user_display_kind=next_display_kind, @@ -4051,6 +4080,7 @@ async def _run_agent_inner( persist_user_message: Optional[Any] = None, persist_user_timestamp: Optional[float] = None, persist_user_display_kind: Optional[str] = None, message_type: Optional[str] = None, persist_user_display_metadata: Optional[dict] = None, + turn_observer=None, ) -> Dict[str, Any]: """Run the agent; returns the full run_conversation result dict. @@ -4076,6 +4106,8 @@ async def _run_agent_inner( persist_user_display_kind=persist_user_display_kind, persist_user_display_metadata=persist_user_display_metadata, ) + turn_runner._observer = turn_observer + turn_ctx.turn_observer = turn_observer _status_thread_metadata = self._run_agent_bind_turn_wiring( turn_ctx, turn_runner, source, event_message_id, disp._native_slack_task_cards, ) @@ -4101,6 +4133,9 @@ async def _run_agent_inner( worker = self._run_agent_start_turn_worker(turn_ctx, turn_runner.run_sync) _executor_task_holder[0] = worker.executor_task # read late by _notify_long_running response = await self._run_agent_await_turn_worker(worker, turn_ctx, _interrupt_detected, interrupt_monitor) + # Execution owns terminal state, before TTS, delivery or queued recursion can suspend. + if turn_observer is not None: + turn_observer.finish(response) self._run_agent_evict_on_fallback(turn_ctx) # Interrupted OR queued message (/queue)? diff --git a/gateway/run_turn_runner.py b/gateway/run_turn_runner.py index a2ee1c4667e3..bd9612e3a037 100644 --- a/gateway/run_turn_runner.py +++ b/gateway/run_turn_runner.py @@ -71,9 +71,10 @@ class _ExecApprovalDeclined(RuntimeError): class TurnRunner: """Per-turn collaborator carrying ``GatewayRunner._run_agent_inner``'s tool-progress callbacks.""" - def __init__(self, runner: "GatewayRunner", ctx: TurnContext) -> None: + def __init__(self, runner: "GatewayRunner", ctx: TurnContext, observer=None) -> None: self._runner = runner self._ctx = ctx + self._observer = observer # ── shared thread→loop plumbing ───────────────────────────────────────────────────────── @@ -753,6 +754,43 @@ async def send_progress_messages(self): # ── ID-bearing lifecycle callbacks (agent thread) ─────────────────────────────────────── + @staticmethod + def _compose_callbacks(existing, *callbacks): + """Compose structured callbacks with sibling failure isolation.""" + + if getattr(existing, "_gateway_turn_fanout", False): + existing = getattr(existing, "_gateway_prior_callback", None) + new_callbacks = [callback for callback in callbacks if callable(callback)] + if not new_callbacks: + return existing if callable(existing) else None + ordered = [callback for callback in (existing, *new_callbacks) if callable(callback)] + if not ordered: + return None + + def fanout(*args, **kwargs): + for callback in ordered: + try: + callback(*args, **kwargs) + except Exception: + logger.debug("Structured tool callback failed open", exc_info=True) + + fanout._gateway_turn_fanout = True + fanout._gateway_prior_callback = existing + return fanout + + def wire_structured_tool_callbacks(self, agent) -> None: + """Add observation without replacing current native/voice callbacks.""" + observer = self._observer + active = observer is not None and observer.active + agent.tool_start_callback = self._compose_callbacks( + getattr(agent, "tool_start_callback", None), + observer.tool_started if active else None, + ) + agent.tool_complete_callback = self._compose_callbacks( + getattr(agent, "tool_complete_callback", None), + observer.tool_finished if active else None, + ) + def voice_ack_callback(self, call_id, tool_name, args): """tool_start_callback: speak a one-time ack in the voice channel.""" ctx = self._ctx @@ -1224,6 +1262,8 @@ def _wire_turn_agent_callbacks(self, agent, turn_route, reasoning_config, if (ctx._voice_ack_guild[0] is not None or ctx._native_slack_task_cards) else None ) agent.tool_complete_callback = ctx.native_tool_complete_callback if ctx._native_slack_task_cards else None + if getattr(self, "_observer", None) is not None: + self.wire_structured_tool_callbacks(agent) agent.step_callback = ctx._step_callback_sync if ctx._hooks_ref.loaded_hooks else None agent.stream_delta_callback = stream_delta_cb agent.interim_assistant_callback = interim_assistant_cb if want_interim_messages else None diff --git a/gateway/turn_observer.py b/gateway/turn_observer.py new file mode 100644 index 000000000000..53f082afb51a --- /dev/null +++ b/gateway/turn_observer.py @@ -0,0 +1,293 @@ +"""Platform-neutral lifecycle observation for one Gateway turn. + +The core emits a deliberately small, closed event vocabulary to the selected +platform adapter. Adapters may translate those events at the edge, but prompts, +tool arguments/results, exception text, protocol payloads, and transport state +never enter this contract. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import logging +import re +import uuid +from collections import OrderedDict +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Callable, Literal, Optional + +logger = logging.getLogger(__name__) + +_SAFE_TOOL_NAME = re.compile(r"[A-Za-z][A-Za-z0-9_.:-]{0,63}") +_TOOL_ID_MAP_CAP = 256 + +TurnPhase = Literal[ + "turn_started", + "session_resolved", + "turn_liveness", + "tool_started", + "tool_finished", + "turn_finished", +] +TurnOutcome = Literal["success", "failed", "cancelled", "timed_out"] +ToolStatus = Literal["executing", "completed", "failed"] + + +@dataclass(frozen=True) +class TurnLifecycleEvent: + """Allowlisted metadata for one platform-neutral Gateway lifecycle event.""" + + phase: TurnPhase + platform: str + profile: str + channel_id: str + session_id: str + turn_id: str + started_at: str + triggering_event_id: Optional[str] = None + is_new_session: Optional[bool] = None + tool_call_id: Optional[str] = None + tool_name: Optional[str] = None + tool_status: Optional[ToolStatus] = None + outcome: Optional[TurnOutcome] = None + + +def classify_turn_outcome( + result: Any, + *, + timed_out: bool = False, + exception_type: Optional[type[BaseException]] = None, +) -> TurnOutcome: + """Classify terminal state without exposing result or exception content.""" + + if timed_out: + return "timed_out" + if exception_type is not None and issubclass( + exception_type, asyncio.CancelledError + ): + return "cancelled" + if exception_type is not None or not isinstance(result, dict): + return "failed" + if bool(result.get("interrupted")): + return "cancelled" + if bool(result.get("failed")) or bool(result.get("error")): + return "failed" + return "success" + + +class GatewayTurnObserver: + """Fail-open lifecycle dispatcher scoped to one logical Gateway turn. + + ``route`` is opaque dispatch context (normally the selected platform + adapter). It is passed separately from :class:`TurnLifecycleEvent` so the + event remains serializable, platform-neutral, and privacy-auditable. + """ + + def __init__( + self, + *, + platform: str, + profile: str, + channel_id: str, + session_id: str, + triggering_event_id: Optional[str], + is_new_session: bool, + route: Any, + loop: asyncio.AbstractEventLoop, + is_current: Callable[[], bool], + ) -> None: + self.platform = str(platform or "") + self.profile = str(profile or "") + self.channel_id = str(channel_id or "") + self.session_id = str(session_id or "") + self.triggering_event_id = ( + str(triggering_event_id) if triggering_event_id is not None else None + ) + self.is_new_session = bool(is_new_session) + self.route = route + self.loop = loop + self.is_current = is_current + self.turn_id = uuid.uuid4().hex + self.started_at = ( + datetime + .now(timezone.utc) + .isoformat(timespec="milliseconds") + .replace("+00:00", "Z") + ) + self._active = False + self._started = False + self._terminal_dispatched = False + self._liveness_task: Optional[asyncio.Task] = None + self._tool_id_counter = 0 + self._tool_ids: OrderedDict[bytes, str] = OrderedDict() + + @property + def active(self) -> bool: + return self._active and not self._terminal_dispatched + + def _event(self, phase: TurnPhase, **kwargs: Any) -> TurnLifecycleEvent: + return TurnLifecycleEvent( + phase=phase, + platform=self.platform, + profile=self.profile, + channel_id=self.channel_id, + session_id=self.session_id, + turn_id=self.turn_id, + started_at=self.started_at, + **kwargs, + ) + + def _dispatch(self, event: TurnLifecycleEvent) -> bool: + """Synchronously offer an event to the selected adapter, fail-open.""" + + try: + handler = getattr(self.route, "on_turn_lifecycle", None) + if not callable(handler): + return False + return bool(handler(event)) + except Exception: + logger.debug("Gateway turn observer failed open", exc_info=True) + return False + + def start(self, *, liveness_interval: float = 10.0) -> bool: + """Dispatch turn start and immediately arm liveness when consumed.""" + + if self._started: + return self.active + self._started = True + self._active = self._dispatch( + self._event( + "turn_started", + triggering_event_id=self.triggering_event_id, + ) + ) + if self._active and liveness_interval > 0: + self._liveness_task = self.loop.create_task( + self._liveness_loop(liveness_interval) + ) + return self._active + + def session_resolved(self) -> bool: + if not self.active: + return False + return self._dispatch( + self._event("session_resolved", is_new_session=self.is_new_session) + ) + + async def _liveness_loop(self, interval: float) -> None: + try: + while self.active: + await asyncio.sleep(interval) + if not self.active or not self.is_current(): + return + self._dispatch(self._event("turn_liveness")) + except asyncio.CancelledError: + return + except Exception: + logger.debug("Gateway turn liveness observer failed open", exc_info=True) + + def _dispatch_if_active(self, event: TurnLifecycleEvent) -> None: + if self.active and self.is_current(): + self._dispatch(event) + + def _dispatch_threadsafe(self, event: TurnLifecycleEvent) -> None: + """Preserve worker-thread tool callbacks without leaking tool content.""" + + if not self.active or not self.is_current(): + return + try: + running_loop = asyncio.get_running_loop() + except RuntimeError: + running_loop = None + if running_loop is self.loop: + self._dispatch_if_active(event) + return + try: + self.loop.call_soon_threadsafe(self._dispatch_if_active, event) + except Exception: + logger.debug("Gateway turn observer scheduling failed open", exc_info=True) + + def _safe_tool_call_id(self, value: Any) -> str: + """Return a bounded turn-local correlation id without retaining input.""" + + raw = str(value or "")[:4096].encode("utf-8", errors="replace") + digest = hashlib.blake2s(raw, digest_size=16).digest() + existing = self._tool_ids.get(digest) + if existing is not None: + self._tool_ids.move_to_end(digest) + return existing + self._tool_id_counter += 1 + safe_id = f"tool-{self._tool_id_counter}" + self._tool_ids[digest] = safe_id + while len(self._tool_ids) > _TOOL_ID_MAP_CAP: + self._tool_ids.popitem(last=False) + return safe_id + + @staticmethod + def _safe_tool_name(value: Any) -> str: + """Allow only compact display labels; arbitrary values become ``tool``.""" + + text = str(value or "") + return text if _SAFE_TOOL_NAME.fullmatch(text) else "tool" + + def tool_started(self, call_id: Any, tool_name: Any, _args: Any = None) -> None: + self._dispatch_threadsafe( + self._event( + "tool_started", + tool_call_id=self._safe_tool_call_id(call_id), + tool_name=self._safe_tool_name(tool_name), + tool_status="executing", + ) + ) + + def tool_finished( + self, + call_id: Any, + tool_name: Any, + _args: Any = None, + result: Any = None, + ) -> None: + try: + from agent.display import _detect_tool_failure + + failed, _ = _detect_tool_failure(str(tool_name), result) + except Exception: + failed = False + self._dispatch_threadsafe( + self._event( + "tool_finished", + tool_call_id=self._safe_tool_call_id(call_id), + tool_name=self._safe_tool_name(tool_name), + tool_status="failed" if failed else "completed", + ) + ) + + def finish( + self, + result: Any, + *, + timed_out: bool = False, + exception_type: Optional[type[BaseException]] = None, + ) -> bool: + """Dispatch exactly one local terminal event and stop liveness.""" + + if self._terminal_dispatched or not self._started: + return False + self._terminal_dispatched = True + self._active = False + task = self._liveness_task + self._liveness_task = None + if task is not None and not task.done(): + task.cancel() + return self._dispatch( + self._event( + "turn_finished", + outcome=classify_turn_outcome( + result, + timed_out=timed_out, + exception_type=exception_type, + ), + ) + ) diff --git a/plugins/platforms/buzz/adapter.py b/plugins/platforms/buzz/adapter.py index 2eb5c21087d9..49567ef0132c 100644 --- a/plugins/platforms/buzz/adapter.py +++ b/plugins/platforms/buzz/adapter.py @@ -20,7 +20,7 @@ import tempfile import time from collections import OrderedDict -from datetime import datetime +from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from urllib.parse import urlsplit, urlunsplit @@ -60,6 +60,17 @@ def _scoped_platform_setting(env_name, extra, key): logger = logging.getLogger(__name__) +_ACTIVITY_QUEUE_SIZE = 256 +_ACTIVITY_SEND_TIMEOUT = 2.0 +_ACTIVITY_ACK_TIMEOUT = 30.0 +_ACTIVITY_PENDING_CAP = 1024 +_ACTIVITY_TERMINAL_REPLAY_CAP = 256 +_ACTIVITY_TERMINAL_ACK_RETRY_CAP = 1 +_ACTIVITY_TERMINAL_RETRY_DELAY = 1.0 +_ACTIVITY_TERMINAL_KINDS = frozenset({"turn_completed", "turn_error"}) +_ACTIVITY_INTERNAL_RETRY_KEY = "_hermesAckRetry" +_ACTIVITY_QUEUE_STOP = object() + from gateway.platforms.base import ( BasePlatformAdapter, CachedMedia, SendResult, cache_media_bytes_async, @@ -540,6 +551,36 @@ def __init__(self, config, **kwargs): # never dispatch; allowed_users wins on overlap. self._allowed_pubkeys: set = _pubkey_set(_setting_or("BUZZ_ALLOWED_USERS", extra, "allowed_users", [])) self._reaction_only_pubkeys: set = _pubkey_set(_setting_or("BUZZ_REACTION_ONLY_USERS", extra, "reaction_only_users", [])) + # Optional native Gateway activity observer. The owner pubkey is a + # routing/encryption setting (not a secret); an empty value keeps the + # observer disabled without changing normal Buzz chat delivery. + _activity_owner = str(extra.get("activity_owner_pubkey", "") or "").strip() + self.activity_owner_pubkey = ( + _normalize_user_ref(_activity_owner) if _activity_owner else "" + ) + if _activity_owner and not self.activity_owner_pubkey: + raise ValueError( + "Buzz activity_owner_pubkey must be a valid x-only secp256k1 " + "public key encoded as 64 hex characters or npub" + ) + if self.activity_owner_pubkey: + try: + self.activity_owner_pubkey = ( + _load_nostr_auth().validate_x_only_public_key( + self.activity_owner_pubkey + ) + ) + except ValueError as exc: + raise ValueError( + "Buzz activity_owner_pubkey must be a valid x-only " + "secp256k1 public key encoded as 64 hex characters or npub" + ) from exc + if self.activity_owner_pubkey and self.transport == "poll": + raise ValueError( + "Buzz native activity requires transport 'websocket' or 'auto'; " + "poll-only transport cannot publish encrypted observer events" + ) + # Secret — resolved lazily (never at import time, never logged); connect() re-resolves. self._private_key = self._auth_tag = "" # Identity — filled in by connect() from ``buzz users get`` @@ -547,6 +588,22 @@ def __init__(self, config, **kwargs): self._poll_task: Optional[asyncio.Task] = None self._ws_task: Optional[asyncio.Task] = None self._ws_ready: Optional[asyncio.Event] = None + self._ws_connection = None + self._ws_active = False + self._activity_seq = 0 + self._activity_ws_generation = 0 + self._activity_pending_event_ids: OrderedDict[ + str, tuple[int, asyncio.TimerHandle] + ] = OrderedDict() + self._activity_pending_terminal_payloads: OrderedDict[ + str, Dict[str, Any] + ] = OrderedDict() + self._activity_queue: asyncio.Queue = asyncio.Queue(maxsize=_ACTIVITY_QUEUE_SIZE) + self._activity_sender_task: Optional[asyncio.Task] = None + self._activity_terminal_retry_handle: Optional[asyncio.TimerHandle] = None + self._activity_terminal_replay: OrderedDict[str, Dict[str, Any]] = ( + OrderedDict() + ) self._membership_since = self._poll_count = 0 # Channels the relay permanently rejected ("restricted"); persists across reconnects so we never re-subscribe. # channel_id -> { "chat_type", "last_ts", "seen": OrderedDict[event_id, None], "event_meta": @@ -702,6 +759,9 @@ async def disconnect(self) -> None: self._ws_task = None await cancel_task(self._poll_task) self._poll_task = None + self._ws_active = False + self._ws_connection = None + await self._reset_activity_transport() self._channel_state = {} self._poll_count = 0 @@ -975,6 +1035,8 @@ async def _start_websocket(self) -> bool: import websockets # noqa: F401 (availability probe) self._websocket_url() except Exception as e: + if self.activity_owner_pubkey: + logger.warning("Buzz: native activity is unavailable while using polling") logger.info("Buzz: WebSocket transport unavailable (%s); falling back to polling", e) return False self._ws_ready = asyncio.Event() @@ -1069,6 +1131,333 @@ async def _rediscover_and_subscribe(self, websocket, subscriptions: Dict[str, Op await self._send_channel_subscription(websocket, subscription_id, channel_id) logger.info("Buzz: subscribed to new conversation %s", channel_id) + def on_turn_lifecycle(self, event) -> bool: + return _handle_gateway_turn_lifecycle(event=event, route=self) + + def _enqueue_activity( + self, + kind: str, + *, + channel_id: Optional[str], + session_id: Optional[str], + turn_id: Optional[str], + payload: Optional[Dict[str, Any]] = None, + started_at: Optional[str] = None, + ) -> bool: + """Non-blocking, fail-open enqueue of one encrypted observer frame.""" + + if not self.activity_owner_pubkey: + return False + self._activity_seq += 1 + timestamp = datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace( + "+00:00", "Z" + ) + observer_payload: Dict[str, Any] = { + "seq": self._activity_seq, + "timestamp": timestamp, + "kind": str(kind), + "agentIndex": None, + "channelId": str(channel_id) if channel_id is not None else None, + "sessionId": str(session_id) if session_id is not None else None, + "turnId": str(turn_id) if turn_id is not None else None, + "payload": payload or {}, + } + if started_at is not None: + observer_payload["startedAt"] = str(started_at) + websocket = self._ws_connection + if not self._ws_active or websocket is None: + return self._cache_terminal_activity(observer_payload) + try: + self._activity_queue.put_nowait( + (self._activity_ws_generation, observer_payload) + ) + except asyncio.QueueFull: + if self._cache_terminal_activity(observer_payload): + return True + logger.debug("Buzz: observer activity queue full; dropping frame") + return False + if self._activity_sender_task is None or self._activity_sender_task.done(): + self._activity_sender_task = asyncio.create_task(self._activity_sender_loop()) + return True + + def _cache_terminal_activity(self, observer_payload: Dict[str, Any]) -> bool: + """Retain only the latest terminal frame per turn across reconnects.""" + + if str(observer_payload.get("kind") or "") not in _ACTIVITY_TERMINAL_KINDS: + return False + replay_key = str( + observer_payload.get("turnId") + or observer_payload.get("sessionId") + or observer_payload.get("seq") + ) + self._activity_terminal_replay[replay_key] = observer_payload + self._activity_terminal_replay.move_to_end(replay_key) + while len(self._activity_terminal_replay) > _ACTIVITY_TERMINAL_REPLAY_CAP: + self._activity_terminal_replay.popitem(last=False) + return True + + def _schedule_terminal_activity_retry( + self, observer_payload: Dict[str, Any], generation: int + ) -> None: + """Retain a terminal and schedule at most one delayed retry.""" + + if not self._cache_terminal_activity(observer_payload): + return + retry_count = int( + observer_payload.get(_ACTIVITY_INTERNAL_RETRY_KEY, 0) or 0 + ) + if retry_count >= _ACTIVITY_TERMINAL_ACK_RETRY_CAP: + observer_payload[_ACTIVITY_INTERNAL_RETRY_KEY] = retry_count + 1 + return + observer_payload[_ACTIVITY_INTERNAL_RETRY_KEY] = retry_count + 1 + handle = self._activity_terminal_retry_handle + if handle is not None and not handle.cancelled(): + return + self._activity_terminal_retry_handle = asyncio.get_running_loop().call_later( + _ACTIVITY_TERMINAL_RETRY_DELAY, + self._drain_terminal_activity_retry, + generation, + ) + + def _drain_terminal_activity_retry(self, generation: int) -> None: + """Replay retained terminals only on the still-current live socket.""" + + self._activity_terminal_retry_handle = None + if ( + self._ws_active + and self._ws_connection is not None + and generation == self._activity_ws_generation + ): + self._replay_terminal_activity() + + def _replay_terminal_activity(self, *, new_generation: bool = False) -> None: + """Move bounded terminal frames onto the current WebSocket generation.""" + + if not self._ws_active or self._ws_connection is None: + return + for replay_key, observer_payload in list(self._activity_terminal_replay.items()): + if new_generation: + observer_payload.pop(_ACTIVITY_INTERNAL_RETRY_KEY, None) + if self._activity_queue.full(): + continue + if int(observer_payload.get(_ACTIVITY_INTERNAL_RETRY_KEY, 0)) > _ACTIVITY_TERMINAL_ACK_RETRY_CAP: + continue + self._activity_terminal_replay.pop(replay_key) + self._activity_queue.put_nowait( + (self._activity_ws_generation, observer_payload) + ) + if ( + not self._activity_queue.empty() + and ( + self._activity_sender_task is None + or self._activity_sender_task.done() + ) + ): + self._activity_sender_task = asyncio.create_task( + self._activity_sender_loop() + ) + + async def publish_activity( + self, + kind: str, + *, + channel_id: Optional[str], + session_id: Optional[str], + turn_id: Optional[str], + payload: Optional[Dict[str, Any]] = None, + started_at: Optional[str] = None, + ) -> bool: + """Compatibility wrapper for direct adapter callers and focused tests.""" + + return self._enqueue_activity( + kind, + channel_id=channel_id, + session_id=session_id, + turn_id=turn_id, + payload=payload, + started_at=started_at, + ) + + def _drop_activity_ack(self, event_id: str) -> bool: + event_id = str(event_id) + pending = self._activity_pending_event_ids.pop(event_id, None) + self._activity_pending_terminal_payloads.pop(event_id, None) + if pending is None: + return False + pending[1].cancel() + return True + + def _expire_activity_ack(self, event_id: str, generation: int) -> None: + event_id = str(event_id) + pending = self._activity_pending_event_ids.get(event_id) + if pending is None or pending[0] != generation: + return + self._activity_pending_event_ids.pop(event_id, None) + pending[1].cancel() + terminal_payload = self._activity_pending_terminal_payloads.pop(event_id, None) + if terminal_payload is not None: + # Delivery is unknown when the relay's OK frame is lost. Preserve + # terminal state and allow one delayed same-generation retry. Any + # further timeout waits for a new authenticated WebSocket generation. + retry_count = int( + terminal_payload.get(_ACTIVITY_INTERNAL_RETRY_KEY, 0) or 0 + ) + terminal_payload[_ACTIVITY_INTERNAL_RETRY_KEY] = retry_count + 1 + self._cache_terminal_activity(terminal_payload) + if retry_count < _ACTIVITY_TERMINAL_ACK_RETRY_CAP: + self._replay_terminal_activity() + logger.debug("Buzz: observer activity ACK timed out") + + def _track_activity_ack( + self, + event_id: str, + generation: int, + observer_payload: Optional[Dict[str, Any]] = None, + ) -> None: + self._drop_activity_ack(event_id) + handle = asyncio.get_running_loop().call_later( + _ACTIVITY_ACK_TIMEOUT, + self._expire_activity_ack, + str(event_id), + generation, + ) + event_id = str(event_id) + self._activity_pending_event_ids[event_id] = (generation, handle) + self._activity_pending_event_ids.move_to_end(event_id) + if ( + observer_payload is not None + and str(observer_payload.get("kind") or "") + in _ACTIVITY_TERMINAL_KINDS + ): + self._activity_pending_terminal_payloads[event_id] = observer_payload + self._activity_pending_terminal_payloads.move_to_end(event_id) + while len(self._activity_pending_event_ids) > _ACTIVITY_PENDING_CAP: + evicted_id, (_, evicted_handle) = self._activity_pending_event_ids.popitem( + last=False + ) + evicted_payload = self._activity_pending_terminal_payloads.pop( + evicted_id, None + ) + if evicted_payload is not None: + self._cache_terminal_activity(evicted_payload) + evicted_handle.cancel() + + async def _activity_sender_loop(self) -> None: + """Encrypt and send queued observer frames without blocking Gateway turns.""" + while True: + queued_item = await self._activity_queue.get() + if queued_item is _ACTIVITY_QUEUE_STOP: + self._activity_queue.task_done() + return + generation, observer_payload = queued_item + event_id: Optional[str] = None + sent = False + try: + websocket = self._ws_connection + if ( + not self._ws_active + or websocket is None + or generation != self._activity_ws_generation + ): + self._cache_terminal_activity(observer_payload) + continue + wire_payload = { + key: value + for key, value in observer_payload.items() + if key != _ACTIVITY_INTERNAL_RETRY_KEY + } + event = await asyncio.to_thread( + _load_nostr_auth().build_observer_event, + private_key=self._private_key, + owner_pubkey=str(self.activity_owner_pubkey), + payload=wire_payload, + ) + if ( + not self._ws_active + or websocket is not self._ws_connection + or generation != self._activity_ws_generation + ): + self._cache_terminal_activity(observer_payload) + continue + event_id = str(event["id"]) + # Track before send so a very fast relay OK cannot race ahead + # of correlation state installation. + self._track_activity_ack(event_id, generation, observer_payload) + raw = json.dumps(["EVENT", event], separators=(",", ":")) + await asyncio.wait_for(websocket.send(raw), timeout=_ACTIVITY_SEND_TIMEOUT) + sent = True + except asyncio.CancelledError: + if event_id is not None: + self._drop_activity_ack(event_id) + self._cache_terminal_activity(observer_payload) + raise + except Exception: + if event_id is not None: + self._drop_activity_ack(event_id) + self._schedule_terminal_activity_retry(observer_payload, generation) + logger.debug("Buzz: observer activity publication failed", exc_info=True) + finally: + # Refill retained terminal frames only after forward progress. + # Replaying after a same-generation send failure creates an + # unbounded hot loop on a socket that still looks active. + if sent: + self._replay_terminal_activity() + self._activity_queue.task_done() + + async def _reset_activity_transport(self) -> None: + """Invalidate one WebSocket generation and drop all of its activity.""" + self._activity_ws_generation += 1 + retry_handle = self._activity_terminal_retry_handle + self._activity_terminal_retry_handle = None + if retry_handle is not None: + retry_handle.cancel() + task = self._activity_sender_task + self._activity_sender_task = None + try: + if task and not task.done(): + try: + self._activity_queue.put_nowait(_ACTIVITY_QUEUE_STOP) + except asyncio.QueueFull: + queued = self._activity_queue.get_nowait() + if isinstance(queued, tuple) and len(queued) == 2: + self._cache_terminal_activity(queued[1]) + self._activity_queue.task_done() + self._activity_queue.put_nowait(_ACTIVITY_QUEUE_STOP) + await task + finally: + # Cancellation propagates through the sender, but must not leave + # queue join accounting or ACK deadlines owned by a dead generation. + while True: + try: + queued = self._activity_queue.get_nowait() + if isinstance(queued, tuple) and len(queued) == 2: + self._cache_terminal_activity(queued[1]) + self._activity_queue.task_done() + except asyncio.QueueEmpty: + break + for event_id in list(self._activity_pending_event_ids): + payload = self._activity_pending_terminal_payloads.get(event_id) + if payload is not None: + self._cache_terminal_activity(payload) + self._drop_activity_ack(event_id) + + def _handle_activity_ack(self, message: list) -> bool: + """Correlate an observer EVENT acknowledgment and surface rejection.""" + if len(message) < 3 or message[0] != "OK": + return False + event_id = str(message[1]) + terminal_payload = self._activity_pending_terminal_payloads.get(event_id) + if not self._drop_activity_ack(event_id): + return False + if message[2] is not True: + if terminal_payload is not None: + self._schedule_terminal_activity_retry( + terminal_payload, self._activity_ws_generation + ) + detail = str(message[3]) if len(message) > 3 else "relay rejected event" + logger.warning("Buzz: observer activity rejected by relay: %s", detail) + return True + async def _ws_discovery_loop(self, websocket, subscriptions: Dict[str, Optional[str]]) -> None: """Periodic discovery on the poll cadence: relays don't guarantee a kind-44100 event for every new conversation. Failures retry next tick, except a closed socket: that is the same dead connection the @@ -1105,6 +1494,11 @@ async def _websocket_loop(self) -> None: ) as websocket: await self._authenticate_websocket(websocket) subscriptions = await self._subscribe_websocket(websocket) + # Observer frames ride this socket; a new generation invalidates in-flight ACK state. + self._activity_ws_generation += 1 + self._ws_connection = websocket + self._ws_active = True + self._replay_terminal_activity(new_generation=True) if self._ws_ready is not None: self._ws_ready.set() if reconnecting: @@ -1123,6 +1517,9 @@ async def _websocket_loop(self) -> None: for finished in done: finished.result() finally: + self._ws_active = False + self._ws_connection = None + await self._reset_activity_transport() for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) @@ -1197,6 +1594,8 @@ async def _handle_ws_message(self, websocket, subscriptions: Dict[str, Optional[ self._restricted_channels.add(closed_channel) del subscriptions[sub_id] self._channel_state.pop(closed_channel, None) + elif message[0] == "OK": + self._handle_activity_ack(message) elif message[0] == "NOTICE": logger.warning("Buzz: relay notice: %s", message[-1]) @@ -1981,6 +2380,80 @@ def ask(label: str, env: str) -> str: print_info("Restart the gateway for changes to take effect: hermes gateway restart") +def _handle_gateway_turn_lifecycle(*, event, route=None, **_kwargs): + """Translate neutral Gateway lifecycle metadata into Buzz observer frames.""" + + if not isinstance(route, BuzzAdapter) or not route.activity_owner_pubkey: + return False + + phase = getattr(event, "phase", "") + payload: Dict[str, Any] + kind: str + if phase == "turn_started": + triggering_id = str(getattr(event, "triggering_event_id", "") or "") + payload = { + "source": "channel", + "triggeringEventIds": ( + [triggering_id] + if re.fullmatch(r"[0-9a-fA-F]{64}", triggering_id) + else [] + ), + } + kind = "turn_started" + elif phase == "session_resolved": + payload = { + "sessionId": str(getattr(event, "session_id", "") or ""), + "isNewSession": bool(getattr(event, "is_new_session", False)), + } + kind = "session_resolved" + elif phase == "turn_liveness": + payload = {} + kind = "turn_liveness" + elif phase in {"tool_started", "tool_finished"}: + status = str(getattr(event, "tool_status", "") or "") + payload = { + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": str(getattr(event, "session_id", "") or ""), + "update": { + "sessionUpdate": ( + "tool_call" if phase == "tool_started" else "tool_call_update" + ), + "toolCallId": str(getattr(event, "tool_call_id", "") or ""), + "title": str(getattr(event, "tool_name", "") or ""), + "toolName": str(getattr(event, "tool_name", "") or ""), + "status": status, + "rawInput": {}, + }, + }, + } + kind = "acp_read" + elif phase == "turn_finished": + outcome = str(getattr(event, "outcome", None) or "failed") + if outcome == "success": + kind = "turn_completed" + payload = {} + else: + kind = "turn_error" + payload = {"status": outcome} + else: + return False + + try: + return route._enqueue_activity( + kind, + channel_id=getattr(event, "channel_id", None), + session_id=getattr(event, "session_id", None), + turn_id=getattr(event, "turn_id", None), + started_at=getattr(event, "started_at", None), + payload=payload, + ) + except Exception: + logger.debug("Buzz: Gateway lifecycle translation failed open", exc_info=True) + return False + + def register(ctx): """Plugin entry point: called by the Hermes plugin system.""" ctx.register_platform( diff --git a/plugins/platforms/buzz/nostr_auth.py b/plugins/platforms/buzz/nostr_auth.py index 9b513573f9bf..5232f831926e 100644 --- a/plugins/platforms/buzz/nostr_auth.py +++ b/plugins/platforms/buzz/nostr_auth.py @@ -2,9 +2,12 @@ from __future__ import annotations +import base64 import hashlib +import hmac import json import secrets +import struct import time from typing import Any, Optional @@ -123,6 +126,169 @@ def public_key_hex(private_key: str) -> str: return point[0].to_bytes(32, "big").hex() +def _lift_x(public_key: str) -> tuple[int, int]: + try: + x = int(public_key, 16) + except ValueError as exc: + raise ValueError("public key must be 64 hex characters") from exc + if len(public_key) != 64 or x >= FIELD_ORDER: + raise ValueError("public key must be a valid 32-byte x-only key") + y_squared = (pow(x, 3, FIELD_ORDER) + 7) % FIELD_ORDER + y = pow(y_squared, (FIELD_ORDER + 1) // 4, FIELD_ORDER) + if pow(y, 2, FIELD_ORDER) != y_squared: + raise ValueError("public key is not on secp256k1") + return x, y if y % 2 == 0 else FIELD_ORDER - y + + +def validate_x_only_public_key(public_key: str) -> str: + """Return a normalized x-only key or raise when it is not on secp256k1.""" + + normalized = str(public_key or "").strip().lower() + _lift_x(normalized) + return normalized + + +def _hkdf_expand(prk: bytes, info: bytes, length: int) -> bytes: + output = bytearray() + previous = b"" + counter = 1 + while len(output) < length: + previous = hmac.new(prk, previous + info + bytes([counter]), hashlib.sha256).digest() + output.extend(previous) + counter += 1 + return bytes(output[:length]) + + +def _rotate_left(value: int, shift: int) -> int: + return ((value << shift) & 0xFFFFFFFF) | (value >> (32 - shift)) + + +def _chacha20_xor(key: bytes, nonce: bytes, payload: bytes) -> bytes: + if len(key) != 32 or len(nonce) != 12: + raise ValueError("ChaCha20 requires a 32-byte key and 12-byte nonce") + + def quarter_round(state: list[int], a: int, b: int, c: int, d: int) -> None: + state[a] = (state[a] + state[b]) & 0xFFFFFFFF + state[d] = _rotate_left(state[d] ^ state[a], 16) + state[c] = (state[c] + state[d]) & 0xFFFFFFFF + state[b] = _rotate_left(state[b] ^ state[c], 12) + state[a] = (state[a] + state[b]) & 0xFFFFFFFF + state[d] = _rotate_left(state[d] ^ state[a], 8) + state[c] = (state[c] + state[d]) & 0xFFFFFFFF + state[b] = _rotate_left(state[b] ^ state[c], 7) + + constants = list(struct.unpack("<4I", b"expand 32-byte k")) + key_words = list(struct.unpack("<8I", key)) + nonce_words = list(struct.unpack("<3I", nonce)) + encrypted = bytearray() + for block_index in range((len(payload) + 63) // 64): + initial = constants + key_words + [block_index] + nonce_words + state = initial.copy() + for _ in range(10): + quarter_round(state, 0, 4, 8, 12) + quarter_round(state, 1, 5, 9, 13) + quarter_round(state, 2, 6, 10, 14) + quarter_round(state, 3, 7, 11, 15) + quarter_round(state, 0, 5, 10, 15) + quarter_round(state, 1, 6, 11, 12) + quarter_round(state, 2, 7, 8, 13) + quarter_round(state, 3, 4, 9, 14) + key_stream = struct.pack( + "<16I", + *((word + original) & 0xFFFFFFFF for word, original in zip(state, initial)), + ) + chunk = payload[block_index * 64 : (block_index + 1) * 64] + encrypted.extend(left ^ right for left, right in zip(chunk, key_stream)) + return bytes(encrypted) + + +def _nip44_padded_length(length: int) -> int: + if length <= 32: + return 32 + next_power = 1 << (length - 1).bit_length() + chunk = 32 if next_power <= 256 else next_power // 8 + return chunk * ((length - 1) // chunk + 1) + + +def nip44_encrypt( + plaintext: str, + *, + private_key: str, + recipient_pubkey: str, + nonce: Optional[bytes] = None, +) -> str: + """Encrypt UTF-8 text with NIP-44 v2 for an x-only secp256k1 recipient.""" + encoded = plaintext.encode("utf-8") + if not 1 <= len(encoded) <= 65_535: + raise ValueError("NIP-44 plaintext must contain 1 to 65535 bytes") + nonce = secrets.token_bytes(32) if nonce is None else nonce + if len(nonce) != 32: + raise ValueError("NIP-44 nonce must be 32 bytes") + + shared_point = _point_multiply(decode_private_key(private_key), _lift_x(recipient_pubkey)) + if shared_point is None: # pragma: no cover - validated nonzero keys make this unreachable + raise ValueError("invalid NIP-44 shared point") + shared_x = shared_point[0].to_bytes(32, "big") + conversation_key = hmac.new(b"nip44-v2", shared_x, hashlib.sha256).digest() + message_keys = _hkdf_expand(conversation_key, nonce, 76) + chacha_key = message_keys[:32] + chacha_nonce = message_keys[32:44] + hmac_key = message_keys[44:] + + prefix = len(encoded).to_bytes(2, "big") + padded = prefix + encoded + bytes(_nip44_padded_length(len(encoded)) - len(encoded)) + ciphertext = _chacha20_xor(chacha_key, chacha_nonce, padded) + mac = hmac.new(hmac_key, nonce + ciphertext, hashlib.sha256).digest() + return base64.b64encode(b"\x02" + nonce + ciphertext + mac).decode("ascii") + + +def build_observer_event( + *, + private_key: str, + owner_pubkey: str, + payload: dict[str, Any], + created_at: Optional[int] = None, + nonce: Optional[bytes] = None, + auxiliary_randomness: Optional[bytes] = None, +) -> dict[str, Any]: + """Build a signed, owner-encrypted NIP-AO telemetry event (kind 24200).""" + plaintext = json.dumps(payload, separators=(",", ":"), ensure_ascii=False) + if len(plaintext.encode("utf-8")) > 65_535: + raise ValueError("observer plaintext exceeds 65535 bytes") + content = nip44_encrypt( + plaintext, + private_key=private_key, + recipient_pubkey=owner_pubkey, + nonce=nonce, + ) + pubkey = public_key_hex(private_key) + timestamp = int(time.time()) if created_at is None else int(created_at) + tags = [ + ["p", owner_pubkey.lower()], + ["agent", pubkey], + ["frame", "telemetry"], + ] + serialized = json.dumps( + [0, pubkey, timestamp, 24200, tags, content], + separators=(",", ":"), + ensure_ascii=False, + ).encode() + event_id = hashlib.sha256(serialized).digest() + return { + "id": event_id.hex(), + "pubkey": pubkey, + "created_at": timestamp, + "kind": 24200, + "tags": tags, + "content": content, + "sig": schnorr_sign( + event_id, + private_key, + auxiliary_randomness=auxiliary_randomness, + ).hex(), + } + + def schnorr_sign(message: bytes, private_key: str, *, auxiliary_randomness: Optional[bytes] = None) -> bytes: if len(message) != 32: raise ValueError("BIP-340 signs a 32-byte message") diff --git a/tests/gateway/test_buzz_activity_bridge.py b/tests/gateway/test_buzz_activity_bridge.py new file mode 100644 index 000000000000..0e2731e2062d --- /dev/null +++ b/tests/gateway/test_buzz_activity_bridge.py @@ -0,0 +1,445 @@ +"""Native Gateway lifecycle → Buzz edge-translation tests.""" + +import asyncio +import json +from dataclasses import asdict +from types import SimpleNamespace + +import pytest + +from gateway.config import Platform +from gateway.run import GatewayRunner, TurnRunner +from gateway.session import SessionSource +from gateway.turn_observer import GatewayTurnObserver +from tests.gateway._plugin_adapter_loader import load_plugin_adapter + +_buzz_mod = load_plugin_adapter("buzz") +BuzzAdapter = _buzz_mod.BuzzAdapter +_nostr_auth = _buzz_mod._load_nostr_auth() + + +class _CaptureRoute: + def __init__(self, events): + self.events = events + + def on_turn_lifecycle(self, event): + self.events.append(event) + return True + + +def _observer(loop, *, is_new_session=True, events=None): + return GatewayTurnObserver( + platform="buzz", + profile="default", + channel_id="channel-1", + session_id="session-1", + triggering_event_id="a" * 64, + is_new_session=is_new_session, + route=_CaptureRoute(events) if events is not None else object(), + loop=loop, + is_current=lambda: True, + ) + + +def _capture_lifecycle(): + return [] + + +@pytest.mark.asyncio +async def test_generic_lifecycle_preserves_session_novelty_and_terminal_order( + monkeypatch, +): + events = _capture_lifecycle() + observer = _observer( + asyncio.get_running_loop(), is_new_session=True, events=events + ) + + assert observer.start(liveness_interval=0) is True + assert observer.session_resolved() is True + assert observer.finish({"final_response": "done"}) is True + assert observer.finish({"failed": True}) is False + + assert [event.phase for event in events] == [ + "turn_started", + "session_resolved", + "turn_finished", + ] + assert events[1].is_new_session is True + assert events[2].outcome == "success" + assert len({event.turn_id for event in events}) == 1 + + +@pytest.mark.asyncio +async def test_liveness_starts_with_turn_and_stops_at_terminal(monkeypatch): + events = _capture_lifecycle() + observer = _observer(asyncio.get_running_loop(), events=events) + + observer.start(liveness_interval=0.005) + await asyncio.sleep(0.012) + observer.finish({"final_response": "done"}) + count_after_finish = len(events) + await asyncio.sleep(0.012) + + assert any(event.phase == "turn_liveness" for event in events) + assert len(events) == count_after_finish + assert events[-1].phase == "turn_finished" + + +@pytest.mark.asyncio +async def test_tool_events_are_allowlisted_and_omit_sensitive_content(monkeypatch): + events = _capture_lifecycle() + observer = _observer(asyncio.get_running_loop(), events=events) + observer.start(liveness_interval=0) + + secret_command = "curl -H 'Authorization: Bearer token' /private/path" + secret_result = json.dumps({"exit_code": 1, "output": "private-error"}) + observer.tool_started("call-1", "terminal", {"command": secret_command}) + observer.tool_finished( + "call-1", "terminal", {"command": secret_command}, secret_result + ) + + tool_events = [event for event in events if event.phase.startswith("tool_")] + assert [event.tool_status for event in tool_events] == ["executing", "failed"] + assert [event.tool_call_id for event in tool_events] == ["tool-1", "tool-1"] + serialized = json.dumps([asdict(event) for event in tool_events]) + assert secret_command not in serialized + assert secret_result not in serialized + assert "private-error" not in serialized + + +@pytest.mark.asyncio +@pytest.mark.parametrize("fail", [False, True]) +async def test_proxy_gateway_turn_emits_terminal_without_masking_error(fail): + events = _capture_lifecycle() + route = _CaptureRoute(events) + runner = object.__new__(GatewayRunner) + runner.config = SimpleNamespace(multiplex_profiles=False) + runner._get_proxy_url = lambda: "http://proxy.invalid" + runner._adapter_for_source = lambda _source: route + runner._is_session_run_current = lambda *_args: True + + async def run_proxy(**_kwargs): + if fail: + raise RuntimeError("proxy root cause") + return {"final_response": "done", "completed": True} + + runner._run_agent_via_proxy = run_proxy + source = SessionSource( + platform=Platform("buzz"), + user_id="owner", + chat_id="channel-1", + user_name="owner", + chat_type="channel", + ) + + if fail: + with pytest.raises(RuntimeError, match="proxy root cause"): + await runner._run_agent( + "hello", "", [], source, "session-1", session_key="buzz:channel-1" + ) + else: + result = await runner._run_agent( + "hello", "", [], source, "session-1", session_key="buzz:channel-1" + ) + assert result["final_response"] == "done" + + assert [event.phase for event in events] == [ + "turn_started", + "session_resolved", + "turn_finished", + ] + assert events[-1].outcome == ("failed" if fail else "success") + + +@pytest.mark.asyncio +async def test_gateway_wrapper_reports_early_inner_setup_failure(): + events = _capture_lifecycle() + route = _CaptureRoute(events) + runner = object.__new__(GatewayRunner) + runner.config = SimpleNamespace(multiplex_profiles=False) + runner._adapter_for_source = lambda _source: route + runner._is_session_run_current = lambda *_args: True + + async def fail_during_setup(*_args, **_kwargs): + raise RuntimeError("early setup root cause") + + runner._run_agent_inner = fail_during_setup + source = SessionSource( + platform=Platform("buzz"), + user_id="owner", + chat_id="channel-1", + user_name="owner", + chat_type="channel", + ) + + with pytest.raises(RuntimeError, match="early setup root cause"): + await runner._run_agent("hello", "", [], source, "session-1") + + assert [event.phase for event in events] == [ + "turn_started", + "session_resolved", + "turn_finished", + ] + assert events[-1].outcome == "failed" + + +@pytest.mark.asyncio +async def test_post_execution_cancellation_does_not_rewrite_latched_success(): + events = _capture_lifecycle() + route = _CaptureRoute(events) + runner = object.__new__(GatewayRunner) + runner.config = SimpleNamespace(multiplex_profiles=False) + runner._adapter_for_source = lambda _source: route + runner._is_session_run_current = lambda *_args: True + + async def cancel_after_execution(*_args, **kwargs): + kwargs["turn_observer"].finish( + {"final_response": "done", "completed": True} + ) + raise asyncio.CancelledError + + runner._run_agent_inner = cancel_after_execution + source = SessionSource( + platform=Platform("buzz"), + user_id="owner", + chat_id="channel-1", + user_name="owner", + chat_type="channel", + ) + + with pytest.raises(asyncio.CancelledError): + await runner._run_agent("hello", "", [], source, "session-1") + + terminal = [event for event in events if event.phase == "turn_finished"] + assert len(terminal) == 1 + assert terminal[0].outcome == "success" + + +@pytest.mark.parametrize( + ("result", "exception_type", "outcome"), + [ + ({"final_response": "done"}, None, "success"), + ({"failed": True}, None, "failed"), + ({"interrupted": True}, None, "cancelled"), + ({"final_response": "done"}, RuntimeError, "failed"), + (None, None, "failed"), + ], +) +@pytest.mark.asyncio +async def test_terminal_outcome_matrix(monkeypatch, result, exception_type, outcome): + events = _capture_lifecycle() + observer = _observer(asyncio.get_running_loop(), events=events) + observer.start(liveness_interval=0) + observer.finish(result, exception_type=exception_type) + + terminal = [event for event in events if event.phase == "turn_finished"] + assert len(terminal) == 1 + assert terminal[0].outcome == outcome + + +@pytest.mark.asyncio +async def test_inactivity_timeout_failed_result_reports_timed_out(monkeypatch): + events = _capture_lifecycle() + observer = _observer(asyncio.get_running_loop(), events=events) + observer.start(liveness_interval=0) + + observer.finish({"failed": True}, timed_out=True) + + terminal = [event for event in events if event.phase == "turn_finished"] + assert [event.outcome for event in terminal] == ["timed_out"] + + +@pytest.mark.asyncio +async def test_propagated_task_cancellation_reports_cancelled(monkeypatch): + events = _capture_lifecycle() + observer = _observer(asyncio.get_running_loop(), events=events) + observer.start(liveness_interval=0) + + observer.finish(None, exception_type=asyncio.CancelledError) + + terminal = [event for event in events if event.phase == "turn_finished"] + assert [event.outcome for event in terminal] == ["cancelled"] + + +@pytest.mark.asyncio +async def test_no_listener_keeps_observer_and_liveness_disabled(monkeypatch): + observer = _observer(asyncio.get_running_loop()) + + assert observer.start(liveness_interval=0.001) is False + await asyncio.sleep(0.005) + assert observer.finish({"final_response": "done"}) is False + + +def _buzz_adapter(): + from gateway.config import PlatformConfig + + owner = _nostr_auth.public_key_hex("00" * 31 + "02") + return BuzzAdapter( + PlatformConfig( + enabled=True, + extra={ + "relay_url": "https://test.relay", + "activity_owner_pubkey": owner, + }, + ) + ) + + +@pytest.mark.parametrize( + ("phase", "fields", "expected_kind"), + [ + ("turn_started", {"triggering_event_id": "a" * 64}, "turn_started"), + ("session_resolved", {"is_new_session": True}, "session_resolved"), + ("turn_liveness", {}, "turn_liveness"), + ( + "tool_started", + { + "tool_call_id": "call-1", + "tool_name": "terminal", + "tool_status": "executing", + }, + "acp_read", + ), + ( + "tool_finished", + { + "tool_call_id": "call-1", + "tool_name": "terminal", + "tool_status": "failed", + }, + "acp_read", + ), + ("turn_finished", {"outcome": "success"}, "turn_completed"), + ("turn_finished", {"outcome": "failed"}, "turn_error"), + ("turn_finished", {"outcome": "cancelled"}, "turn_error"), + ("turn_finished", {"outcome": "timed_out"}, "turn_error"), + ], +) +def test_buzz_edge_translates_neutral_lifecycle_only( + monkeypatch, phase, fields, expected_kind +): + adapter = _buzz_adapter() + captured = [] + monkeypatch.setattr( + adapter, + "_enqueue_activity", + lambda kind, **kwargs: captured.append((kind, kwargs)) or True, + ) + event_fields = { + "phase": phase, + "channel_id": "channel-1", + "session_id": "session-1", + "turn_id": "turn-1", + "started_at": "2026-08-04T00:00:00.000Z", + "triggering_event_id": None, + "is_new_session": None, + "tool_call_id": None, + "tool_name": None, + "tool_status": None, + "outcome": None, + } + event_fields.update(fields) + event = SimpleNamespace(**event_fields) + + assert _buzz_mod._handle_gateway_turn_lifecycle(event=event, route=adapter) + assert captured[0][0] == expected_kind + serialized = json.dumps(captured[0][1]) + assert "Authorization" not in serialized + assert "/private/" not in serialized + + if phase == "session_resolved": + assert captured[0][1]["payload"]["isNewSession"] is True + if phase.startswith("tool_"): + update = captured[0][1]["payload"]["params"]["update"] + assert update["rawInput"] == {} + assert update["toolCallId"] == "call-1" + if phase == "turn_finished" and fields["outcome"] != "success": + assert captured[0][1]["payload"]["status"] == fields["outcome"] + + +def test_buzz_edge_ignores_non_buzz_routes(): + event = SimpleNamespace(phase="turn_started") + assert not _buzz_mod._handle_gateway_turn_lifecycle( + event=event, route=SimpleNamespace() + ) + + +@pytest.mark.asyncio +async def test_structured_callbacks_wire_for_active_observer_and_preserve_existing( + monkeypatch, +): + calls = [] + events = _capture_lifecycle() + observer = _observer(asyncio.get_running_loop(), events=events) + assert observer.active is False + + context = SimpleNamespace(_voice_ack_guild=[None]) + runner = TurnRunner(SimpleNamespace(), context, observer) + + def prior_start(*_args): + calls.append(("prior-start", ())) + raise RuntimeError("must not suppress observer") + + def prior_finish(*_args): + calls.append(("prior-finish", ())) + + agent = SimpleNamespace( + tool_start_callback=prior_start, + tool_complete_callback=prior_finish, + ) + observer.start(liveness_interval=0) + runner.wire_structured_tool_callbacks(agent) + + assert callable(agent.tool_start_callback) + assert callable(agent.tool_complete_callback) + agent.tool_start_callback("call-1", "terminal", {"secret": "value"}) + agent.tool_complete_callback("call-1", "terminal", {}, "result") + + assert [name for name, _ in calls] == ["prior-start", "prior-finish"] + assert [event.phase for event in events] == [ + "turn_started", + "tool_started", + "tool_finished", + ] + + +def test_inactive_observer_does_not_wrap_existing_tool_callbacks(): + loop = asyncio.new_event_loop() + try: + observer = _observer(loop, events=[]) + context = SimpleNamespace(_voice_ack_guild=[None]) + runner = TurnRunner(SimpleNamespace(), context, observer) + prior_start = lambda *_args: None + prior_finish = lambda *_args: None + agent = SimpleNamespace( + tool_start_callback=prior_start, + tool_complete_callback=prior_finish, + ) + + runner.wire_structured_tool_callbacks(agent) + + assert agent.tool_start_callback is prior_start + assert agent.tool_complete_callback is prior_finish + finally: + loop.close() + + +def test_gateway_core_has_no_platform_activity_transport_coupling(): + """Core emits neutral lifecycle events but never knows Buzz wire semantics.""" + + import inspect + import gateway.run as gateway_run + import gateway.turn_context as turn_context + + source = inspect.getsource(gateway_run.TurnRunner) + inspect.getsource( + turn_context.TurnContext + ) + for forbidden in ( + "publish_activity", + "activity_owner_pubkey", + "acp_read", + "session/update", + "Nostr", + "Buzz activity", + ): + assert forbidden not in source diff --git a/tests/gateway/test_buzz_websocket.py b/tests/gateway/test_buzz_websocket.py index 56a15e6d58fa..dce2e731c9a1 100644 --- a/tests/gateway/test_buzz_websocket.py +++ b/tests/gateway/test_buzz_websocket.py @@ -8,7 +8,9 @@ import asyncio import json +import sys import time +from types import SimpleNamespace import pytest @@ -25,7 +27,7 @@ nostr_auth = _ilu.module_from_spec(_spec) _spec.loader.exec_module(nostr_auth) -SELF_PUBKEY = "9fd5c7ba6d3ef224da78f541e0fcb9c50f72cc63edb19aae76ac6a0474dfa860" +SELF_PUBKEY = "f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9" # public_key_hex(TEST_PRIVATE_KEY) # BIP-340 test vector 0 private key TEST_PRIVATE_KEY = "00" * 31 + "03" CHANNEL = "ccc2bc1a-7a82-5a8f-8c4e-57a070cbe7cd" @@ -85,6 +87,77 @@ def test_build_auth_event_shape_and_owner_tag(): assert event["pubkey"] == nostr_auth.public_key_hex(TEST_PRIVATE_KEY) +def test_nip44_encrypt_matches_official_vector(): + """Observer payload encryption must be byte-compatible with NIP-44 v2.""" + sender_private_key = "00" * 31 + "01" + recipient_private_key = "00" * 31 + "02" + recipient_pubkey = nostr_auth.public_key_hex(recipient_private_key) + + payload = nostr_auth.nip44_encrypt( + "a", + private_key=sender_private_key, + recipient_pubkey=recipient_pubkey, + nonce=bytes.fromhex("00" * 31 + "01"), + ) + + assert payload == ( + "AgAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABee0G5VSK0/9YypIObAtD" + "KfYEAjD35uVkHyB0F4DwrcNaCXlCWZKaArsGrY6M9wnuTMxWfp1RTN9Xga8no+" + "kF5Vsb" + ) + + +def test_nip44_encrypt_rejects_non_spec_plaintext_above_65535_bytes(): + with pytest.raises(ValueError, match="1 to 65535 bytes"): + nostr_auth.nip44_encrypt( + "a" * 65_536, + private_key="00" * 31 + "01", + recipient_pubkey=nostr_auth.public_key_hex("00" * 31 + "02"), + nonce=bytes(32), + ) + + +def test_build_observer_event_encrypts_and_signs_nip_ao_shape(monkeypatch): + owner_private_key = "00" * 31 + "02" + owner_pubkey = nostr_auth.public_key_hex(owner_private_key) + captured = {} + + def fake_encrypt(plaintext, **kwargs): + captured["plaintext"] = plaintext + captured.update(kwargs) + return "encrypted-observer-payload" + + monkeypatch.setattr(nostr_auth, "nip44_encrypt", fake_encrypt) + payload = { + "seq": 1, + "timestamp": "2026-08-03T14:00:00.000Z", + "kind": "turn_started", + "agentIndex": None, + "channelId": CHANNEL, + "sessionId": "session-1", + "turnId": "turn-1", + "payload": {"source": "channel"}, + } + + event = nostr_auth.build_observer_event( + private_key=TEST_PRIVATE_KEY, + owner_pubkey=owner_pubkey, + payload=payload, + created_at=1_700_000_000, + auxiliary_randomness=bytes(32), + ) + + assert event["kind"] == 24200 + assert event["content"] == "encrypted-observer-payload" + assert ["p", owner_pubkey] in event["tags"] + assert ["agent", event["pubkey"]] in event["tags"] + assert ["frame", "telemetry"] in event["tags"] + assert json.loads(captured["plaintext"]) == payload + assert captured["private_key"] == TEST_PRIVATE_KEY + assert captured["recipient_pubkey"] == owner_pubkey + assert len(bytes.fromhex(event["sig"])) == 64 + + # ── Adapter WS wiring ───────────────────────────────────────────────────── @@ -795,3 +868,666 @@ def fake_connect(*args, **kwargs): assert started, "discovery task was never started with the connection" assert all(t.done() for t in started), "discovery task outlived its connection" + + +@pytest.mark.asyncio +async def test_publish_activity_sends_encrypted_observer_event_over_active_websocket(): + owner_private_key = "00" * 31 + "02" + owner_pubkey = nostr_auth.public_key_hex(owner_private_key) + adapter = _make_adapter({"activity_owner_pubkey": owner_pubkey}) + websocket = _FakeWebSocket() + adapter._ws_active = True + adapter._ws_connection = websocket + + published = await adapter.publish_activity( + "turn_started", + channel_id=CHANNEL, + session_id="session-1", + turn_id="turn-1", + started_at="2026-08-03T14:00:00.000Z", + payload={"source": "channel"}, + ) + + assert published is True + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + assert len(websocket.sent) == 1 + frame = websocket.sent[0] + assert frame[0] == "EVENT" + assert frame[1]["kind"] == 24200 + assert ["p", owner_pubkey] in frame[1]["tags"] + assert ["agent", frame[1]["pubkey"]] in frame[1]["tags"] + assert ["frame", "telemetry"] in frame[1]["tags"] + adapter._activity_sender_task.cancel() + with pytest.raises(asyncio.CancelledError): + await adapter._activity_sender_task + + +@pytest.mark.asyncio +async def test_publish_activity_does_not_wait_for_backpressured_websocket(): + owner_pubkey = nostr_auth.public_key_hex("2".zfill(64)) + adapter = _make_adapter(extra={"activity_owner_pubkey": owner_pubkey}) + + class _BackpressuredWebSocket: + async def send(self, _payload): + await asyncio.Event().wait() + + adapter._ws_connection = _BackpressuredWebSocket() + adapter._ws_active = True + + published = await asyncio.wait_for( + adapter.publish_activity( + "turn_started", + channel_id="channel-1", + session_id="session-1", + turn_id="turn-1", + payload={}, + ), + timeout=0.05, + ) + + assert published is True + assert adapter._activity_sender_task is not None + adapter._activity_sender_task.cancel() + with pytest.raises(asyncio.CancelledError): + await adapter._activity_sender_task + + +def test_activity_owner_pubkey_rejects_malformed_config(): + with pytest.raises(ValueError, match="activity_owner_pubkey"): + _make_adapter({"activity_owner_pubkey": "not-a-pubkey"}) + + +def test_activity_owner_pubkey_rejects_poll_only_transport(): + owner_pubkey = nostr_auth.public_key_hex("00" * 31 + "02") + with pytest.raises(ValueError, match="requires transport"): + _make_adapter( + {"activity_owner_pubkey": owner_pubkey, "transport": "poll"} + ) + + +@pytest.mark.asyncio +async def test_activity_auto_fallback_reports_websocket_requirement( + monkeypatch, caplog +): + owner_pubkey = nostr_auth.public_key_hex("00" * 31 + "02") + adapter = _make_adapter({"activity_owner_pubkey": owner_pubkey}) + + def invalid_websocket_url(): + raise ValueError("websocket unavailable") + + monkeypatch.setattr(adapter, "_websocket_url", invalid_websocket_url) + with caplog.at_level("WARNING"): + assert await adapter._start_websocket() is False + + assert "native activity is unavailable while using polling" in caplog.text + + +@pytest.mark.asyncio +async def test_terminal_send_failure_retries_once_on_same_live_socket(monkeypatch): + owner_pubkey = nostr_auth.public_key_hex("00" * 31 + "02") + adapter = _make_adapter({"activity_owner_pubkey": owner_pubkey}) + monkeypatch.setattr(_buzz_mod, "_ACTIVITY_TERMINAL_RETRY_DELAY", 0.01) + + event_counter = 0 + + def build_event(**kwargs): + nonlocal event_counter + event_counter += 1 + return { + "id": f"terminal-event-{event_counter}", + "kind": 24200, + "payload": kwargs["payload"], + } + + monkeypatch.setattr( + _buzz_mod, + "_load_nostr_auth", + lambda: SimpleNamespace(build_observer_event=build_event), + ) + + class FlakyWebSocket: + def __init__(self): + self.send_count = 0 + self.delivered = [] + + async def send(self, raw): + self.send_count += 1 + if self.send_count == 1: + raise TimeoutError("relay backpressure") + self.delivered.append(json.loads(raw)) + + websocket = FlakyWebSocket() + adapter._ws_active = True + adapter._ws_connection = websocket + generation = adapter._activity_ws_generation + + assert await adapter.publish_activity( + "turn_completed", + channel_id=CHANNEL, + session_id="session-1", + turn_id="turn-1", + ) is True + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + assert websocket.send_count == 1 + assert list(adapter._activity_terminal_replay) == ["turn-1"] + assert not adapter._activity_pending_event_ids + + await asyncio.sleep(0.03) + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + assert websocket.send_count == 2 + assert adapter._ws_connection is websocket + assert adapter._activity_ws_generation == generation + assert [frame[1]["payload"]["kind"] for frame in websocket.delivered] == [ + "turn_completed" + ] + assert not adapter._activity_terminal_replay + + await adapter._reset_activity_transport() + + +@pytest.mark.asyncio +async def test_reset_activity_transport_propagates_cancellation(): + adapter = _make_adapter() + sender_blocked = asyncio.Event() + adapter._activity_sender_task = asyncio.create_task(sender_blocked.wait()) + + reset_task = asyncio.create_task(adapter._reset_activity_transport()) + await asyncio.sleep(0) + reset_task.cancel() + + with pytest.raises(asyncio.CancelledError): + await reset_task + assert reset_task.cancelled() + + +@pytest.mark.asyncio +async def test_disconnect_completes_while_websocket_loop_is_resetting(): + adapter = _make_adapter() + reset_entered = asyncio.Event() + sender_blocked = asyncio.Event() + adapter._activity_sender_task = asyncio.create_task(sender_blocked.wait()) + + async def reconnecting_websocket_loop(): + while True: + try: + raise ConnectionError("relay disconnected") + except Exception: + reset_entered.set() + await adapter._reset_activity_transport() + await asyncio.sleep(3600) + + adapter._ws_task = asyncio.create_task(reconnecting_websocket_loop()) + await asyncio.wait_for(reset_entered.wait(), timeout=1) + await asyncio.sleep(0) + + disconnect_task = asyncio.create_task(adapter.disconnect()) + try: + await asyncio.wait_for(asyncio.shield(disconnect_task), timeout=0.1) + finally: + if not disconnect_task.done(): + if adapter._ws_task and not adapter._ws_task.done(): + adapter._ws_task.cancel() + disconnect_task.cancel() + try: + await disconnect_task + except asyncio.CancelledError: + pass + + +@pytest.mark.asyncio +async def test_observer_relay_rejection_is_correlated_and_logged(caplog): + owner_pubkey = nostr_auth.public_key_hex("2".zfill(64)) + adapter = _make_adapter(extra={"activity_owner_pubkey": owner_pubkey}) + websocket = _FakeWebSocket() + adapter._ws_connection = websocket + adapter._ws_active = True + + assert await adapter.publish_activity( + "turn_started", + channel_id="channel-1", + session_id="session-1", + turn_id="turn-1", + started_at="2026-08-03T14:00:00.000Z", + payload={}, + ) is True + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + event_id = websocket.sent[0][1]["id"] + assert event_id in adapter._activity_pending_event_ids + + with caplog.at_level("WARNING"): + assert adapter._handle_activity_ack( + ["OK", event_id, False, "restricted: not authorized"] + ) is True + + assert event_id not in adapter._activity_pending_event_ids + assert "restricted: not authorized" in caplog.text + adapter._activity_sender_task.cancel() + with pytest.raises(asyncio.CancelledError): + await adapter._activity_sender_task + + +@pytest.mark.asyncio +async def test_rejected_terminal_is_retained_and_retried_once(monkeypatch, caplog): + owner_pubkey = nostr_auth.public_key_hex("2".zfill(64)) + adapter = _make_adapter(extra={"activity_owner_pubkey": owner_pubkey}) + websocket = _FakeWebSocket() + adapter._ws_connection = websocket + adapter._ws_active = True + monkeypatch.setattr(_buzz_mod, "_ACTIVITY_TERMINAL_RETRY_DELAY", 0.01) + + event_counter = 0 + + def build_event(**kwargs): + nonlocal event_counter + event_counter += 1 + return { + "id": f"terminal-event-{event_counter}", + "kind": 24200, + "payload": kwargs["payload"], + } + + monkeypatch.setattr( + _buzz_mod, + "_load_nostr_auth", + lambda: SimpleNamespace(build_observer_event=build_event), + ) + + assert await adapter.publish_activity( + "turn_completed", + channel_id=CHANNEL, + session_id="session-1", + turn_id="turn-1", + ) + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + + with caplog.at_level("WARNING"): + assert adapter._handle_activity_ack( + ["OK", "terminal-event-1", False, "rate-limited: slow down"] + ) is True + + assert list(adapter._activity_terminal_replay) == ["turn-1"] + await asyncio.sleep(0.03) + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + assert len(websocket.sent) == 2 + + assert adapter._handle_activity_ack( + ["OK", "terminal-event-2", False, "rate-limited: slow down"] + ) is True + await asyncio.sleep(0.03) + assert len(websocket.sent) == 2 + assert list(adapter._activity_terminal_replay) == ["turn-1"] + assert "rate-limited: slow down" in caplog.text + + await adapter._reset_activity_transport() + + +@pytest.mark.asyncio +async def test_disconnect_drops_stale_activity_queue_and_pending_acks(): + adapter = _make_adapter() + adapter._track_activity_ack("event-id", adapter._activity_ws_generation) + _, timer = adapter._activity_pending_event_ids["event-id"] + adapter._activity_queue.put_nowait({"kind": "turn_liveness"}) + + await adapter.disconnect() + + assert adapter._activity_queue.empty() + assert not adapter._activity_pending_event_ids + assert timer.cancelled() + + +@pytest.mark.asyncio +async def test_terminal_during_disconnect_replays_once_after_reconnect(monkeypatch): + owner_pubkey = nostr_auth.public_key_hex("00" * 31 + "02") + adapter = _make_adapter({"activity_owner_pubkey": owner_pubkey}) + old_websocket = _FakeWebSocket() + adapter._ws_active = True + adapter._ws_connection = old_websocket + + counter = 0 + + def build_event(**kwargs): + nonlocal counter + counter += 1 + return { + "id": f"event-{counter}", + "kind": 24200, + "payload": kwargs["payload"], + } + + monkeypatch.setattr( + _buzz_mod, + "_load_nostr_auth", + lambda: SimpleNamespace(build_observer_event=build_event), + ) + + assert adapter._enqueue_activity( + "turn_started", + channel_id=CHANNEL, + session_id="session-1", + turn_id="turn-1", + ) + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + assert [frame[1]["payload"]["kind"] for frame in old_websocket.sent] == [ + "turn_started" + ] + + adapter._ws_active = False + adapter._ws_connection = None + await adapter._reset_activity_transport() + assert not adapter._enqueue_activity( + "turn_liveness", + channel_id=CHANNEL, + session_id="session-1", + turn_id="turn-1", + ) + assert adapter._enqueue_activity( + "turn_completed", + channel_id=CHANNEL, + session_id="session-1", + turn_id="turn-1", + ) + + new_websocket = _FakeWebSocket() + adapter._activity_ws_generation += 1 + adapter._ws_connection = new_websocket + adapter._ws_active = True + adapter._replay_terminal_activity() + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + + assert [frame[1]["payload"]["kind"] for frame in new_websocket.sent] == [ + "turn_completed" + ] + assert not adapter._activity_terminal_replay + adapter._replay_terminal_activity() + await asyncio.sleep(0) + assert len(new_websocket.sent) == 1 + adapter._activity_sender_task.cancel() + with pytest.raises(asyncio.CancelledError): + await adapter._activity_sender_task + + +@pytest.mark.asyncio +async def test_websocket_loop_replays_terminal_after_real_reconnect(monkeypatch): + owner_pubkey = nostr_auth.public_key_hex("00" * 31 + "02") + adapter = _make_adapter({"activity_owner_pubkey": owner_pubkey}) + second_delivery = asyncio.Event() + + class RelaySocket(_FakeWebSocket): + def __init__(self, *, disconnect_after_terminal): + super().__init__() + self.disconnect_after_terminal = disconnect_after_terminal + self.terminal_sent = asyncio.Event() + + async def send(self, raw): + frame = json.loads(raw) + self.sent.append(frame) + if frame[0] == "EVENT" and frame[1].get("kind") == 24200: + self.terminal_sent.set() + + def __aiter__(self): + async def frames(): + await self.terminal_sent.wait() + if not self.disconnect_after_terminal: + second_delivery.set() + await asyncio.Future() + if False: + yield "" + + return frames() + + first_socket = RelaySocket(disconnect_after_terminal=True) + second_socket = RelaySocket(disconnect_after_terminal=False) + sockets = iter((first_socket, second_socket)) + + class RelayConnection: + def __init__(self, websocket): + self.websocket = websocket + + async def __aenter__(self): + return self.websocket + + async def __aexit__(self, exc_type, exc, traceback): + return False + + def connect(*args, **kwargs): + return RelayConnection(next(sockets)) + + monkeypatch.setitem(sys.modules, "websockets", SimpleNamespace(connect=connect)) + terminal_payload = { + "kind": "turn_completed", + "seq": 1, + "timestamp": "2026-08-03T14:00:00.000Z", + "channelId": CHANNEL, + "sessionId": "session-1", + "turnId": "turn-1", + "payload": {}, + } + assert adapter._cache_terminal_activity(terminal_payload) + + websocket_task = asyncio.create_task(adapter._websocket_loop()) + try: + await asyncio.wait_for(second_delivery.wait(), timeout=3) + first_terminal = [ + frame for frame in first_socket.sent if frame[0] == "EVENT" + ] + second_terminal = [ + frame for frame in second_socket.sent if frame[0] == "EVENT" + ] + assert len(first_terminal) == 1 + assert len(second_terminal) == 1 + assert adapter._activity_ws_generation >= 3 + finally: + if not websocket_task.done(): + websocket_task.cancel() + try: + await websocket_task + except asyncio.CancelledError: + pass + + +def test_terminal_replay_is_bounded_and_keeps_latest_turns(monkeypatch): + owner_pubkey = nostr_auth.public_key_hex("00" * 31 + "02") + adapter = _make_adapter({"activity_owner_pubkey": owner_pubkey}) + monkeypatch.setattr(_buzz_mod, "_ACTIVITY_TERMINAL_REPLAY_CAP", 2) + + for turn_id in ("turn-1", "turn-2", "turn-3"): + assert adapter._enqueue_activity( + "turn_error", + channel_id=CHANNEL, + session_id="session-1", + turn_id=turn_id, + payload={"status": "failed"}, + ) + + assert list(adapter._activity_terminal_replay) == ["turn-2", "turn-3"] + + +@pytest.mark.asyncio +async def test_unacked_terminal_is_recovered_when_socket_disconnects(monkeypatch): + owner_pubkey = nostr_auth.public_key_hex("00" * 31 + "02") + adapter = _make_adapter({"activity_owner_pubkey": owner_pubkey}) + old_websocket = _FakeWebSocket() + adapter._ws_active = True + adapter._ws_connection = old_websocket + + counter = 0 + + def build_event(**kwargs): + nonlocal counter + counter += 1 + return { + "id": f"event-{counter}", + "kind": 24200, + "payload": kwargs["payload"], + } + + monkeypatch.setattr( + _buzz_mod, + "_load_nostr_auth", + lambda: SimpleNamespace(build_observer_event=build_event), + ) + + assert adapter._enqueue_activity( + "turn_completed", + channel_id=CHANNEL, + session_id="session-1", + turn_id="turn-1", + ) + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + assert "event-1" in adapter._activity_pending_event_ids + + adapter._ws_active = False + adapter._ws_connection = None + await adapter._reset_activity_transport() + assert not adapter._activity_pending_event_ids + assert list(adapter._activity_terminal_replay) == ["turn-1"] + + new_websocket = _FakeWebSocket() + adapter._activity_ws_generation += 1 + adapter._ws_connection = new_websocket + adapter._ws_active = True + adapter._replay_terminal_activity() + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + + assert [frame[1]["payload"]["kind"] for frame in new_websocket.sent] == [ + "turn_completed" + ] + adapter._activity_sender_task.cancel() + with pytest.raises(asyncio.CancelledError): + await adapter._activity_sender_task + + +@pytest.mark.asyncio +async def test_activity_sender_drops_frame_when_websocket_generation_changes(monkeypatch): + owner_pubkey = nostr_auth.public_key_hex("00" * 31 + "02") + adapter = _make_adapter({"activity_owner_pubkey": owner_pubkey}) + old_websocket = _FakeWebSocket() + new_websocket = _FakeWebSocket() + adapter._ws_active = True + adapter._ws_connection = old_websocket + adapter._activity_ws_generation = 1 + + def build_during_reconnect(**kwargs): + adapter._activity_ws_generation = 2 + adapter._ws_connection = new_websocket + return {"id": "event-id"} + + monkeypatch.setattr( + _buzz_mod, + "_load_nostr_auth", + lambda: SimpleNamespace(build_observer_event=build_during_reconnect), + ) + adapter._activity_queue.put_nowait((1, {"kind": "turn_liveness"})) + adapter._activity_sender_task = asyncio.create_task(adapter._activity_sender_loop()) + + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + adapter._activity_sender_task.cancel() + with pytest.raises(asyncio.CancelledError): + await adapter._activity_sender_task + + assert old_websocket.sent == [] + assert new_websocket.sent == [] + assert not adapter._activity_pending_event_ids + + +@pytest.mark.asyncio +async def test_activity_ack_expires_on_deadline_and_late_ack_is_ignored(monkeypatch): + owner_pubkey = nostr_auth.public_key_hex("00" * 31 + "02") + adapter = _make_adapter({"activity_owner_pubkey": owner_pubkey}) + websocket = _FakeWebSocket() + adapter._ws_active = True + adapter._ws_connection = websocket + monkeypatch.setattr(_buzz_mod, "_ACTIVITY_ACK_TIMEOUT", 0.01) + + assert await adapter.publish_activity( + "turn_started", + channel_id=CHANNEL, + session_id="session-1", + turn_id="turn-1", + payload={}, + ) + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + event_id = websocket.sent[0][1]["id"] + assert event_id in adapter._activity_pending_event_ids + + await asyncio.sleep(0.03) + assert event_id not in adapter._activity_pending_event_ids + assert adapter._handle_activity_ack(["OK", event_id, True, "late"]) is False + + adapter._activity_sender_task.cancel() + with pytest.raises(asyncio.CancelledError): + await adapter._activity_sender_task + + +@pytest.mark.asyncio +async def test_unacked_terminal_is_retained_on_ack_timeout(monkeypatch): + owner_pubkey = nostr_auth.public_key_hex("00" * 31 + "02") + adapter = _make_adapter({"activity_owner_pubkey": owner_pubkey}) + websocket = _FakeWebSocket() + adapter._ws_active = True + adapter._ws_connection = websocket + monkeypatch.setattr(_buzz_mod, "_ACTIVITY_ACK_TIMEOUT", 0.01) + captured_payloads = [] + + def build_event(**kwargs): + captured_payloads.append(kwargs["payload"]) + return {"id": f"event-{len(captured_payloads)}"} + + monkeypatch.setattr( + _buzz_mod, + "_load_nostr_auth", + lambda: SimpleNamespace(build_observer_event=build_event), + ) + + assert await adapter.publish_activity( + "turn_completed", + channel_id=CHANNEL, + session_id="session-1", + turn_id="turn-1", + ) + await asyncio.wait_for(adapter._activity_queue.join(), timeout=1) + event_id = websocket.sent[0][1]["id"] + await asyncio.sleep(0.03) + + assert event_id not in adapter._activity_pending_event_ids + assert list(adapter._activity_terminal_replay) == ["turn-1"] + assert len(websocket.sent) == 2 + assert all("_hermesAckRetry" not in payload for payload in captured_payloads) + adapter._activity_sender_task.cancel() + with pytest.raises(asyncio.CancelledError): + await adapter._activity_sender_task + + +@pytest.mark.asyncio +async def test_activity_pending_ack_cap_cancels_evicted_deadline(monkeypatch): + adapter = _make_adapter() + monkeypatch.setattr(_buzz_mod, "_ACTIVITY_PENDING_CAP", 2) + + terminal_payload = { + "kind": "turn_completed", + "turnId": "turn-1", + "sessionId": "session-1", + } + adapter._track_activity_ack("first", 1, terminal_payload) + first_timer = adapter._activity_pending_event_ids["first"][1] + adapter._track_activity_ack("second", 1) + adapter._track_activity_ack("third", 1) + + assert list(adapter._activity_pending_event_ids) == ["second", "third"] + assert first_timer.cancelled() + assert list(adapter._activity_terminal_replay) == ["turn-1"] + await adapter._reset_activity_transport() + + +def test_activity_owner_pubkey_rejects_non_curve_x_coordinate(): + with pytest.raises(ValueError, match="activity_owner_pubkey"): + _make_adapter({"activity_owner_pubkey": "f" * 64}) + + +def test_activity_owner_pubkey_is_config_only(monkeypatch): + env_owner = nostr_auth.public_key_hex("00" * 31 + "02") + monkeypatch.setenv("BUZZ_ACTIVITY_OWNER_PUBKEY", env_owner) + + adapter = _make_adapter() + + assert adapter.activity_owner_pubkey == "" diff --git a/tests/gateway/test_turn_observer.py b/tests/gateway/test_turn_observer.py new file mode 100644 index 000000000000..d5c4f35c2784 --- /dev/null +++ b/tests/gateway/test_turn_observer.py @@ -0,0 +1,95 @@ +"""Load-bearing tests for the platform-neutral Gateway lifecycle seam.""" + +import asyncio +from dataclasses import fields + +import pytest + +from gateway.turn_observer import GatewayTurnObserver, TurnLifecycleEvent + + +def test_event_contract_is_closed_and_platform_neutral(): + names = {field.name for field in fields(TurnLifecycleEvent)} + assert names == { + "phase", + "platform", + "profile", + "channel_id", + "session_id", + "turn_id", + "started_at", + "triggering_event_id", + "is_new_session", + "tool_call_id", + "tool_name", + "tool_status", + "outcome", + } + assert not any("buzz" in name.lower() for name in names) + assert "metadata" not in names + + +@pytest.mark.asyncio +async def test_observer_uses_adapter_scoped_seam_and_fails_open(): + received = [] + + class Route: + def on_turn_lifecycle(self, event): + received.append(event) + raise RuntimeError("observer failure must not break chat") + + route = Route() + observer = GatewayTurnObserver( + platform="buzz", + profile="default", + channel_id="channel-1", + session_id="session-1", + triggering_event_id=None, + is_new_session=False, + route=route, + loop=asyncio.get_running_loop(), + is_current=lambda: True, + ) + + assert observer.start(liveness_interval=0) is False + assert len(received) == 1 + assert received[0].phase == "turn_started" + + +@pytest.mark.asyncio +async def test_tool_identifiers_are_bounded_opaque_and_privacy_safe(): + received = [] + + class Route: + def on_turn_lifecycle(self, event): + received.append(event) + return True + + observer = GatewayTurnObserver( + platform="buzz", + profile="default", + channel_id="channel-1", + session_id="session-1", + triggering_event_id=None, + is_new_session=False, + route=Route(), + loop=asyncio.get_running_loop(), + is_current=lambda: True, + ) + observer.start(liveness_interval=0) + secret_id = "private-token/" + "x" * 5000 + secret_name = "terminal\n/private/path/" + "y" * 5000 + + observer.tool_started(secret_id, secret_name, {"password": "do-not-copy"}) + observer.tool_finished(secret_id, secret_name, {}, "private result") + + tool_events = [event for event in received if event.phase.startswith("tool_")] + assert len(tool_events) == 2 + assert tool_events[0].tool_call_id == tool_events[1].tool_call_id == "tool-1" + assert tool_events[0].tool_name == tool_events[1].tool_name == "tool" + assert len(tool_events[0].tool_call_id) <= 32 + assert len(tool_events[0].tool_name) <= 64 + serialized = repr(tool_events) + assert "private-token" not in serialized + assert "/private/path" not in serialized + assert "do-not-copy" not in serialized diff --git a/website/docs/integrations/buzz.md b/website/docs/integrations/buzz.md index c0514bf99e47..360510492552 100644 --- a/website/docs/integrations/buzz.md +++ b/website/docs/integrations/buzz.md @@ -16,6 +16,7 @@ Hermes integrates with Buzz three ways. Pick by where Hermes runs and what you w | **Hermes runs** | On your desktop, launched by Buzz | On a server, launched by `buzz-acp` | In your own gateway, alongside Telegram/Discord/etc. | | **Best for** | Trying Hermes inside Buzz Desktop with zero config | A hosted agent identity when Buzz owns the transport | Full Hermes: memory, skills, approvals, cron, sessions | | **Inbound** | ACP stdio | ACP stdio (via relay WebSocket) | NIP-42-authenticated Nostr WebSocket (poll fallback) | +| **Owner Activity** | Managed by Desktop | `BUZZ_ACP_RELAY_OBSERVER` | Optional encrypted native Gateway Activity over the authenticated WebSocket | | **Setup** | Automatic discovery | `buzz-acp` env vars | `hermes gateway setup` → Buzz | ## ① Buzz Desktop managed runtime @@ -36,7 +37,7 @@ The spawned Hermes uses the same config, credentials, memory, and skills as `her ## ③ Native gateway platform (recommended for full Hermes) -The bundled `buzz` platform plugin makes Buzz a normal Hermes messaging platform — channels, DMs, mention gating, threaded replies, reactions, images, and cron delivery (`deliver=buzz`), with Hermes' own approvals, memory, and session management intact. Inbound arrives over a persistent NIP-42-authenticated Nostr WebSocket (dependency-free BIP-340 signing) with automatic fallback to CLI polling; outbound goes through the `buzz` CLI. +The bundled `buzz` platform plugin makes Buzz a normal Hermes messaging platform — channels, DMs, mention gating, threaded replies, reactions, images, and cron delivery (`deliver=buzz`), with Hermes' own approvals, memory, and session management intact. Inbound arrives over a persistent NIP-42-authenticated Nostr WebSocket (dependency-free BIP-340 signing) with automatic fallback to CLI polling; outbound goes through the `buzz` CLI. Native Gateway turns can also publish optional, owner-encrypted NIP-AO Activity on that authenticated WebSocket, without routing execution through ACP. ```bash hermes gateway setup # pick Buzz diff --git a/website/docs/user-guide/messaging/buzz.md b/website/docs/user-guide/messaging/buzz.md index 0d030b72664a..d7eeff5e5fbb 100644 --- a/website/docs/user-guide/messaging/buzz.md +++ b/website/docs/user-guide/messaging/buzz.md @@ -36,6 +36,7 @@ gateway: 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) + activity_owner_pubkey: "" # optional owner npub/hex; enables encrypted View activity events allowed_users: [] # empty = allow all; hex pubkeys or npubs ``` @@ -58,6 +59,21 @@ BUZZ_PRIVATE_KEY=nsec1... | `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 | +| `BUZZ_AUTH_TAG` | — | Owner-signed NIP-OA attestation for this agent identity; required by hosted relays for owner-authorized activity | + +## Native Gateway activity + +Set `gateway.platforms.buzz.extra.activity_owner_pubkey` in `config.yaml` to publish native Hermes turn and tool lifecycle activity for Buzz's **View activity** panel. This non-secret behavior setting is intentionally configuration-only; environment variables remain reserved for credentials and deployment concerns. Hermes remains the execution engine: this observer stream does not route the turn through Buzz ACP. Activity is published on the authenticated WebSocket. Automatic polling fallback can continue ordinary chat while Activity is unavailable; explicit `transport: poll` cannot be combined with Activity. + +Activity events are ephemeral NIP-AO events (kind `24200`), encrypted to the owner with NIP-44 and signed by the configured agent identity. Tool activity contains only a bounded tool name, a turn-local opaque call ID, and status. Hermes deliberately omits provider call IDs, tool arguments, results, model text, credentials, code, queries, and local paths. + +Encryption protects event contents, not traffic metadata. The relay can still see the outer `p`, `agent`, and `frame` tags, creation time, ciphertext size, and event cadence, which reveal the owner-agent relationship and approximate turn/tool timing. NIP-44 does not provide forward secrecy, and “ephemeral” is relay retention policy rather than cryptographic deletion. + +Activity transport is bounded and fail-open. Ordinary progress and liveness frames are dropped while the WebSocket is unavailable. A terminal completion, cancellation, timeout, or failure that occurs during a temporary outage is retained in a bounded, terminal-only replay buffer and sent once on reconnect so a previously observed turn cannot remain stuck as working indefinitely. + +The setting is optional and fail-open. If it is absent, no observer events are emitted. If encryption, signing, WebSocket delivery, or relay acceptance fails, the normal Hermes turn and Buzz reply continue unaffected. A malformed owner key is rejected at startup rather than silently disabling activity. + +The Buzz relay must recognize the signing identity as an agent owned by that owner and must authorize the owner as an observer. Signing a valid kind-`24200` event alone does not grant relay authorization. ## Recommended default settings @@ -80,6 +96,7 @@ gateway: - ccc2bc1a-7a82-5a8f-8c4e-57a070cbe7cd home_channel: ccc2bc1a-7a82-5a8f-8c4e-57a070cbe7cd poll_interval: 4 # seconds between inbound poll sweeps (default 4 — balances latency vs. relay load) + activity_owner_pubkey: "" # optional owner npub/hex; enables encrypted View activity events 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 if allow_all_users is true; otherwise restrict to listed npubs/hex pubkeys @@ -157,6 +174,8 @@ Check status with `hermes gateway status` — Buzz connection state is reported - **`BUZZ_*` env vars are available in terminal tool children for Buzz sessions** — the agent can invoke the `buzz` CLI directly (e.g. `buzz messages send ...`) because `BUZZ_PRIVATE_KEY`, `BUZZ_AUTH_TAG`, `BUZZ_RELAY_URL`, and the other `BUZZ_*` variables are passed through to terminal subprocesses when the session's platform is `buzz` or the process is a Buzz Desktop managed agent (`BUZZ_MANAGED_AGENT`). Non-Buzz sessions on the same host, `execute_code`, and other non-terminal spawns remain sealed. - **Inbound streaming has a watchdog.** On the WebSocket transport a connection that goes quiet for five minutes, or whose socket the relay closed underneath us, is torn down and reconnected with backoff; while that happens the gateway health (`/health/detailed`, dashboard status) reports Buzz as `retrying`, not `connected`. On the `poll` transport the adapter polls `buzz messages get` per watched channel every `poll_interval` seconds (default 4), so expect up to one interval of latency. + +- **Inbound is WebSocket-first with polling fallback.** `transport: auto` (the default) uses an authenticated WebSocket and falls back to CLI polling if initial WebSocket authentication is unavailable. `transport: websocket` requires WebSocket startup; `transport: poll` forces polling. Native Activity requires WebSocket delivery and cannot be enabled with explicit poll-only transport. - 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.