diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py
index 45a88f8decf0d..730ca3535fd59 100644
--- a/agent/chat_completion_helpers.py
+++ b/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
diff --git a/agent/plugin_stream_hooks.py b/agent/plugin_stream_hooks.py
new file mode 100644
index 0000000000000..a5c62e1d518d6
--- /dev/null
+++ b/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)
diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py
index 6ca393fca53c1..4c598fe5ae719 100644
--- a/hermes_cli/plugins.py
+++ b/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()
 
 
diff --git a/run_agent.py b/run_agent.py
index bcacec3909b69..c012590aed92f 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -4749,19 +4749,69 @@ def _interim_content_was_streamed(self, content: str) -> bool:
 
     def _emit_interim_assistant_message(self, assistant_msg: Dict[str, Any]) -> None:
         """Surface a real mid-turn assistant commentary message to the UI layer."""
-        cb = getattr(self, "interim_assistant_callback", None)
-        if cb is None or not isinstance(assistant_msg, dict):
+        if not isinstance(assistant_msg, dict):
             return
         content = assistant_msg.get("content")
         visible = self._strip_think_blocks(content or "").strip()
         if not visible or visible == "(empty)":
             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)
         except Exception:
             logger.debug("interim_assistant_callback error", exc_info=True)
 
+    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)."""
         # If a tool iteration set the break flag, prepend a single paragraph
@@ -4812,6 +4862,17 @@ def _fire_stream_delta(self, text: str) -> None:
                 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)
 
@@ -4823,6 +4884,18 @@ def _fire_reasoning_delta(self, text: str) -> None:
                 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.
@@ -4841,6 +4914,13 @@ def _fire_tool_gen_started(self, tool_name: str) -> None:
 
     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
diff --git a/scripts/release.py b/scripts/release.py
index 571009f4f0cab..f65d703765356 100755
--- a/scripts/release.py
+++ b/scripts/release.py
@@ -45,6 +45,7 @@
 
 # Auto-extracted from noreply emails + manual overrides
 AUTHOR_MAP = {
+    "dnethusahan.h05@gmail.com": "deaneeth",  # PR #64317 (plugins: streaming output observer hooks; #64161)
     "41409874+2751738943@users.noreply.github.com": "2751738943",  # PR #54785 salvage (tui: post-turn completion ownership routing)
     "Burgunthy@users.noreply.github.com": "Burgunthy",  # PR #20096 salvage (gateway: profile-based routing for inbound messages)
     "75556242+webtecnica@users.noreply.github.com": "webtecnica",  # PR #63360 salvage (nous: restore inference-api base_url)
diff --git a/tests/run_agent/test_plugin_stream_hooks.py b/tests/run_agent/test_plugin_stream_hooks.py
new file mode 100644
index 0000000000000..708e7528103ee
--- /dev/null
+++ b/tests/run_agent/test_plugin_stream_hooks.py
@@ -0,0 +1,356 @@
+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,
+        on_tool_start,
+        on_reasoning_delta,
+        on_interrupt_check,
+    ):
+        assert on_text_delta is not None
+        assert on_tool_start is not None
+        assert on_interrupt_check() is False
+        assert on_reasoning_delta is not None
+        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"
diff --git a/website/docs/user-guide/features/hooks.md b/website/docs/user-guide/features/hooks.md
index f38ed9343b96d..23911d6c5769a 100644
--- a/website/docs/user-guide/features/hooks.md
+++ b/website/docs/user-guide/features/hooks.md
@@ -382,6 +382,10 @@ def register(ctx):
 | [`post_tool_call`](#post_tool_call) | After any tool returns | ignored |
 | [`pre_llm_call`](#pre_llm_call) | Once per turn, before the tool-calling loop | `{"context": str}` to prepend context to the user message |
 | [`post_llm_call`](#post_llm_call) | Once per turn, after the tool-calling loop | ignored |
+| [`on_stream_start`](#streaming-output-hooks) | A streaming LLM response starts | ignored |
+| [`on_stream_delta`](#streaming-output-hooks) | A normalized streaming text delta is produced | ignored |
+| [`on_stream_end`](#streaming-output-hooks) | A streaming LLM response finishes or errors | ignored |
+| [`on_interim_message`](#streaming-output-hooks) | A mid-loop assistant message is surfaced before the final answer | ignored |
 | [`pre_verify`](#pre_verify) | Once per turn when the agent edited code, before it verifies/finishes | `{"action": "continue", "message": str}` to keep going |
 | [`on_session_start`](#on_session_start) | New session created (first turn only) | ignored |
 | [`on_session_end`](#on_session_end) | Session ends | ignored |
@@ -398,6 +402,54 @@ def register(ctx):
 
 ---
 
+### 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.
diff --git a/website/docs/user-guide/features/plugins.md b/website/docs/user-guide/features/plugins.md
index ed8012325b6e5..b2d4b27eb757c 100644
--- a/website/docs/user-guide/features/plugins.md
+++ b/website/docs/user-guide/features/plugins.md
@@ -195,6 +195,10 @@ Plugins can register callbacks for these lifecycle events. See the **[Event Hook
 | [`post_tool_call`](/user-guide/features/hooks#post_tool_call) | After any tool returns |
 | [`pre_llm_call`](/user-guide/features/hooks#pre_llm_call) | Once per turn, before the LLM loop — can return `{"context": "..."}` to [inject context into the user message](/user-guide/features/hooks#pre_llm_call) |
 | [`post_llm_call`](/user-guide/features/hooks#post_llm_call) | Once per turn, after the LLM loop (successful turns only) |
+| [`on_stream_start`](/user-guide/features/hooks#streaming-output-hooks) | A streaming LLM response starts |
+| [`on_stream_delta`](/user-guide/features/hooks#streaming-output-hooks) | A normalized text delta is produced during streaming |
+| [`on_stream_end`](/user-guide/features/hooks#streaming-output-hooks) | A streaming LLM response finishes or errors |
+| [`on_interim_message`](/user-guide/features/hooks#streaming-output-hooks) | A mid-loop assistant message is surfaced before the final answer |
 | [`on_session_start`](/user-guide/features/hooks#on_session_start) | New session created (first turn only) |
 | [`on_session_end`](/user-guide/features/hooks#on_session_end) | End of every `run_conversation` call + CLI exit handler |
 | [`on_session_finalize`](/user-guide/features/hooks#on_session_finalize) | CLI/gateway tears down an active session (`/new`, GC, CLI quit) |