andy/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

phase 1: hermes_events.py — generic plugin event bus

Browse Source 
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.
This commit is contained in:
Ben
2026-05-28 13:28:07 +10:00
parent 11d93096b3
commit 2316d2a225
2 changed files with 471 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/events.md
+173
View File
@@ -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.<name>` | 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.
hermes_events.py
+298
View File
@@ -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": "<topic>", # always the topic the publisher used
"ts": <unix-seconds-float>, # auto-stamped if missing
"src": "<first-topic-segment>", # 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"]