feat(plugins): add streaming output observer hooks
Salvage of PR #64317 (@deaneeth) onto current main, implementing #64161: observer-only on_stream_start / on_stream_delta / on_stream_end / on_interim_message plugin hooks dispatched through a host-owned bounded queue (one worker per callback) so plugin callbacks never run inline on the token path. Reasoning deltas are opt-in via plugins.stream_reasoning_deltas.
This commit is contained in:
Dineth Hettiarachchi
Teknium
parent
c5097da12b
commit
00f4da01ec
|
|
@ -2748,6 +2748,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
|
||||
|
|
@ -2763,8 +2798,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
|
||||
|
||||
|
|
@ -2873,6 +2914,14 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
|
|||
token = writer_token["value"]
|
||||
return token is None or stream_writer_is_current(agent, token)
|
||||
|
||||
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
|
||||
|
||||
stream = relay_llm.stream(
|
||||
dict(api_kwargs),
|
||||
_open_bedrock_stream,
|
||||
|
|
@ -2906,7 +2955,9 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
|
|||
{"stream": stream},
|
||||
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,
|
||||
on_event=lambda: _bedrock_last_event.__setitem__("t", time.time()),
|
||||
)
|
||||
|
|
@ -2917,82 +2968,88 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
|
|||
if stream is not None:
|
||||
stream.close()
|
||||
|
||||
t = threading.Thread(
|
||||
target=_context_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")
|
||||
# Liveness watchdog: no Bedrock event for longer than the stale
|
||||
# timeout means the stream has wedged (open socket, keep-alives but
|
||||
# no data, or a silently hung provider). Without this the worker
|
||||
# blocks in ``for event in event_stream`` indefinitely.
|
||||
_stale_elapsed = time.time() - _bedrock_last_event["t"]
|
||||
if _stale_elapsed > _bedrock_stale_timeout:
|
||||
logger.warning(
|
||||
"Bedrock stream stale for %.0fs (threshold %.0fs) — no events "
|
||||
"received. region=%s model=%s. Aborting call.",
|
||||
_stale_elapsed, _bedrock_stale_timeout,
|
||||
_bedrock_region, api_kwargs.get("modelId", "unknown"),
|
||||
)
|
||||
agent._buffer_status(
|
||||
f"⚠️ No events from Bedrock for {int(_stale_elapsed)}s "
|
||||
f"(model: {api_kwargs.get('modelId', 'unknown')}). Aborting..."
|
||||
)
|
||||
# Count the stale kill in the SAME cross-turn breaker as the
|
||||
# OpenAI/Anthropic path (#58962).
|
||||
_bump_stale_streak(agent)
|
||||
# Best-effort: evict the region's cached bedrock-runtime client
|
||||
# so the NEXT call reconnects with a fresh pool. NOTE: this does
|
||||
# NOT abort the in-flight botocore EventStream the worker thread
|
||||
# is blocked on — botocore exposes no external cancellation for
|
||||
# it — so the daemon worker keeps reading until its socket read
|
||||
# ultimately errors. We therefore end THIS call by raising
|
||||
# below and let the streak+give-up breaker escalate across turns.
|
||||
try:
|
||||
from agent.bedrock_adapter import invalidate_runtime_client
|
||||
invalidate_runtime_client(_bedrock_region)
|
||||
except Exception as _inval_exc:
|
||||
logger.debug(
|
||||
"bedrock: stale client eviction failed: %s", _inval_exc
|
||||
_emit_stream_start()
|
||||
try:
|
||||
t = threading.Thread(
|
||||
target=_context_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")
|
||||
# Liveness watchdog: no Bedrock event for longer than the stale
|
||||
# timeout means the stream has wedged (open socket, keep-alives but
|
||||
# no data, or a silently hung provider). Without this the worker
|
||||
# blocks in ``for event in event_stream`` indefinitely.
|
||||
_stale_elapsed = time.time() - _bedrock_last_event["t"]
|
||||
if _stale_elapsed > _bedrock_stale_timeout:
|
||||
logger.warning(
|
||||
"Bedrock stream stale for %.0fs (threshold %.0fs) — no events "
|
||||
"received. region=%s model=%s. Aborting call.",
|
||||
_stale_elapsed, _bedrock_stale_timeout,
|
||||
_bedrock_region, api_kwargs.get("modelId", "unknown"),
|
||||
)
|
||||
# Reset the timer so a repeated trip (should the worker somehow
|
||||
# survive) waits a fresh interval rather than re-firing instantly.
|
||||
_bedrock_last_event["t"] = time.time()
|
||||
# Escalate across turns: raises RuntimeError once the streak
|
||||
# crosses HERMES_STREAM_STALE_GIVEUP, so a persistently wedged
|
||||
# Bedrock provider aborts fast instead of re-waiting the timeout.
|
||||
_check_stale_giveup(agent)
|
||||
# Streak still under the give-up threshold: end THIS call with a
|
||||
# TimeoutError so the outer retry loop / next turn re-evaluates
|
||||
# and the streak carries forward. Break rather than keep polling
|
||||
# a worker we cannot abort.
|
||||
result["error"] = TimeoutError(
|
||||
f"Bedrock stream produced no events for {int(_stale_elapsed)}s "
|
||||
f"(threshold {int(_bedrock_stale_timeout)}s) — aborting stalled "
|
||||
f"stream so the retry/fallback path can recover."
|
||||
)
|
||||
break
|
||||
# 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"]
|
||||
# Success — clear the cross-turn breaker (#58962): Bedrock proved
|
||||
# responsive. Mirrors the OpenAI/Anthropic success reset below so a
|
||||
# recovered provider doesn't carry a stale streak into later turns.
|
||||
if result["response"] is not None:
|
||||
_reset_stale_streak(agent)
|
||||
return result["response"]
|
||||
agent._buffer_status(
|
||||
f"⚠️ No events from Bedrock for {int(_stale_elapsed)}s "
|
||||
f"(model: {api_kwargs.get('modelId', 'unknown')}). Aborting..."
|
||||
)
|
||||
# Count the stale kill in the SAME cross-turn breaker as the
|
||||
# OpenAI/Anthropic path (#58962).
|
||||
_bump_stale_streak(agent)
|
||||
# Best-effort: evict the region's cached bedrock-runtime client
|
||||
# so the NEXT call reconnects with a fresh pool. NOTE: this does
|
||||
# NOT abort the in-flight botocore EventStream the worker thread
|
||||
# is blocked on — botocore exposes no external cancellation for
|
||||
# it — so the daemon worker keeps reading until its socket read
|
||||
# ultimately errors. We therefore end THIS call by raising
|
||||
# below and let the streak+give-up breaker escalate across turns.
|
||||
try:
|
||||
from agent.bedrock_adapter import invalidate_runtime_client
|
||||
invalidate_runtime_client(_bedrock_region)
|
||||
except Exception as _inval_exc:
|
||||
logger.debug(
|
||||
"bedrock: stale client eviction failed: %s", _inval_exc
|
||||
)
|
||||
# Reset the timer so a repeated trip (should the worker somehow
|
||||
# survive) waits a fresh interval rather than re-firing instantly.
|
||||
_bedrock_last_event["t"] = time.time()
|
||||
# Escalate across turns: raises RuntimeError once the streak
|
||||
# crosses HERMES_STREAM_STALE_GIVEUP, so a persistently wedged
|
||||
# Bedrock provider aborts fast instead of re-waiting the timeout.
|
||||
_check_stale_giveup(agent)
|
||||
# Streak still under the give-up threshold: end THIS call with a
|
||||
# TimeoutError so the outer retry loop / next turn re-evaluates
|
||||
# and the streak carries forward. Break rather than keep polling
|
||||
# a worker we cannot abort.
|
||||
result["error"] = TimeoutError(
|
||||
f"Bedrock stream produced no events for {int(_stale_elapsed)}s "
|
||||
f"(threshold {int(_bedrock_stale_timeout)}s) — aborting stalled "
|
||||
f"stream so the retry/fallback path can recover."
|
||||
)
|
||||
break
|
||||
# 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"]
|
||||
# Success — clear the cross-turn breaker (#58962): Bedrock proved
|
||||
# responsive. Mirrors the OpenAI/Anthropic success reset below so a
|
||||
# recovered provider doesn't carry a stale streak into later turns.
|
||||
if result["response"] is not None:
|
||||
_reset_stale_streak(agent)
|
||||
_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": []}
|
||||
|
||||
|
|
@ -4008,6 +4065,7 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
|
|||
if agent._interrupt_requested:
|
||||
_cancel_current_stream_attempt("interrupt_before_stream_retry")
|
||||
raise InterruptedError("Agent interrupted before stream retry")
|
||||
_emit_stream_start()
|
||||
try:
|
||||
if agent.api_mode == "anthropic_messages":
|
||||
# #67142: per-request client (credential refresh happens
|
||||
|
|
@ -4022,8 +4080,14 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
|
|||
result["response"] = _call_anthropic(request_client)
|
||||
else:
|
||||
result["response"] = _call_chat_completions(stream_attempt_id)
|
||||
_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))
|
||||
_close_managed_stream()
|
||||
# If the main poll loop force-closed this request because
|
||||
# of an interrupt, the resulting transport error is the
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
@ -155,6 +155,13 @@ VALID_HOOKS: Set[str] = {
|
|||
"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
|
||||
|
|
@ -2835,6 +2842,10 @@ class PluginManager:
|
|||
"""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 render_system_prompt_sections(
|
||||
self, session_info: Mapping[str, Any]
|
||||
) -> List[RenderedPluginSystemPromptSection]:
|
||||
|
|
@ -3245,6 +3256,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()
|
||||
|
||||
|
||||
|
|
|
|||
84
run_agent.py
84
run_agent.py
|
|
@ -6355,8 +6355,7 @@ class AIAgent:
|
|||
when the only streamed text was unrelated mid-turn commentary. (#65919
|
||||
review: response-loss blocker)
|
||||
"""
|
||||
cb = getattr(self, "interim_assistant_callback", None)
|
||||
if cb is None or not isinstance(assistant_msg, dict):
|
||||
if not isinstance(assistant_msg, dict):
|
||||
return
|
||||
commentary_parts = self._extract_codex_interim_visible_parts(assistant_msg)
|
||||
undelivered_parts: List[str] = []
|
||||
|
|
@ -6383,6 +6382,25 @@ class AIAgent:
|
|||
):
|
||||
return
|
||||
already_streamed = self._interim_content_was_streamed(visible)
|
||||
try:
|
||||
from agent.plugin_stream_hooks import enqueue_plugin_stream_hook
|
||||
|
||||
enqueue_plugin_stream_hook(
|
||||
"on_interim_message",
|
||||
turn_id=getattr(self, "_current_turn_id", "") or "",
|
||||
iteration=int(getattr(self, "_api_call_count", 0) or 0),
|
||||
session_id=self.session_id or "",
|
||||
model=self.model or "",
|
||||
provider=self.provider or "",
|
||||
surface=self.platform or "cli",
|
||||
text=visible,
|
||||
already_streamed=already_streamed,
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("on_interim_message plugin hook enqueue failed", exc_info=True)
|
||||
cb = getattr(self, "interim_assistant_callback", None)
|
||||
if cb is None:
|
||||
return
|
||||
try:
|
||||
cb(visible, already_streamed=already_streamed)
|
||||
if undelivered_parts:
|
||||
|
|
@ -6468,6 +6486,38 @@ class AIAgent:
|
|||
where, _n,
|
||||
)
|
||||
|
||||
def _stream_hook_base_payload(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"turn_id": getattr(self, "_current_turn_id", "") or "",
|
||||
"iteration": int(getattr(self, "_api_call_count", 0) or 0),
|
||||
"session_id": self.session_id or "",
|
||||
"model": self.model or "",
|
||||
"provider": self.provider or "",
|
||||
"surface": self.platform or "cli",
|
||||
}
|
||||
|
||||
def _emit_stream_start(self) -> None:
|
||||
try:
|
||||
from agent.plugin_stream_hooks import enqueue_plugin_stream_hook
|
||||
|
||||
enqueue_plugin_stream_hook("on_stream_start", **self._stream_hook_base_payload())
|
||||
except Exception:
|
||||
logger.debug("on_stream_start plugin hook enqueue failed", exc_info=True)
|
||||
|
||||
def _emit_stream_end(self, *, final_text: str, finished: bool, error: str | None) -> None:
|
||||
try:
|
||||
from agent.plugin_stream_hooks import enqueue_plugin_stream_hook
|
||||
|
||||
enqueue_plugin_stream_hook(
|
||||
"on_stream_end",
|
||||
**self._stream_hook_base_payload(),
|
||||
final_text=final_text,
|
||||
finished=finished,
|
||||
error=error,
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("on_stream_end plugin hook enqueue failed", exc_info=True)
|
||||
|
||||
def _fire_stream_delta(self, text: str) -> None:
|
||||
"""Fire all registered stream delta callbacks (display + TTS)."""
|
||||
# Single-writer guard (#65991): a superseded stream must not interleave
|
||||
|
|
@ -6523,6 +6573,17 @@ class AIAgent:
|
|||
delivered = True
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
from agent.plugin_stream_hooks import enqueue_plugin_stream_hook
|
||||
|
||||
enqueue_plugin_stream_hook(
|
||||
"on_stream_delta",
|
||||
**self._stream_hook_base_payload(),
|
||||
delta=text,
|
||||
kind="text",
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("on_stream_delta plugin hook enqueue failed", exc_info=True)
|
||||
if delivered:
|
||||
self._record_streamed_assistant_text(text)
|
||||
|
||||
|
|
@ -6539,6 +6600,18 @@ class AIAgent:
|
|||
cb(text)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
from agent.plugin_stream_hooks import enqueue_plugin_stream_hook, stream_reasoning_deltas_enabled
|
||||
|
||||
if stream_reasoning_deltas_enabled():
|
||||
enqueue_plugin_stream_hook(
|
||||
"on_stream_delta",
|
||||
**self._stream_hook_base_payload(),
|
||||
delta=text,
|
||||
kind="reasoning",
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("reasoning on_stream_delta plugin hook enqueue failed", exc_info=True)
|
||||
|
||||
def _fire_tool_gen_started(self, tool_name: str) -> None:
|
||||
"""Notify display layer that the model is generating tool call arguments.
|
||||
|
|
@ -6557,6 +6630,13 @@ class AIAgent:
|
|||
|
||||
def _has_stream_consumers(self) -> bool:
|
||||
"""Return True if any streaming consumer is registered."""
|
||||
try:
|
||||
from agent.plugin_stream_hooks import has_stream_observer_hooks
|
||||
|
||||
if has_stream_observer_hooks():
|
||||
return True
|
||||
except Exception:
|
||||
logger.debug("plugin stream hook consumer check failed", exc_info=True)
|
||||
return (
|
||||
self.stream_delta_callback is not None
|
||||
or getattr(self, "_stream_callback", None) is not None
|
||||
|
|
|
|||
|
|
@ -0,0 +1,359 @@
|
|||
import threading
|
||||
import time
|
||||
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
def _agent():
|
||||
from run_agent import AIAgent
|
||||
|
||||
return AIAgent(
|
||||
api_key="test-key",
|
||||
base_url="https://openrouter.ai/api/v1",
|
||||
provider="openrouter",
|
||||
model="test/model",
|
||||
quiet_mode=True,
|
||||
skip_context_files=True,
|
||||
skip_memory=True,
|
||||
)
|
||||
|
||||
|
||||
def _wait_for(predicate, timeout=1.0):
|
||||
deadline = time.monotonic() + timeout
|
||||
while time.monotonic() < deadline:
|
||||
if predicate():
|
||||
return
|
||||
time.sleep(0.01)
|
||||
assert predicate()
|
||||
|
||||
|
||||
def _make_stream_chunk(content=None, finish_reason=None):
|
||||
delta = SimpleNamespace(content=content, reasoning_content=None, reasoning=None, tool_calls=None)
|
||||
choice = SimpleNamespace(delta=delta, finish_reason=finish_reason)
|
||||
return SimpleNamespace(choices=[choice], model="test/model")
|
||||
|
||||
|
||||
def _callbacks(callbacks_by_hook):
|
||||
return lambda name: tuple(callbacks_by_hook.get(name, ()))
|
||||
|
||||
|
||||
def test_stream_observer_hooks_are_valid_plugin_hooks():
|
||||
from hermes_cli.plugins import VALID_HOOKS
|
||||
|
||||
assert {
|
||||
"on_stream_start",
|
||||
"on_stream_delta",
|
||||
"on_stream_end",
|
||||
"on_interim_message",
|
||||
}.issubset(VALID_HOOKS)
|
||||
|
||||
|
||||
def test_stream_delta_plugin_hook_is_queued_off_token_path(monkeypatch):
|
||||
from agent.plugin_stream_hooks import shutdown_plugin_stream_hook_dispatcher
|
||||
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
calls = []
|
||||
|
||||
def on_stream_delta(**kwargs):
|
||||
time.sleep(0.2)
|
||||
calls.append(("on_stream_delta", kwargs))
|
||||
|
||||
monkeypatch.setattr("hermes_cli.plugins.iter_hook_callbacks", _callbacks({"on_stream_delta": [on_stream_delta]}))
|
||||
|
||||
agent = _agent()
|
||||
|
||||
started = time.monotonic()
|
||||
agent._fire_stream_delta("hello")
|
||||
elapsed = time.monotonic() - started
|
||||
|
||||
assert elapsed < 0.05
|
||||
_wait_for(lambda: calls)
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
|
||||
assert calls[0][0] == "on_stream_delta"
|
||||
assert calls[0][1]["delta"] == "hello"
|
||||
assert calls[0][1]["kind"] == "text"
|
||||
assert calls[0][1]["model"] == "test/model"
|
||||
assert calls[0][1]["provider"] == "openrouter"
|
||||
|
||||
|
||||
def test_stream_delta_plugin_hook_error_does_not_break_streaming(monkeypatch):
|
||||
from agent.plugin_stream_hooks import shutdown_plugin_stream_hook_dispatcher
|
||||
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
ui_deltas = []
|
||||
|
||||
def on_stream_delta(**_kwargs):
|
||||
raise RuntimeError("plugin failed")
|
||||
|
||||
monkeypatch.setattr("hermes_cli.plugins.iter_hook_callbacks", _callbacks({"on_stream_delta": [on_stream_delta]}))
|
||||
|
||||
agent = _agent()
|
||||
agent.stream_delta_callback = ui_deltas.append
|
||||
|
||||
agent._fire_stream_delta("still visible")
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
|
||||
assert ui_deltas == ["still visible"]
|
||||
|
||||
|
||||
def test_stream_hook_queue_drops_oldest_pending_event_when_full(monkeypatch):
|
||||
from agent.plugin_stream_hooks import enqueue_plugin_stream_hook, shutdown_plugin_stream_hook_dispatcher
|
||||
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
monkeypatch.setattr("agent.plugin_stream_hooks._QUEUE_SIZE", 1)
|
||||
delivered = []
|
||||
first_delivered = threading.Event()
|
||||
release_worker = threading.Event()
|
||||
|
||||
def on_stream_delta(**kwargs):
|
||||
delivered.append(kwargs["delta"])
|
||||
first_delivered.set()
|
||||
release_worker.wait(timeout=1.0)
|
||||
|
||||
monkeypatch.setattr("hermes_cli.plugins.iter_hook_callbacks", _callbacks({"on_stream_delta": [on_stream_delta]}))
|
||||
|
||||
assert enqueue_plugin_stream_hook("on_stream_delta", delta="first") is True
|
||||
assert first_delivered.wait(timeout=1.0)
|
||||
assert enqueue_plugin_stream_hook("on_stream_delta", delta="second") is True
|
||||
assert enqueue_plugin_stream_hook("on_stream_delta", delta="third") is True
|
||||
|
||||
release_worker.set()
|
||||
_wait_for(lambda: "third" in delivered)
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
|
||||
assert delivered == ["first", "third"]
|
||||
|
||||
|
||||
def test_stream_hook_queue_isolated_per_consumer(monkeypatch):
|
||||
from agent.plugin_stream_hooks import enqueue_plugin_stream_hook, shutdown_plugin_stream_hook_dispatcher
|
||||
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
monkeypatch.setattr("agent.plugin_stream_hooks._QUEUE_SIZE", 1)
|
||||
slow_delivered = []
|
||||
fast_delivered = []
|
||||
slow_started = threading.Event()
|
||||
release_slow = threading.Event()
|
||||
|
||||
def slow_consumer(**kwargs):
|
||||
slow_delivered.append(kwargs["delta"])
|
||||
slow_started.set()
|
||||
release_slow.wait(timeout=1.0)
|
||||
|
||||
def fast_consumer(**kwargs):
|
||||
fast_delivered.append(kwargs["delta"])
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.plugins.iter_hook_callbacks",
|
||||
_callbacks({"on_stream_delta": [slow_consumer, fast_consumer]}),
|
||||
)
|
||||
|
||||
assert enqueue_plugin_stream_hook("on_stream_delta", delta="first") is True
|
||||
assert slow_started.wait(timeout=1.0)
|
||||
_wait_for(lambda: fast_delivered == ["first"])
|
||||
assert enqueue_plugin_stream_hook("on_stream_delta", delta="second") is True
|
||||
_wait_for(lambda: fast_delivered == ["first", "second"])
|
||||
assert enqueue_plugin_stream_hook("on_stream_delta", delta="third") is True
|
||||
|
||||
_wait_for(lambda: fast_delivered == ["first", "second", "third"])
|
||||
release_slow.set()
|
||||
_wait_for(lambda: "third" in slow_delivered)
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
|
||||
assert slow_delivered == ["first", "third"]
|
||||
|
||||
|
||||
def test_reasoning_stream_delta_plugin_hook_is_opt_in(monkeypatch):
|
||||
from agent.plugin_stream_hooks import shutdown_plugin_stream_hook_dispatcher
|
||||
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
calls = []
|
||||
|
||||
def on_stream_delta(**kwargs):
|
||||
calls.append(("on_stream_delta", kwargs))
|
||||
|
||||
monkeypatch.setattr("hermes_cli.plugins.iter_hook_callbacks", _callbacks({"on_stream_delta": [on_stream_delta]}))
|
||||
|
||||
agent = _agent()
|
||||
agent._fire_reasoning_delta("private chain")
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
|
||||
assert calls == []
|
||||
|
||||
with patch("hermes_cli.config.cfg_get", return_value=True):
|
||||
agent._fire_reasoning_delta("visible reasoning")
|
||||
_wait_for(lambda: calls)
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
|
||||
assert calls[0][0] == "on_stream_delta"
|
||||
assert calls[0][1]["kind"] == "reasoning"
|
||||
assert calls[0][1]["delta"] == "visible reasoning"
|
||||
|
||||
|
||||
def test_interim_message_plugin_hook_is_queued(monkeypatch):
|
||||
from agent.plugin_stream_hooks import shutdown_plugin_stream_hook_dispatcher
|
||||
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
calls = []
|
||||
|
||||
def on_interim_message(**kwargs):
|
||||
calls.append(("on_interim_message", kwargs))
|
||||
|
||||
monkeypatch.setattr("hermes_cli.plugins.iter_hook_callbacks", _callbacks({"on_interim_message": [on_interim_message]}))
|
||||
|
||||
agent = _agent()
|
||||
agent._emit_interim_assistant_message({"content": "I will inspect the files first."})
|
||||
_wait_for(lambda: calls)
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
|
||||
assert calls[0][0] == "on_interim_message"
|
||||
assert calls[0][1]["text"] == "I will inspect the files first."
|
||||
assert calls[0][1]["already_streamed"] is False
|
||||
|
||||
|
||||
def test_stream_plugin_hook_counts_as_stream_consumer(monkeypatch):
|
||||
monkeypatch.setattr("hermes_cli.plugins.iter_hook_callbacks", _callbacks({"on_stream_delta": [lambda **_kwargs: None]}))
|
||||
|
||||
agent = _agent()
|
||||
|
||||
assert agent._has_stream_consumers() is True
|
||||
|
||||
|
||||
def test_interim_message_plugin_hook_does_not_count_as_stream_consumer(monkeypatch):
|
||||
monkeypatch.setattr("hermes_cli.plugins.iter_hook_callbacks", _callbacks({"on_interim_message": [lambda **_kwargs: None]}))
|
||||
|
||||
agent = _agent()
|
||||
|
||||
assert agent._has_stream_consumers() is False
|
||||
|
||||
|
||||
def test_stream_lifecycle_plugin_hooks_are_queued(monkeypatch):
|
||||
from agent.plugin_stream_hooks import shutdown_plugin_stream_hook_dispatcher
|
||||
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
calls = []
|
||||
|
||||
def on_stream_start(**kwargs):
|
||||
calls.append(("on_stream_start", kwargs))
|
||||
|
||||
def on_stream_end(**kwargs):
|
||||
calls.append(("on_stream_end", kwargs))
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.plugins.iter_hook_callbacks",
|
||||
_callbacks({"on_stream_start": [on_stream_start], "on_stream_end": [on_stream_end]}),
|
||||
)
|
||||
|
||||
agent = _agent()
|
||||
agent._emit_stream_start()
|
||||
agent._emit_stream_end(final_text="done", finished=True, error=None)
|
||||
_wait_for(lambda: len(calls) == 2)
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
|
||||
assert [call[0] for call in calls] == ["on_stream_start", "on_stream_end"]
|
||||
assert calls[0][1]["model"] == "test/model"
|
||||
assert calls[1][1]["final_text"] == "done"
|
||||
assert calls[1][1]["finished"] is True
|
||||
assert calls[1][1]["error"] is None
|
||||
|
||||
|
||||
@patch("run_agent.AIAgent._create_request_openai_client")
|
||||
@patch("run_agent.AIAgent._close_request_openai_client")
|
||||
def test_chat_completion_stream_emits_lifecycle_hooks(_mock_close, mock_create, monkeypatch):
|
||||
from agent.plugin_stream_hooks import shutdown_plugin_stream_hook_dispatcher
|
||||
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
calls = []
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.plugins.iter_hook_callbacks",
|
||||
_callbacks(
|
||||
{
|
||||
"on_stream_start": [lambda **kwargs: calls.append(("on_stream_start", kwargs))],
|
||||
"on_stream_delta": [lambda **kwargs: calls.append(("on_stream_delta", kwargs))],
|
||||
"on_stream_end": [lambda **kwargs: calls.append(("on_stream_end", kwargs))],
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
mock_client = SimpleNamespace(
|
||||
chat=SimpleNamespace(
|
||||
completions=SimpleNamespace(
|
||||
create=lambda **_kwargs: iter([
|
||||
_make_stream_chunk(content="hello "),
|
||||
_make_stream_chunk(content="world"),
|
||||
_make_stream_chunk(finish_reason="stop"),
|
||||
])
|
||||
)
|
||||
)
|
||||
)
|
||||
mock_create.return_value = mock_client
|
||||
|
||||
agent = _agent()
|
||||
agent.api_mode = "chat_completions"
|
||||
response = agent._interruptible_streaming_api_call({})
|
||||
|
||||
_wait_for(lambda: [call[0] for call in calls].count("on_stream_end") == 1)
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
|
||||
assert response.choices[0].message.content == "hello world"
|
||||
assert [call[0] for call in calls] == [
|
||||
"on_stream_start",
|
||||
"on_stream_delta",
|
||||
"on_stream_delta",
|
||||
"on_stream_end",
|
||||
]
|
||||
assert calls[-1][1]["final_text"] == "hello world"
|
||||
assert calls[-1][1]["finished"] is True
|
||||
|
||||
|
||||
def test_bedrock_reasoning_delta_reaches_plugin_only_observer(monkeypatch):
|
||||
from agent.plugin_stream_hooks import shutdown_plugin_stream_hook_dispatcher
|
||||
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
calls = []
|
||||
|
||||
def on_stream_delta(**kwargs):
|
||||
calls.append(kwargs)
|
||||
|
||||
monkeypatch.setattr("hermes_cli.plugins.iter_hook_callbacks", _callbacks({"on_stream_delta": [on_stream_delta]}))
|
||||
monkeypatch.setattr("hermes_cli.config.cfg_get", lambda *_args, **_kwargs: True)
|
||||
monkeypatch.setattr(
|
||||
"agent.bedrock_adapter._get_bedrock_runtime_client",
|
||||
lambda _region: SimpleNamespace(converse_stream=lambda **_kwargs: {"stream": []}),
|
||||
)
|
||||
monkeypatch.setattr("agent.bedrock_adapter.is_stale_connection_error", lambda _exc: False)
|
||||
monkeypatch.setattr("agent.bedrock_adapter.is_streaming_access_denied_error", lambda _exc: False)
|
||||
monkeypatch.setattr("agent.bedrock_adapter.invalidate_runtime_client", lambda *_args, **_kwargs: None)
|
||||
|
||||
def stream_converse_with_callbacks(
|
||||
_raw_response,
|
||||
*,
|
||||
on_text_delta=None,
|
||||
on_tool_start=None,
|
||||
on_reasoning_delta=None,
|
||||
on_interrupt_check=None,
|
||||
on_event=None,
|
||||
**_kwargs,
|
||||
):
|
||||
# Main's Bedrock path also invokes this as a Relay finalizer with the
|
||||
# intercepted-event replay; only the live pass wires callbacks.
|
||||
if on_reasoning_delta is not None:
|
||||
assert on_tool_start is not None
|
||||
assert on_interrupt_check() is False
|
||||
on_reasoning_delta("bedrock reasoning")
|
||||
return SimpleNamespace(choices=[], usage=None, stop_reason="end_turn")
|
||||
|
||||
monkeypatch.setattr("agent.bedrock_adapter.stream_converse_with_callbacks", stream_converse_with_callbacks)
|
||||
|
||||
agent = _agent()
|
||||
agent.api_mode = "bedrock_converse"
|
||||
agent.reasoning_callback = None
|
||||
agent.stream_delta_callback = None
|
||||
|
||||
agent._interruptible_streaming_api_call({"__bedrock_region__": "us-east-1", "__bedrock_converse__": True})
|
||||
_wait_for(lambda: calls)
|
||||
shutdown_plugin_stream_hook_dispatcher()
|
||||
|
||||
assert calls[0]["kind"] == "reasoning"
|
||||
assert calls[0]["delta"] == "bedrock reasoning"
|
||||
|
|
@ -448,6 +448,10 @@ Payload fields below are the exact event-specific fields supplied by each call s
|
|||
| `pre_api_request` | Observer | Per provider attempt, immediately before the request; return ignored. | `task_id`, `turn_id`, `api_request_id`, `session_id`, `user_message`, `conversation_history`, `platform`, `model`, `provider`, `base_url`, `api_mode`, `api_call_count`, `retry_count`, `request_messages`, `message_count`, `tool_count`, `approx_input_tokens`, `request_char_count`, `max_tokens`, `started_at`, `middleware_trace`, `request` | High sensitivity: legacy `user_message`, `conversation_history`, and `request_messages` are intentionally raw; prefer sanitized `request`. |
|
||||
| `post_api_request` | Observer | After normalized provider success; return ignored. | `task_id`, `turn_id`, `api_request_id`, `session_id`, `platform`, `model`, `provider`, `base_url`, `api_mode`, `api_call_count`, `api_duration`, `started_at`, `ended_at`, `finish_reason`, `message_count`, `response_model`, `response`, `usage`, `assistant_message`, `assistant_content_chars`, `assistant_tool_call_count` | Sanitized `response` is available, but raw normalized `assistant_message` may contain model/user content; `usage` is accounting data. |
|
||||
| `api_request_error` | Observer | On each failed provider attempt; return ignored. | `task_id`, `turn_id`, `api_request_id`, `session_id`, `platform`, `model`, `provider`, `base_url`, `api_mode`, `api_call_count`, `api_duration`, `started_at`, `ended_at`, `status_code`, `retry_count`, `max_retries`, `retryable`, `reason`, `error`, `request` | Error text may contain provider/user data; `request` is intended to be sanitized. |
|
||||
| `on_stream_start` | Observer | Dispatched when a streaming LLM response begins; delivered off the token path via a host-owned bounded queue with one worker per callback; return ignored. | `turn_id`, `iteration`, `session_id`, `model`, `provider`, `surface` | Identifiers and routing metadata only. |
|
||||
| `on_stream_delta` | Observer | Dispatched per normalized streaming text delta via the bounded observer queue; a stalled callback drops only its own oldest events; return ignored. | `delta`, `kind` (`text` or `reasoning`), `turn_id`, `iteration`, `session_id`, `model`, `provider`, `surface` | Delta text is raw model output; reasoning deltas require the `plugins.stream_reasoning_deltas` opt-in. |
|
||||
| `on_stream_end` | Observer | Dispatched when a streaming response finishes or errors, after the stream closes; return ignored. | `final_text`, `finished`, `error`, `turn_id`, `iteration`, `session_id`, `model`, `provider`, `surface` | Full assembled response text; error text may include provider data. |
|
||||
| `on_interim_message` | Observer | Dispatched when a mid-loop assistant message is surfaced before the final answer (streaming or non-streaming); return ignored. | `text`, `already_streamed`, `turn_id`, `iteration`, `session_id`, `model`, `provider`, `surface` | Full interim assistant text. |
|
||||
| `on_session_start` | Observer | First turn of a new session; return ignored. | `session_id`, `model`, `platform` | Identifiers and routing metadata only. |
|
||||
| `on_session_end` | Observer | Canonically at each turn finalization; CLI/TUI exits have additional reduced legacy shapes. Return ignored. | Canonical: `session_id`, `task_id`, `turn_id`, `completed`, `failed`, `interrupted`, `turn_exit_reason`, `model`, `platform`; exit paths may add `reason`/`api_request_id` and omit fields. | IDs, model/platform, and outcome; canonical payload has no message body. |
|
||||
| `on_session_finalize` | Observer | CLI/TUI/gateway teardown through `finalize_session`; gateway shutdown or expiry may finalize without a reset. Return ignored. | Surface-dependent `session_id`, `platform`, optionally `reason`, `old_session_id`, `new_session_id` | Session and routing identifiers. |
|
||||
|
|
@ -465,6 +469,54 @@ Payload fields below are the exact event-specific fields supplied by each call s
|
|||
|
||||
---
|
||||
|
||||
### Streaming output hooks
|
||||
|
||||
These observer-only hooks let plugins consume streaming LLM output for telemetry, live dashboards, or TTS pipelines without changing the response. They are delivered through host-owned bounded queues with one background worker per registered callback, so plugin callbacks never run inline on the token path. If one callback stalls, only that callback's queue can fill and drop its oldest pending observer event; other observers continue receiving events independently.
|
||||
|
||||
Register them like any other plugin hook:
|
||||
|
||||
```python
|
||||
def on_delta(delta, kind, model, provider, **kwargs):
|
||||
if kind == "text":
|
||||
print(delta, end="", flush=True)
|
||||
|
||||
def register(ctx):
|
||||
ctx.register_hook("on_stream_delta", on_delta)
|
||||
```
|
||||
|
||||
Common fields for all four hooks:
|
||||
|
||||
| Parameter | Type | Description |
|
||||
|-----------|------|-------------|
|
||||
| `turn_id` | `str` | Opaque turn identifier, when available |
|
||||
| `iteration` | `int` | Current API-call/tool-loop iteration |
|
||||
| `session_id` | `str` | Current Hermes session id |
|
||||
| `model` | `str` | Active model identifier |
|
||||
| `provider` | `str` | Active provider name |
|
||||
| `surface` | `str` | Calling surface, e.g. `cli`, `discord`, `telegram` |
|
||||
|
||||
Additional fields:
|
||||
|
||||
| Hook | Extra fields |
|
||||
|------|--------------|
|
||||
| `on_stream_start` | none |
|
||||
| `on_stream_delta` | `delta: str`, `kind: "text" | "reasoning"` |
|
||||
| `on_stream_end` | `final_text: str`, `finished: bool`, `error: str | None` |
|
||||
| `on_interim_message` | `text: str`, `already_streamed: bool` |
|
||||
|
||||
`on_interim_message` can also fire after a non-streaming response, so registering only that hook does not force a provider call onto streaming transport.
|
||||
|
||||
Reasoning deltas are not exposed to plugins by default. Opt in explicitly:
|
||||
|
||||
```yaml
|
||||
plugins:
|
||||
stream_reasoning_deltas: true
|
||||
```
|
||||
|
||||
Return values are ignored. To keep the stream fast, callbacks should enqueue their own work and return quickly. Exceptions are logged and do not stop the stream.
|
||||
|
||||
---
|
||||
|
||||
### `pre_tool_call`
|
||||
|
||||
Fires **immediately before** every tool execution — built-in tools and plugin tools alike.
|
||||
|
|
|
|||
|
|
@ -265,7 +265,7 @@ Plugins can register the 24 lifecycle events currently accepted by `hermes_cli.p
|
|||
|---|---|
|
||||
| **Directive/control** | `pre_tool_call`, `pre_llm_call`, `pre_verify`, `pre_gateway_dispatch` |
|
||||
| **Transform** | `transform_tool_result`, `transform_terminal_output`, `transform_llm_output` |
|
||||
| **Observer** | `post_tool_call`, `post_llm_call`, `pre_api_request`, `post_api_request`, `api_request_error`, `on_session_start`, `on_session_end`, `on_session_finalize`, `on_session_reset`, `on_skill_lifecycle`, `subagent_start`, `subagent_stop`, `pre_approval_request`, `post_approval_response`, `kanban_task_claimed`, `kanban_task_completed`, `kanban_task_blocked` |
|
||||
| **Observer** | `post_tool_call`, `post_llm_call`, `pre_api_request`, `post_api_request`, `api_request_error`, `on_stream_start`, `on_stream_delta`, `on_stream_end`, `on_interim_message`, `on_session_start`, `on_session_end`, `on_session_finalize`, `on_session_reset`, `on_skill_lifecycle`, `subagent_start`, `subagent_stop`, `pre_approval_request`, `post_approval_response`, `kanban_task_claimed`, `kanban_task_completed`, `kanban_task_blocked` |
|
||||
|
||||
These categories describe current behavior rather than defining future naming rules. Plugin middleware remains a separate registry/surface.
|
||||
## Plugin types
|
||||
|
|
|
|||
Loading…
Reference in New Issue