feat(gateway): opt-in SSE side-channel for live token streaming to external chat clients

gateway/keryx_stream.py

@@ -0,0 +1,248 @@
+"""Keryx side-channel stream hub — the server half of Keryx's dual-tier streaming.
+
+Installed as ``gateway/keryx_stream.py`` inside the hermes-agent tree (see install.py; this is a
+REINSTALL-FRAGILE patch — re-run install.py after ``hermes update``).
+
+What it does
+============
+The Keryx Android client opens a transient SSE subscription (``GET /keryx/stream?platform=matrix&
+chat_id=<room>`` on the API server, Bearer-authed with API_SERVER_KEY) right before sending a
+command into a Matrix room. While that subscriber is attached:
+
+  * every assistant-text delta the ``GatewayStreamConsumer`` receives is mirrored to the SSE
+    channel (``event: delta``) for live token rendering in the app;
+  * protocol edits to the homeserver are suppressed — the room receives only the single final
+    committed message (no m.replace database bloat);
+  * ``event: stop`` fires when the turn's stream finishes, telling the client to hold its overlay
+    until the final Matrix event syncs in.
+
+When no subscriber is attached and ``FALLBACK_EDITS`` is True, Matrix falls back to
+smart-throttled native m.replace edits driven by the normal streaming config
+(``streaming.edit_interval`` / ``streaming.buffer_threshold`` — tune to 1.2s / 60 in config.yaml).
+Set ``FALLBACK_EDITS = False`` to restore final-message-only behaviour when Keryx is offline.
+
+Thread-safety: ``publish_threadsafe`` is called from the agent's sync worker thread; delivery hops
+onto each subscriber's event loop via ``call_soon_threadsafe``. Queues are bounded — a stalled
+subscriber drops its own events, never blocks the agent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import threading
+from typing import Any, Dict, List, Optional, Tuple
+
+logger = logging.getLogger("gateway.keryx_stream")
+
+# Opt-in fallback tier: when set (KERYX_STREAM_FALLBACK_EDITS=1), a Matrix chat WITHOUT a live
+# side-channel subscriber gets throttled protocol (m.replace) edit streaming instead of the
+# buffer-only default. Off by default so this module changes no existing gateway behaviour.
+FALLBACK_EDITS = os.getenv("KERYX_STREAM_FALLBACK_EDITS", "").strip().lower() in {"1", "true", "yes", "on"}
+
+# Per-subscriber event buffer. Generous relative to token rate x ping interval; overflow drops
+# oldest-first semantics are approximated by dropping the incoming event for that subscriber.
+_QUEUE_MAX = 2048
+
+
+class _Subscription:
+    __slots__ = ("queue", "loop")
+
+    def __init__(self, queue: "asyncio.Queue[Tuple[str, Optional[str]]]", loop: asyncio.AbstractEventLoop):
+        self.queue = queue
+        self.loop = loop
+
+
+class KeryxStreamHub:
+    """In-process pub/sub keyed by (platform, chat_id)."""
+
+    def __init__(self) -> None:
+        self._subs: Dict[Tuple[str, str], List[_Subscription]] = {}
+        self._lock = threading.Lock()
+
+    @staticmethod
+    def _key(platform: str, chat_id: str) -> Tuple[str, str]:
+        return (str(platform).strip().lower(), str(chat_id).strip())
+
+    def subscribe(self, platform: str, chat_id: str) -> _Subscription:
+        sub = _Subscription(asyncio.Queue(maxsize=_QUEUE_MAX), asyncio.get_running_loop())
+        key = self._key(platform, chat_id)
+        with self._lock:
+            self._subs.setdefault(key, []).append(sub)
+        logger.info("keryx subscriber attached: %s", key)
+        return sub
+
+    def unsubscribe(self, platform: str, chat_id: str, sub: _Subscription) -> None:
+        key = self._key(platform, chat_id)
+        with self._lock:
+            lst = self._subs.get(key)
+            if lst and sub in lst:
+                lst.remove(sub)
+                if not lst:
+                    del self._subs[key]
+        logger.info("keryx subscriber detached: %s", key)
+
+    def has_subscribers(self, platform: str, chat_id: str) -> bool:
+        with self._lock:
+            return bool(self._subs.get(self._key(platform, chat_id)))
+
+    def publish_threadsafe(self, platform: str, chat_id: str, event: str, text: Optional[str]) -> None:
+        """Mirror one stream event to every subscriber. Never raises, never blocks."""
+        key = self._key(platform, chat_id)
+        with self._lock:
+            subs = list(self._subs.get(key, ()))
+        for sub in subs:
+            try:
+                sub.loop.call_soon_threadsafe(self._offer, sub.queue, (event, text))
+            except Exception:
+                # Subscriber's loop is gone — it will be pruned when its handler exits.
+                pass
+
+    @staticmethod
+    def _offer(queue: "asyncio.Queue[Tuple[str, Optional[str]]]", item: Tuple[str, Optional[str]]) -> None:
+        try:
+            queue.put_nowait(item)
+        except asyncio.QueueFull:
+            logger.debug("keryx subscriber queue full; dropping %s", item[0])
+
+
+hub = KeryxStreamHub()
+
+
+def drain_coalesced(
+    queue: "asyncio.Queue[Tuple[str, Optional[str]]]",
+    first: Tuple[str, Optional[str]],
+) -> Tuple[List[Tuple[str, Optional[str]]], bool]:
+    """Merge a burst of queued token deltas into as few frames as possible.
+
+    Takes the item already pulled from [queue] ([first]) plus everything currently queued
+    (non-blocking) and returns ``(frames, stop)``: an ordered list of ``(event, text)`` frames
+    ready to write, and whether a ``stop`` was seen (the caller then closes the channel).
+
+    Consecutive ``delta`` events are concatenated into a single ``delta`` frame; non-delta
+    boundaries (``segment``/``stop``) flush the accumulator and pass through in order. This is
+    byte-exact — delta concatenation is associative — and bounds the write rate to how fast the
+    consumer drains, so a brain generating faster than a remote client drains can't back the
+    per-subscriber queue up to _QUEUE_MAX and lose tokens to overflow. A dropped token would
+    break the client's stream/commit reconciliation (accumulated stream no longer byte-matches
+    the committed message), which is exactly what this coalescing prevents.
+    """
+    pending: List[Tuple[str, Optional[str]]] = [first]
+    while True:
+        try:
+            pending.append(queue.get_nowait())
+        except asyncio.QueueEmpty:
+            break
+
+    frames: List[Tuple[str, Optional[str]]] = []
+    buf: List[str] = []
+    stop = False
+    for event, text in pending:
+        if event == "delta":
+            buf.append(text or "")
+            continue
+        if buf:
+            frames.append(("delta", "".join(buf)))
+            buf = []
+        frames.append((event, text))
+        if event == "stop":
+            stop = True
+            break
+    if buf:
+        frames.append(("delta", "".join(buf)))
+    return frames, stop
+
+
+def _platform_of(adapter: Any) -> str:
+    """Stable lowercase platform key for an adapter ("matrix", "telegram", …)."""
+    try:
+        return str(adapter.platform.value).lower()
+    except Exception:
+        return str(getattr(adapter, "name", "")).lower()
+
+
+def publish_delta(adapter: Any, chat_id: Any, text: str) -> None:
+    """Called from GatewayStreamConsumer.on_delta (agent worker thread)."""
+    hub.publish_threadsafe(_platform_of(adapter), str(chat_id), "delta", text)
+
+
+def publish_segment(adapter: Any, chat_id: Any) -> None:
+    hub.publish_threadsafe(_platform_of(adapter), str(chat_id), "segment", None)
+
+
+def publish_stop(adapter: Any, chat_id: Any, final_text: Optional[str] = None) -> None:
+    hub.publish_threadsafe(_platform_of(adapter), str(chat_id), "stop", final_text)
+
+
+def suppress_protocol_edits(adapter: Any, chat_id: Any, default_buffer_only: bool) -> bool:
+    """Decide whether the stream consumer should skip interval/threshold homeserver edits.
+
+    Live Keryx subscriber → True (the side-channel carries tokens; commit only the final).
+    No subscriber on Matrix with FALLBACK_EDITS → False (throttled m.replace fallback tier).
+    Anything else → whatever the gateway decided ([default_buffer_only]).
+    """
+    platform = _platform_of(adapter)
+    if hub.has_subscribers(platform, str(chat_id)):
+        return True
+    if default_buffer_only and FALLBACK_EDITS and platform == "matrix":
+        return False
+    return default_buffer_only
+
+
+def make_stream_handler(check_auth):
+    """Build the aiohttp handler for ``GET /keryx/stream`` (wired in api_server.py).
+
+    [check_auth] is ApiServerAdapter._check_auth — same Bearer key as every other route.
+    """
+    from aiohttp import web
+
+    async def handle_keryx_stream(request: "web.Request") -> "web.StreamResponse":
+        auth_err = check_auth(request)
+        if auth_err is not None:
+            return auth_err
+        platform = request.query.get("platform", "matrix")
+        chat_id = request.query.get("chat_id", "").strip()
+        if not chat_id:
+            return web.json_response({"error": {"message": "chat_id is required"}}, status=400)
+
+        resp = web.StreamResponse(
+            status=200,
+            headers={
+                "Content-Type": "text/event-stream",
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no",
+            },
+        )
+        await resp.prepare(request)
+        sub = hub.subscribe(platform, chat_id)
+        try:
+            while True:
+                try:
+                    first = await asyncio.wait_for(sub.queue.get(), timeout=20.0)
+                except asyncio.TimeoutError:
+                    # Keepalive: keeps NATs open and lets a dead client surface as a write error.
+                    await resp.write(b"event: ping\ndata: {}\n\n")
+                    continue
+
+                # Coalesce whatever else is queued into as few frames as possible so a fast brain
+                # can't overflow the bounded queue and drop tokens (see drain_coalesced).
+                frames, stop = drain_coalesced(sub.queue, first)
+                for event, text in frames:
+                    payload = json.dumps({"text": text} if text is not None else {})
+                    await resp.write(f"event: {event}\ndata: {payload}\n\n".encode("utf-8"))
+                if stop:
+                    break  # transient channel: one turn per subscription
+        except (ConnectionResetError, asyncio.CancelledError):
+            pass
+        finally:
+            hub.unsubscribe(platform, chat_id, sub)
+        try:
+            await resp.write_eof()
+        except Exception:
+            pass
+        return resp
+
+    return handle_keryx_stream

gateway/platforms/api_server.py

@@ -4742,6 +4742,11 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
             self._app = web.Application(middlewares=mws, client_max_size=MAX_REQUEST_BYTES)
             assert self._app is not None
             self._app.router.add_get("/health", self._handle_health)
+            try:
+                from gateway.keryx_stream import make_stream_handler
+                self._app.router.add_get("/keryx/stream", make_stream_handler(self._check_auth))
+            except Exception:
+                logger.debug("keryx stream route unavailable", exc_info=True)
             self._app.router.add_get("/health/detailed", self._handle_health_detailed)
             self._app.router.add_get("/v1/health", self._handle_health)
             self._app.router.add_get("/v1/models", self._handle_models)

gateway/stream_consumer.py

@@ -314,6 +314,11 @@ def has_delivered_text(self, text: str) -> bool:
 
     def on_segment_break(self) -> None:
         """Finalize the current stream segment and start a fresh message."""
+        try:
+            from gateway import keryx_stream as _keryx
+            _keryx.publish_segment(self.adapter, self.chat_id)
+        except Exception:
+            pass
         self._queue.put(_NEW_SEGMENT)
 
     def on_commentary(self, text: str) -> None:
@@ -366,12 +371,22 @@ def on_delta(self, text: str) -> None:
         appears below any tool-progress messages the gateway sent in between.
         """
         if text:
+            try:
+                from gateway import keryx_stream as _keryx
+                _keryx.publish_delta(self.adapter, self.chat_id, text)
+            except Exception:
+                pass
             self._queue.put(text)
         elif text is None:
             self.on_segment_break()
 
     def finish(self) -> None:
         """Signal that the stream is complete."""
+        try:
+            from gateway import keryx_stream as _keryx
+            _keryx.publish_stop(self.adapter, self.chat_id)
+        except Exception:
+            pass
         self._queue.put(_DONE)
 
     # ── Think-block filtering ────────────────────────────────────────
@@ -618,7 +633,15 @@ async def run(self) -> None:
                     or got_segment_break
                     or commentary_text is not None
                 )
-                if not self.cfg.buffer_only:
+                _keryx_buffer_only = self.cfg.buffer_only
+                try:
+                    from gateway import keryx_stream as _keryx
+                    _keryx_buffer_only = _keryx.suppress_protocol_edits(
+                        self.adapter, self.chat_id, self.cfg.buffer_only
+                    )
+                except Exception:
+                    pass
+                if not _keryx_buffer_only:
                     should_edit = should_edit or (
                         (elapsed >= self._current_edit_interval
                             and self._accumulated)