From 2316d2a2258fa53b2ad2f834d30c8a13127a2b54 Mon Sep 17 00:00:00 2001 From: Ben Date: Thu, 28 May 2026 13:28:07 +1000 Subject: [PATCH] =?UTF-8?q?phase=201:=20hermes=5Fevents.py=20=E2=80=94=20g?= =?UTF-8?q?eneric=20plugin=20event=20bus?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a process-local pub/sub bus that lets sources (TUI sidecar, gateway hooks, agent loop, etc.) publish lifecycle events to subscribers (plugins, observability, debug taps) without either side knowing about the other. The orb is the first consumer of this facility, but the API is generic — any plugin can subscribe via its plugin_api.py. Design ------ - Sync publish() — callable from any context, including async coroutines (no await needed). Async publishers in gateway/hooks.py and agent hot-paths can use it without restructuring. - Mixed-mode subscribers — sync callbacks fire in the publisher's stack; async callbacks are scheduled via asyncio.create_task() when a loop is detected, dropped with a warning otherwise. - Glob patterns — `*` matches one segment, `**` matches any number. - Auto-stamped envelope — bus fills in `type`/`ts`/`src` when missing, preserves them when the publisher pre-stamps (for cross-process relays). - Exception isolation — a raising subscriber never poisons others or bubbles back to the publisher. Files ----- - hermes_events.py — the bus module (~270 lines, well-commented) - docs/events.md — public taxonomy + envelope spec + EXPERIMENTAL disclaimer at the top Stability --------- docs/events.md carries the explicit "experimental, breakage expected" notice from the plan. Topic names and payload shapes may change without a deprecation cycle until v1.0 lands. The API shape (publish/subscribe/ unsubscribe + envelope rule + glob syntax) is intended to be stable. Phase 0 baseline ---------------- Verified pristine HEAD passes tests/gateway/test_hooks.py + tests/test_tui_gateway_server.py (207 tests). No baseline-fix commit needed. Build status: hermes_events module imports cleanly; smoke-tested publish/subscribe/glob/envelope/exception-isolation in a subprocess. --- docs/events.md | 173 +++++++++++++++++++++++++++ hermes_events.py | 298 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 471 insertions(+) create mode 100644 docs/events.md create mode 100644 hermes_events.py diff --git a/docs/events.md b/docs/events.md new file mode 100644 index 0000000000..c6ffe16873 --- /dev/null +++ b/docs/events.md @@ -0,0 +1,173 @@ +# Hermes Events Bus + +> ⚠️ **EXPERIMENTAL.** Topic names and payload shapes in this document are +> explicitly prone to change. The bus is shipped now so plugin authors can +> start consuming it, but you should expect to update your topic globs and +> payload-field accesses over the next few releases. We will declare a +> stable v1.0 schema once the orb (and at least one more consumer) have +> shaken out the design — at that point we'll add a proper deprecation +> policy. Until then, breakage is on the menu. + +`hermes_events` is the in-process pub/sub bus that lets core components +(the embedded TUI, the gateway, the agent loop, the cron scheduler, ...) +broadcast lifecycle events that plugins (and other core components) can +subscribe to. It is the substrate that drives the orb's `SceneState` +machine and is intended to be the substrate that drives any future +"what is Hermes doing right now" widget, observability shipper, or +debug-tap plugin. + +## API surface + +```python +from hermes_events import publish, subscribe, unsubscribe + +# Sources publish — always sync, even from inside async contexts. +publish("tui.tool.start", {"name": "web_search", "session_id": "abc"}) +publish("gateway.agent.start", {"platform": "telegram", "session_id": "xyz"}) + +# Plugins subscribe with glob patterns. +handle = subscribe("tui.tool.*", on_tool_event) # one segment after tool. +handle = subscribe("gateway.agent.*", on_agent_event) +handle = subscribe("tui.**", any_tui_topic) # any number of segments +handle = subscribe("**", on_anything) # firehose +unsubscribe(handle) +``` + +- Topics are `.`-segmented. +- Pattern globs: `*` matches one segment; `**` matches zero or more segments. +- `publish()` is **synchronous**. Async publishers (`async def emit(): ...`) + just call sync `publish()` from inside their coroutine — no `await` needed. +- Subscribers may be sync or async. Sync subscribers fire immediately in the + publisher's stack. Async subscribers are scheduled via + `asyncio.create_task()` if a running event loop is detected. **If no loop + is running** (unit tests outside `pytest-asyncio`, startup-before-loop), + async subscribers are dropped for that emit with a warning log line; sync + subscribers still fire. +- Subscriber exceptions are logged but never raised back to the publisher. + +## Envelope + +Every published payload is wrapped before delivery: + +```python +{ + "type": "tui.tool.start", # always the topic the publisher used + "ts": 1716938400.123, # unix seconds (float) + "src": "tui", # first dot-segment of `type` + # ...source-specific fields, freely evolving during the experimental phase +} +``` + +`type`, `ts`, and `src` are always present. The bus auto-stamps `ts` (current +`time.time()`) and `src` (first segment of the topic) only if the publisher +omits them. **Relayed events** (e.g. the dashboard receiving a frame from a +remote gateway process) should pre-populate `ts` and `src` so the originating +values are preserved — the bus does not overwrite existing keys. + +## Current topic taxonomy + +All topic names below are **experimental**. Plugins should subscribe to +the most general pattern that meets their needs and tolerate the addition +of new sibling topics over time. + +### `tui.*` — emitted by the dashboard's TUI sidecar + +| Topic | When it fires | +|---|---| +| `tui.message.start` | The model begins emitting a new assistant message. | +| `tui.message.delta` | A streaming-token chunk of the current assistant message. | +| `tui.message.complete` | The assistant message has finished streaming. | +| `tui.tool.start` | A tool call has begun execution. | +| `tui.tool.progress` | A long-running tool reports incremental output. | +| `tui.tool.complete` | A tool call finished (success or failure). | +| `tui.tool.generating` | The model is mid-stream of constructing a tool call's arguments. | +| `tui.reasoning.delta` | A streaming-token chunk of the model's reasoning content. | +| `tui.reasoning.available` | A full reasoning block became available (non-streaming case). | +| `tui.error` | An error frame propagated through the TUI. | + +Common payload fields (best-effort, not guaranteed): +- `session_id` (str) — TUI session this event belongs to. +- `name` (str) — for tool events, the tool's name. +- `preview` (str) — for tool events, a short preview of args or output. + +### `gateway.*` — emitted by the messaging gateway + +| Topic | When it fires | +|---|---| +| `gateway.startup` | The gateway process has finished initializing. | +| `gateway.agent.start` | The agent has begun processing a user message. | +| `gateway.agent.step` | Each iteration of the agent's tool-calling loop. | +| `gateway.agent.end` | The agent has finished processing a user message. | +| `gateway.session.start` | A new session was created (first message of a new session). | +| `gateway.session.end` | A session ended (user ran `/new` or `/reset`). | +| `gateway.session.reset` | A session reset completed; a new session entry was created. | +| `gateway.command.` | Any slash command was executed. Wildcard-friendly via `gateway.command.*`. | + +Common payload fields: +- `platform` (str) — `"telegram"`, `"discord"`, `"slack"`, etc. +- `session_id` (str) — the gateway session this event belongs to. + +### Future namespaces + +Topics under namespaces we haven't shipped yet are reserved. If you see +events fire under a namespace not listed here, that's a leak from in-flight +work — please don't subscribe to it; the topic may be renamed or removed +before the next release. + +## Cross-process delivery + +Each process has its own bus instance. Bridges ship events between +processes: + +- **TUI sidecar → dashboard:** the sidecar runs as a PTY-spawned subprocess + of the dashboard. A default `subscribe("**", ship_via_pub_ws)` on the TUI + side forwards every event to the dashboard via the `/api/pub` WebSocket, + which re-publishes them onto the dashboard's local bus. +- **Gateway → dashboard:** the gateway process opens a WebSocket to the + dashboard at startup and ships its events the same way. If the dashboard + is offline, the bridge silently no-ops; gateway runs are never blocked. +- **Same-process publishers** (e.g. an embedded gateway in the dashboard + process) hit the local bus directly — no bridge involved. + +This means a plugin's `plugin_api.py` running in the dashboard process +sees events from all three sources via a single `subscribe(...)` call. +Subscription patterns do not change based on where events originated. + +## For plugin authors + +A typical plugin backend wires up subscriptions in its `register()` (or +at import time, since `plugin_api.py` is imported once at dashboard +startup): + +```python +from fastapi import APIRouter +from hermes_events import subscribe + +router = APIRouter() + +def _on_tool_event(envelope: dict) -> None: + # Process the envelope, push to your WebSocket subscribers, etc. + ... + +subscribe("tui.tool.*", _on_tool_event) +subscribe("gateway.agent.*", _on_tool_event) +``` + +If you need to subscribe asynchronously (e.g. to drive a per-subscriber +`asyncio.Queue` in a long-lived FastAPI WebSocket handler), register an +`async def` callback — the bus will `asyncio.create_task()` it inside the +dashboard's running event loop. + +For a fully-worked example, see `plugins/orb/dashboard/plugin_api.py`. + +## Stability declaration + +Until further notice (read: until a `docs/events-v1.md` lands and this +file is renamed to `docs/events-experimental.md`), assume any topic name +or payload field may be renamed, removed, or split into multiple events +in any release. + +The **shape of the bus itself** (the `publish` / `subscribe` / +`unsubscribe` Python API, the envelope auto-stamping rule, the glob +pattern syntax) is more stable — we intend to preserve it through v1.0. +But the **content** flowing through it is the experimental part. diff --git a/hermes_events.py b/hermes_events.py new file mode 100644 index 0000000000..eb6c294429 --- /dev/null +++ b/hermes_events.py @@ -0,0 +1,298 @@ +""" +Hermes Agent — process-local pub/sub event bus. + +Plugins and core components can publish events on a shared in-process bus and +subscribe to topic patterns. Sources publish; subscribers listen; neither knows +about the other. + +EXPERIMENTAL — topic names and payload shapes are explicitly prone to change +until we publish a v1.0 schema. See ``docs/events.md`` for the current taxonomy +and the breakage notice. Third-party plugins are welcome to subscribe today, +but should expect to update their topic globs over the next few releases. + +Basic usage +----------- + +:: + + from hermes_events import publish, subscribe, unsubscribe + + # Sources publish — always sync, even from inside async contexts. + publish("tui.tool.start", {"name": "web_search", "session_id": "abc"}) + + # Subscribers register glob patterns. ``*`` matches one topic segment; + # ``**`` matches any number of segments. + handle = subscribe("tui.tool.*", on_tool_event) # tui.tool.start/.complete/... + handle = subscribe("tui.**", on_any_tui) # any tui.* topic + handle = subscribe("**", everything) + unsubscribe(handle) + +Envelope +-------- + +Every published payload is wrapped in a minimal envelope before being delivered +to subscribers: + +:: + + { + "type": "", # always the topic the publisher used + "ts": , # auto-stamped if missing + "src": "", # auto-stamped if missing + ...source-specific fields, freely evolving in the experimental phase + } + +Publishers that are *relaying* events from another process should pre-populate +``ts`` and ``src`` so the original values are preserved (the bus only fills in +missing keys; it never overwrites them). + +Async semantics +--------------- + +``publish()`` is synchronous. Sync subscribers fire immediately in the +publisher's stack. Async subscribers are scheduled via +``asyncio.create_task()`` when a running event loop is detected. When no loop +is running (unit tests outside of an async test, startup-before-loop), async +subscribers are dropped for that emit with a single warning log line — sync +subscribers still fire normally. + +This means async publishers (e.g. ``gateway.hooks.HookRegistry.emit()``) just +call sync ``publish()`` from inside their coroutine — no ``await`` needed. + +Exceptions in subscribers are logged but never raised back to the publisher. +""" + +from __future__ import annotations + +import asyncio +import inspect +import logging +import threading +import time +from collections.abc import Awaitable, Callable +from typing import Any + +logger = logging.getLogger(__name__) + + +# A subscriber callback. Receives the envelope dict and may return either +# ``None`` (sync) or an awaitable (async). +SubscriberCallback = Callable[[dict], "Awaitable[None] | None"] + + +class _Subscription: + """Internal record of one subscription. Returned (opaque) to callers as + the handle for ``unsubscribe()``. + """ + + __slots__ = ("pattern", "segments", "callback", "is_async") + + def __init__(self, pattern: str, callback: SubscriberCallback) -> None: + self.pattern = pattern + self.segments = pattern.split(".") + self.callback = callback + # ``iscoroutinefunction`` doesn't catch partials, bound methods of + # async functions in all cases, or callable objects with an async + # ``__call__``. ``inspect.iscoroutinefunction`` handles the common + # cases; we also probe by calling and inspecting if a callable + # object's __call__ is async-tagged. Keep this conservative — a + # false-negative just runs an async subscriber synchronously and + # discards the coroutine (with a warning), which is recoverable; + # a false-positive schedules a sync function as a task, which + # would explode in create_task(). + self.is_async = inspect.iscoroutinefunction(callback) or inspect.iscoroutinefunction( + getattr(callback, "__call__", None) + ) + + +# Module-level state. The bus is a process-local singleton — we deliberately +# don't expose a class or constructor to avoid the "which bus is this?" +# confusion. Inside one Python process, there is exactly one bus. +_subscriptions: list[_Subscription] = [] +_lock = threading.RLock() + + +def _matches(topic_segments: list[str], pattern_segments: list[str]) -> bool: + """Glob-match a topic against a pattern. + + ``*`` matches exactly one segment. ``**`` matches zero or more segments. + Anywhere a literal segment appears in the pattern, the topic must match + exactly at that position. + + Recursive implementation; patterns are short (typically 1-3 segments). + """ + # ``ti`` indexes into topic; ``pi`` indexes into pattern. + def _rec(ti: int, pi: int) -> bool: + # Both exhausted simultaneously → match. + if pi == len(pattern_segments): + return ti == len(topic_segments) + + seg = pattern_segments[pi] + + if seg == "**": + # Greedy: try consuming 0..N topic segments. + # First, try matching with ``**`` consuming zero topic segments. + if _rec(ti, pi + 1): + return True + # Then try consuming one more topic segment and recurse on the + # same ``**`` position. + if ti < len(topic_segments) and _rec(ti + 1, pi): + return True + return False + + if ti == len(topic_segments): + # Pattern still has segments to consume but topic is exhausted. + return False + + if seg == "*" or seg == topic_segments[ti]: + return _rec(ti + 1, pi + 1) + + return False + + return _rec(0, 0) + + +def subscribe(pattern: str, callback: SubscriberCallback) -> _Subscription: + """Subscribe a callback to a topic glob pattern. + + Returns a subscription handle. Pass it to :func:`unsubscribe` to remove. + + The same callback may subscribe to the same pattern more than once; each + call returns a distinct handle. The bus does not deduplicate. + """ + if not pattern: + raise ValueError("subscribe pattern must be a non-empty string") + + sub = _Subscription(pattern, callback) + with _lock: + _subscriptions.append(sub) + return sub + + +def unsubscribe(handle: _Subscription) -> bool: + """Remove a subscription. Returns True if the handle was found, False + otherwise (idempotent — unsubscribing a removed handle is not an error). + """ + with _lock: + try: + _subscriptions.remove(handle) + return True + except ValueError: + return False + + +def publish(topic: str, payload: dict | None = None) -> None: + """Publish an event on a topic. + + ``topic`` is a ``.``-segmented string (e.g. ``"tui.tool.start"``). + ``payload`` is a dict of source-specific fields; the bus wraps it with the + minimal envelope (``type``, ``ts``, ``src``) before delivery. + + Publishers that are relaying events from another process may pre-populate + ``ts`` and/or ``src`` in the payload to preserve the originating values; + the bus only fills in missing keys. + + This call always returns synchronously. Sync subscribers fire immediately; + async subscribers are scheduled via ``asyncio.create_task()`` when a + running loop is available. If no loop is running, async subscribers are + dropped with a single warning log line per emit. + """ + if not topic: + raise ValueError("publish topic must be a non-empty string") + + # Build the delivered envelope. We do this once and share the dict across + # all subscribers — subscribers that mutate it are misusing the API, but + # we can't realistically guard against that without copying for every + # delivery (which would balloon cost for the ``**`` firehose case). + envelope = dict(payload) if payload else {} + envelope.setdefault("type", topic) + envelope.setdefault("ts", time.time()) + envelope.setdefault("src", topic.split(".", 1)[0]) + + topic_segments = topic.split(".") + + # Snapshot subscribers under the lock, then release before invoking + # callbacks — a subscriber that calls publish() or subscribe() from + # inside its callback must not deadlock on a sync callback. + with _lock: + matches = [ + sub + for sub in _subscriptions + if _matches(topic_segments, sub.segments) + ] + + if not matches: + return + + # Probe for a running event loop ONCE per publish call, not per + # subscriber. ``get_running_loop()`` only succeeds if we're being called + # from inside a coroutine or task; otherwise it raises ``RuntimeError``. + try: + loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop() + except RuntimeError: + loop = None + + dropped_async = 0 + + for sub in matches: + try: + if sub.is_async: + if loop is None: + dropped_async += 1 + continue + # Invoke the coroutine factory to get a coroutine, then schedule + # it on the running loop. The task object is intentionally not + # awaited — the bus is fire-and-forget. + coro = sub.callback(envelope) + if inspect.iscoroutine(coro): + loop.create_task(coro) + else: + result = sub.callback(envelope) + # A sync subscriber that accidentally returns a coroutine + # (e.g. someone changed sync→async without updating the + # registration) would leak the coroutine. Warn and discard. + if inspect.iscoroutine(result): + logger.warning( + "hermes_events: subscriber for %r returned a coroutine " + "but was not registered as async; coroutine dropped", + sub.pattern, + ) + result.close() + except Exception: + # Subscriber exceptions are logged but never propagated. One + # bad subscriber must not break the publisher's main path or + # starve the rest of the subscribers on this emit. + logger.exception( + "hermes_events: subscriber for pattern %r raised on topic %r", + sub.pattern, + topic, + ) + + if dropped_async: + logger.warning( + "hermes_events: dropped %d async subscriber(s) on topic %r " + "(no running asyncio loop)", + dropped_async, + topic, + ) + + +# ----------------------------------------------------------------------------- +# Testing aids — not part of the documented public API. +# ----------------------------------------------------------------------------- + + +def _reset_for_tests() -> None: + """Clear all subscriptions. For unit tests only — must NOT be called from + production code or third-party plugins.""" + with _lock: + _subscriptions.clear() + + +def _subscriber_count() -> int: + """Return the current subscriber count. For unit tests only.""" + with _lock: + return len(_subscriptions) + + +__all__ = ["publish", "subscribe", "unsubscribe"]