feat(streaming): publish observed local runtime model to SSE and journal

api/routes.py

@@ -4067,6 +4067,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict |
         "tool_calls": tool_calls,
         "last_assistant_text": assistant_text,
         "last_reasoning_text": reasoning_text,
+        "runtime_model": runtime_model_from_events(session_id, stream_id, events),
         "activity_burst_anchors": activity_burst_anchors,
         "current_activity_burst_id": current_activity_burst_id,
         "current_live_segment_seq": current_live_segment_seq,
@@ -10839,6 +10840,7 @@ from api.streaming import (
 )
 from api.gateway_chat import _run_gateway_chat_streaming, webui_gateway_chat_enabled
 from api.run_journal import (
+    runtime_model_from_events,
     _parse_run_journal_event_id as _shared_parse_run_journal_event_id,
     _summary_from_events,
     bound_run_journal_snapshot_args,

api/run_journal.py

@@ -10,6 +10,7 @@ import os
 import re
 import threading
 import time
+from copy import deepcopy
 from collections import OrderedDict
 from pathlib import Path
 from typing import Any, Iterable
@@ -130,7 +131,7 @@ def _get_cached_summary(path: Path) -> dict | None:
             _SUMMARY_CACHE.pop(key, None)
             return None
         _SUMMARY_CACHE.move_to_end(key)
-        return dict(summary)
+        return deepcopy(summary)
 
 
 def _cache_summary(
@@ -147,7 +148,7 @@ def _cache_summary(
         return
     key = str(path)
     with _SUMMARY_CACHE_LOCK:
-        _SUMMARY_CACHE[key] = (signature, dict(summary))
+        _SUMMARY_CACHE[key] = (signature, deepcopy(summary))
         _SUMMARY_CACHE.move_to_end(key)
         while len(_SUMMARY_CACHE) > _SUMMARY_CACHE_MAX_ENTRIES:
             _SUMMARY_CACHE.popitem(last=False)
@@ -534,6 +535,37 @@ def select_authoritative_terminal_event(events: Iterable[dict]) -> dict | None:
     )
 
 
+def runtime_model_from_events(session_id: str, stream_id: str, events: Iterable[dict]) -> dict | None:
+    """Project only observed serving identity for this journal owner.
+
+    A fallback warning or malformed later observation invalidates old evidence;
+    the configured selection and free-text status never supply serving identity.
+    """
+    observed = None
+    for event in events:
+        if not isinstance(event, dict) or event.get("session_id") != session_id or event.get("run_id") != stream_id:
+            continue
+        payload = event.get("payload")
+        if not isinstance(payload, dict):
+            if event.get("event") == "runtime_model":
+                observed = None
+            continue
+        if event.get("event") == "warning" and payload.get("type") == "fallback":
+            observed = None
+        elif event.get("event") == "runtime_model":
+            observed = None
+            if (payload.get("session_id") != session_id or payload.get("stream_id") != stream_id
+                or not isinstance(payload.get("model"), str) or not payload["model"].strip()
+                or not isinstance(payload.get("fallback_active"), bool)
+                or payload.get("phase") not in ("observed_output", "route_observed")):
+                continue
+            observed = {key: payload[key] for key in (
+                "session_id", "stream_id", "model", "fallback_active", "phase")}
+            if isinstance(payload.get("provider"), str) and payload["provider"].strip():
+                observed["provider"] = payload["provider"]
+    return observed
+
+
 def _summary_from_events(session_id: str, run_id: str, events: Iterable[dict]) -> dict:
     ordered = [event for event in events if isinstance(event, dict)]
     last = ordered[-1] if ordered else None
@@ -549,6 +581,7 @@ def _summary_from_events(session_id: str, run_id: str, events: Iterable[dict]) -
         "terminal": bool(terminal),
         "terminal_state": status,
         "last_event": (last or {}).get("event"),
+        "runtime_model": runtime_model_from_events(session_id, run_id, ordered),
     }
 
 

api/streaming.py

@@ -891,6 +891,9 @@ def _is_fallback_lifecycle_message(kind: str, message: str) -> bool:
             or 'falling back' in m
             or 'fallback activated' in m
             or 'trying fallback' in m
+            or 'model fallback:' in m
+            or 'switched to fallback' in m
+            or 'primary model restored:' in m
         )
     )
 
@@ -9734,6 +9737,37 @@ def _run_agent_streaming(
         except Exception:
             logger.debug("Failed to put event to queue")
 
+    _last_runtime_model_identity = None
+    _runtime_model_session_id = session_id
+
+    def _observe_runtime_model():
+        """Publish this turn's Agent identity at output, never at an attempted route."""
+        nonlocal _last_runtime_model_identity
+        raw_model = getattr(agent, 'model', None)
+        if not isinstance(raw_model, str) or not raw_model.strip():
+            return
+        raw_provider = getattr(agent, 'provider', None)
+        provider = str(raw_provider).strip().lstrip('@').lower() if isinstance(raw_provider, str) else ''
+        # The Agent may carry a provider-qualified routing hint in its model.
+        from api.config import _parse_provider_qualified_model_id
+        parsed = _parse_provider_qualified_model_id(raw_model.strip())
+        model_id = (parsed[0] if parsed else raw_model).strip()
+        if not model_id:
+            return
+        fallback_active = getattr(agent, '_provider_fallback_active', None) is True
+        identity = (provider, model_id.lower(), fallback_active)
+        if identity == _last_runtime_model_identity:
+            return
+        payload = {
+            'session_id': _runtime_model_session_id, 'stream_id': stream_id,
+            'model': model_id, 'fallback_active': fallback_active,
+            'phase': 'observed_output',
+        }
+        if provider:
+            payload['provider'] = provider
+        put('runtime_model', payload)
+        _last_runtime_model_identity = identity
+
     # #5940: capture a terminal (non-retryable) provider error the Agent emits via
     # its lifecycle status_callback. The Agent aborts a non-retryable API error
     # (e.g. HTTP 400 "invalid model / no credentials") with
@@ -9754,6 +9788,7 @@ def _run_agent_streaming(
         turn-completion classifier can report the real cause instead of the
         generic no_response fallback. All other lifecycle messages are dropped.
         """
+        nonlocal _last_runtime_model_identity
         _message = str(message or '').strip()
         _kind = str(kind or '').strip().lower()
         if not _message:
@@ -9778,6 +9813,7 @@ def _run_agent_streaming(
         # show them as warnings via the existing messages.js 'warning' listener.
         _is_fallback_notice = _is_fallback_lifecycle_message(_kind, _message)
         if _is_fallback_notice:
+            _last_runtime_model_identity = None
             put('warning', {'type': 'fallback', 'message': _message})
 
     # xsession wakeup misroute root fix (Option 1): pre-init so the outer
@@ -10325,6 +10361,8 @@ def _run_agent_streaming(
                 nonlocal _token_sent
                 if text is None:
                     return  # end-of-stream sentinel
+                if text:
+                    _observe_runtime_model()
                 # #4729: visible output is starting — flush any buffered reasoning tail
                 # first so the live Thinking stream is complete before/at the transition.
                 _flush_reasoning_buffer()
@@ -10368,6 +10406,8 @@ def _run_agent_streaming(
                     # partial window is not lost when the reasoning phase ends.
                     _flush_reasoning_buffer()
                     return
+                if text:
+                    _observe_runtime_model()
                 _tool_boundary_advanced = False
                 reasoning_delta = str(text)
                 # Some runtimes mirror user-visible progress text through the
@@ -11570,6 +11610,11 @@ def _run_agent_streaming(
                     if isinstance(_m, dict) and _m.get('role') == 'assistant':
                         _answer = str(_m.get('content', ''))
                         break
+                if (_answer.strip() and not result.get('error')
+                    and not _agent_result_terminal_failure(result)
+                    and not getattr(agent, '_last_error', None)
+                    and not _captured_terminal_error[0]):
+                    _observe_runtime_model()
                 # /btw is intentionally non-persistent, but its terminal SSE
                 # payload is still public output.  Project the ephemeral
                 # session before enqueueing it so raw Agent ``api_content`` or
@@ -12272,6 +12317,8 @@ def _run_agent_streaming(
                         # the catch-all label, hint, and provider details.
                         return  # apperror already closes the stream on the client side
 
+                _observe_runtime_model()
+
                 # ── Handle context compression side effects ──
                 # Also detect compression via the result dict or compressor state
                 if not _compressed:
@@ -12478,7 +12525,9 @@ def _run_agent_streaming(
                 # mutates agent.model when a fallback fires, so the pre-run
                 # resolved_model would mis-attribute exactly the turns where
                 # attribution matters most.
-                _used_model = getattr(agent, 'model', None) or resolved_model or model
+                # The configured selection is not proof that it served this turn.
+                _observed_model = getattr(agent, 'model', None)
+                _used_model = _observed_model.strip() if isinstance(_observed_model, str) else None
                 if _gateway_routing:
                     s.gateway_routing = _gateway_routing
                     _history = list(getattr(s, 'gateway_routing_history', None) or [])
@@ -13364,6 +13413,9 @@ def _run_agent_streaming(
                     if 'credential_pool' in _agent_params:
                         _heal_kwargs['credential_pool'] = _runtime_bundle['credential_pool']
                     _heal_agent = _AIAgent(**_heal_kwargs)
+                    # Delta callbacks close over `agent`; the replacement, not
+                    # the failed original, owns any successful retry output.
+                    agent = _heal_agent
                     _agent_sig = _compute_agent_cache_signature(
                         resolved_model,
                         resolved_api_key,
@@ -13491,6 +13543,7 @@ def _run_agent_streaming(
                                             s, tool_calls=s.tool_calls
                                         )
                                     )
+                            _observe_runtime_model()
                             if _done_session_payload is not None:
                                 put('done', {
                                     'session': _done_session_payload,

docs/rfcs/session-sse-contract-v1.md

@@ -162,7 +162,7 @@ gate rather than inventing values without source support.
 ## Authoritative emitted events (`/api/chat/stream`)
 
 These are the **real wire `event:` names** emitted by `api/streaming.py` today
-(23 names). Clients and docs must use this table — not the semantic draft above —
+(24 names). Clients and docs must use this table — not the semantic draft above —
 when integrating with the live chat SSE relay.
 
 | Wire name | Role |
@@ -179,6 +179,7 @@ when integrating with the live chat SSE relay.
 | `title` | Session title update (often after `done`) |
 | `title_status` | Title generation status / skip reason |
 | `warning` | Non-fatal provider/fallback warning |
+| `runtime_model` | Local Agent serving identity observed at output or successful completion |
 | `apperror` | Terminal application error (no trailing `stream_end`) |
 | `cancel` | Run cancelled |
 | `done` | Turn finalized (session payload); title/`stream_end` may follow |
@@ -199,6 +200,31 @@ The semantic taxonomy table remains a draft for the proposed per-session
 endpoint vocabulary and must be confirmed during maintainer review before that
 endpoint claims parity.
 
+### Observed local runtime model
+
+The local worker emits `runtime_model` with
+`{session_id, stream_id, model, provider?, fallback_active, phase}`. The IDs
+refer to the original run-journal owner even if compression rotates the Agent's
+session. `phase="observed_output"` means the Agent's own model was read at a
+nonempty token/reasoning callback or after successful completion (including a
+non-streaming reply or credential self-heal). It does not promise that a turn
+which emitted partial output will finish successfully. Missing Agent model
+sends no observation; the configured selection is not used as serving proof.
+`fallback_active` is true only when the Agent explicitly reports its fallback
+flag. Consecutive identical observations are deduplicated within the turn.
+
+Fallback lifecycle warnings invalidate older serving evidence and reset the
+deduplication; status text alone never establishes the replacement model. A
+fresh observation after the warning reestablishes it, including when a buffered
+success notice arrives after output. Durable replay retains the event IDs.
+The journal summary and HTTP `runtime_journal_snapshot.runtime_model` project
+the latest valid observation from the same session/stream event window, or null
+if a warning or malformed latest observation invalidated it. No previous run's
+footer or selected route fills an unknown current run. This is a local-worker
+producer; gateway attribution and frontend presentation are separate concerns
+(see #6272 and #7181). `route_observed` remains reserved for route-only data,
+not successful output.
+
 ## Cursor and resume semantics
 
 `Last-Event-ID` is the standard SSE reconnect header. Clients send the last

tests/test_issue5121_provider_auth_terminal_error.py

@@ -423,6 +423,72 @@ def test_auth_retry_success_does_not_append_error_turn(tmp_path, monkeypatch):
     assert not any(msg.get("_error") for msg in saved.messages)
 
 
+def test_auth_retry_success_observes_replacement_agent_runtime(tmp_path, monkeypatch):
+    from api import run_journal
+
+    session = _prepare_session("auth_runtime", "stream_auth_runtime", pending_user_message="Please retry")
+    Base = _build_auth_failure_agent(token_text="")
+
+    class ReplacementAgent(Base):
+        def __init__(self, **kwargs):
+            super().__init__(**kwargs)
+            self.model = "original" if self.__class__.runs == 0 else "healed-model"
+            self.provider = "old-provider" if self.__class__.runs == 0 else "healed-provider"
+
+    monkeypatch.setattr(run_journal, "_default_session_dir", lambda: tmp_path / "sessions")
+    q = queue.Queue()
+    streaming.STREAMS["stream_auth_runtime"] = q
+    config.STREAM_PARTIAL_TEXT["stream_auth_runtime"] = ""
+    heal_rt = {"provider": "test-provider", "api_key": "fresh-key", "base_url": None}
+    with mock.patch.object(streaming, "get_session", return_value=session), \
+         mock.patch.object(streaming, "_get_ai_agent", return_value=ReplacementAgent), \
+         mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \
+         mock.patch("api.config.get_config", return_value={}), \
+         mock.patch("api.config._resolve_cli_toolsets", return_value=[]), \
+         mock.patch.object(streaming, "_attempt_credential_self_heal", return_value=heal_rt):
+        streaming._run_agent_streaming(session.session_id, "Please retry", "test-model", str(tmp_path), "stream_auth_runtime")
+
+    events = _queue_events(q)
+    assert any(name == "done" for name, _ in events)
+    assert [(item["model"], item["provider"]) for name, item in events if name == "runtime_model"] == [
+        ("healed-model", "healed-provider")]
+    summary = run_journal.latest_run_summary("auth_runtime", "stream_auth_runtime", session_dir=tmp_path / "sessions")
+    assert summary["runtime_model"]["model"] == "healed-model"
+
+
+def test_auth_exception_retry_observes_replacement_agent(tmp_path, monkeypatch):
+    session = _prepare_session("auth_exception_runtime", "stream_auth_exception_runtime", pending_user_message="Retry")
+
+    class RetryAgent(MockAgent):
+        runs = 0
+        def __init__(self, **kwargs):
+            super().__init__(**kwargs)
+            self.model = "original" if type(self).runs == 0 else "retry-model"
+            self.provider = "original-provider" if type(self).runs == 0 else "retry-provider"
+        def run_conversation(self, **kwargs):
+            type(self).runs += 1
+            if type(self).runs == 1:
+                raise RuntimeError("401 unauthorized")
+            return {"messages": list(kwargs.get("conversation_history") or []) + [
+                {"role": "assistant", "content": "Recovered answer"}]}
+
+    q = queue.Queue()
+    streaming.STREAMS["stream_auth_exception_runtime"] = q
+    config.STREAM_PARTIAL_TEXT["stream_auth_exception_runtime"] = ""
+    heal_rt = {"provider": "test-provider", "api_key": "fresh-key", "base_url": None}
+    with mock.patch.object(streaming, "get_session", return_value=session), \
+         mock.patch.object(streaming, "_get_ai_agent", return_value=RetryAgent), \
+         mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \
+         mock.patch("api.config.get_config", return_value={}), \
+         mock.patch("api.config._resolve_cli_toolsets", return_value=[]), \
+         mock.patch.object(streaming, "_attempt_credential_self_heal", return_value=heal_rt):
+        streaming._run_agent_streaming(session.session_id, "Retry", "test-model", str(tmp_path), "stream_auth_exception_runtime")
+    events = _queue_events(q)
+    assert any(name == "done" for name, _ in events)
+    assert [(item["model"], item["provider"]) for name, item in events if name == "runtime_model"] == [
+        ("retry-model", "retry-provider")]
+
+
 def test_auth_exception_retry_structured_failure_still_emits_error(
     tmp_path,
     monkeypatch,

tests/test_issue6068_used_model_footer.py

@@ -140,10 +140,11 @@ console.log(JSON.stringify({{
 
 def test_streaming_stamps_used_model_on_assistant_message_and_usage_payload():
     assert "_dm['_usedModel'] = _used_model" in STREAMING_PY
-    # The served model must be read from the agent AFTER the run — the agent
-    # mutates agent.model when a fallback fires, so the pre-run resolved_model
-    # would mis-attribute fallback turns.
-    assert "_used_model = getattr(agent, 'model', None) or resolved_model or model" in STREAMING_PY
+    # Read the served model from the agent after the run (including fallback),
+    # without treating a requested/configured selection as proof of service.
+    assert "_observed_model = getattr(agent, 'model', None)" in STREAMING_PY
+    assert "_used_model = _observed_model.strip() if isinstance(_observed_model, str) else None" in STREAMING_PY
+    assert "_used_model = getattr(agent, 'model', None) or resolved_model or model" not in STREAMING_PY
     assert "usage['used_model'] = _used_model" in STREAMING_PY
 
 

tests/test_runtime_model_observation.py

@@ -0,0 +1,183 @@
+"""Observed Agent runtime, durable replay and HTTP reconnect projection."""
+import queue
+import sys
+import types
+from unittest import mock
+
+import pytest
+
+from api.run_journal import RunJournalWriter, latest_run_summary, read_run_events
+
+
+def observation(model="backup", *, session_id="session-a", stream_id="run-a"):
+    return {"session_id": session_id, "stream_id": stream_id, "model": model,
+            "provider": "custom", "fallback_active": True, "phase": "observed_output"}
+
+
+def test_journal_runtime_is_owned_and_invalidated(tmp_path):
+    writer = RunJournalWriter("session-a", "run-a", session_dir=tmp_path)
+    writer.append_sse_event("token", {"text": "answer"})
+    first = writer.append_sse_event("runtime_model", observation())
+    assert read_run_events("session-a", "run-a", session_dir=tmp_path)["events"][-1] == first
+    summary = latest_run_summary("session-a", "run-a", session_dir=tmp_path)
+    assert summary["runtime_model"] == observation()
+    summary["runtime_model"]["model"] = "corrupted"
+    assert latest_run_summary("session-a", "run-a", session_dir=tmp_path)["runtime_model"] == observation()
+    writer.append_sse_event("warning", {"type": "fallback", "message": "Trying another route"})
+    assert latest_run_summary("session-a", "run-a", session_dir=tmp_path)["runtime_model"] is None
+    writer.append_sse_event("runtime_model", {**observation(), "stream_id": "foreign"})
+    assert latest_run_summary("session-a", "run-a", session_dir=tmp_path)["runtime_model"] is None
+    writer.append_sse_event("runtime_model", observation("recovered"))
+    assert latest_run_summary("session-a", "run-a", session_dir=tmp_path)["runtime_model"]["model"] == "recovered"
+    writer.append_sse_event("runtime_model", "malformed observation")
+    assert latest_run_summary("session-a", "run-a", session_dir=tmp_path)["runtime_model"] is None
+    writer.append_sse_event("runtime_model", observation("recovered-again"))
+    assert latest_run_summary("session-a", "run-a", session_dir=tmp_path)["runtime_model"]["model"] == "recovered-again"
+    assert latest_run_summary("session-a", "another-run", session_dir=tmp_path)["runtime_model"] is None
+
+
+@pytest.mark.parametrize("suffix,expected", [("", "backup"), ("warning", None), ("foreign", None)])
+def test_http_snapshot_projects_exact_journal_window(tmp_path, monkeypatch, suffix, expected):
+    import api.routes as routes
+    writer = RunJournalWriter("session-a", "run-a", session_dir=tmp_path)
+    writer.append_sse_event("runtime_model", observation())
+    writer.append_sse_event("token", {"text": "answer"})
+    if suffix == "warning":
+        writer.append_sse_event("warning", {"type": "fallback"})
+    elif suffix == "foreign":
+        writer.append_sse_event("runtime_model", {**observation(), "session_id": "foreign"})
+    monkeypatch.setattr(routes, "find_run_summary", lambda rid: latest_run_summary("session-a", rid, session_dir=tmp_path))
+    monkeypatch.setattr(routes, "read_run_events", lambda sid, rid: read_run_events(sid, rid, session_dir=tmp_path))
+    snapshot = routes._run_journal_live_snapshot("run-a")
+    assert snapshot["last_assistant_text"] == "answer"
+    value = snapshot["runtime_model"]
+    assert (value["model"] if value else None) == expected
+    assert routes._runtime_journal_snapshot_for_session_payload(snapshot)["runtime_model"] == value
+
+
+def run_worker(steps, *, fail=False, ephemeral=False, tmp_path=None, monkeypatch=None):
+    """Drive the production local worker; fake only external Agent/provider calls."""
+    from api import streaming, run_journal
+    from api.config import SESSION_AGENT_CACHE
+    SESSION_AGENT_CACHE.clear()
+    monkeypatch.setattr(run_journal, "_default_session_dir", lambda: tmp_path)
+    class Session:
+        session_id = "session-a"
+        title = "Session"
+        workspace = "/tmp"
+        model = "primary"
+        model_provider = "anthropic"
+        profile = None
+        personality = None
+        messages = [{"role": "user", "content": "old"}, {"role": "assistant", "content": "old answer"}]
+        context_messages = messages
+        input_tokens = output_tokens = cache_read_tokens = cache_write_tokens = 0
+        estimated_cost = 0.0
+        tool_calls = []
+        gateway_routing = None
+        gateway_routing_history = []
+        active_stream_id = "run-a"
+        pending_user_message = None
+        pending_attachments = []
+        pending_started_at = None
+        context_length = threshold_tokens = last_prompt_tokens = 0
+        llm_title_generated = True
+        def save(self, *args, **kwargs): pass
+        def compact(self): return {"session_id": self.session_id}
+
+    class Agent:
+        def __init__(self, model=None, provider=None, session_id=None,
+                     stream_delta_callback=None, reasoning_callback=None,
+                     status_callback=None, **kwargs):
+            self.model, self.provider, self.session_id = model, provider, session_id
+            self.stream_delta_callback = stream_delta_callback
+            self.reasoning_callback = reasoning_callback
+            self.status_callback = status_callback
+            self._provider_fallback_active = False
+            self.context_compressor = None
+            self.session_prompt_tokens = self.session_completion_tokens = 0
+            self.session_estimated_cost_usd = None
+            self.session_cache_read_tokens = self.session_cache_write_tokens = 0
+            self.reasoning_config = self.ephemeral_system_prompt = self._last_error = None
+        def run_conversation(self, **kwargs):
+            for model, provider, fallback, channel, text in steps:
+                self.model, self.provider, self._provider_fallback_active = model, provider, fallback
+                if channel == "token": self.stream_delta_callback(text)
+                elif channel == "reasoning": self.reasoning_callback(text)
+                elif channel == "status": self.status_callback("lifecycle", text)
+            if fail: return {"error": "provider unavailable", "messages": []}
+            return {"messages": kwargs.get("conversation_history", []) + [
+                {"role": "user", "content": kwargs["persist_user_message"]},
+                {"role": "assistant", "content": "served answer"}]}
+        def interrupt(self, message): pass
+
+    runtime_module = types.ModuleType("hermes_cli.runtime_provider")
+    runtime_module.resolve_runtime_provider = mock.Mock(return_value={
+        "provider": "anthropic", "base_url": None, "api_key": "sk-test",
+        "api_mode": "chat_completions", "command": None, "args": [], "credential_pool": None})
+    cli_module = types.ModuleType("hermes_cli")
+    cli_module.runtime_provider = runtime_module
+    state_module = types.ModuleType("hermes_state")
+    state_module.SessionDB = mock.Mock(return_value=None)
+    injected = {"hermes_cli": cli_module, "hermes_cli.runtime_provider": runtime_module,
+                "hermes_state": state_module}
+    sentinel = object()
+    saved = {name: sys.modules.get(name, sentinel) for name in injected}
+    sys.modules.update(injected)
+    q = queue.Queue()
+    s = Session()
+    try:
+        with mock.patch.object(streaming, "get_session", return_value=s), \
+             mock.patch.object(streaming, "_get_ai_agent", return_value=Agent), \
+             mock.patch.object(streaming, "resolve_model_provider", return_value=("primary", "anthropic", None)), \
+             mock.patch("api.config.get_config", return_value={}), \
+             mock.patch("api.config._resolve_cli_toolsets", return_value=[]):
+            streaming.STREAMS["run-a"] = q
+            streaming._run_agent_streaming("session-a", "new turn", "primary", "/tmp", "run-a", ephemeral=ephemeral)
+    finally:
+        for name, original in saved.items():
+            if original is sentinel: sys.modules.pop(name, None)
+            else: sys.modules[name] = original
+        streaming.STREAMS.pop("run-a", None)
+        SESSION_AGENT_CACHE.clear()
+    return s, list(q.queue)
+
+
+def test_worker_observes_serving_runtime_not_requested(tmp_path, monkeypatch):
+    s, events = run_worker([
+        ("primary", "anthropic", False, "reasoning", "thinking"),
+        ("backup", "custom", True, "token", "answer"),
+        ("backup", "custom", True, "status", "Switched to fallback: backup (custom)"),
+    ], tmp_path=tmp_path, monkeypatch=monkeypatch)
+    names = [name for name, _ in events]
+    assert names.count("runtime_model") == 3  # primary, backup, post-warning correction
+    observed = [value for name, value in events if name == "runtime_model"]
+    assert [value["model"] for value in observed] == ["primary", "backup", "backup"]
+    assert observed[-1] == observation()
+    assert names.index("runtime_model") < names.index("reasoning")
+    assert names.index("warning") < len(names) - 1 - names[::-1].index("runtime_model")
+    assert latest_run_summary("session-a", "run-a", session_dir=tmp_path)["runtime_model"] == observation()
+    assert s.model == "primary"  # selection is not overwritten
+
+
+def test_worker_does_not_observe_attempt_or_missing_model(tmp_path, monkeypatch):
+    _, events = run_worker([
+        ("backup", "custom", True, "status", "Trying fallback backup"),
+        (None, "custom", True, "token", "partial"),
+    ], fail=True, tmp_path=tmp_path, monkeypatch=monkeypatch)
+    assert not any(name == "runtime_model" for name, _ in events)
+
+
+def test_missing_agent_model_does_not_stamp_requested_as_served(tmp_path, monkeypatch):
+    s, events = run_worker([(None, "custom", True, "token", "partial")],
+                           tmp_path=tmp_path, monkeypatch=monkeypatch)
+    assert not any(name == "runtime_model" for name, _ in events)
+    assert "_usedModel" not in s.messages[-1]
+    usage = next(value["usage"] for name, value in events if name == "done")
+    assert not usage.get("used_model")
+
+
+def test_worker_nonstreaming_success_observes_fallback(tmp_path, monkeypatch):
+    _, events = run_worker([("backup", "custom", True, "route", "")],
+                           tmp_path=tmp_path, monkeypatch=monkeypatch, ephemeral=True)
+    assert [value["model"] for name, value in events if name == "runtime_model"] == ["backup"]