feat(plugins): add streaming output observer hooks

agent/chat_completion_helpers.py

@@ -2029,6 +2029,41 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
     if agent._interrupt_requested:
         raise InterruptedError("Agent interrupted before streaming API call")
 
+    def _stream_final_text(response) -> str:
+        try:
+            choices = getattr(response, "choices", None)
+            first_choice = choices[0] if isinstance(choices, (list, tuple)) and choices else None
+            message = getattr(first_choice, "message", None)
+            content = getattr(message, "content", None)
+            if isinstance(content, str):
+                return content
+        except Exception:
+            pass
+        try:
+            content = getattr(response, "content", None)
+            if isinstance(content, str):
+                return content
+            if isinstance(content, list):
+                parts = []
+                for part in content:
+                    text = getattr(part, "text", None)
+                    if isinstance(text, str):
+                        parts.append(text)
+                return "".join(parts)
+        except Exception:
+            pass
+        return ""
+
+    def _emit_stream_start() -> None:
+        emit = getattr(agent, "_emit_stream_start", None)
+        if emit is not None:
+            emit()
+
+    def _emit_stream_end(*, final_text: str, finished: bool, error: str | None) -> None:
+        emit = getattr(agent, "_emit_stream_end", None)
+        if emit is not None:
+            emit(final_text=final_text, finished=finished, error=error)
+
     # Cron and other non-interactive, nested-pool contexts deadlock on the
     # spawned worker thread (#62151). They also have no stream consumer, so the
     # deltas this path produces go nowhere. Delegate to the non-streaming entry
@@ -2044,8 +2079,14 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
         # ensure on_first_delta reaches it. Store it on the instance
         # temporarily so _run_codex_stream can pick it up.
         agent._codex_on_first_delta = on_first_delta
+        _emit_stream_start()
         try:
-            return agent._interruptible_api_call(api_kwargs)
+            response = agent._interruptible_api_call(api_kwargs)
+            _emit_stream_end(final_text=_stream_final_text(response), finished=True, error=None)
+            return response
+        except Exception as exc:
+            _emit_stream_end(final_text="", finished=False, error=str(exc))
+            raise
         finally:
             agent._codex_on_first_delta = None
 
@@ -2122,35 +2163,51 @@ def _on_reasoning(text):
                     _fire_first()
                     agent._fire_reasoning_delta(text)
 
+                try:
+                    from agent.plugin_stream_hooks import has_reasoning_stream_observer_hooks
+
+                    plugin_reasoning_observer = has_reasoning_stream_observer_hooks()
+                except Exception:
+                    logger.debug("plugin reasoning stream observer check failed", exc_info=True)
+                    plugin_reasoning_observer = False
+
                 result["response"] = stream_converse_with_callbacks(
                     raw_response,
                     on_text_delta=_on_text if agent._has_stream_consumers() else None,
                     on_tool_start=_on_tool,
-                    on_reasoning_delta=_on_reasoning if agent.reasoning_callback or agent.stream_delta_callback else None,
+                    on_reasoning_delta=_on_reasoning
+                    if agent.reasoning_callback or agent.stream_delta_callback or plugin_reasoning_observer
+                    else None,
                     on_interrupt_check=lambda: agent._interrupt_requested,
                 )
             except Exception as e:
                 result["error"] = e
 
-        t = threading.Thread(target=_bedrock_call, daemon=True)
-        t.start()
-        while t.is_alive():
-            t.join(timeout=0.3)
+        _emit_stream_start()
+        try:
+            t = threading.Thread(target=_bedrock_call, daemon=True)
+            t.start()
+            while t.is_alive():
+                t.join(timeout=0.3)
+                if agent._interrupt_requested:
+                    raise InterruptedError("Agent interrupted during Bedrock API call")
+            # Worker exited before the poll loop observed the interrupt flag. The
+            # Bedrock stream callback breaks out and returns a PARTIAL response
+            # without raising on interrupt (see bedrock_adapter.py
+            # stream_converse_with_callbacks / on_interrupt_check), so result[
+            # "response"] is populated with error=None and the in-loop raise above
+            # never fires. Re-check here so /stop is not silently swallowed on the
+            # Bedrock path - mirrors the post-worker guard on the main streaming
+            # loop. (#59999 area)
             if agent._interrupt_requested:
-                raise InterruptedError("Agent interrupted during Bedrock API call")
-        # Worker exited before the poll loop observed the interrupt flag. The
-        # Bedrock stream callback breaks out and returns a PARTIAL response
-        # without raising on interrupt (see bedrock_adapter.py
-        # stream_converse_with_callbacks / on_interrupt_check), so result[
-        # "response"] is populated with error=None and the in-loop raise above
-        # never fires. Re-check here so /stop is not silently swallowed on the
-        # Bedrock path — mirrors the post-worker guard on the main streaming
-        # loop. (#59999 area)
-        if agent._interrupt_requested:
-            raise InterruptedError("Agent interrupted during Bedrock API call (post-worker)")
-        if result["error"] is not None:
-            raise result["error"]
-        return result["response"]
+                raise InterruptedError("Agent interrupted during Bedrock API call (post-worker)")
+            if result["error"] is not None:
+                raise result["error"]
+            _emit_stream_end(final_text=_stream_final_text(result["response"]), finished=True, error=None)
+            return result["response"]
+        except Exception as exc:
+            _emit_stream_end(final_text="", finished=False, error=str(exc))
+            raise
 
     result = {"response": None, "error": None, "partial_tool_names": []}
 
@@ -2811,14 +2868,21 @@ def _call():
                 # causing multi-minute delays between /stop and response.
                 if agent._interrupt_requested:
                     raise InterruptedError("Agent interrupted before stream retry")
+                _emit_stream_start()
                 try:
                     if agent.api_mode == "anthropic_messages":
                         agent._try_refresh_anthropic_client_credentials()
                         result["response"] = _call_anthropic()
                     else:
                         result["response"] = _call_chat_completions()
+                    _emit_stream_end(
+                        final_text=_stream_final_text(result["response"]),
+                        finished=True,
+                        error=None,
+                    )
                     return  # success
                 except Exception as e:
+                    _emit_stream_end(final_text="", finished=False, error=str(e))
                     # If the main poll loop force-closed this request because
                     # of an interrupt, the resulting transport error is the
                     # expected consequence of our own close — NOT a transient

agent/plugin_stream_hooks.py

@@ -0,0 +1,176 @@
+"""Asynchronous per-consumer plugin observers for streaming LLM output."""
+
+from __future__ import annotations
+
+import logging
+import queue
+import threading
+from dataclasses import dataclass
+from typing import Any, Callable
+
+from hermes_cli.middleware import OBSERVER_SCHEMA_VERSION
+
+logger = logging.getLogger(__name__)
+
+_QUEUE_SIZE = 1024
+_STOP = object()
+
+
+@dataclass
+class _ConsumerDispatcher:
+    hook_name: str
+    callback: Callable[..., Any]
+    events: "queue.Queue[dict[str, Any] | object]"
+    thread: threading.Thread | None = None
+
+
+_dispatcher_lock = threading.Lock()
+_dispatchers: dict[tuple[str, int], _ConsumerDispatcher] = {}
+
+
+def _callback_name(callback: Callable[..., Any]) -> str:
+    return getattr(callback, "__name__", repr(callback))
+
+
+def _worker(dispatcher: _ConsumerDispatcher) -> None:
+    while True:
+        item = dispatcher.events.get()
+        try:
+            if item is _STOP:
+                return
+            payload = dict(item)
+            payload.setdefault("telemetry_schema_version", OBSERVER_SCHEMA_VERSION)
+            try:
+                dispatcher.callback(**payload)
+            except Exception as exc:
+                logger.warning(
+                    "Hook '%s' callback %s raised: %s",
+                    dispatcher.hook_name,
+                    _callback_name(dispatcher.callback),
+                    exc,
+                )
+        finally:
+            dispatcher.events.task_done()
+
+
+def _registered_callbacks(hook_name: str) -> tuple[Callable[..., Any], ...]:
+    try:
+        from hermes_cli import plugins
+
+        return plugins.iter_hook_callbacks(hook_name)
+    except Exception:
+        logger.debug("plugin stream hook callback lookup failed: %s", hook_name, exc_info=True)
+        return ()
+
+
+def _stop_dispatcher(dispatcher: _ConsumerDispatcher, timeout: float = 1.0) -> None:
+    try:
+        dispatcher.events.put_nowait(_STOP)
+    except queue.Full:
+        try:
+            dispatcher.events.get_nowait()
+            dispatcher.events.task_done()
+        except queue.Empty:
+            pass
+        try:
+            dispatcher.events.put_nowait(_STOP)
+        except queue.Full:
+            pass
+    if dispatcher.thread is not None:
+        dispatcher.thread.join(timeout=timeout)
+
+
+def _dispatchers_for(hook_name: str) -> list[_ConsumerDispatcher]:
+    callbacks = _registered_callbacks(hook_name)
+    if not callbacks:
+        return []
+
+    callback_ids = {id(callback) for callback in callbacks}
+    stale: list[_ConsumerDispatcher] = []
+    ready: list[_ConsumerDispatcher] = []
+    with _dispatcher_lock:
+        for key, dispatcher in list(_dispatchers.items()):
+            key_hook_name, callback_id = key
+            if key_hook_name == hook_name and callback_id not in callback_ids:
+                stale.append(_dispatchers.pop(key))
+
+        for callback in callbacks:
+            key = (hook_name, id(callback))
+            dispatcher = _dispatchers.get(key)
+            if dispatcher is None or dispatcher.thread is None or not dispatcher.thread.is_alive():
+                events: "queue.Queue[dict[str, Any] | object]" = queue.Queue(maxsize=_QUEUE_SIZE)
+                dispatcher = _ConsumerDispatcher(
+                    hook_name=hook_name,
+                    callback=callback,
+                    events=events,
+                )
+                dispatcher.thread = threading.Thread(
+                    target=_worker,
+                    args=(dispatcher,),
+                    daemon=True,
+                    name=f"plugin-stream-hook:{hook_name}",
+                )
+                dispatcher.thread.start()
+                _dispatchers[key] = dispatcher
+            ready.append(dispatcher)
+
+    for dispatcher in stale:
+        _stop_dispatcher(dispatcher, timeout=0.2)
+    return ready
+
+
+def enqueue_plugin_stream_hook(hook_name: str, **payload: Any) -> bool:
+    """Queue an observer hook for each consumer without running plugin code inline."""
+    queued = False
+    item = dict(payload)
+    for dispatcher in _dispatchers_for(hook_name):
+        try:
+            dispatcher.events.put_nowait(item)
+            queued = True
+            continue
+        except queue.Full:
+            try:
+                dispatcher.events.get_nowait()
+                dispatcher.events.task_done()
+            except queue.Empty:
+                pass
+        try:
+            dispatcher.events.put_nowait(item)
+            queued = True
+        except queue.Full:
+            logger.debug(
+                "plugin stream hook queue full after drop-oldest: %s callback=%s",
+                hook_name,
+                _callback_name(dispatcher.callback),
+            )
+    return queued
+
+
+def has_stream_observer_hooks() -> bool:
+    return any(_registered_callbacks(name) for name in ("on_stream_start", "on_stream_delta", "on_stream_end"))
+
+
+def has_reasoning_stream_observer_hooks() -> bool:
+    return stream_reasoning_deltas_enabled() and bool(_registered_callbacks("on_stream_delta"))
+
+
+def stream_reasoning_deltas_enabled() -> bool:
+    """Return True only when the user opted plugins into reasoning deltas."""
+    try:
+        from hermes_cli import config as config_mod
+
+        config = config_mod.load_config()
+        return bool(config_mod.cfg_get(config, "plugins", "stream_reasoning_deltas", default=False))
+    except Exception:
+        logger.debug("failed to read plugins.stream_reasoning_deltas", exc_info=True)
+        return False
+
+
+def shutdown_plugin_stream_hook_dispatcher(timeout: float = 1.0) -> None:
+    """Stop background stream hook dispatchers; used by tests and clean shutdown paths."""
+    global _dispatchers
+    with _dispatcher_lock:
+        dispatchers = list(_dispatchers.values())
+        _dispatchers = {}
+    for dispatcher in dispatchers:
+        _stop_dispatcher(dispatcher, timeout=timeout)

hermes_cli/plugins.py

@@ -143,6 +143,13 @@ def _install_plugin_debug_handler(force: bool = False) -> None:
     "transform_llm_output",
     "pre_llm_call",
     "post_llm_call",
+    # Streaming LLM output observer hooks. Fired asynchronously off the token
+    # path by agent.plugin_stream_hooks; callbacks observe immutable normalized
+    # text/lifecycle payloads and cannot transform the stream.
+    "on_stream_start",
+    "on_stream_delta",
+    "on_stream_end",
+    "on_interim_message",
     # Verification-loop gate. Fired once per turn when the agent has edited code
     # and is about to verify/finish (after the verify-on-stop guard). A callback
     # may keep the agent going — run a check, defer it, tidy the diff — instead
@@ -1930,6 +1937,10 @@ def has_hook(self, hook_name: str) -> bool:
         """Return True when at least one callback is registered for a hook."""
         return bool(self._hooks.get(hook_name))
 
+    def iter_hook_callbacks(self, hook_name: str) -> tuple[Callable, ...]:
+        """Return a stable snapshot of callbacks registered for a hook."""
+        return tuple(self._hooks.get(hook_name, ()))
+
     def has_middleware(self, kind: str) -> bool:
         """Return True when at least one callback is registered for middleware."""
         return bool(self._middleware.get(kind))
@@ -2076,6 +2087,11 @@ def has_hook(hook_name: str) -> bool:
     return get_plugin_manager().has_hook(hook_name)
 
 
+def iter_hook_callbacks(hook_name: str) -> tuple[Callable, ...]:
+    """Return a stable snapshot of callbacks registered for a hook."""
+    return get_plugin_manager().iter_hook_callbacks(hook_name)
+
+
 _thread_tool_whitelist = threading.local()