From c45cab3779e822676c53cfab0d0d075234debaf2 Mon Sep 17 00:00:00 2001 From: Anthony Ruiz Date: Wed, 23 Sep 2026 11:01:53 +0000 Subject: [PATCH 1/2] feat(streaming): publish observed local runtime model to SSE and journal --- api/routes.py | 2 + api/run_journal.py | 37 +++- api/streaming.py | 55 +++++- docs/rfcs/session-sse-contract-v1.md | 28 ++- ..._issue5121_provider_auth_terminal_error.py | 66 +++++++ tests/test_issue6068_used_model_footer.py | 9 +- tests/test_runtime_model_observation.py | 183 ++++++++++++++++++ 7 files changed, 372 insertions(+), 8 deletions(-) create mode 100644 tests/test_runtime_model_observation.py diff --git a/api/routes.py b/api/routes.py index 62486aea..73116a2a 100644 --- a/api/routes.py +++ b/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, diff --git a/api/run_journal.py b/api/run_journal.py index 48cd8ef7..231d6699 100644 --- a/api/run_journal.py +++ b/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), } diff --git a/api/streaming.py b/api/streaming.py index 4dfa5b63..38363132 100644 --- a/api/streaming.py +++ b/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, diff --git a/docs/rfcs/session-sse-contract-v1.md b/docs/rfcs/session-sse-contract-v1.md index e6248e03..9abbc1a2 100644 --- a/docs/rfcs/session-sse-contract-v1.md +++ b/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 diff --git a/tests/test_issue5121_provider_auth_terminal_error.py b/tests/test_issue5121_provider_auth_terminal_error.py index 19c9c817..09ada5b9 100644 --- a/tests/test_issue5121_provider_auth_terminal_error.py +++ b/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, diff --git a/tests/test_issue6068_used_model_footer.py b/tests/test_issue6068_used_model_footer.py index 7ba04c53..ef0fd09b 100644 --- a/tests/test_issue6068_used_model_footer.py +++ b/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 diff --git a/tests/test_runtime_model_observation.py b/tests/test_runtime_model_observation.py new file mode 100644 index 00000000..9c76e45f --- /dev/null +++ b/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"] From 6d830802bb39e768481a389c83a3fc3f192f9c34 Mon Sep 17 00:00:00 2001 From: Anthony Ruiz Date: Wed, 23 Sep 2026 11:22:23 +0000 Subject: [PATCH 2/2] test(journal): cover malformed runtime model reconnect invalidation --- tests/test_runtime_model_observation.py | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/tests/test_runtime_model_observation.py b/tests/test_runtime_model_observation.py index 9c76e45f..7c1b87e3 100644 --- a/tests/test_runtime_model_observation.py +++ b/tests/test_runtime_model_observation.py @@ -1,4 +1,5 @@ """Observed Agent runtime, durable replay and HTTP reconnect projection.""" +import json import queue import sys import types @@ -36,6 +37,28 @@ def test_journal_runtime_is_owned_and_invalidated(tmp_path): assert latest_run_summary("session-a", "another-run", session_dir=tmp_path)["runtime_model"] is None +@pytest.mark.parametrize("malformed", ["bad observation", ["bad observation"], None]) +def test_non_object_observation_invalidates_summary_and_http_snapshot(tmp_path, monkeypatch, malformed): + from api import routes, run_journal + writer = RunJournalWriter("session-a", "run-a", session_dir=tmp_path) + writer.append_sse_event("runtime_model", observation("first")) + writer.append_sse_event("token", malformed) + assert latest_run_summary("session-a", "run-a", session_dir=tmp_path)["runtime_model"]["model"] == "first" + writer.append_sse_event("runtime_model", malformed if malformed is not None else {"placeholder": True}) + if malformed is None: + # The writer normalizes None to {}; exercise a legacy/raw null journal row. + path = run_journal._run_path("session-a", "run-a", session_dir=tmp_path) + rows = [json.loads(line) for line in path.read_text().splitlines()] + rows[-1]["payload"] = None + path.write_text("\n".join(json.dumps(row) for row in rows) + "\n") + assert latest_run_summary("session-a", "run-a", session_dir=tmp_path)["runtime_model"] is None + 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["runtime_model"] is None + assert routes._runtime_journal_snapshot_for_session_payload(snapshot)["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