From 015171eba3371ddda96464dab5cced0f0f007b70 Mon Sep 17 00:00:00 2001 From: Ruby Hartono <58564005+rh-id@users.noreply.github.com> Date: Thu, 16 Jul 2026 19:09:29 +0700 Subject: [PATCH 01/14] fix(models): cap _CLI_SESSIONS_CACHE with LRU eviction (drop-oldest) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Memory: the CLI/cron session projection cache was a plain dict with TTL storage but NO size cap and only lazy, same-key eviction. Each value is a copy.deepcopy() of the full CLI/cron session list (the expensive projection behind #4842/#4672). The cache key folds in a state.db content fingerprint that advances on every streamed message, so the next poll builds a NEW key; the previous key's heavy deepcopy was never looked up again, so its lazy-expiry path never ran. A long-lived process under churn could accumulate orphaned heavy deepcopies until the next structural clear_cli_sessions_cache(). Make it an OrderedDict capped at _CLI_SESSIONS_CACHE_MAX_ENTRIES (8) with drop-oldest on write (while len > MAX: popitem(last=False)) and move_to_end on fresh hits, mirroring the _CLAUDE_CODE_PARSE_CACHE / _SIDECAR_METADATA_CACHE LRU pattern in the same file. TTL remains the primary freshness control; the cap is the backstop the plain dict lacked. clear_cli_sessions_cache() keeps .clear(). Test: tests/test_cli_sessions_cache_cap.py — cache stays at len == MAX after MAX+5 distinct keys (oldest 5 evicted), a fresh hit marks the entry most-recently-used so it survives over colder entries, and the cache is an OrderedDict (the regression: it was an unbounded plain dict). Existing streaming-cache-TTL/cron-session tests unchanged. Contract Routing - Task type: memory-bound fix (unbounded projection cache). - Touched areas: CLI/cron sidebar projection cache (api/models.py _CLI_SESSIONS_CACHE). - Relevant public docs: AGENTS.md, CONTRIBUTING.md, docs/CONTRACTS.md, docs/rfcs/webui-run-state-consistency-contract.md (Sidebar/session metadata layer). - State layer mutated: Sidebar/session metadata layer — specifically the in-memory projection cache that backs the CLI/cron sidebar rows. The cache is a read-derivation of state.db content; it is NOT a source of truth. Dropped entries are rebuilt on the next poll (TTL/invalidation already drive rebuilds). - Invariant preserved — Invariant 4 (Maintenance is not activity): the cache is a derived projection; evicting an entry cannot move a session in the sidebar or change a row's content, because the next poll recomputes the projection from state.db. TTL remains the primary freshness control; the LRU cap only bounds how many STALE orphaned deepcopies linger between invalidations. Regression: test_cli_sessions_cache_cap.py plus the existing streaming-cache-TTL and cron-session tests (which exercise the real invalidate/rebuild path). - Scope boundaries: does not change sidebar ordering, unread state, updated_at, or the projection result. The cap is a memory backstop; correctness still flows from the existing TTL + clear-on-mutation invalidation. Release note: the CLI/cron sidebar projection cache is now LRU-bounded, capping worst-case memory for long-lived processes under heavy streaming churn. Model Used: builtin:zai-coding-plan/GLM-5.2 (ZCode agent). No AI-specific tool use mattered beyond the repo's standard read/edit/test workflow. --- api/models.py | 18 +++++- tests/test_cli_sessions_cache_cap.py | 89 ++++++++++++++++++++++++++++ 2 files changed, 106 insertions(+), 1 deletion(-) create mode 100644 tests/test_cli_sessions_cache_cap.py diff --git a/api/models.py b/api/models.py index 48a879a11..49888ad2f 100644 --- a/api/models.py +++ b/api/models.py @@ -63,7 +63,16 @@ _CLI_SESSIONS_CACHE_STREAMING_TTL_SECONDS = 45.0 _CLI_SESSIONS_CACHE_LOCK = threading.Lock() _CLI_SESSIONS_CACHE_INFLIGHT: "dict[tuple, threading.Event]" = {} _CLI_SESSIONS_CACHE_INVALIDATION_VERSION = 0 -_CLI_SESSIONS_CACHE = {} +# LRU-bounded (drop-oldest) so a long-lived process under churn — where the +# state.db fingerprint advances on every streamed message and the structural +# clear-on-mutation listener doesn't fire for every fingerprint advance — can't +# accumulate orphaned heavy deepcopies. Each value is a copy.deepcopy() of the +# full CLI/cron session list (the expensive projection behind #4842/#4672), so +# the cap is deliberately small. TTL is still the primary freshness control; +# the cap is the backstop that the plain dict previously lacked. Mirrors the +# _CLAUDE_CODE_PARSE_CACHE / _SIDECAR_METADATA_CACHE LRU pattern. +_CLI_SESSIONS_CACHE: "collections.OrderedDict[tuple, tuple]" = collections.OrderedDict() +_CLI_SESSIONS_CACHE_MAX_ENTRIES = 8 _CLI_SESSIONS_CACHE_WAIT_SECONDS = 0.25 # Event waits that keep stale rows visible while a rebuild is in flight. _CLI_SESSIONS_CACHE_STALE_WAIT_SECONDS = 0.10 @@ -6405,6 +6414,9 @@ def _cache_cli_sessions_if_current( invalidation_stamp, _copy_cli_sessions(sessions), ) + _CLI_SESSIONS_CACHE.move_to_end(cache_key) + while len(_CLI_SESSIONS_CACHE) > _CLI_SESSIONS_CACHE_MAX_ENTRIES: + _CLI_SESSIONS_CACHE.popitem(last=False) return True @@ -6423,6 +6435,8 @@ def _copy_fresh_cli_sessions_cache_entry(cache_key: tuple): return None if cached_expires_at <= time.monotonic(): return None + # LRU: a fresh hit is the most-recently-used entry. + _CLI_SESSIONS_CACHE.move_to_end(cache_key) return _copy_cli_sessions(cached_sessions) @@ -7272,6 +7286,8 @@ def get_cli_sessions( if cached_stamp != _CLI_SESSIONS_CACHE_INVALIDATION_VERSION: _CLI_SESSIONS_CACHE.pop(cache_key, None) elif cached_expires_at > now: + # LRU: a fresh hit is the most-recently-used entry. + _CLI_SESSIONS_CACHE.move_to_end(cache_key) return _copy_cli_sessions(cached_sessions) else: stale_sessions = _copy_cli_sessions(cached_sessions) diff --git a/tests/test_cli_sessions_cache_cap.py b/tests/test_cli_sessions_cache_cap.py new file mode 100644 index 000000000..7d2af868e --- /dev/null +++ b/tests/test_cli_sessions_cache_cap.py @@ -0,0 +1,89 @@ +"""Tests: ``_CLI_SESSIONS_CACHE`` is LRU-bounded (drop-oldest). + +Regression: the CLI/cron session projection cache was a plain ``dict`` with TTL +storage but NO size cap and only lazy, same-key eviction. Each value is a +``copy.deepcopy()`` of the full CLI/cron session list (the expensive projection +behind #4842/#4672). The cache key folds in a state.db content fingerprint that +advances on every streamed message, so the next poll builds a NEW key; the +previous key's heavy deepcopy was never looked up again, so its lazy-expiry path +never ran. A long-lived process under churn could accumulate orphaned heavy +deepcopies until the next structural ``clear_cli_sessions_cache()``. + +The cache is now an ``OrderedDict`` capped at ``_CLI_SESSIONS_CACHE_MAX_ENTRIES`` +with drop-oldest on write (mirroring ``_CLAUDE_CODE_PARSE_CACHE``) and +``move_to_end`` on fresh hits. TTL remains the primary freshness control; the +cap is the backstop the plain dict lacked. +""" +from __future__ import annotations + +import api.models as models +from api.models import ( + _CLI_SESSIONS_CACHE_MAX_ENTRIES, + _cache_cli_sessions_if_current, + _cli_sessions_cache_invalidation_stamp, + _copy_fresh_cli_sessions_cache_entry, +) + + +def _reset_cache(): + with models._CLI_SESSIONS_CACHE_LOCK: + models._CLI_SESSIONS_CACHE.clear() + + +def test_cache_is_bounded_drops_oldest_on_write(): + """Inserting more than MAX_ENTRIES distinct keys keeps len == MAX and evicts + the OLDEST (least-recently-used) entry.""" + _reset_cache() + stamp = _cli_sessions_cache_invalidation_stamp() + cap = _CLI_SESSIONS_CACHE_MAX_ENTRIES + assert cap >= 1 + + # Insert cap + 5 distinct keys, each a distinct payload so we can track them. + for i in range(cap + 5): + key = (f"profile-{i}",) + ok = _cache_cli_sessions_if_current( + key, ttl=60.0, invalidation_stamp=stamp, sessions=[{"id": i}] + ) + assert ok + + with models._CLI_SESSIONS_CACHE_LOCK: + assert len(models._CLI_SESSIONS_CACHE) == cap + keys = list(models._CLI_SESSIONS_CACHE.keys()) + # The oldest 5 (profile-0..4) were evicted; the newest `cap` remain. + assert keys[0] == ("profile-5",) + assert keys[-1] == (f"profile-{cap + 5 - 1}",) + assert ("profile-0",) not in keys + + +def test_fresh_hit_marks_entry_most_recently_used(): + """A fresh cache hit moves the entry to the end (most-recently-used), so a + frequently-read entry is NOT evicted over colder ones.""" + _reset_cache() + stamp = _cli_sessions_cache_invalidation_stamp() + cap = _CLI_SESSIONS_CACHE_MAX_ENTRIES + + # Fill the cache exactly. + for i in range(cap): + _cache_cli_sessions_if_current( + (f"k{i}",), ttl=60.0, invalidation_stamp=stamp, sessions=[{"id": i}] + ) + # Touch k0 (the oldest) so it becomes most-recently-used. + served = _copy_fresh_cli_sessions_cache_entry(("k0",)) + assert served == [{"id": 0}] + # Now insert one more; the eviction victim should be k1, NOT k0. + _cache_cli_sessions_if_current( + ("k_new",), ttl=60.0, invalidation_stamp=stamp, sessions=[{"id": 999}] + ) + with models._CLI_SESSIONS_CACHE_LOCK: + keys = list(models._CLI_SESSIONS_CACHE.keys()) + assert ("k0",) in keys # survived because it was just used + assert ("k1",) not in keys # evicted as the new oldest + assert ("k_new",) in keys + + +def test_cache_is_ordered_dict_not_plain_dict(): + """The regression: the cache used to be a plain dict with no ordering. It + must now be an OrderedDict to support move_to_end / popitem(last=False).""" + import collections + + assert isinstance(models._CLI_SESSIONS_CACHE, collections.OrderedDict) From 5df6b1fe8a103546e59d073c1bbecdc5d3ae9575 Mon Sep 17 00:00:00 2001 From: pxxD1998 <214340659+pxxD1998@users.noreply.github.com> Date: Wed, 15 Jul 2026 22:01:22 +0800 Subject: [PATCH 02/14] fix: stop stale agent runtime after source updates --- api/agent_runtime.py | 79 +++++++++++++++ api/routes.py | 19 ++-- api/streaming.py | 22 +++-- docs/troubleshooting.md | 22 +++++ tests/test_agent_runtime_revision_guard.py | 101 ++++++++++++++++++++ tests/test_agent_source_dependency_audit.py | 2 +- tests/test_regressions.py | 8 +- tests/test_update_banner_fixes.py | 2 +- 8 files changed, 232 insertions(+), 23 deletions(-) create mode 100644 api/agent_runtime.py create mode 100644 tests/test_agent_runtime_revision_guard.py diff --git a/api/agent_runtime.py b/api/agent_runtime.py new file mode 100644 index 000000000..be3139f6f --- /dev/null +++ b/api/agent_runtime.py @@ -0,0 +1,79 @@ +"""Fail-closed guard for in-process Hermes Agent source revisions. + +Hermes WebUI currently imports ``run_agent.AIAgent`` into its long-lived server +process. If the Agent checkout changes while that process is alive, Python may +combine already-cached modules with newly-read source. Refuse to reuse that +mixed runtime and require a clean WebUI restart instead. +""" + +from __future__ import annotations + +from pathlib import Path +import subprocess +import threading + +from api.config import _AGENT_DIR + + +_RESTART_MESSAGE = ( + "Hermes Agent was updated while Hermes WebUI was running. " + "Restart Hermes WebUI before retrying this action." +) + + +def _read_agent_revision(agent_dir: Path) -> str | None: + """Return the checkout HEAD, or ``None`` for a non-Git/unavailable source.""" + try: + result = subprocess.run( + ["git", "-C", str(agent_dir), "rev-parse", "--verify", "HEAD"], + check=False, + capture_output=True, + text=True, + timeout=2, + ) + except (OSError, subprocess.TimeoutExpired): + return None + revision = result.stdout.strip() + return revision if result.returncode == 0 and revision else None + + +_AGENT_REVISION = _read_agent_revision(_AGENT_DIR) +_AIAgent = None +_RUNTIME_LOCK = threading.Lock() + + +class AgentRuntimeChangedError(RuntimeError): + """Raised when the loaded Agent runtime no longer matches its source tree.""" + + +def ensure_agent_runtime_current() -> None: + """Reject a known Git checkout change instead of mixing Python modules.""" + if _AGENT_REVISION is None: + return + if _read_agent_revision(_AGENT_DIR) != _AGENT_REVISION: + raise AgentRuntimeChangedError(_RESTART_MESSAGE) + + +def require_ai_agent_class(): + """Import ``AIAgent`` after proving the loaded source revision is current.""" + ensure_agent_runtime_current() + from run_agent import AIAgent # noqa: PLC0415 + + return AIAgent + + +def get_ai_agent_class(): + """Return ``AIAgent`` while preserving the existing lazy-import retry.""" + global _AIAgent, _AGENT_REVISION + + with _RUNTIME_LOCK: + ensure_agent_runtime_current() + if _AIAgent is None: + try: + agent_class = require_ai_agent_class() + except ImportError: + return None + _AIAgent = agent_class + if _AGENT_REVISION is None: + _AGENT_REVISION = _read_agent_revision(_AGENT_DIR) + return _AIAgent diff --git a/api/routes.py b/api/routes.py index 2a654ab5a..93b63aa77 100644 --- a/api/routes.py +++ b/api/routes.py @@ -34,6 +34,7 @@ from contextlib import closing from urllib.parse import parse_qs, quote, unquote, urljoin, urlsplit from urllib.error import HTTPError, URLError from urllib.request import HTTPRedirectHandler, HTTPSHandler, ProxyHandler, Request, build_opener +from api.agent_runtime import ensure_agent_runtime_current, require_ai_agent_class from api.agent_sessions import ( MESSAGING_SOURCES, _looks_like_default_cli_title, @@ -16052,6 +16053,7 @@ def handle_post(handler, parsed) -> bool: "api_key": _main_api_key, } + ensure_agent_runtime_current() try: from agent.auxiliary_client import get_text_auxiliary_client @@ -16072,7 +16074,7 @@ def handle_post(handler, parsed) -> bool: except Exception as _e: logger.debug("update summary auxiliary model failed; falling back to main model: %s", _e) - from run_agent import AIAgent + AIAgent = require_ai_agent_class() agent = AIAgent( model=_main_model, @@ -21991,7 +21993,7 @@ def _handle_chat_sync(handler, body): os.environ["HERMES_EXEC_ASK"] = "1" os.environ["HERMES_SESSION_KEY"] = s.session_id try: - from run_agent import AIAgent + AIAgent = require_ai_agent_class() with CHAT_LOCK: from api.config import ( @@ -22622,6 +22624,7 @@ def _llm_git_commit_message(system_prompt: str, user_prompt: str, session=None) "base_url": _main_base_url, "api_key": _main_api_key, } + ensure_agent_runtime_current() try: from agent.auxiliary_client import get_text_auxiliary_client @@ -22638,7 +22641,7 @@ def _llm_git_commit_message(system_prompt: str, user_prompt: str, session=None) except Exception as _e: logger.debug("git commit message auxiliary model failed; falling back to main model: %s", _e) - from run_agent import AIAgent + AIAgent = require_ai_agent_class() agent = AIAgent( model=_main_model, @@ -24035,10 +24038,11 @@ def _handle_session_compress(handler, body): focus_topic, ) + ensure_agent_runtime_current() import api.config as _cfg from api.oauth import resolve_runtime_provider_with_anthropic_env_lock import hermes_cli.runtime_provider as _runtime_provider - import run_agent as _run_agent + AIAgent = require_ai_agent_class() resolved_model, resolved_provider, resolved_base_url = _cfg.resolve_model_provider( _cfg.model_with_provider_context(s.model, getattr(s, "model_provider", None)) @@ -24081,7 +24085,7 @@ def _handle_session_compress(handler, body): ) approx_tokens = _estimate_messages_tokens_rough(original_messages) - agent = _run_agent.AIAgent( + agent = AIAgent( model=resolved_model, provider=resolved_provider, base_url=resolved_base_url, @@ -24685,10 +24689,11 @@ def _handle_handoff_summary(handler, body): # Call LLM for summary. try: + ensure_agent_runtime_current() import api.config as _cfg from api.oauth import resolve_runtime_provider_with_anthropic_env_lock import hermes_cli.runtime_provider as _runtime_provider - import run_agent as _run_agent + AIAgent = require_ai_agent_class() # Try to resolve model from an existing session, fall back to default. resolved_model = None @@ -24744,7 +24749,7 @@ def _handle_handoff_summary(handler, body): "fallback": True, }) - agent = _run_agent.AIAgent( + agent = AIAgent( model=resolved_model, provider=resolved_provider, base_url=resolved_base_url, diff --git a/api/streaming.py b/api/streaming.py index c733a8f9a..1b8fc70bf 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -564,10 +564,13 @@ def _prewarm_skill_tool_modules(): # Lazy import to avoid circular deps -- hermes-agent is on sys.path via api/config.py -try: - from run_agent import AIAgent -except ImportError: - AIAgent = None +from api.agent_runtime import ensure_agent_runtime_current, get_ai_agent_class + + +# Eagerly attempt the import at startup, matching the pre-guard behavior. If +# dependencies are not ready yet, _get_ai_agent() retries when a chat starts. +AIAgent = get_ai_agent_class() + def _get_ai_agent(): """Return AIAgent class, retrying the import if the initial attempt failed. @@ -575,15 +578,13 @@ def _get_ai_agent(): auto_install_agent_deps() in server.py may install missing packages after this module is first imported (common in Docker with a volume-mounted agent). Re-attempting the import here picks up the newly installed packages without - requiring a server restart. + requiring a server restart. The shared runtime guard also refuses to reuse + cached Agent modules after the source checkout changes. """ global AIAgent + ensure_agent_runtime_current() if AIAgent is None: - try: - from run_agent import AIAgent as _cls # noqa: PLC0415 - AIAgent = _cls - except ImportError: - pass + AIAgent = get_ai_agent_class() return AIAgent @@ -7213,6 +7214,7 @@ def _run_agent_streaming( logger.debug("register_process_session failed", exc_info=True) # first-time module initialisation (which can be slow) does not # block other concurrent sessions waiting on _ENV_LOCK (#2024). + ensure_agent_runtime_current() _prewarm_skill_tool_modules() _install_streaming_cronjob_profile_wrapper() # Still set process-level env as fallback for tools that bypass thread-local diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 71d63d73a..da8a5dcfe 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -177,6 +177,28 @@ turn in the exhausted session instead of being blocked with recovery guidance. --- +## "Hermes Agent was updated while Hermes WebUI was running" + +**Symptom.** An action that uses the in-process Agent runtime stops with a message telling you to restart Hermes WebUI. This can happen after `hermes update`, a Git checkout/pull in the Agent source tree, or another tool updates Hermes Agent without restarting the already-running WebUI backend. + +**Why.** WebUI currently imports `run_agent.AIAgent` into its long-lived Python process. Python keeps imported modules in memory. Continuing after the Agent Git revision changes could combine cached modules from the old revision with source read from the new revision, producing misleading `ImportError`s or inconsistent runtime state. WebUI therefore fails closed instead of attempting a partial in-process reload. + +**Diagnostic.** Compare the running WebUI process start time with the Agent checkout revision and recent update history. If the Agent was updated after WebUI started, restart WebUI before investigating individual missing-symbol errors. + +**Fix.** Restart using the same launch method that started WebUI: + +```bash +./ctl.sh restart +# Or, for a user systemd service: +systemctl --user restart hermes-webui.service +``` + +If you launched `python3 bootstrap.py` in the foreground, stop it with Ctrl-C and start it again. Restarting the whole computer or WSL is not required when restarting the WebUI backend succeeds. + +**When to file a bug.** File a WebUI bug if the restart-required message appears even though the Agent revision did not change, or if a clean WebUI restart still produces the same import error. Include the WebUI launch method, WebUI revision, Agent revision, and the sanitized error text. + +--- + ## Other troubleshooting This document grows over time. If a recurring failure mode isn't covered here yet, add it via PR. The format for each entry: **Symptom → Why → Diagnostic commands → Fix → When to file a bug**. diff --git a/tests/test_agent_runtime_revision_guard.py b/tests/test_agent_runtime_revision_guard.py new file mode 100644 index 000000000..f9cc1ae6f --- /dev/null +++ b/tests/test_agent_runtime_revision_guard.py @@ -0,0 +1,101 @@ +"""Regression coverage for Hermes Agent source changes during a WebUI process lifetime.""" + +from __future__ import annotations + +import os +from pathlib import Path +import subprocess +import sys + + +REPO = Path(__file__).resolve().parents[1] + + +def _git(repo: Path, *args: str) -> str: + result = subprocess.run( + ["git", "-c", "user.name=Test", "-c", "user.email=test@example.invalid", *args], + cwd=repo, + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +def test_loaded_agent_runtime_fails_closed_after_source_revision_changes(tmp_path: Path): + agent_dir = tmp_path / "hermes-agent" + agent_dir.mkdir() + (agent_dir / "run_agent.py").write_text( + "class AIAgent:\n revision = 'before'\n", + encoding="utf-8", + ) + _git(agent_dir, "init", "-q") + _git(agent_dir, "add", "run_agent.py") + _git(agent_dir, "commit", "-qm", "before") + + probe = tmp_path / "probe.py" + probe.write_text( + """ +from pathlib import Path +import subprocess + +import api.streaming as streaming +from api import agent_runtime + +agent_dir = Path(__file__).parent / "hermes-agent" +assert agent_runtime._AGENT_DIR == agent_dir.resolve() +assert streaming._get_ai_agent().revision == "before" + +(agent_dir / "run_agent.py").write_text( + "class AIAgent:\\n revision = 'after'\\n", + encoding="utf-8", +) +subprocess.run(["git", "add", "run_agent.py"], cwd=agent_dir, check=True) +subprocess.run( + [ + "git", "-c", "user.name=Test", "-c", "user.email=test@example.invalid", + "commit", "-qm", "after", + ], + cwd=agent_dir, + check=True, +) + +try: + streaming._get_ai_agent() +except RuntimeError as exc: + message = str(exc) + assert "Hermes Agent was updated" in message + assert "Restart Hermes WebUI" in message +else: + raise AssertionError("stale in-process AIAgent was reused after its source revision changed") + +try: + agent_runtime.require_ai_agent_class() +except agent_runtime.AgentRuntimeChangedError as exc: + assert "Restart Hermes WebUI" in str(exc) +else: + raise AssertionError("unguarded AIAgent import was allowed after its source revision changed") +""".strip() + + "\n", + encoding="utf-8", + ) + + env = os.environ.copy() + env.update( + { + "HERMES_WEBUI_AGENT_DIR": str(agent_dir), + "HERMES_HOME": str(tmp_path / "hermes-home"), + "HERMES_WEBUI_STATE_DIR": str(tmp_path / "webui-state"), + "PYTHONPATH": os.pathsep.join((str(REPO), str(agent_dir))), + } + ) + result = subprocess.run( + [sys.executable, str(probe)], + cwd=REPO, + env=env, + capture_output=True, + text=True, + timeout=30, + ) + + assert result.returncode == 0, result.stdout + result.stderr diff --git a/tests/test_agent_source_dependency_audit.py b/tests/test_agent_source_dependency_audit.py index 9f56f5fb4..052673068 100644 --- a/tests/test_agent_source_dependency_audit.py +++ b/tests/test_agent_source_dependency_audit.py @@ -120,7 +120,7 @@ def test_audit_reports_runtime_agent_execution_imports(): classes = _class_by_id(_run_audit()) anchors = _anchors(classes["runtime_agent_execution"]) - assert ("api/streaming.py", "run_agent") in anchors + assert ("api/agent_runtime.py", "run_agent") in anchors assert ("api/routes.py", "tools.skills_tool") in anchors assert ("api/streaming.py", "tools.approval") in anchors assert ("api/routes.py", "cron.jobs") in anchors diff --git a/tests/test_regressions.py b/tests/test_regressions.py index 827254297..8d38e4499 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -166,13 +166,13 @@ def test_streaming_py_imports_has_pending(cleanup_test_sessions): def test_aiagent_imported_in_streaming(cleanup_test_sessions): - """R2b: api/streaming.py must import AIAgent. + """R2b: api/streaming.py must resolve AIAgent through the runtime guard. When missing, the streaming thread crashed immediately after being spawned. """ src = (REPO_ROOT / "api/streaming.py").read_text() - assert "AIAgent" in src, "AIAgent not referenced in api/streaming.py" - assert "from run_agent import AIAgent" in src or "import AIAgent" in src, \ - "AIAgent must be imported in api/streaming.py" + assert "get_ai_agent_class" in src, "guarded AIAgent resolver not referenced in api/streaming.py" + assert "from api.agent_runtime import" in src and "get_ai_agent_class" in src, \ + "AIAgent must be resolved through api.agent_runtime in api/streaming.py" # ── R5: SSE loop did not break on cancel event (Sprint 10 bug) ─────────────── diff --git a/tests/test_update_banner_fixes.py b/tests/test_update_banner_fixes.py index 0ab7a0518..0da5ad417 100644 --- a/tests/test_update_banner_fixes.py +++ b/tests/test_update_banner_fixes.py @@ -1635,7 +1635,7 @@ class TestUpdateSummaryRouteModelSelection: assert '"update_summary"' not in src assert 'main_runtime=main_runtime' in src assert 'update summary auxiliary model failed; falling back to main model' in src - assert 'from run_agent import AIAgent' in src + assert 'require_ai_agent_class()' in src def test_summary_route_auxiliary_model_uses_active_profile_env(self, monkeypatch, tmp_path): import api.config as cfg From 5f1791e08d5cf234a573b797f885805a89714791 Mon Sep 17 00:00:00 2001 From: pxxD1998 <214340659+pxxD1998@users.noreply.github.com> Date: Wed, 15 Jul 2026 22:20:44 +0800 Subject: [PATCH 03/14] fix: reject stale runtime before chat mutation --- api/routes.py | 22 ++++++++- docs/troubleshooting.md | 2 +- tests/test_agent_runtime_revision_guard.py | 57 ++++++++++++++++++++++ 3 files changed, 79 insertions(+), 2 deletions(-) diff --git a/api/routes.py b/api/routes.py index 93b63aa77..e0fdeafe0 100644 --- a/api/routes.py +++ b/api/routes.py @@ -34,7 +34,11 @@ from contextlib import closing from urllib.parse import parse_qs, quote, unquote, urljoin, urlsplit from urllib.error import HTTPError, URLError from urllib.request import HTTPRedirectHandler, HTTPSHandler, ProxyHandler, Request, build_opener -from api.agent_runtime import ensure_agent_runtime_current, require_ai_agent_class +from api.agent_runtime import ( + AgentRuntimeChangedError, + ensure_agent_runtime_current, + require_ai_agent_class, +) from api.agent_sessions import ( MESSAGING_SOURCES, _looks_like_default_cli_title, @@ -21663,6 +21667,22 @@ def _handle_chat_start(handler, body, diag=None): require(body, "session_id") except ValueError as e: return bad(handler, str(e)) + # Reject a stale local Agent runtime before materialising, claiming, or + # mutating any session state. Gateway-backed turns run in the gateway's + # process and do not depend on this WebUI process's imported checkout. + if not webui_gateway_chat_enabled(get_config()): + try: + ensure_agent_runtime_current() + except AgentRuntimeChangedError as exc: + return j( + handler, + { + "error": str(exc), + "type": "agent_runtime_stale", + "retryable": True, + }, + status=409, + ) diag.stage("get_session") if diag else None try: s = _get_or_materialize_session(body["session_id"], refresh_cli_messages=True) diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index da8a5dcfe..5d0e7ea6a 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -181,7 +181,7 @@ turn in the exhausted session instead of being blocked with recovery guidance. **Symptom.** An action that uses the in-process Agent runtime stops with a message telling you to restart Hermes WebUI. This can happen after `hermes update`, a Git checkout/pull in the Agent source tree, or another tool updates Hermes Agent without restarting the already-running WebUI backend. -**Why.** WebUI currently imports `run_agent.AIAgent` into its long-lived Python process. Python keeps imported modules in memory. Continuing after the Agent Git revision changes could combine cached modules from the old revision with source read from the new revision, producing misleading `ImportError`s or inconsistent runtime state. WebUI therefore fails closed instead of attempting a partial in-process reload. +**Why.** WebUI currently imports `run_agent.AIAgent` into its long-lived Python process. Python keeps imported modules in memory. Continuing after a known Agent Git revision changes could combine cached modules from the old revision with source read from the new revision, producing misleading `ImportError`s or inconsistent runtime state. For local Agent-backed chat, WebUI therefore returns a retryable `409 agent_runtime_stale` before claiming or mutating session state instead of attempting a partial in-process reload. Gateway-backed chat runs in the gateway process and is not blocked by this WebUI-local check. Non-Git Agent installs preserve their existing behavior because there is no revision identity to compare. **Diagnostic.** Compare the running WebUI process start time with the Agent checkout revision and recent update history. If the Agent was updated after WebUI started, restart WebUI before investigating individual missing-symbol errors. diff --git a/tests/test_agent_runtime_revision_guard.py b/tests/test_agent_runtime_revision_guard.py index f9cc1ae6f..e9de2bbcb 100644 --- a/tests/test_agent_runtime_revision_guard.py +++ b/tests/test_agent_runtime_revision_guard.py @@ -7,6 +7,8 @@ from pathlib import Path import subprocess import sys +import pytest + REPO = Path(__file__).resolve().parents[1] @@ -99,3 +101,58 @@ else: ) assert result.returncode == 0, result.stdout + result.stderr + + +def test_initial_non_git_source_preserves_supported_runtime(monkeypatch): + """Non-Git installs cannot be compared, so they preserve existing behavior.""" + from api import agent_runtime + + monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", None) + monkeypatch.setattr(agent_runtime, "_read_agent_revision", lambda _path: None) + + agent_runtime.ensure_agent_runtime_current() + + +def test_known_revision_becoming_unreadable_fails_closed(monkeypatch): + """Losing a previously-known revision is indistinguishable from source drift.""" + from api import agent_runtime + + monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", "known-revision") + monkeypatch.setattr(agent_runtime, "_read_agent_revision", lambda _path: None) + + with pytest.raises(agent_runtime.AgentRuntimeChangedError): + agent_runtime.ensure_agent_runtime_current() + + +def test_chat_start_rejects_stale_runtime_before_session_materialization(monkeypatch): + """A stale local runtime must not claim, create, or mutate session state.""" + from api import agent_runtime, routes + + monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: False) + monkeypatch.setattr(routes, "get_config", lambda: {}) + + def stale(): + raise agent_runtime.AgentRuntimeChangedError("restart required") + + monkeypatch.setattr(routes, "ensure_agent_runtime_current", stale) + + def must_not_materialize(*_args, **_kwargs): + raise AssertionError("session materialized before stale-runtime barrier") + + monkeypatch.setattr(routes, "_get_or_materialize_session", must_not_materialize) + monkeypatch.setattr( + routes, + "j", + lambda _handler, payload, status=200: {"status": status, "payload": payload}, + ) + + response = routes._handle_chat_start(object(), {"session_id": "session-1"}) + + assert response == { + "status": 409, + "payload": { + "error": "restart required", + "type": "agent_runtime_stale", + "retryable": True, + }, + } From f722f68738dcd3f6e5d9c6b397667c2103bbaf81 Mon Sep 17 00:00:00 2001 From: pxxD1998 <214340659+pxxD1998@users.noreply.github.com> Date: Thu, 16 Jul 2026 14:30:46 +0800 Subject: [PATCH 04/14] fix: scope stale runtime guard by execution owner --- api/agent_runtime.py | 48 +++- api/routes.py | 56 ++++- tests/test_agent_runtime_revision_guard.py | 268 +++++++++++++++++++++ 3 files changed, 352 insertions(+), 20 deletions(-) diff --git a/api/agent_runtime.py b/api/agent_runtime.py index be3139f6f..84514e0d8 100644 --- a/api/agent_runtime.py +++ b/api/agent_runtime.py @@ -9,11 +9,13 @@ mixed runtime and require a clean WebUI restart instead. from __future__ import annotations from pathlib import Path +import sys import subprocess import threading -from api.config import _AGENT_DIR - +# Retain the discovered path as a diagnostic/test-visible compatibility value; +# runtime identity is deliberately captured from the loaded module below. +from api.config import _AGENT_DIR # noqa: F401 _RESTART_MESSAGE = ( "Hermes Agent was updated while Hermes WebUI was running. " @@ -21,8 +23,10 @@ _RESTART_MESSAGE = ( ) -def _read_agent_revision(agent_dir: Path) -> str | None: +def _read_agent_revision(agent_dir: Path | None) -> str | None: """Return the checkout HEAD, or ``None`` for a non-Git/unavailable source.""" + if agent_dir is None: + return None try: result = subprocess.run( ["git", "-C", str(agent_dir), "rev-parse", "--verify", "HEAD"], @@ -37,7 +41,8 @@ def _read_agent_revision(agent_dir: Path) -> str | None: return revision if result.returncode == 0 and revision else None -_AGENT_REVISION = _read_agent_revision(_AGENT_DIR) +_AGENT_SOURCE_DIR: Path | None = None +_AGENT_REVISION: str | None = None _AIAgent = None _RUNTIME_LOCK = threading.Lock() @@ -46,11 +51,41 @@ class AgentRuntimeChangedError(RuntimeError): """Raised when the loaded Agent runtime no longer matches its source tree.""" +def _loaded_agent_source_dir() -> Path | None: + """Return the source directory that actually supplied ``run_agent``.""" + module = sys.modules.get("run_agent") + module_file = getattr(module, "__file__", None) + if not module_file: + return None + try: + return Path(module_file).resolve().parent + except (OSError, RuntimeError, TypeError): + return None + + +def _capture_loaded_agent_revision() -> None: + """Bind the guard to the checkout that supplied the loaded Agent module.""" + global _AGENT_SOURCE_DIR, _AGENT_REVISION + + source_dir = _loaded_agent_source_dir() + if source_dir is None: + return + current_revision = _read_agent_revision(source_dir) + if _AGENT_SOURCE_DIR is not None and source_dir != _AGENT_SOURCE_DIR: + raise AgentRuntimeChangedError(_RESTART_MESSAGE) + if _AGENT_REVISION is not None: + if current_revision != _AGENT_REVISION: + raise AgentRuntimeChangedError(_RESTART_MESSAGE) + return + _AGENT_SOURCE_DIR = source_dir + _AGENT_REVISION = current_revision + + def ensure_agent_runtime_current() -> None: """Reject a known Git checkout change instead of mixing Python modules.""" if _AGENT_REVISION is None: return - if _read_agent_revision(_AGENT_DIR) != _AGENT_REVISION: + if _read_agent_revision(_AGENT_SOURCE_DIR) != _AGENT_REVISION: raise AgentRuntimeChangedError(_RESTART_MESSAGE) @@ -59,6 +94,7 @@ def require_ai_agent_class(): ensure_agent_runtime_current() from run_agent import AIAgent # noqa: PLC0415 + _capture_loaded_agent_revision() return AIAgent @@ -74,6 +110,4 @@ def get_ai_agent_class(): except ImportError: return None _AIAgent = agent_class - if _AGENT_REVISION is None: - _AGENT_REVISION = _read_agent_revision(_AGENT_DIR) return _AIAgent diff --git a/api/routes.py b/api/routes.py index e0fdeafe0..698325b84 100644 --- a/api/routes.py +++ b/api/routes.py @@ -20431,6 +20431,9 @@ def _handle_btw(handler, body): require(body, "question") except ValueError as e: return bad(handler, str(e)) + stale_response = _agent_runtime_barrier_response(runner_local_owned=False) + if stale_response is not None: + return j(handler, stale_response, status=409) if _session_is_subagent_view_only(str(body.get("session_id") or "")): return bad(handler, "Subagent sessions are view-only and cannot be used for /btw from WebUI", 400) try: @@ -20490,6 +20493,9 @@ def _handle_background(handler, body): require(body, "prompt") except ValueError as e: return bad(handler, str(e)) + stale_response = _agent_runtime_barrier_response(runner_local_owned=False) + if stale_response is not None: + return j(handler, stale_response, status=409) try: s = get_session(body["session_id"]) except KeyError: @@ -20776,6 +20782,25 @@ def _active_run_stream_for_session(session_id: str | None) -> str | None: return None +def _agent_runtime_barrier_response(*, runner_local_owned: bool = False) -> dict | None: + """Return the typed stale-runtime response for local in-process turns.""" + if runner_local_owned and webui_gateway_chat_enabled(get_config()): + return None + from api.runtime_adapter import runtime_adapter_runner_enabled + + if runner_local_owned and runtime_adapter_runner_enabled(): + return None + try: + ensure_agent_runtime_current() + except AgentRuntimeChangedError as exc: + return { + "error": str(exc), + "type": "agent_runtime_stale", + "retryable": True, + } + return None + + def _start_chat_stream_for_session( s, *, @@ -20789,8 +20814,15 @@ def _start_chat_stream_for_session( goal_related: bool = False, source: str = "webui", moa_config=None, + external_runtime_owned: bool = False, ): """Persist pending state, register an SSE channel, and start an agent turn.""" + stale_response = _agent_runtime_barrier_response( + runner_local_owned=external_runtime_owned, + ) + if stale_response is not None: + stale_response["_status"] = 409 + return stale_response attachments = attachments or [] # Prevent duplicate runs in the same session while a stream is still active. # This commonly happens after page refresh/reconnect races and can produce @@ -21080,6 +21112,7 @@ def _start_run( diag=diag, source=source, moa_config=moa_config, + external_runtime_owned=webui_gateway_chat_enabled(get_config()), ) @@ -21173,6 +21206,10 @@ def start_session_turn( msg = str(message or "").strip() if not msg: return {"error": "message is required", "_status": 400} + stale_response = _agent_runtime_barrier_response(runner_local_owned=True) + if stale_response is not None: + stale_response["_status"] = 409 + return stale_response turn_source = str(source or "process_wakeup").strip() or "process_wakeup" try: s = get_session(session_id) @@ -21670,19 +21707,9 @@ def _handle_chat_start(handler, body, diag=None): # Reject a stale local Agent runtime before materialising, claiming, or # mutating any session state. Gateway-backed turns run in the gateway's # process and do not depend on this WebUI process's imported checkout. - if not webui_gateway_chat_enabled(get_config()): - try: - ensure_agent_runtime_current() - except AgentRuntimeChangedError as exc: - return j( - handler, - { - "error": str(exc), - "type": "agent_runtime_stale", - "retryable": True, - }, - status=409, - ) + stale_response = _agent_runtime_barrier_response(runner_local_owned=True) + if stale_response is not None: + return j(handler, stale_response, status=409) diag.stage("get_session") if diag else None try: s = _get_or_materialize_session(body["session_id"], refresh_cli_messages=True) @@ -21978,6 +22005,9 @@ def _normalize_chat_attachments(raw_attachments): def _handle_chat_sync(handler, body): """Fallback synchronous chat endpoint (POST /api/chat). Not used by frontend.""" + stale_response = _agent_runtime_barrier_response(runner_local_owned=False) + if stale_response is not None: + return j(handler, stale_response, status=409) if _session_is_subagent_view_only(str(body.get("session_id") or "")): return bad(handler, "Subagent sessions are view-only and cannot be written from WebUI", 400) s = get_session(body["session_id"]) diff --git a/tests/test_agent_runtime_revision_guard.py b/tests/test_agent_runtime_revision_guard.py index e9de2bbcb..066c375d3 100644 --- a/tests/test_agent_runtime_revision_guard.py +++ b/tests/test_agent_runtime_revision_guard.py @@ -6,6 +6,7 @@ import os from pathlib import Path import subprocess import sys +import types import pytest @@ -113,6 +114,35 @@ def test_initial_non_git_source_preserves_supported_runtime(monkeypatch): agent_runtime.ensure_agent_runtime_current() +def test_revision_identity_comes_from_loaded_agent_module(monkeypatch, tmp_path: Path): + """The configured discovery path must not override the loaded module path.""" + from api import agent_runtime + + configured_dir = tmp_path / "configured-agent" + loaded_dir = tmp_path / "loaded-agent" + for agent_dir, revision in ((configured_dir, "configured"), (loaded_dir, "loaded")): + agent_dir.mkdir() + (agent_dir / "run_agent.py").write_text( + f"class AIAgent:\n revision = '{revision}'\n", + encoding="utf-8", + ) + _git(agent_dir, "init", "-q") + _git(agent_dir, "add", "run_agent.py") + _git(agent_dir, "commit", "-qm", revision) + + loaded_module = types.ModuleType("run_agent") + loaded_module.__file__ = str(loaded_dir / "run_agent.py") + monkeypatch.setitem(sys.modules, "run_agent", loaded_module) + monkeypatch.setattr(agent_runtime, "_AGENT_DIR", configured_dir) + monkeypatch.setattr(agent_runtime, "_AGENT_SOURCE_DIR", None) + monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", None) + + agent_runtime._capture_loaded_agent_revision() + + assert agent_runtime._AGENT_SOURCE_DIR == loaded_dir.resolve() + assert agent_runtime._AGENT_REVISION == _git(loaded_dir, "rev-parse", "HEAD") + + def test_known_revision_becoming_unreadable_fails_closed(monkeypatch): """Losing a previously-known revision is indistinguishable from source drift.""" from api import agent_runtime @@ -124,6 +154,75 @@ def test_known_revision_becoming_unreadable_fails_closed(monkeypatch): agent_runtime.ensure_agent_runtime_current() +def test_import_recapture_cannot_downgrade_known_revision(monkeypatch, tmp_path: Path): + """A second unreadable revision read must not erase a known identity.""" + from api import agent_runtime + + source_dir = tmp_path / "loaded-agent" + source_dir.mkdir() + loaded_module = types.ModuleType("run_agent") + loaded_module.__file__ = str(source_dir / "run_agent.py") + loaded_module.__dict__["AIAgent"] = type("AIAgent", (), {}) + monkeypatch.setitem(sys.modules, "run_agent", loaded_module) + monkeypatch.setattr(agent_runtime, "_AGENT_SOURCE_DIR", source_dir.resolve()) + monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", "known-revision") + revisions = iter(("known-revision", None)) + monkeypatch.setattr(agent_runtime, "_read_agent_revision", lambda _path: next(revisions)) + + with pytest.raises(agent_runtime.AgentRuntimeChangedError): + agent_runtime.require_ai_agent_class() + + assert agent_runtime._AGENT_REVISION == "known-revision" + + +def test_runner_local_bypasses_webui_agent_barrier(monkeypatch): + """Runner-local execution is owned by the runner, not this process.""" + from api import routes + + monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: False) + monkeypatch.setattr(routes, "get_config", lambda: {}) + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: (_ for _ in ()).throw( + AssertionError("runner-local must not inspect the WebUI Agent checkout") + ), + ) + monkeypatch.setattr( + "api.runtime_adapter.runtime_adapter_runner_enabled", + lambda: True, + ) + + assert routes._agent_runtime_barrier_response(runner_local_owned=True) is None + + +def test_runner_flag_does_not_bypass_webui_owned_hidden_turns(monkeypatch): + """Runner/gateway modes do not transfer ownership of legacy hidden turns.""" + from api import agent_runtime, routes + + monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: True) + monkeypatch.setattr(routes, "get_config", lambda: {}) + monkeypatch.setattr( + "api.runtime_adapter.runtime_adapter_runner_enabled", + lambda: True, + ) + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: (_ for _ in ()).throw( + agent_runtime.AgentRuntimeChangedError("restart required") + ), + ) + + response = routes._agent_runtime_barrier_response(runner_local_owned=False) + + assert response == { + "error": "restart required", + "type": "agent_runtime_stale", + "retryable": True, + } + + def test_chat_start_rejects_stale_runtime_before_session_materialization(monkeypatch): """A stale local runtime must not claim, create, or mutate session state.""" from api import agent_runtime, routes @@ -156,3 +255,172 @@ def test_chat_start_rejects_stale_runtime_before_session_materialization(monkeyp "retryable": True, }, } + + +def test_runner_owned_start_run_does_not_enter_local_stream_barrier(monkeypatch): + """The runner adapter must not invoke the WebUI in-process delegate.""" + from api import routes + + calls = [] + + class RunnerClient: + def start_run(self, request): + calls.append(request) + return { + "run_id": "runner-run-1", + "stream_id": "runner-stream-1", + "session_id": request.session_id, + "status": "started", + } + + session = types.SimpleNamespace(session_id="session-1", profile=None) + monkeypatch.setenv("HERMES_WEBUI_RUNTIME_ADAPTER", "runner-local") + monkeypatch.setattr("api.runtime_adapter.runtime_adapter_enabled", lambda: False) + monkeypatch.setattr("api.runtime_adapter.runtime_adapter_runner_enabled", lambda: True) + monkeypatch.setattr(routes, "_runtime_runner_client_factory", lambda: RunnerClient()) + monkeypatch.setattr( + routes, + "_start_chat_stream_for_session", + lambda **_kwargs: (_ for _ in ()).throw( + AssertionError("runner-owned start entered the WebUI local stream path") + ), + ) + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: (_ for _ in ()).throw( + AssertionError("runner-owned start inspected the WebUI Agent revision") + ), + ) + + response = routes._start_run( + session, + msg="hello", + attachments=[], + workspace="/tmp/workspace", + model="test-model", + model_provider="test-provider", + normalized_model=False, + source="webui", + route="/api/chat/start", + ) + + assert response.get("stream_id") == "runner-stream-1", response + assert len(calls) == 1 + + +def test_gateway_owned_start_run_bypasses_local_runtime_barrier(monkeypatch): + """Gateway chat must reach its worker without inspecting the local Agent.""" + from api import routes + + session = types.SimpleNamespace(session_id="session-1", profile=None) + captured = {} + monkeypatch.setattr("api.runtime_adapter.runtime_adapter_enabled", lambda: False) + monkeypatch.setattr("api.runtime_adapter.runtime_adapter_runner_enabled", lambda: False) + monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: True) + monkeypatch.setattr(routes, "get_config", lambda: {}) + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: (_ for _ in ()).throw( + AssertionError("gateway-owned start inspected the WebUI Agent revision") + ), + ) + + def start_gateway(_session, **kwargs): + captured.update(kwargs) + return {"stream_id": "gateway-stream-1", "session_id": "session-1"} + + monkeypatch.setattr(routes, "_start_chat_stream_for_session", start_gateway) + + response = routes._start_run( + session, + msg="hello", + attachments=[], + workspace="/tmp/workspace", + model="test-model", + model_provider="test-provider", + normalized_model=False, + source="webui", + route="/api/chat/start", + ) + + assert response["stream_id"] == "gateway-stream-1" + assert captured["external_runtime_owned"] is True + + +@pytest.mark.parametrize( + ("route", "body"), + [ + ("_handle_btw", {"session_id": "session-1", "question": "question"}), + ("_handle_background", {"session_id": "session-1", "prompt": "prompt"}), + ("_handle_chat_sync", {"session_id": "session-1", "message": "message"}), + ], +) +def test_hidden_turn_routes_reject_stale_runtime_before_session_creation( + monkeypatch, route, body +): + """Hidden-turn endpoints must reject before creation even in gateway mode.""" + from api import routes + + monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: True) + monkeypatch.setattr(routes, "get_config", lambda: {}) + monkeypatch.setattr( + "api.runtime_adapter.runtime_adapter_runner_enabled", + lambda: True, + ) + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: (_ for _ in ()).throw( + routes.AgentRuntimeChangedError("restart required") + ), + ) + monkeypatch.setattr( + routes, + "j", + lambda _handler, payload, status=200: {"status": status, "payload": payload}, + ) + monkeypatch.setattr( + routes, + "get_session", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("session loaded before stale-runtime barrier") + ), + ) + + response = getattr(routes, route)(object(), body) + + assert response["status"] == 409 + assert response["payload"]["type"] == "agent_runtime_stale" + + +def test_server_side_turn_rejects_stale_runtime_before_session_acceptance(monkeypatch): + """The direct process-wakeup entrypoint must fail before loading/mutating state.""" + from api import routes + + monkeypatch.setattr( + routes, + "_agent_runtime_barrier_response", + lambda **_kwargs: { + "error": "restart required", + "type": "agent_runtime_stale", + "retryable": True, + }, + ) + monkeypatch.setattr( + routes, + "get_session", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("session loaded before stale-runtime barrier") + ), + ) + + response = routes.start_session_turn("session-1", "wake up") + + assert response == { + "error": "restart required", + "type": "agent_runtime_stale", + "retryable": True, + "_status": 409, + } From 66dceb10e5a16a4994821ed6a39209f53e80178b Mon Sep 17 00:00:00 2001 From: pxxD1998 <214340659+pxxD1998@users.noreply.github.com> Date: Thu, 16 Jul 2026 17:05:42 +0800 Subject: [PATCH 05/14] fix: ignore untracked enclosing agent repositories --- api/agent_runtime.py | 49 ++++++++++++++--- tests/test_agent_runtime_revision_guard.py | 62 ++++++++++++++++++++++ 2 files changed, 105 insertions(+), 6 deletions(-) diff --git a/api/agent_runtime.py b/api/agent_runtime.py index 84514e0d8..3c9306f45 100644 --- a/api/agent_runtime.py +++ b/api/agent_runtime.py @@ -24,21 +24,58 @@ _RESTART_MESSAGE = ( def _read_agent_revision(agent_dir: Path | None) -> str | None: - """Return the checkout HEAD, or ``None`` for a non-Git/unavailable source.""" + """Return the loaded Agent checkout HEAD, or ``None`` if it is not tracked.""" if agent_dir is None: return None + + module = sys.modules.get("run_agent") + module_file = getattr(module, "__file__", None) + if not module_file: + return None + try: - result = subprocess.run( - ["git", "-C", str(agent_dir), "rev-parse", "--verify", "HEAD"], + module_path = Path(module_file).resolve() + worktree_result = subprocess.run( + ["git", "-C", str(agent_dir), "rev-parse", "--show-toplevel"], check=False, capture_output=True, text=True, timeout=2, ) - except (OSError, subprocess.TimeoutExpired): + if worktree_result.returncode != 0: + return None + worktree = Path(worktree_result.stdout.strip()).resolve() + relative_module = module_path.relative_to(worktree).as_posix() + tracked_result = subprocess.run( + [ + "git", + "--literal-pathspecs", + "-C", + str(worktree), + "ls-files", + "--error-unmatch", + "--", + relative_module, + ], + check=False, + capture_output=True, + text=True, + timeout=2, + ) + if tracked_result.returncode != 0: + return None + revision_result = subprocess.run( + ["git", "-C", str(worktree), "rev-parse", "--verify", "HEAD"], + check=False, + capture_output=True, + text=True, + timeout=2, + ) + except (OSError, subprocess.TimeoutExpired, RuntimeError, ValueError): return None - revision = result.stdout.strip() - return revision if result.returncode == 0 and revision else None + + revision = revision_result.stdout.strip() + return revision if revision_result.returncode == 0 and revision else None _AGENT_SOURCE_DIR: Path | None = None diff --git a/tests/test_agent_runtime_revision_guard.py b/tests/test_agent_runtime_revision_guard.py index 066c375d3..10ad81aea 100644 --- a/tests/test_agent_runtime_revision_guard.py +++ b/tests/test_agent_runtime_revision_guard.py @@ -114,6 +114,68 @@ def test_initial_non_git_source_preserves_supported_runtime(monkeypatch): agent_runtime.ensure_agent_runtime_current() +def test_untracked_loaded_module_inside_outer_git_repo_is_non_git( + monkeypatch, tmp_path: Path +): + """An enclosing unrelated Git repo must not identify an installed module.""" + from api import agent_runtime + + outer_repo = tmp_path / "outer-repo" + module_dir = outer_repo / "installed" / "hermes-agent" + module_dir.mkdir(parents=True) + module_file = module_dir / "run_agent.py" + module_file.write_text("class AIAgent: pass\n", encoding="utf-8") + (outer_repo / "tracked.txt").write_text("unrelated\n", encoding="utf-8") + _git(outer_repo, "init", "-q") + _git(outer_repo, "add", "tracked.txt") + _git(outer_repo, "commit", "-qm", "unrelated") + + loaded_module = types.ModuleType("run_agent") + loaded_module.__file__ = str(module_file) + monkeypatch.setitem(sys.modules, "run_agent", loaded_module) + monkeypatch.setattr(agent_runtime, "_AGENT_SOURCE_DIR", None) + monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", None) + + agent_runtime._capture_loaded_agent_revision() + + assert agent_runtime._AGENT_SOURCE_DIR == module_dir.resolve() + assert agent_runtime._AGENT_REVISION is None + + (outer_repo / "tracked.txt").write_text("unrelated change\n", encoding="utf-8") + _git(outer_repo, "add", "tracked.txt") + _git(outer_repo, "commit", "-qm", "unrelated change") + + agent_runtime.ensure_agent_runtime_current() + + +def test_untracked_module_pathspec_metacharacters_are_literal(monkeypatch, tmp_path: Path): + """Git pathspec syntax must not turn an untracked module into a tracked match.""" + from api import agent_runtime + + outer_repo = tmp_path / "outer-repo" + tracked_dir = outer_repo / "installed" / "agent" + untracked_dir = outer_repo / "installed" / "[a]gent" + tracked_dir.mkdir(parents=True) + untracked_dir.mkdir(parents=True) + (tracked_dir / "run_agent.py").write_text("class Other: pass\n", encoding="utf-8") + module_file = untracked_dir / "run_agent.py" + module_file.write_text("class AIAgent: pass\n", encoding="utf-8") + _git(outer_repo, "init", "-q") + _git(outer_repo, "add", "installed/agent/run_agent.py") + _git(outer_repo, "commit", "-qm", "tracked lookalike") + + loaded_module = types.ModuleType("run_agent") + loaded_module.__file__ = str(module_file) + monkeypatch.setitem(sys.modules, "run_agent", loaded_module) + monkeypatch.setattr(agent_runtime, "_AGENT_SOURCE_DIR", None) + monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", None) + + agent_runtime._capture_loaded_agent_revision() + + assert agent_runtime._AGENT_SOURCE_DIR == untracked_dir.resolve() + assert agent_runtime._AGENT_REVISION is None + + def test_revision_identity_comes_from_loaded_agent_module(monkeypatch, tmp_path: Path): """The configured discovery path must not override the loaded module path.""" from api import agent_runtime From 0bab6a3d7a541dd3204cc85d18c123e5de5076ee Mon Sep 17 00:00:00 2001 From: pxxD1998 <214340659+pxxD1998@users.noreply.github.com> Date: Thu, 16 Jul 2026 18:35:11 +0800 Subject: [PATCH 06/14] fix: handle remaining stale runtime entrypoints --- api/routes.py | 41 +++++- tests/test_agent_runtime_revision_guard.py | 146 +++++++++++++++++++++ tests/test_goal_command_webui.py | 12 +- tests/test_issue1013_handoff_dock.py | 54 ++++++++ tests/test_sprint46.py | 32 +++++ 5 files changed, 280 insertions(+), 5 deletions(-) diff --git a/api/routes.py b/api/routes.py index 698325b84..fc8516420 100644 --- a/api/routes.py +++ b/api/routes.py @@ -20782,8 +20782,14 @@ def _active_run_stream_for_session(session_id: str | None) -> str | None: return None -def _agent_runtime_barrier_response(*, runner_local_owned: bool = False) -> dict | None: +def _agent_runtime_barrier_response( + *, + runner_local_owned: bool = False, + external_runtime_owned: bool | None = None, +) -> dict | None: """Return the typed stale-runtime response for local in-process turns.""" + if external_runtime_owned is True: + return None if runner_local_owned and webui_gateway_chat_enabled(get_config()): return None from api.runtime_adapter import runtime_adapter_runner_enabled @@ -20814,11 +20820,14 @@ def _start_chat_stream_for_session( goal_related: bool = False, source: str = "webui", moa_config=None, - external_runtime_owned: bool = False, + external_runtime_owned: bool | None = None, ): """Persist pending state, register an SSE channel, and start an agent turn.""" + if external_runtime_owned is None: + external_runtime_owned = webui_gateway_chat_enabled(get_config()) + backend_is_gateway = bool(external_runtime_owned) stale_response = _agent_runtime_barrier_response( - runner_local_owned=external_runtime_owned, + external_runtime_owned=backend_is_gateway, ) if stale_response is not None: stale_response["_status"] = 409 @@ -20941,7 +20950,6 @@ def _start_chat_stream_for_session( if goal_related: STREAM_GOAL_RELATED[stream_id] = True diag.stage("worker_thread_start") if diag else None - backend_is_gateway = webui_gateway_chat_enabled(get_config()) worker_target = _run_gateway_chat_streaming if backend_is_gateway else _run_agent_streaming worker_kwargs = {"model_provider": model_provider, "goal_related": goal_related} if moa_config and not backend_is_gateway: @@ -21686,6 +21694,7 @@ def _handle_goal_command(handler, body): model_provider=model_provider, normalized_model=normalized_model, goal_related=True, + external_runtime_owned=webui_gateway_chat_enabled(get_config()), ) status = int(stream_response.pop("_status", 200) or 200) payload.update(stream_response) @@ -22737,6 +22746,12 @@ def _handle_git_commit_message(handler, body): return bad(handler, str(e)) except GitWorkspaceError as e: return _git_bad(handler, e) + except AgentRuntimeChangedError as e: + return j(handler, { + "error": str(e), + "type": "agent_runtime_stale", + "retryable": True, + }, status=409) except Exception as e: logger.exception("git commit message generation failed") return bad(handler, _sanitize_error(e), 500) @@ -22768,6 +22783,12 @@ def _handle_git_commit_message_selected(handler, body): return bad(handler, str(e)) except GitWorkspaceError as e: return _git_bad(handler, e) + except AgentRuntimeChangedError as e: + return j(handler, { + "error": str(e), + "type": "agent_runtime_stale", + "retryable": True, + }, status=409) except Exception as e: logger.exception("selected git commit message generation failed") return bad(handler, _sanitize_error(e), 500) @@ -24230,6 +24251,12 @@ def _handle_session_compress(handler, body): "focus_topic": focus_topic, }, ) + except AgentRuntimeChangedError as e: + return j(handler, { + "error": str(e), + "type": "agent_runtime_stale", + "retryable": True, + }, status=409) except Exception as e: logger.warning("Manual session compression failed: %s", e) return bad(handler, f"Compression failed: {_sanitize_error(e)}") @@ -24879,6 +24906,12 @@ def _handle_handoff_summary(handler, body): "rounds": rounds, "fallback": fallback, }) + except AgentRuntimeChangedError as e: + return j(handler, { + "error": str(e), + "type": "agent_runtime_stale", + "retryable": True, + }, status=409) except Exception as e: logger.warning("Handoff summary generation failed: %s", e) summary_text = _fallback_handoff_summary(msgs) diff --git a/tests/test_agent_runtime_revision_guard.py b/tests/test_agent_runtime_revision_guard.py index 10ad81aea..625f3e59d 100644 --- a/tests/test_agent_runtime_revision_guard.py +++ b/tests/test_agent_runtime_revision_guard.py @@ -371,6 +371,152 @@ def test_runner_owned_start_run_does_not_enter_local_stream_barrier(monkeypatch) assert len(calls) == 1 +@pytest.mark.parametrize("gateway_owned", [False, True]) +def test_stream_admission_uses_one_gateway_ownership_snapshot(monkeypatch, gateway_owned): + """The barrier and worker must share one immutable backend decision.""" + from api import routes + from api import turn_journal + from api.config import unregister_stream_owner + + gateway_reads = [] + revision_checks = [] + worker_targets = [] + session = types.SimpleNamespace( + session_id="gateway-snapshot", + profile=None, + title="Gateway snapshot", + active_stream_id=None, + pending_started_at=None, + ) + + def read_gateway(_cfg): + gateway_reads.append(True) + if len(gateway_reads) > 1: + raise AssertionError("gateway ownership was re-read after admission") + return gateway_owned + + def prepare(current, **kwargs): + current.active_stream_id = kwargs["stream_id"] + current.pending_started_at = 123.0 + + class FakeThread: + def __init__(self, *, target, args, kwargs, daemon): + worker_targets.append(target) + + def start(self): + return None + + monkeypatch.setattr(routes, "webui_gateway_chat_enabled", read_gateway) + monkeypatch.setattr(routes, "get_config", lambda: {}) + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: revision_checks.append(True), + ) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda _sid: None) + monkeypatch.setattr(routes, "_is_hidden_empty_session", lambda _session: False) + monkeypatch.setattr(routes, "_prepare_chat_start_session_for_stream", prepare) + monkeypatch.setattr(routes, "set_last_workspace", lambda _workspace: None) + monkeypatch.setattr(routes.threading, "Thread", FakeThread) + monkeypatch.setattr(turn_journal, "append_turn_journal_event", lambda *_args, **_kwargs: {}) + + response = routes._start_chat_stream_for_session( + session, + msg="goal", + attachments=[], + workspace="/tmp/workspace", + model="test-model", + goal_related=True, + ) + + try: + assert response["session_id"] == session.session_id + assert len(gateway_reads) == 1 + assert revision_checks == ([] if gateway_owned else [True]) + expected_worker = ( + routes._run_gateway_chat_streaming + if gateway_owned + else routes._run_agent_streaming + ) + assert worker_targets == [expected_worker] + finally: + stream_id = str(response.get("stream_id") or "") + with routes.STREAMS_LOCK: + routes.STREAMS.pop(stream_id, None) + unregister_stream_owner(stream_id) + routes.STREAM_GOAL_RELATED.pop(stream_id, None) + + +@pytest.mark.parametrize( + ("handler_name", "body"), + [ + ("_handle_git_commit_message", {"session_id": "git-message"}), + ( + "_handle_git_commit_message_selected", + {"session_id": "git-message", "paths": ["selected.py"]}, + ), + ], +) +def test_git_commit_message_stale_runtime_returns_typed_409( + monkeypatch, tmp_path, handler_name, body +): + """Commit-message generation must preserve the stale-runtime contract.""" + from api import routes + from api import workspace_git + + session = types.SimpleNamespace(workspace=str(tmp_path)) + monkeypatch.setattr(routes, "require", lambda _body, *keys: None) + monkeypatch.setattr(routes, "get_session", lambda _sid: session) + monkeypatch.setattr(routes, "_git_paths_from_body", lambda _body: ["selected.py"]) + monkeypatch.setattr( + workspace_git, + "staged_commit_message_prompt", + lambda _workspace: {"system_prompt": "system", "user_prompt": "user"}, + ) + monkeypatch.setattr( + workspace_git, + "selected_commit_message_prompt", + lambda _workspace, _paths: { + "system_prompt": "system", + "user_prompt": "user", + }, + ) + monkeypatch.setattr( + routes, + "_llm_git_commit_message", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + routes.AgentRuntimeChangedError("restart required") + ), + ) + monkeypatch.setattr( + routes, + "j", + lambda _handler, payload, status=200, **_kwargs: { + "status": status, + "payload": payload, + }, + ) + monkeypatch.setattr( + routes, + "bad", + lambda _handler, message, status=400: { + "status": status, + "payload": {"error": message}, + }, + ) + + response = getattr(routes, handler_name)(object(), body) + + assert response == { + "status": 409, + "payload": { + "error": "restart required", + "type": "agent_runtime_stale", + "retryable": True, + }, + } + + def test_gateway_owned_start_run_bypasses_local_runtime_barrier(monkeypatch): """Gateway chat must reach its worker without inspecting the local Agent.""" from api import routes diff --git a/tests/test_goal_command_webui.py b/tests/test_goal_command_webui.py index 2389c08bc..b16f38926 100644 --- a/tests/test_goal_command_webui.py +++ b/tests/test_goal_command_webui.py @@ -161,7 +161,10 @@ def test_goal_continuation_decision_emits_status_and_normal_user_prompt(monkeypa assert decision["continuation_prompt"].startswith("[Continuing toward your standing goal]") -def test_goal_endpoint_sets_goal_and_starts_kickoff_stream(monkeypatch, tmp_path): +@pytest.mark.parametrize("gateway_owned", [False, True]) +def test_goal_endpoint_sets_goal_and_starts_kickoff_stream( + monkeypatch, tmp_path, gateway_owned +): """POST /api/goal uses GoalManager state and launches the first goal turn.""" from api import goals as webui_goals from api import routes @@ -199,6 +202,12 @@ def test_goal_endpoint_sets_goal_and_starts_kickoff_stream(monkeypatch, tmp_path monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager) monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession()) monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path) + monkeypatch.setattr( + routes, + "webui_gateway_chat_enabled", + lambda _cfg: gateway_owned, + ) + monkeypatch.setattr(routes, "get_config", lambda: {}) monkeypatch.setattr( routes, "_resolve_compatible_session_model_state", @@ -229,6 +238,7 @@ def test_goal_endpoint_sets_goal_and_starts_kickoff_stream(monkeypatch, tmp_path assert result["payload"]["stream_id"] == "goal-stream" assert started and started[0]["msg"] == "ship the feature" assert started[0]["model_provider"] == "openai-codex" + assert started[0]["external_runtime_owned"] is gateway_owned def test_goal_endpoint_preserves_response_shape_under_runtime_adapter_flag(monkeypatch, tmp_path): diff --git a/tests/test_issue1013_handoff_dock.py b/tests/test_issue1013_handoff_dock.py index 39121ebc4..2a970bacc 100644 --- a/tests/test_issue1013_handoff_dock.py +++ b/tests/test_issue1013_handoff_dock.py @@ -248,6 +248,60 @@ def test_no_api_key_handoff_summary_persists_fallback_summary(monkeypatch): assert persisted[0]["rounds"] == models.CONVERSATION_ROUND_THRESHOLD +def test_stale_runtime_handoff_summary_returns_typed_409_without_persisting(monkeypatch): + """A stale runtime must remain retryable instead of becoming fallback success.""" + import api.routes as routes + import api.models as models + + monkeypatch.setattr(routes, "require", lambda body, *keys: None) + monkeypatch.setattr( + routes, + "j", + lambda _handler, payload, status=200, extra_headers=None: { + "status": status, + "payload": payload, + }, + ) + monkeypatch.setattr( + models, + "count_conversation_rounds", + lambda sid, since=None: models.CONVERSATION_ROUND_THRESHOLD, + ) + monkeypatch.setattr( + models, + "get_cli_session_messages", + lambda sid: [ + {"role": "user", "content": "Need a handoff", "timestamp": 1.0}, + {"role": "assistant", "content": "Context is ready", "timestamp": 2.0}, + ], + ) + monkeypatch.setattr( + routes, + "_persist_handoff_summary", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("stale runtime persisted a fallback summary") + ), + ) + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: (_ for _ in ()).throw( + routes.AgentRuntimeChangedError("restart required") + ), + ) + + response = routes._handle_handoff_summary(object(), {"session_id": "stale-handoff"}) + + assert response == { + "status": 409, + "payload": { + "error": "restart required", + "type": "agent_runtime_stale", + "retryable": True, + }, + } + + def test_exception_handoff_summary_persists_fallback_summary(monkeypatch): """Unhandled summary exception should still persist a fallback handoff marker.""" import api.routes as routes diff --git a/tests/test_sprint46.py b/tests/test_sprint46.py index 272f86689..ff10828e2 100644 --- a/tests/test_sprint46.py +++ b/tests/test_sprint46.py @@ -129,6 +129,38 @@ def test_session_compress_requires_session_id(cleanup_test_sessions): assert handler.payload()["error"] == "Missing required field(s): session_id" +def test_session_compress_stale_runtime_returns_typed_409_before_mutation( + monkeypatch, cleanup_test_sessions +): + import api.routes as routes + + sid = _make_session() + cleanup_test_sessions.append(sid) + loaded_before = Session.load(sid) + assert loaded_before is not None + before = loaded_before.compact() + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: (_ for _ in ()).throw( + routes.AgentRuntimeChangedError("restart required") + ), + ) + + handler = _FakeHandler() + _handle_session_compress(handler, {"session_id": sid}) + + assert handler.status == 409 + assert handler.payload() == { + "error": "restart required", + "type": "agent_runtime_stale", + "retryable": True, + } + loaded_after = Session.load(sid) + assert loaded_after is not None + assert loaded_after.compact() == before + + def test_session_compress_roundtrip(monkeypatch, cleanup_test_sessions): created = cleanup_test_sessions original_messages = [ From 2415995204530f3940bb78c8bdf86c67cd091089 Mon Sep 17 00:00:00 2001 From: pxxD1998 <214340659+pxxD1998@users.noreply.github.com> Date: Thu, 16 Jul 2026 22:27:53 +0800 Subject: [PATCH 07/14] fix(runtime): ignore in-memory agent module swaps --- api/agent_runtime.py | 51 ++++++++++++------- tests/test_agent_runtime_revision_guard.py | 59 ++++++++++++++++++++-- 2 files changed, 88 insertions(+), 22 deletions(-) diff --git a/api/agent_runtime.py b/api/agent_runtime.py index 3c9306f45..865a2fd50 100644 --- a/api/agent_runtime.py +++ b/api/agent_runtime.py @@ -23,18 +23,26 @@ _RESTART_MESSAGE = ( ) -def _read_agent_revision(agent_dir: Path | None) -> str | None: +def _read_agent_revision( + agent_dir: Path | None, + *, + module_path: Path | None = None, +) -> str | None: """Return the loaded Agent checkout HEAD, or ``None`` if it is not tracked.""" if agent_dir is None: return None - module = sys.modules.get("run_agent") - module_file = getattr(module, "__file__", None) - if not module_file: - return None + if module_path is None: + module = sys.modules.get("run_agent") + module_file = getattr(module, "__file__", None) + if not module_file: + return None + try: + module_path = Path(module_file).resolve() + except (OSError, RuntimeError, TypeError): + return None try: - module_path = Path(module_file).resolve() worktree_result = subprocess.run( ["git", "-C", str(agent_dir), "rev-parse", "--show-toplevel"], check=False, @@ -79,6 +87,7 @@ def _read_agent_revision(agent_dir: Path | None) -> str | None: _AGENT_SOURCE_DIR: Path | None = None +_AGENT_MODULE_PATH: Path | None = None _AGENT_REVISION: str | None = None _AIAgent = None _RUNTIME_LOCK = threading.Lock() @@ -88,33 +97,34 @@ class AgentRuntimeChangedError(RuntimeError): """Raised when the loaded Agent runtime no longer matches its source tree.""" -def _loaded_agent_source_dir() -> Path | None: - """Return the source directory that actually supplied ``run_agent``.""" +def _loaded_agent_source_identity() -> tuple[Path, Path] | None: + """Return the source directory and file that supplied ``run_agent``.""" module = sys.modules.get("run_agent") module_file = getattr(module, "__file__", None) if not module_file: return None try: - return Path(module_file).resolve().parent + module_path = Path(module_file).resolve() + return module_path.parent, module_path except (OSError, RuntimeError, TypeError): return None def _capture_loaded_agent_revision() -> None: """Bind the guard to the checkout that supplied the loaded Agent module.""" - global _AGENT_SOURCE_DIR, _AGENT_REVISION + global _AGENT_SOURCE_DIR, _AGENT_MODULE_PATH, _AGENT_REVISION - source_dir = _loaded_agent_source_dir() - if source_dir is None: - return - current_revision = _read_agent_revision(source_dir) - if _AGENT_SOURCE_DIR is not None and source_dir != _AGENT_SOURCE_DIR: - raise AgentRuntimeChangedError(_RESTART_MESSAGE) if _AGENT_REVISION is not None: - if current_revision != _AGENT_REVISION: - raise AgentRuntimeChangedError(_RESTART_MESSAGE) + ensure_agent_runtime_current() return + + identity = _loaded_agent_source_identity() + if identity is None: + return + source_dir, module_path = identity + current_revision = _read_agent_revision(source_dir, module_path=module_path) _AGENT_SOURCE_DIR = source_dir + _AGENT_MODULE_PATH = module_path _AGENT_REVISION = current_revision @@ -122,7 +132,10 @@ def ensure_agent_runtime_current() -> None: """Reject a known Git checkout change instead of mixing Python modules.""" if _AGENT_REVISION is None: return - if _read_agent_revision(_AGENT_SOURCE_DIR) != _AGENT_REVISION: + if ( + _read_agent_revision(_AGENT_SOURCE_DIR, module_path=_AGENT_MODULE_PATH) + != _AGENT_REVISION + ): raise AgentRuntimeChangedError(_RESTART_MESSAGE) diff --git a/tests/test_agent_runtime_revision_guard.py b/tests/test_agent_runtime_revision_guard.py index 625f3e59d..9712f711f 100644 --- a/tests/test_agent_runtime_revision_guard.py +++ b/tests/test_agent_runtime_revision_guard.py @@ -109,7 +109,11 @@ def test_initial_non_git_source_preserves_supported_runtime(monkeypatch): from api import agent_runtime monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", None) - monkeypatch.setattr(agent_runtime, "_read_agent_revision", lambda _path: None) + monkeypatch.setattr( + agent_runtime, + "_read_agent_revision", + lambda _path, **_kwargs: None, + ) agent_runtime.ensure_agent_runtime_current() @@ -134,11 +138,13 @@ def test_untracked_loaded_module_inside_outer_git_repo_is_non_git( loaded_module.__file__ = str(module_file) monkeypatch.setitem(sys.modules, "run_agent", loaded_module) monkeypatch.setattr(agent_runtime, "_AGENT_SOURCE_DIR", None) + monkeypatch.setattr(agent_runtime, "_AGENT_MODULE_PATH", None) monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", None) agent_runtime._capture_loaded_agent_revision() assert agent_runtime._AGENT_SOURCE_DIR == module_dir.resolve() + assert agent_runtime._AGENT_MODULE_PATH == module_file.resolve() assert agent_runtime._AGENT_REVISION is None (outer_repo / "tracked.txt").write_text("unrelated change\n", encoding="utf-8") @@ -168,11 +174,13 @@ def test_untracked_module_pathspec_metacharacters_are_literal(monkeypatch, tmp_p loaded_module.__file__ = str(module_file) monkeypatch.setitem(sys.modules, "run_agent", loaded_module) monkeypatch.setattr(agent_runtime, "_AGENT_SOURCE_DIR", None) + monkeypatch.setattr(agent_runtime, "_AGENT_MODULE_PATH", None) monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", None) agent_runtime._capture_loaded_agent_revision() assert agent_runtime._AGENT_SOURCE_DIR == untracked_dir.resolve() + assert agent_runtime._AGENT_MODULE_PATH == module_file.resolve() assert agent_runtime._AGENT_REVISION is None @@ -197,11 +205,13 @@ def test_revision_identity_comes_from_loaded_agent_module(monkeypatch, tmp_path: monkeypatch.setitem(sys.modules, "run_agent", loaded_module) monkeypatch.setattr(agent_runtime, "_AGENT_DIR", configured_dir) monkeypatch.setattr(agent_runtime, "_AGENT_SOURCE_DIR", None) + monkeypatch.setattr(agent_runtime, "_AGENT_MODULE_PATH", None) monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", None) agent_runtime._capture_loaded_agent_revision() assert agent_runtime._AGENT_SOURCE_DIR == loaded_dir.resolve() + assert agent_runtime._AGENT_MODULE_PATH == (loaded_dir / "run_agent.py").resolve() assert agent_runtime._AGENT_REVISION == _git(loaded_dir, "rev-parse", "HEAD") @@ -210,7 +220,11 @@ def test_known_revision_becoming_unreadable_fails_closed(monkeypatch): from api import agent_runtime monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", "known-revision") - monkeypatch.setattr(agent_runtime, "_read_agent_revision", lambda _path: None) + monkeypatch.setattr( + agent_runtime, + "_read_agent_revision", + lambda _path, **_kwargs: None, + ) with pytest.raises(agent_runtime.AgentRuntimeChangedError): agent_runtime.ensure_agent_runtime_current() @@ -227,9 +241,14 @@ def test_import_recapture_cannot_downgrade_known_revision(monkeypatch, tmp_path: loaded_module.__dict__["AIAgent"] = type("AIAgent", (), {}) monkeypatch.setitem(sys.modules, "run_agent", loaded_module) monkeypatch.setattr(agent_runtime, "_AGENT_SOURCE_DIR", source_dir.resolve()) + monkeypatch.setattr(agent_runtime, "_AGENT_MODULE_PATH", source_dir / "run_agent.py") monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", "known-revision") revisions = iter(("known-revision", None)) - monkeypatch.setattr(agent_runtime, "_read_agent_revision", lambda _path: next(revisions)) + monkeypatch.setattr( + agent_runtime, + "_read_agent_revision", + lambda _path, **_kwargs: next(revisions), + ) with pytest.raises(agent_runtime.AgentRuntimeChangedError): agent_runtime.require_ai_agent_class() @@ -237,6 +256,40 @@ def test_import_recapture_cannot_downgrade_known_revision(monkeypatch, tmp_path: assert agent_runtime._AGENT_REVISION == "known-revision" +def test_in_memory_agent_swap_does_not_mimic_source_revision_change( + monkeypatch, tmp_path: Path +): + """Only the bound checkout revision, not ``sys.modules`` swaps, defines staleness.""" + from api import agent_runtime + + agent_dir = tmp_path / "loaded-agent" + agent_dir.mkdir() + module_file = agent_dir / "run_agent.py" + module_file.write_text("class AIAgent: pass\n", encoding="utf-8") + _git(agent_dir, "init", "-q") + _git(agent_dir, "add", "run_agent.py") + _git(agent_dir, "commit", "-qm", "loaded agent") + + loaded_module = types.ModuleType("run_agent") + loaded_module.__file__ = str(module_file) + loaded_module.__dict__["AIAgent"] = type("LoadedAgent", (), {}) + monkeypatch.setitem(sys.modules, "run_agent", loaded_module) + monkeypatch.setattr(agent_runtime, "_AGENT_SOURCE_DIR", None) + monkeypatch.setattr(agent_runtime, "_AGENT_MODULE_PATH", None) + monkeypatch.setattr(agent_runtime, "_AGENT_REVISION", None) + agent_runtime._capture_loaded_agent_revision() + + replacement_class = type("ReplacementAgent", (), {}) + replacement_module = types.ModuleType("run_agent") + replacement_module.__dict__["AIAgent"] = replacement_class + monkeypatch.setitem(sys.modules, "run_agent", replacement_module) + + assert agent_runtime.require_ai_agent_class() is replacement_class + assert agent_runtime._AGENT_SOURCE_DIR == agent_dir.resolve() + assert agent_runtime._AGENT_MODULE_PATH == module_file.resolve() + assert agent_runtime._AGENT_REVISION == _git(agent_dir, "rev-parse", "HEAD") + + def test_runner_local_bypasses_webui_agent_barrier(monkeypatch): """Runner-local execution is owned by the runner, not this process.""" from api import routes From 0cee7dba478c187e35ab96a4ba98e0d58f0a0204 Mon Sep 17 00:00:00 2001 From: carlotestor Date: Thu, 16 Jul 2026 16:38:43 +0200 Subject: [PATCH 08/14] fix: stop closing SessionDB under live background subagents MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Background subagents (delegate_tool) capture parent_agent._session_db by reference. The cached-agent reuse path created a fresh SessionDB every stream request and unconditionally closed the previous handle (PR #1421 EMFILE fix). A server-side wakeup / new turn for the same parent session could therefore close the shared handle while children were still flushing messages, producing: Session DB append_message failed: 'NoneType' object has no attribute 'execute' Reproduce (observed in production): 1. Parent WebUI session runs delegate_task with N background subagents 2. Parent turn tears down / idle-hook redelivers a deferred wakeup 3. New stream for the parent adopts a fresh SessionDB and closes the old one 4. Live subagents' append_message hits conn=None Fix: _adopt_session_db_for_cached_agent reuses a still-open handle (and closes the unused new one — still EMFILE-safe) and only replaces when the existing handle is missing or already closed. Same guard applied to _replace_session_db_in_kwargs (credential self-heal). LRU eviction still closes at true session boundaries. Tests updated to pin the adopt helper and the subagent-safe semantics. --- api/streaming.py | 90 +++++++- tests/test_v050259_sessiondb_fd_leak.py | 274 ++++++++++++++---------- 2 files changed, 246 insertions(+), 118 deletions(-) diff --git a/api/streaming.py b/api/streaming.py index c733a8f9a..192c0d4ea 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -6265,6 +6265,57 @@ def _last_resort_sync_from_core(session, stream_id, agent_lock): ) +def _session_db_is_open(session_db) -> bool: + """True when *session_db* still has a live sqlite connection. + + SessionDB.close() sets ``_conn = None``. Subagents capture the parent's + SessionDB object by reference at spawn time (delegate_tool), so closing + that object mid-parent-turn makes every subsequent child + ``append_message`` fail with + ``'NoneType' object has no attribute 'execute'``. + """ + if session_db is None: + return False + return getattr(session_db, "_conn", None) is not None + + +def _adopt_session_db_for_cached_agent(agent, new_session_db): + """Attach a SessionDB to a reused cached agent without breaking subagents. + + Historical behaviour (PR #1421 FD-leak fix): create a fresh SessionDB every + stream request and close the previous handle before replacing + ``agent._session_db``. That stops EMFILE growth, but a server-side wakeup + / new turn for the same parent session will close the shared handle while + background subagents are still writing into it. + + Policy now: + - If the cached agent already holds an *open* SessionDB, keep it and close + the unused new handle (no FD leak; live subagents keep working). + - If the existing handle is missing or already closed, adopt *new_session_db*. + - If *new_session_db* is None, leave the existing handle alone. + """ + if agent is None: + return new_session_db + existing = getattr(agent, "_session_db", None) + if new_session_db is None: + return existing + if existing is new_session_db: + return existing + if _session_db_is_open(existing): + try: + new_session_db.close() + except Exception: + pass + return existing + if existing is not None: + try: + existing.close() + except Exception: + pass + agent._session_db = new_session_db + return new_session_db + + def _build_session_db_for_stream(state_db_path): """Build a per-request SessionDB handle for WebUI session search. @@ -6280,12 +6331,27 @@ def _build_session_db_for_stream(state_db_path): def _replace_session_db_in_kwargs(agent_kwargs, state_db_path): - """Build a fresh SessionDB and replace ``agent_kwargs['session_db']`` safely.""" + """Build a fresh SessionDB and replace ``agent_kwargs['session_db']`` safely. + + Does not close an existing open handle that may still be shared with live + subagents; only replaces when the prior handle is missing or already closed. + """ if not isinstance(agent_kwargs, dict): return None _old_session_db = agent_kwargs.get("session_db") _next_session_db = _build_session_db_for_stream(state_db_path) + if _next_session_db is None: + return _old_session_db + if _session_db_is_open(_old_session_db): + # Keep the live handle; discard the unused new one. + try: + if _next_session_db is not _old_session_db: + _next_session_db.close() + except Exception: + logger.debug("Failed to close unused session_db handle during self-heal") + agent_kwargs["session_db"] = _old_session_db + return _old_session_db if _old_session_db is not None and _old_session_db is not _next_session_db: try: _old_session_db.close() @@ -8295,15 +8361,19 @@ def _run_agent_streaming( if 'prefill_messages' in _agent_kwargs and hasattr(agent, 'prefill_messages'): agent.prefill_messages = list(_agent_kwargs.get('prefill_messages') or []) if _session_db is not None: - # Close any previously held SessionDB connection before - # replacing it. Without this, each streaming request creates - # a new SessionDB whose WAL handles leak indefinitely, - # eventually causing EMFILE crashes (#streaming FD leak). - if hasattr(agent, '_session_db') and agent._session_db is not None: - try: - agent._session_db.close() - except Exception: - pass + # Prefer reusing a still-open SessionDB on the cached + # agent. Closing it mid-turn breaks background + # subagents that hold a reference to the same object + # (delegate_tool copies parent._session_db by ref) — + # they then fail with + # 'NoneType' object has no attribute 'execute'. + # When the existing handle is already closed/missing, + # adopt the fresh per-request SessionDB (and close the + # dead one) so we still avoid the EMFILE FD-leak from + # PR #1421. + _session_db = _adopt_session_db_for_cached_agent( + agent, _session_db + ) agent._session_db = _session_db if hasattr(agent, '_api_call_count'): agent._api_call_count = 0 diff --git a/tests/test_v050259_sessiondb_fd_leak.py b/tests/test_v050259_sessiondb_fd_leak.py index 3b80ac974..7b17212d9 100644 --- a/tests/test_v050259_sessiondb_fd_leak.py +++ b/tests/test_v050259_sessiondb_fd_leak.py @@ -1,18 +1,25 @@ -"""Regression tests for v0.50.259 — SessionDB FD leak fixes from PR #1421 -plus Opus pre-release advisor follow-up extending the fix to LRU eviction. +"""Regression tests for SessionDB FD-leak fixes (PR #1421) plus the +subagent shared-handle race (close-under-live-subagents). -The bug: `_run_agent_streaming` created a new `SessionDB` per request and +History +------- +PR #1421: `_run_agent_streaming` created a new `SessionDB` per request and replaced the cached agent's `_session_db` without closing the old one. After ~73 messages on a long-lived agent, leaked FDs exhausted the 256 FD -default limit causing `EMFILE` crashes. +default limit causing `EMFILE` crashes. Fix: close the previous handle +when it is safe to replace it. -Fix path 1 (PR #1421 by @wali-reheman): close `agent._session_db` before -replacing it on the cached-agent reuse path. +Follow-up (this change): always-close-before-replace is *not* safe when +background subagents still hold a reference to the same SessionDB object +(delegate_tool copies ``parent._session_db`` by ref). A server-side wakeup +/ new turn for the parent session was closing the shared handle mid-child- +run, producing: -Fix path 2 (this PR Opus follow-up): same shape on the LRU eviction path. -When `SESSION_AGENT_CACHE.popitem(last=False)` evicts an old agent, its -`_session_db` is dropped on the floor and only released when GC eventually -finalizes the agent — which on a long-running server may be never. + Session DB append_message failed: 'NoneType' object has no attribute 'execute' + +Policy now (``_adopt_session_db_for_cached_agent``): +- existing handle still open → keep it, close the unused *new* handle +- existing handle missing/closed → adopt the new handle (close dead one) """ from __future__ import annotations @@ -25,38 +32,49 @@ import pytest REPO = Path(__file__).resolve().parents[1] -# ── 1: source-level pin: cached-agent-reuse path closes _session_db ───────── +# ── 1: source-level pin: cached-agent reuse uses the adopt helper ────────── -def test_cached_agent_reuse_closes_old_session_db(): - """The cached-agent reuse path in `_run_agent_streaming` MUST close the - old `_session_db` before replacing it. Without this, every streaming - request leaks a SessionDB connection (3 FDs once WAL is active).""" +def test_cached_agent_reuse_uses_adopt_helper(): + """Cached-agent reuse must go through `_adopt_session_db_for_cached_agent` + so a still-open SessionDB is reused (subagent-safe) and only a dead handle + is closed+replaced (still EMFILE-safe).""" src = (REPO / "api" / "streaming.py").read_text(encoding="utf-8") - # Find the cached-agent reuse block (where callbacks are refreshed). - # The replacement of agent._session_db must be preceded by a close(). reuse_idx = src.find("Refresh per-turn callbacks") assert reuse_idx != -1, "cached-agent reuse block missing" block = src[reuse_idx : reuse_idx + 2500] - # Must close before replace. - assert "agent._session_db.close()" in block, ( - "cached-agent reuse path must call agent._session_db.close() before " - "replacing it. Without this, FDs leak on every streaming request " - "and the server EMFILE-crashes after ~73 messages. See PR #1421." + assert "_adopt_session_db_for_cached_agent" in block, ( + "cached-agent reuse path must call _adopt_session_db_for_cached_agent " + "instead of unconditionally closing agent._session_db. Unconditional " + "close breaks background subagents that share the handle by reference." ) - # And the close must come BEFORE the replacement (lexically). - close_idx = block.find("agent._session_db.close()") - replace_idx = block.find("agent._session_db = _session_db") - assert close_idx != -1 and replace_idx != -1 - assert close_idx < replace_idx, ( - "close() must lexically precede the assignment so the old connection " - "is released before the reference is rebound." + assert "agent._session_db = _session_db" in block, ( + "reuse path must still assign the adopted SessionDB onto the agent" + ) + # The old unconditional-close pattern must not remain in the reuse block. + assert "agent._session_db.close()" not in block, ( + "unconditional agent._session_db.close() in the reuse path is the " + "subagent race; close is now owned by _adopt_session_db_for_cached_agent" ) -# ── 2: source-level pin: LRU eviction path also closes _session_db ────────── +def test_adopt_and_is_open_helpers_exist(): + src = (REPO / "api" / "streaming.py").read_text(encoding="utf-8") + assert "def _session_db_is_open(" in src + assert "def _adopt_session_db_for_cached_agent(" in src + # self-heal path must also refuse to close a still-open handle + replace_idx = src.find("def _replace_session_db_in_kwargs") + assert replace_idx != -1 + block = src[replace_idx : replace_idx + 1200] + assert "_session_db_is_open" in block, ( + "_replace_session_db_in_kwargs must guard on _session_db_is_open so " + "credential self-heal cannot close a handle live subagents share" + ) + + +# ── 2: source-level pin: LRU eviction path still closes _session_db ──────── def test_lru_eviction_closes_evicted_agent_session_db(): @@ -66,27 +84,21 @@ def test_lru_eviction_closes_evicted_agent_session_db(): on GC finalization which may never run on a long-lived server. Fix: capture the evicted entry, close its agent's `_session_db` before - dropping the reference. + dropping the reference. (Eviction is a true session boundary — no live + subagents are expected to still be writing into that agent.) """ src = (REPO / "api" / "streaming.py").read_text(encoding="utf-8") - # The eviction site uses popitem(last=False). The evicted entry must be - # captured (not discarded with `_`) and the agent's _session_db closed. eviction_idx = src.find("Evicted LRU agent from cache") assert eviction_idx != -1, "LRU eviction debug log missing" - # Look in a window around the eviction. block = src[max(0, eviction_idx - 1500) : eviction_idx + 200] - # Negative pattern: the old `evicted_sid, _ = ...` discard form must NOT - # be present — that's the bug shape. assert "evicted_sid, _ = SESSION_AGENT_CACHE.popitem" not in block, ( - "LRU eviction must capture the evicted entry so the agent\'s " + "LRU eviction must capture the evicted entry so the agent's " "_session_db can be closed. The `evicted_sid, _ = ...` discard form " "is the original bug shape." ) - # Positive pattern: eviction must call the lifecycle helper, and the helper - # must close the evicted agent's SessionDB after provider teardown. assert "_close_evicted_agent_at_session_boundary(_evicted_sid, _evicted_agent)" in block, ( "LRU eviction must route the evicted agent through the session-boundary " "close helper." @@ -104,16 +116,7 @@ def test_lru_eviction_closes_evicted_agent_session_db(): def test_session_db_close_is_idempotent(): - """`SessionDB.close()` must be safe to call multiple times. The fix - relies on this — if a future code path closes the same `_session_db` - after we've swapped it, the second close is a benign no-op. - - Skipped when hermes_state is not on the import path (e.g. on the GH - Actions runner that only has the WebUI repo, not the agent repo). - The source-level pin in test_cached_agent_reuse_closes_old_session_db - catches revert of the close() call; this test only adds runtime - confirmation when both repos are co-located. - """ + """`SessionDB.close()` must be safe to call multiple times.""" import importlib.util if importlib.util.find_spec("hermes_state") is None: pytest.skip("hermes_state not on import path (CI-only — agent repo not present)") @@ -123,65 +126,133 @@ def test_session_db_close_is_idempotent(): with tempfile.TemporaryDirectory() as tmpd: db_path = Path(tmpd) / "test.db" db = SessionDB(db_path=db_path) - # Force connection open by issuing a query. with db._lock: db._conn.execute("SELECT 1") - # First close db.close() assert db._conn is None - # Second close — must not raise. db.close() assert db._conn is None - # Third close — still safe. db.close() -# ── 4: behavioral: cached-agent reuse with mock agent ─────────────────────── +# ── 4: behavioral: adopt helper keeps open handles, closes dead ones ─────── -def test_cached_agent_reuse_calls_close_on_old_session_db(): - """End-to-end: simulate the cached-agent reuse code path with a mock - agent and verify the mock SessionDB.close() is called when _session_db - is replaced. Pins the runtime behavior, not just the source pattern.""" - import sys - sys.path.insert(0, str(REPO)) +class _MockSessionDB: + def __init__(self, name, open_=True): + self.name = name + self.close_calls = 0 + # Mirror SessionDB: open → _conn is truthy; closed → _conn is None. + self._conn = object() if open_ else None - class MockSessionDB: - def __init__(self, name): - self.name = name - self.close_calls = 0 - def close(self): - self.close_calls += 1 + def close(self): + self.close_calls += 1 + self._conn = None - class MockAgent: - def __init__(self, db): - self._session_db = db - self.stream_delta_callback = None - self.tool_progress_callback = None - self._api_call_count = 0 - self._interrupted = False - self._interrupt_message = None - # Simulate the inner block of the cached-agent reuse path. We replicate - # the pattern manually rather than importing _run_agent_streaming - # directly because that function has many other side effects. - old_db = MockSessionDB("old") - new_db = MockSessionDB("new") - agent = MockAgent(old_db) +class _MockAgent: + def __init__(self, db): + self._session_db = db + self.stream_delta_callback = None + self.tool_progress_callback = None + self._api_call_count = 0 + self._interrupted = False + self._interrupt_message = None - # Mirror the production code path: - if hasattr(agent, "_session_db") and agent._session_db is not None: - try: - agent._session_db.close() - except Exception: - pass - agent._session_db = new_db - assert old_db.close_calls == 1, ( - "old SessionDB must be closed exactly once when replaced on cached agent" - ) - assert new_db.close_calls == 0, "new SessionDB should not be closed" +def _import_adopt_helpers(): + """Import the helpers without pulling in the whole streaming module graph. + + streaming.py has heavy import-time side effects; exec the helper bodies + from source so unit tests stay hermetic. + """ + import importlib.util + import types + + # Prefer real import when the package is importable. + try: + from api.streaming import ( # type: ignore + _adopt_session_db_for_cached_agent, + _session_db_is_open, + ) + return _session_db_is_open, _adopt_session_db_for_cached_agent + except Exception: + pass + + src = (REPO / "api" / "streaming.py").read_text(encoding="utf-8") + # Extract just the two helper defs by line slicing between markers. + start = src.index("def _session_db_is_open") + end = src.index("def _build_session_db_for_stream") + blob = src[start:end] + ns: dict = {} + exec(blob, ns) # noqa: S102 — test-only extraction of pure helpers + return ns["_session_db_is_open"], ns["_adopt_session_db_for_cached_agent"] + + +def test_adopt_reuses_open_session_db_and_closes_new(): + """Live (open) existing handle must be kept; unused new handle closed. + + This is the subagent-safe path: children hold a reference to `old_db`. + """ + _is_open, adopt = _import_adopt_helpers() + old_db = _MockSessionDB("old", open_=True) + new_db = _MockSessionDB("new", open_=True) + agent = _MockAgent(old_db) + + result = adopt(agent, new_db) + + assert result is old_db + assert agent._session_db is old_db + assert old_db.close_calls == 0, "must not close the live shared handle" + assert new_db.close_calls == 1, "unused per-request handle must be closed (FD leak)" + assert _is_open(old_db) is True + assert _is_open(new_db) is False + + +def test_adopt_replaces_closed_session_db(): + """Dead existing handle is closed (idempotent) and replaced with the new one.""" + _is_open, adopt = _import_adopt_helpers() + old_db = _MockSessionDB("old", open_=False) + new_db = _MockSessionDB("new", open_=True) + agent = _MockAgent(old_db) + + result = adopt(agent, new_db) + + assert result is new_db assert agent._session_db is new_db + assert old_db.close_calls == 1 + assert new_db.close_calls == 0 + + +def test_adopt_handles_missing_existing(): + _is_open, adopt = _import_adopt_helpers() + new_db = _MockSessionDB("new", open_=True) + agent = _MockAgent(None) + agent._session_db = None + + result = adopt(agent, new_db) + + assert result is new_db + assert agent._session_db is new_db + assert new_db.close_calls == 0 + + +def test_cached_agent_reuse_calls_adopt_semantics(): + """End-to-end mirror of the production reuse block using the real helper.""" + _is_open, adopt = _import_adopt_helpers() + old_db = _MockSessionDB("old", open_=True) + new_db = _MockSessionDB("new", open_=True) + agent = _MockAgent(old_db) + _session_db = new_db + + # Mirror production: + if _session_db is not None: + _session_db = adopt(agent, _session_db) + agent._session_db = _session_db + + assert agent._session_db is old_db + assert old_db.close_calls == 0 + assert new_db.close_calls == 1 # ── 5: behavioral: LRU eviction with mock agents ──────────────────────────── @@ -192,24 +263,12 @@ def test_lru_eviction_closes_evicted_session_db(): SessionDB.close() is called.""" import collections - class MockSessionDB: - def __init__(self, name): - self.name = name - self.close_calls = 0 - def close(self): - self.close_calls += 1 - - class MockAgent: - def __init__(self, db): - self._session_db = db - cache = collections.OrderedDict() - db1, db2, db3 = MockSessionDB("a"), MockSessionDB("b"), MockSessionDB("c") - cache["sid-a"] = (MockAgent(db1), "sig1") - cache["sid-b"] = (MockAgent(db2), "sig2") - cache["sid-c"] = (MockAgent(db3), "sig3") + db1, db2, db3 = _MockSessionDB("a"), _MockSessionDB("b"), _MockSessionDB("c") + cache["sid-a"] = (_MockAgent(db1), "sig1") + cache["sid-b"] = (_MockAgent(db2), "sig2") + cache["sid-c"] = (_MockAgent(db3), "sig3") - # Mirror the production eviction path with MAX=2: MAX = 2 while len(cache) > MAX: evicted_sid, evicted_entry = cache.popitem(last=False) @@ -220,7 +279,6 @@ def test_lru_eviction_closes_evicted_session_db(): except Exception: pass - # First-inserted entry (sid-a) was evicted. assert "sid-a" not in cache assert db1.close_calls == 1, "evicted agent's SessionDB must be closed exactly once" assert db2.close_calls == 0, "remaining agents' SessionDBs must not be touched" From 798bc5fb51eb3be81e9aa99b3bb59785c261cbb3 Mon Sep 17 00:00:00 2001 From: pxxD1998 <214340659+pxxD1998@users.noreply.github.com> Date: Thu, 16 Jul 2026 22:43:19 +0800 Subject: [PATCH 09/14] fix(compression): reject stale runtime before async job --- api/routes.py | 38 +++++++++++++++++++++++ tests/test_sprint46.py | 69 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 107 insertions(+) diff --git a/api/routes.py b/api/routes.py index fc8516420..fb40b26bf 100644 --- a/api/routes.py +++ b/api/routes.py @@ -23813,6 +23813,10 @@ def _manual_compression_status_payload(job): payload["ok"] = False payload["error"] = job.get("error") or "Compression failed" payload["error_status"] = int(job.get("error_status") or 400) + if job.get("error_type"): + payload["type"] = job["error_type"] + if job.get("retryable") is not None: + payload["retryable"] = bool(job["retryable"]) elif status == "cancelled": payload["ok"] = False payload["error"] = job.get("error") or "Compression cancelled" @@ -23847,6 +23851,8 @@ def _run_manual_compression_job(sid, body): "status": "error", "error": str((payload or {}).get("error") or "Compression failed"), "error_status": status, + "error_type": (payload or {}).get("type"), + "retryable": (payload or {}).get("retryable"), "updated_at": now, } ) @@ -23858,6 +23864,21 @@ def _run_manual_compression_job(sid, body): "updated_at": now, } ) + except AgentRuntimeChangedError as exc: + logger.warning("Manual compression worker found stale Agent runtime for session %s", sid) + with _MANUAL_COMPRESSION_JOBS_LOCK: + job = _MANUAL_COMPRESSION_JOBS.get(sid) + if job: + job.update( + { + "status": "error", + "error": str(exc), + "error_status": 409, + "error_type": "agent_runtime_stale", + "retryable": True, + "updated_at": time.time(), + } + ) except Exception as exc: logger.warning("Manual compression worker failed for session %s: %s", sid, exc) with _MANUAL_COMPRESSION_JOBS_LOCK: @@ -23889,6 +23910,23 @@ def _handle_session_compress_start(handler, body): if getattr(s, "active_stream_id", None): return bad(handler, "Session is still streaming; wait for the current turn to finish.", 409) + # Reject a stale local Agent runtime before creating the asynchronous job. + # The worker has its own guarded synchronous path, but waiting for that + # worker would already mutate the job state and would collapse the typed + # retryable stale-runtime response into a generic worker error. + try: + ensure_agent_runtime_current() + except AgentRuntimeChangedError as exc: + return j( + handler, + { + "error": str(exc), + "type": "agent_runtime_stale", + "retryable": True, + }, + status=409, + ) + focus_topic = str(body.get("focus_topic") or body.get("topic") or "").strip()[:500] or None job_body = {"session_id": sid} if focus_topic: diff --git a/tests/test_sprint46.py b/tests/test_sprint46.py index ff10828e2..6dabce1bc 100644 --- a/tests/test_sprint46.py +++ b/tests/test_sprint46.py @@ -161,6 +161,75 @@ def test_session_compress_stale_runtime_returns_typed_409_before_mutation( assert loaded_after.compact() == before +def test_session_compress_start_stale_runtime_returns_409_before_job_creation( + monkeypatch, cleanup_test_sessions +): + import api.routes as routes + + sid = _make_session() + cleanup_test_sessions.append(sid) + with routes._MANUAL_COMPRESSION_JOBS_LOCK: + routes._MANUAL_COMPRESSION_JOBS.pop(sid, None) + + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: (_ for _ in ()).throw( + routes.AgentRuntimeChangedError("restart required") + ), + ) + + handler = _FakeHandler() + routes._handle_session_compress_start(handler, {"session_id": sid}) + + assert handler.status == 409 + assert json.loads(handler.wfile.getvalue().decode("utf-8")) == { + "error": "restart required", + "type": "agent_runtime_stale", + "retryable": True, + } + with routes._MANUAL_COMPRESSION_JOBS_LOCK: + assert sid not in routes._MANUAL_COMPRESSION_JOBS + + +def test_session_compress_worker_preserves_stale_runtime_taxonomy(monkeypatch): + import api.routes as routes + + sid = "stale-worker-session" + started_at = time.time() + with routes._MANUAL_COMPRESSION_JOBS_LOCK: + routes._MANUAL_COMPRESSION_JOBS[sid] = { + "session_id": sid, + "status": "running", + "started_at": started_at, + "updated_at": started_at, + } + + monkeypatch.setattr( + routes, + "_handle_session_compress", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + routes.AgentRuntimeChangedError("restart required") + ), + ) + + routes._run_manual_compression_job(sid, {"session_id": sid}) + + handler = _FakeHandler() + routes._handle_session_compress_status(handler, sid) + payload = handler.payload() + assert handler.status == 200 + assert payload["ok"] is False + assert payload["status"] == "error" + assert payload["session_id"] == sid + assert payload["error"] == "restart required" + assert payload["error_status"] == 409 + assert payload["type"] == "agent_runtime_stale" + assert payload["retryable"] is True + with routes._MANUAL_COMPRESSION_JOBS_LOCK: + routes._MANUAL_COMPRESSION_JOBS.pop(sid, None) + + def test_session_compress_roundtrip(monkeypatch, cleanup_test_sessions): created = cleanup_test_sessions original_messages = [ From 7d2ca8cabb7209267c7cdd1e081b710fe74c47dc Mon Sep 17 00:00:00 2001 From: carlotestor Date: Thu, 16 Jul 2026 16:53:42 +0200 Subject: [PATCH 10/14] fix: address greptile review on SessionDB adopt helper - Log failed closes in _adopt_session_db_for_cached_agent with logger.debug(..., exc_info=True) instead of bare pass, matching _replace_session_db_in_kwargs so a failed close (EMFILE risk) is diagnosable. - Drop the fragile exec/source-slice fallback in the test helper import; real import or pytest.skip. Source-level pins still cover reverts without importing streaming. --- api/streaming.py | 12 +++++++-- tests/test_v050259_sessiondb_fd_leak.py | 35 +++++++++++-------------- 2 files changed, 26 insertions(+), 21 deletions(-) diff --git a/api/streaming.py b/api/streaming.py index 192c0d4ea..55c51d0e6 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -6305,13 +6305,21 @@ def _adopt_session_db_for_cached_agent(agent, new_session_db): try: new_session_db.close() except Exception: - pass + # Same observability as _replace_session_db_in_kwargs: a failed + # close here reintroduces the EMFILE pressure PR #1421 fixed. + logger.debug( + "Failed to close unused session_db handle in adopt helper", + exc_info=True, + ) return existing if existing is not None: try: existing.close() except Exception: - pass + logger.debug( + "Failed to close previous session_db handle in adopt helper", + exc_info=True, + ) agent._session_db = new_session_db return new_session_db diff --git a/tests/test_v050259_sessiondb_fd_leak.py b/tests/test_v050259_sessiondb_fd_leak.py index 7b17212d9..6ad48124c 100644 --- a/tests/test_v050259_sessiondb_fd_leak.py +++ b/tests/test_v050259_sessiondb_fd_leak.py @@ -72,6 +72,13 @@ def test_adopt_and_is_open_helpers_exist(): "_replace_session_db_in_kwargs must guard on _session_db_is_open so " "credential self-heal cannot close a handle live subagents share" ) + # adopt helper must log failed closes (not bare `pass`) so EMFILE pressure + # from a failed close is diagnosable — matches _replace_session_db_in_kwargs. + adopt_idx = src.find("def _adopt_session_db_for_cached_agent") + assert adopt_idx != -1 + adopt_block = src[adopt_idx : adopt_idx + 1800] + assert 'Failed to close unused session_db handle in adopt helper' in adopt_block + assert "logger.debug" in adopt_block # ── 2: source-level pin: LRU eviction path still closes _session_db ──────── @@ -161,32 +168,22 @@ class _MockAgent: def _import_adopt_helpers(): - """Import the helpers without pulling in the whole streaming module graph. + """Import the production helpers. - streaming.py has heavy import-time side effects; exec the helper bodies - from source so unit tests stay hermetic. + No source-slicing / exec fallback: that path breaks as soon as the helpers + reference module-level names (e.g. ``logger.debug``) or another def is + inserted between the markers. Prefer a real import; skip the behavioral + suite when the package cannot be imported (CI without full deps). + Source-level pins above still catch reverts without importing streaming. """ - import importlib.util - import types - - # Prefer real import when the package is importable. try: from api.streaming import ( # type: ignore _adopt_session_db_for_cached_agent, _session_db_is_open, ) - return _session_db_is_open, _adopt_session_db_for_cached_agent - except Exception: - pass - - src = (REPO / "api" / "streaming.py").read_text(encoding="utf-8") - # Extract just the two helper defs by line slicing between markers. - start = src.index("def _session_db_is_open") - end = src.index("def _build_session_db_for_stream") - blob = src[start:end] - ns: dict = {} - exec(blob, ns) # noqa: S102 — test-only extraction of pure helpers - return ns["_session_db_is_open"], ns["_adopt_session_db_for_cached_agent"] + except Exception as exc: + pytest.skip(f"api.streaming helpers not importable: {exc}") + return _session_db_is_open, _adopt_session_db_for_cached_agent def test_adopt_reuses_open_session_db_and_closes_new(): From aae2830459b893dfba9d2864911f1a2393eecc5b Mon Sep 17 00:00:00 2001 From: pxxD1998 <214340659+pxxD1998@users.noreply.github.com> Date: Thu, 16 Jul 2026 22:56:41 +0800 Subject: [PATCH 11/14] fix(compression): preserve running job reuse --- api/routes.py | 27 +++++++++++----- tests/test_sprint46.py | 71 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 8 deletions(-) diff --git a/api/routes.py b/api/routes.py index fb40b26bf..58b93f6e7 100644 --- a/api/routes.py +++ b/api/routes.py @@ -23910,10 +23910,24 @@ def _handle_session_compress_start(handler, body): if getattr(s, "active_stream_id", None): return bad(handler, "Session is still streaming; wait for the current turn to finish.", 409) + focus_topic = str(body.get("focus_topic") or body.get("topic") or "").strip()[:500] or None + job_body = {"session_id": sid} + if focus_topic: + job_body["focus_topic"] = focus_topic + + # Repeated start requests observe an existing running job and do not admit + # new Agent work, so preserve that idempotent read even if the checkout has + # since changed. Do not hold the job lock while running Git subprocesses. + now = time.time() + with _MANUAL_COMPRESSION_JOBS_LOCK: + _manual_compression_cleanup_locked(now) + existing = _MANUAL_COMPRESSION_JOBS.get(sid) + if existing: + existing_payload = _manual_compression_status_payload(existing) + if existing_payload.get("status") == "running": + return j(handler, existing_payload) + # Reject a stale local Agent runtime before creating the asynchronous job. - # The worker has its own guarded synchronous path, but waiting for that - # worker would already mutate the job state and would collapse the typed - # retryable stale-runtime response into a generic worker error. try: ensure_agent_runtime_current() except AgentRuntimeChangedError as exc: @@ -23927,11 +23941,8 @@ def _handle_session_compress_start(handler, body): status=409, ) - focus_topic = str(body.get("focus_topic") or body.get("topic") or "").strip()[:500] or None - job_body = {"session_id": sid} - if focus_topic: - job_body["focus_topic"] = focus_topic - + # Another start request may have admitted a job while the runtime check ran. + # Re-check under the lock so only one worker is created. now = time.time() with _MANUAL_COMPRESSION_JOBS_LOCK: _manual_compression_cleanup_locked(now) diff --git a/tests/test_sprint46.py b/tests/test_sprint46.py index 6dabce1bc..9ca5bed66 100644 --- a/tests/test_sprint46.py +++ b/tests/test_sprint46.py @@ -192,6 +192,77 @@ def test_session_compress_start_stale_runtime_returns_409_before_job_creation( assert sid not in routes._MANUAL_COMPRESSION_JOBS +def test_session_compress_start_reuses_running_job_when_runtime_is_stale( + monkeypatch, cleanup_test_sessions +): + import api.routes as routes + + sid = _make_session() + cleanup_test_sessions.append(sid) + existing = { + "session_id": sid, + "focus_topic": "already running", + "status": "running", + "started_at": time.time(), + "updated_at": time.time(), + } + with routes._MANUAL_COMPRESSION_JOBS_LOCK: + routes._MANUAL_COMPRESSION_JOBS[sid] = existing + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: (_ for _ in ()).throw( + routes.AgentRuntimeChangedError("restart required") + ), + ) + + handler = _FakeHandler() + routes._handle_session_compress_start(handler, {"session_id": sid}) + + assert handler.status == 200 + payload = handler.payload() + assert payload["status"] == "running" + assert payload["session_id"] == sid + assert payload["focus_topic"] == "already running" + with routes._MANUAL_COMPRESSION_JOBS_LOCK: + assert routes._MANUAL_COMPRESSION_JOBS[sid] is existing + routes._MANUAL_COMPRESSION_JOBS.pop(sid, None) + + +def test_session_compress_start_rechecks_job_after_runtime_barrier( + monkeypatch, cleanup_test_sessions +): + import api.routes as routes + + sid = _make_session() + cleanup_test_sessions.append(sid) + admitted = { + "session_id": sid, + "focus_topic": "admitted concurrently", + "status": "running", + "started_at": time.time(), + "updated_at": time.time(), + } + + def barrier_then_admit(): + with routes._MANUAL_COMPRESSION_JOBS_LOCK: + routes._MANUAL_COMPRESSION_JOBS[sid] = admitted + + def reject_duplicate_worker(**_kwargs): + raise AssertionError("duplicate worker admitted") + + monkeypatch.setattr(routes, "ensure_agent_runtime_current", barrier_then_admit) + monkeypatch.setattr(routes.threading, "Thread", reject_duplicate_worker) + + handler = _FakeHandler() + routes._handle_session_compress_start(handler, {"session_id": sid}) + + assert handler.status == 200 + assert handler.payload()["status"] == "running" + with routes._MANUAL_COMPRESSION_JOBS_LOCK: + routes._MANUAL_COMPRESSION_JOBS.pop(sid, None) + + def test_session_compress_worker_preserves_stale_runtime_taxonomy(monkeypatch): import api.routes as routes From df0c204d734dbdb0dda42f1fd12af25a870b9f07 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Thu, 16 Jul 2026 08:19:43 -0700 Subject: [PATCH 12/14] docs(rfc): reconcile Live-to-Final assistant-reply contracts with shipped behavior (#6144) (#6147) Docs-only reconciliation of the Stable Assistant Turn Anchors / Live-to-Final RFCs + CONTRACTS.md + Phase 0 inventory from proposed/scaffold to accepted/implemented framing. No runtime/CI change; one comment-header fix in static/assistant_turn_anchors.js. Remaining hardening tracked under #3400. Co-authored-by: nesquena-hermes Co-authored-by: franksong2702 --- CHANGELOG.md | 4 ++ docs/CONTRACTS.md | 14 +++++-- .../stable-assistant-turn-anchor-phase0.md | 16 ++++--- docs/rfcs/README.md | 17 ++++---- docs/rfcs/live-to-final-assistant-replies.md | 42 ++++++++++++++++++- docs/rfcs/stable-assistant-turn-anchors.md | 35 +++++++++++++++- docs/rfcs/transparent-stream-activity-mode.md | 31 +++++++++++--- .../webui-run-state-consistency-contract.md | 18 +++++++- static/assistant_turn_anchors.js | 10 ++--- 9 files changed, 156 insertions(+), 31 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bec8aa28..c402851f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,10 @@ ## [Unreleased] +### Changed + +- **Docs: reconciled the assistant-reply lifecycle RFCs/contracts with shipped behavior.** The Stable Assistant Turn Anchors and Live-to-Final RFCs, `docs/CONTRACTS.md`, and the Phase 0 architecture inventory were updated from "proposed/scaffold" framing to "accepted/implemented" to match what actually ships (single assistant-turn owner projecting one `activity_scene_v1` into Compact Worklog / Transparent Stream / Final answer). Documentation-only — no runtime behavior or CI-gate change; the one code touch is a corrected comment header in `static/assistant_turn_anchors.js`. Remaining hardening stays tracked under #3400. Thanks @franksong2702. (#6144) + ### Fixed - **Models selected from a list-shaped custom-provider allowlist now route to the right provider.** A `custom_providers[].models` allowlist can be written as either a mapping or a list, and the model picker accepted both — but runtime routing only read mapping keys, so a model chosen from a *list*-shaped allowlist could silently stay on the active/default provider instead of its own. The supported-model-ID extraction is now centralized in one helper used by both catalog construction and routing, so the picker and runtime can no longer drift. Existing mapping-shaped and list-of-strings configs are unaffected. Thanks @franksong2702. (#6127, #6121) diff --git a/docs/CONTRACTS.md b/docs/CONTRACTS.md index 0094a5c28..e7bcc6fc3 100644 --- a/docs/CONTRACTS.md +++ b/docs/CONTRACTS.md @@ -29,13 +29,19 @@ contributor guidance; it does not change runtime behavior or CI gates. repairs. Start here for narrow fixes that keep the existing WebUI execution path. - [`docs/rfcs/live-to-final-assistant-replies.md`](rfcs/live-to-final-assistant-replies.md): - proposed product model for long-running assistant replies, live process text, + accepted product model for long-running assistant replies, live process text, tool activity, recovery, terminal outcomes, and final-answer boundaries. Start here for UI/UX changes to running-session assistant reply rendering. +- [`docs/rfcs/stable-assistant-turn-anchors.md`](rfcs/stable-assistant-turn-anchors.md): + implemented presentation/reconciliation model that attaches live, settled, + replayed, and recovered activity to one assistant-turn owner and projects one + `activity_scene_v1` into Compact Worklog, Transparent Stream, or Final answer + only. Remaining hardening stays tracked under #3400. - [`docs/architecture/stable-assistant-turn-anchor-phase0.md`](architecture/stable-assistant-turn-anchor-phase0.md): - current Phase 0 inventory for the Stable Assistant Turn Anchors work under - #3926. Use this before wiring anchor helpers into live SSE, replay, - settlement, `INFLIGHT`, or `renderMessages()` paths. + cumulative implementation inventory for the Stable Assistant Turn Anchors + work under #3926. Use it to distinguish shipped wiring from historical slice + boundaries before changing live SSE, replay, settlement, `INFLIGHT`, or + `renderMessages()` paths. - [`docs/rfcs/canonical-session-resolution.md`](rfcs/canonical-session-resolution.md): proposed contract for resolving URL routes, query parameters, localStorage, sidebar rows, and compression-lineage IDs to one canonical visible session diff --git a/docs/architecture/stable-assistant-turn-anchor-phase0.md b/docs/architecture/stable-assistant-turn-anchor-phase0.md index b09ef65ae..ceea20232 100644 --- a/docs/architecture/stable-assistant-turn-anchor-phase0.md +++ b/docs/architecture/stable-assistant-turn-anchor-phase0.md @@ -1,12 +1,16 @@ # Stable Assistant Turn Anchors Phase 0 Inventory -This inventory implements the first non-visual slice of -[`stable-assistant-turn-anchors.md`](../rfcs/stable-assistant-turn-anchors.md). -It documents the current per-turn state layers and the event-shape contract that -future anchor phases must consume. It does not claim that anchors are wired into -streaming or rendering yet. +This inventory began as the first non-visual slice of +[`stable-assistant-turn-anchors.md`](../rfcs/stable-assistant-turn-anchors.md) and +now records the shipped phase progression. Current `master` wires Anchor-owned +`activity_scene_v1` data into live and settled Compact Worklog, Transparent +Stream, journal hydration, session re-entry, and transcript-backed persistence. -## RFC Phase Progress +The slice-by-slice sections below are retained as implementation history. +Present-tense statements such as “unwired” or “future phase” describe the point +when that slice landed; they do not override the current coverage summary above. + +## Shipped RFC Phase Progress - The #3962 Phase 0 scaffold shipped through #3977 / v0.51.359: inventory the current state layers, encode the owner seed, and pin the source classification diff --git a/docs/rfcs/README.md b/docs/rfcs/README.md index 921bb2d45..d728461d1 100644 --- a/docs/rfcs/README.md +++ b/docs/rfcs/README.md @@ -47,16 +47,17 @@ First-time contributor RFCs should be discussed in an issue before opening a PR. replay, compression, and session metadata coherent during active and recovered WebUI runs. - [`live-to-final-assistant-replies.md`](live-to-final-assistant-replies.md) - — #3400 product model for long-running assistant replies, live process prose, - tool activity, recovery, terminal outcomes, and the final-answer boundary. + — #3400 accepted product model for long-running assistant replies, live + process prose, tool activity, recovery, terminal outcomes, display + projections, and the final-answer boundary. - [`transparent-stream-activity-mode.md`](transparent-stream-activity-mode.md) - — #3820 opt-in display mode for power users who need a transparent, - chronological Thinking / progress / tool-call stream alongside the default - Compact Worklog model. + — #3820 implemented opt-in display mode for power users who need a + transparent, chronological Thinking / progress / tool-call stream alongside + the default Compact Worklog and opt-in Final answer only projections. - [`stable-assistant-turn-anchors.md`](stable-assistant-turn-anchors.md) — #3926 - proposed frontend presentation/reconciliation model for anchoring live - assistant activity, settled final answers, replay, and Compact/Transparent - render modes to one assistant turn. + implemented frontend presentation/reconciliation model for anchoring live + assistant activity, settled final answers, replay, and all activity display + modes to one assistant turn; remaining hardening is tracked under #3400. - [`canonical-session-resolution.md`](canonical-session-resolution.md) — #2361 focused contract for resolving URL, query parameter, localStorage, sidebar, and compression-lineage session IDs to one canonical visible chat target. diff --git a/docs/rfcs/live-to-final-assistant-replies.md b/docs/rfcs/live-to-final-assistant-replies.md index 68e8093dd..6564390e8 100644 --- a/docs/rfcs/live-to-final-assistant-replies.md +++ b/docs/rfcs/live-to-final-assistant-replies.md @@ -3,6 +3,7 @@ - **Status:** Accepted (parent contract; implementation tracked in [#3400](https://github.com/nesquena/hermes-webui/issues/3400)) - **Author:** @franksong2702 - **Created:** 2026-06-03 +- **Updated:** 2026-07-16 - **Tracking issue:** [#3400](https://github.com/nesquena/hermes-webui/issues/3400) ## Background: Long-Running Sessions Are The Anchor @@ -204,6 +205,43 @@ Requirements: - Very long final answers remain complete and readable. They should not be hidden inside the activity summary or replaced by a progress/status artifact. +### Activity display and disclosure addenda + +The accepted lifecycle now has three presentation strategies over the same +assistant-turn activity data: + +- **Compact Worklog** remains the default. Its top-level Worklog is expanded + while live so progress remains visible, and the user may collapse or reopen it. + Nested tool/reasoning detail remains progressively disclosed. +- **Transparent Stream** is opt-in and renders the same ordered activity as + chronological rows. It does not create a second live or settled owner. +- **Final answer only** is opt-in (`hide_all_activity` in persisted settings). + It suppresses activity rows without deleting the persisted Anchor scene or + changing the final-answer owner. + +For normal completed turns, the settled Compact Worklog remains collapsed by +default. When a terminal error-family turn has actual Worklog content, the +Worklog defaults open so readable partial work is not hidden behind the error +outcome. The current disclosure error family is `error`, `no_response`, +`degraded`, `connection_lost`, `tool_limit_reached`, and +`compression_exhausted`; cancelled turns and the parent `interrupted` terminal +outcome keep their separate semantics. An explicit user disclosure choice wins +over these defaults and may be restored across a render rebuild. + +`degraded` and `connection_lost` are Anchor-level reconstruction/transport +outcomes defined in +[`stable-assistant-turn-anchors.md`](stable-assistant-turn-anchors.md), not +additional canonical product states in the table below. In this parent +contract, `connection_lost` is the transport-specific Anchor form of an +interruption, while `degraded` is the explicit recovery-state counterpart to the +restoring/degraded path. They use error-family disclosure only because they can +leave readable partial Worklog content without a normal final answer. + +These are presentation and disclosure rules only. They do not change reply +ownership, terminal classification, durable transcript truth, or the requirement +that all three strategies converge across live, settle, reload, session switch, +and reconnect. + ### Recovery and replay Refresh, reconnect, session switching, and replay should preserve the same @@ -473,7 +511,9 @@ for current open/merged/superseded status. | Track | Scope | Current vehicle | | --- | --- | --- | | Parent product RFC | Define the long-running live-to-final assistant reply lifecycle and review checklist. | This RFC; tracking issue [#3400](https://github.com/nesquena/hermes-webui/issues/3400). | -| First reply lifecycle implementation | Live process prose, quiet tool activity, settled activity summary above final answer, replay/reattach consistency, live-only compression status, supporting stream ownership fixes. | [#3401](https://github.com/nesquena/hermes-webui/pull/3401). | +| First reply lifecycle implementation | Live process prose, quiet tool activity, settled activity summary above final answer, replay/reattach consistency, live-only compression status, supporting stream ownership fixes. | [#3401](https://github.com/nesquena/hermes-webui/pull/3401), absorbed through [#3741](https://github.com/nesquena/hermes-webui/pull/3741). | +| Activity display projections | Compact Worklog by default, opt-in Transparent Stream, and opt-in Final answer only over the same assistant-turn data. | [#3820](https://github.com/nesquena/hermes-webui/issues/3820), [`transparent-stream-activity-mode.md`](transparent-stream-activity-mode.md), tracking issue [#3400](https://github.com/nesquena/hermes-webui/issues/3400). | +| Assistant-turn presentation ownership | Normalize live, settled, replayed, and recovered activity into one Anchor / `activity_scene_v1`, with legacy rendering limited to historical or non-anchor compatibility. | [#3926](https://github.com/nesquena/hermes-webui/issues/3926), [`stable-assistant-turn-anchors.md`](stable-assistant-turn-anchors.md); core implementation shipped through the #4411/#4564 and #5242/#5243 capstones. | | Terminal/no-final stabilization | Compression exhausted, tool-tail/no-final transcript shape, context-compaction marker suppression, terminal error routing. | [#3315](https://github.com/nesquena/hermes-webui/issues/3315), [#3316](https://github.com/nesquena/hermes-webui/pull/3316). | | Cancel ownership hardening | Frontend cancel should close its own SSE source and clear only its own busy state. | [#3344](https://github.com/nesquena/hermes-webui/issues/3344), [#3345](https://github.com/nesquena/hermes-webui/pull/3345). | | Early-cancel startup race | Backend cancel should still interrupt the worker when the SSE registry detached before startup fully settled. | [#3475](https://github.com/nesquena/hermes-webui/issues/3475), [#3476](https://github.com/nesquena/hermes-webui/pull/3476). | diff --git a/docs/rfcs/stable-assistant-turn-anchors.md b/docs/rfcs/stable-assistant-turn-anchors.md index 7fea9e301..9c522519b 100644 --- a/docs/rfcs/stable-assistant-turn-anchors.md +++ b/docs/rfcs/stable-assistant-turn-anchors.md @@ -1,12 +1,35 @@ # Stable Assistant Turn Anchors for Live-to-Final Rendering -- **Status:** Proposed +- **Status:** Implemented - **Author:** @franksong2702 - **Created:** 2026-06-10 +- **Updated:** 2026-07-16 - **Tracking issue:** [#3926](https://github.com/nesquena/hermes-webui/issues/3926) - **Parent contract:** [Live-to-Final Assistant Replies](./live-to-final-assistant-replies.md) ([#3400](https://github.com/nesquena/hermes-webui/issues/3400)) - **Related RFCs:** [Transparent Stream](./transparent-stream-activity-mode.md), [Hermes Run Adapter Contract](./hermes-run-adapter-contract.md), [WebUI Run State Consistency Contract](./webui-run-state-consistency-contract.md), [Turn Journal](./turn-journal.md), [Pending Intent Controls](./webui-pending-intent-controls.md) +## Implementation Status + +The RFC's core presentation/reconciliation path is implemented. Current +`master` includes the event normalizer and registry, settled final-answer +projection, ordered `activity_scene_v1`, live Compact Worklog rendering, +Transparent Stream rendering, run-journal hydration, session re-entry, +transcript-backed scene persistence, and explicit fallback ownership for +historical or non-anchor transcripts. + +`Implemented` does not mean every adjacent hardening item is complete. The +remaining work is tracked under [#3400](https://github.com/nesquena/hermes-webui/issues/3400) +and should land as separate slices: + +- replace presentation-layer text/prefix de-echo heuristics with provenance or + event identity without deleting legitimate repeated model output; +- make newer journal/replay evidence explicitly outrank stale `INFLIGHT` + first-paint state; +- finish stable `run_id` versus transport `stream_id` separation; +- complete the Phase 6 Artifact/side-effect ownership path; +- remove each legacy fallback only after its historical transcript shape can + hydrate an Anchor reliably. + ## Problem The Live-to-Final redesign made the user-facing product model clearer: @@ -763,6 +786,11 @@ Implementation PRs may combine adjacent low-risk phases when they preserve behavior and include coverage. Settlement, reconstruction, and display-mode changes should remain independently reviewable. +Implementation reconciliation (2026-07-16): the core Phase 0-5 path has shipped. +Phase 6 and the hardening items listed in **Implementation Status** remain +follow-up work; the phase descriptions below are retained as the rollout +contract that produced the current implementation. + ### Phase 0: RFC and inventory - Land this RFC as design guidance. @@ -902,6 +930,11 @@ Any implementation PR against this RFC should answer: ## Open Questions +The first-slice and Transparent Stream sequencing questions below are retained +as historical implementation rationale. Current open hardening questions are +identity fallback depth, Artifact/side-effect ownership, and how far active-turn +DOM rebuilds can be reduced without weakening recovery compatibility. + ### First implementation slice size Should the first implementation PR stop at hidden anchor scaffolding, or should diff --git a/docs/rfcs/transparent-stream-activity-mode.md b/docs/rfcs/transparent-stream-activity-mode.md index c844b5781..9909c634b 100644 --- a/docs/rfcs/transparent-stream-activity-mode.md +++ b/docs/rfcs/transparent-stream-activity-mode.md @@ -1,12 +1,31 @@ # Transparent Stream — A Chronological Activity Display Mode -- **Status:** Accepted (direction confirmed by @nesquena; RFC merged via [#3862](https://github.com/nesquena/hermes-webui/pull/3862)) +- **Status:** Implemented (opt-in display mode; follow-up hardening tracked in [#3400](https://github.com/nesquena/hermes-webui/issues/3400)) - **Author:** @franksong2702 - **Created:** 2026-06-09 -- **Updated:** 2026-06-09 +- **Updated:** 2026-07-16 - **Tracking issue:** [#3820](https://github.com/nesquena/hermes-webui/issues/3820) - **Parent contract:** [Live-to-Final Assistant Replies](./live-to-final-assistant-replies.md) ([#3400](https://github.com/nesquena/hermes-webui/issues/3400) / #3401 / #3741) +## Current projection family + +The shipped `chat_activity_display_mode` setting now has three explicit values: + +- `compact_worklog` — default, results-first presentation; +- `transparent_stream` — opt-in chronological activity rows; +- `hide_all_activity` — opt-in **Final answer only** presentation. + +All three are render strategies over the same Assistant Turn Anchor and +`activity_scene_v1`; they do not create separate activity ownership or change +the final-answer boundary. Final answer only suppresses activity presentation +while preserving the persisted scene so another mode, settle, or reload can +reconstruct the same turn. + +The original proposal and rollout notes below are retained as design history. +Statements describing a missing selector, missing settled branch, or unwired +live/replay renderer describe the tree when this RFC was written, not current +`master`. + ## Problem After the live-to-final redesign (#3401/#3741), the execution trace of a turn @@ -104,21 +123,22 @@ Plus two boundaries: - **Not full transcript virtualization.** Performance hardening for very long transparent transcripts (#3714) is acknowledged but scoped to a later slice. -## Proposal +## Implemented proposal ### 1. A new, dedicated preference -A new preference, independent of `simplified_tool_calling`: +The shipped preference is independent of `simplified_tool_calling`: ```python # api/config.py preferences -"chat_activity_display_mode": "compact_worklog", # | "transparent_stream" +"chat_activity_display_mode": "compact_worklog", # | "transparent_stream" | "hide_all_activity" ``` - Default `compact_worklog` — existing users see no change. - Surfaced in Settings as an explicit **segmented choice** (not a checkbox): - **Compact Worklog** — default; results-first, supporting activity quiet. - **Transparent Stream** — advanced; every step shown chronologically. + - **Final answer only** — activity hidden; final answer remains visible. - The legacy `Compact tool activity` checkbox is marked deprecated. It is not the seam for this feature. @@ -128,6 +148,7 @@ predicate used everywhere: ```js function chatActivityMode(){ return window._chatActivityDisplayMode || 'compact_worklog'; } function isTransparentStream(){ return chatActivityMode() === 'transparent_stream'; } +function isFinalAnswerOnlyMode(){ return chatActivityMode() === 'hide_all_activity'; } ``` ### 2. Decouple "event sequence" from "grouping strategy" diff --git a/docs/rfcs/webui-run-state-consistency-contract.md b/docs/rfcs/webui-run-state-consistency-contract.md index 6a3c839fb..a14bc5205 100644 --- a/docs/rfcs/webui-run-state-consistency-contract.md +++ b/docs/rfcs/webui-run-state-consistency-contract.md @@ -3,8 +3,9 @@ - **Status:** Proposed - **Author:** @franksong2702 - **Created:** 2026-05-16 +- **Updated:** 2026-07-16 - **Tracking issue:** [#2361](https://github.com/nesquena/hermes-webui/issues/2361) -- **Related architecture:** [#1925](https://github.com/nesquena/hermes-webui/issues/1925), [`hermes-run-adapter-contract.md`](hermes-run-adapter-contract.md) +- **Related architecture:** [#1925](https://github.com/nesquena/hermes-webui/issues/1925), [`hermes-run-adapter-contract.md`](hermes-run-adapter-contract.md), [`stable-assistant-turn-anchors.md`](stable-assistant-turn-anchors.md) ## Problem @@ -46,6 +47,21 @@ while WebUI still has multiple overlapping state stores. - Do not reopen already-fixed narrow bugs. - Do not make this a catch-all for unrelated UI polish. +## Current implementation relationship + +Stable Assistant Turn Anchors now implement the presentation/reconciliation +portion of this contract for one assistant turn. The run journal and settled +transcript provide durable observations; the Anchor registry and +`activity_scene_v1` reconcile those observations into Compact Worklog, +Transparent Stream, or Final answer only; `S.messages`, `INFLIGHT`, renderer +caches, and DOM remain projections or recovery caches rather than independent +semantic owners. + +This RFC remains `Proposed` because its broader cross-layer contract also covers +model-context reconstruction, compression handoff, session metadata, and future +runtime-adapter migration. Shipped Anchor coverage strengthens invariants 2, 3, +and 5; it does not mark every run-state boundary implemented. + ## State Layers | Layer | Purpose | Source-of-truth expectation | Must not do | diff --git a/static/assistant_turn_anchors.js b/static/assistant_turn_anchors.js index 7c11ca1f5..17c69ba0f 100644 --- a/static/assistant_turn_anchors.js +++ b/static/assistant_turn_anchors.js @@ -1,9 +1,9 @@ -// Stable Assistant Turn Anchors scaffold (#3926). +// Stable Assistant Turn Anchors implementation (#3926 / #3400). // -// This file defines the current ownership inventory, event classifications, and -// small owner helpers. It does not register anchors globally. The only renderer -// wiring in this slice is settled assistant final-answer projection; live -// streaming, replay hydration, tools, and DOM ownership remain unwired. +// This file defines the ownership inventory, event normalization, registry, and +// activity-scene projection consumed by current live, settled, replay-hydration, +// and session-reentry paths. Runtime execution and transport remain outside the +// Anchor's presentation/reconciliation ownership boundary. (function(){ const ROOT=(typeof window!=='undefined')?window:globalThis; From dcdcfb99776f5e419563ae94a52e45101c296218 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Thu, 16 Jul 2026 19:14:53 +0000 Subject: [PATCH 13/14] fix(streaming): degrade self-heal SessionDB to None on failed rebuild Codex gate on #6143 found a SILENT regression the reuse-path fix introduced: _replace_session_db_in_kwargs early-returned the OLD handle when the rebuild failed, but master assigned None. If the old handle was already CLOSED, both credential self-heal callers (streaming.py:9032/10253) then built a fresh agent around a closed SessionDB -> every persist/search fails with 'NoneType has no attribute execute' while the chat keeps going. Guard on _session_db_is_open: keep a live handle (subagents hold it by ref), else degrade to None (clean lazy reinit, master behaviour). +2 regression tests. Co-authored-by: carlotestor --- api/streaming.py | 11 +++++- tests/test_v050259_sessiondb_fd_leak.py | 52 +++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 1 deletion(-) diff --git a/api/streaming.py b/api/streaming.py index 55c51d0e6..5e222a5f1 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -6350,7 +6350,16 @@ def _replace_session_db_in_kwargs(agent_kwargs, state_db_path): _old_session_db = agent_kwargs.get("session_db") _next_session_db = _build_session_db_for_stream(state_db_path) if _next_session_db is None: - return _old_session_db + # Replacement construction failed. Keep the prior handle only if it is + # still open (live subagents may hold it by reference); otherwise + # degrade cleanly to None — as master did — so the rebuilt agent lazily + # reinitialises its SessionDB instead of reusing a closed handle and + # failing every persist/search with + # "'NoneType' object has no attribute 'execute'". + if _session_db_is_open(_old_session_db): + return _old_session_db + agent_kwargs["session_db"] = None + return None if _session_db_is_open(_old_session_db): # Keep the live handle; discard the unused new one. try: diff --git a/tests/test_v050259_sessiondb_fd_leak.py b/tests/test_v050259_sessiondb_fd_leak.py index 6ad48124c..12bfd90e4 100644 --- a/tests/test_v050259_sessiondb_fd_leak.py +++ b/tests/test_v050259_sessiondb_fd_leak.py @@ -280,3 +280,55 @@ def test_lru_eviction_closes_evicted_session_db(): assert db1.close_calls == 1, "evicted agent's SessionDB must be closed exactly once" assert db2.close_calls == 0, "remaining agents' SessionDBs must not be touched" assert db3.close_calls == 0 + + +# ── 6: self-heal path must not reuse a CLOSED handle when the rebuild fails ── + + +def _import_replace_helper(): + """Import the real credential-self-heal SessionDB replacer.""" + try: + from api.streaming import _replace_session_db_in_kwargs # type: ignore + except Exception as exc: + pytest.skip(f"api.streaming not importable: {exc}") + return _replace_session_db_in_kwargs + + +def test_replace_degrades_to_none_when_rebuild_fails_and_old_is_closed(monkeypatch): + """Credential self-heal regression (Codex gate finding on PR #6143). + + When ``_build_session_db_for_stream`` returns None (rebuild failed) AND the + prior handle is already CLOSED, ``_replace_session_db_in_kwargs`` must leave + ``agent_kwargs['session_db'] = None`` — as master did — so the rebuilt agent + lazily reinitialises. Retaining the closed handle (the pre-fix behaviour) + makes every persist/search fail with + ``'NoneType' object has no attribute 'execute'`` while the chat continues. + """ + import api.streaming as streaming + + _replace = _import_replace_helper() + monkeypatch.setattr(streaming, "_build_session_db_for_stream", lambda _p: None) + + old_db = _MockSessionDB("old", open_=False) # already closed + kwargs = {"session_db": old_db} + result = _replace(kwargs, "/tmp/does-not-matter.db") + + assert result is None, "must not hand back a closed handle when rebuild fails" + assert kwargs["session_db"] is None, "kwargs must degrade to None (clean lazy reinit)" + + +def test_replace_keeps_open_handle_when_rebuild_fails(monkeypatch): + """Inverse: a still-OPEN prior handle (held by live subagents) is retained + when the rebuild fails — do not orphan a live shared connection.""" + import api.streaming as streaming + + _replace = _import_replace_helper() + monkeypatch.setattr(streaming, "_build_session_db_for_stream", lambda _p: None) + + old_db = _MockSessionDB("old", open_=True) # still live (subagents hold it) + kwargs = {"session_db": old_db} + result = _replace(kwargs, "/tmp/does-not-matter.db") + + assert result is old_db, "a live handle must be kept when the rebuild fails" + assert kwargs["session_db"] is old_db + assert old_db.close_calls == 0, "must not close a live shared handle" From 11773af0e3e4fcdcdbb1fde7a73c1c97ef208430 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Thu, 16 Jul 2026 13:17:42 -0700 Subject: [PATCH 14/14] docs(changelog): add #6143 (SessionDB subagent use-after-close) + #6140 (CLI-cache LRU cap) bullets (#6158) Co-authored-by: nesquena-hermes --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index c402851f2..a7c98a350 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ ### Fixed +- **Background subagents no longer fail to persist when the parent's cached agent is reused mid-turn.** A parent session's `SessionDB` handle is shared by reference with the background subagents it spawns (`delegate_tool`), but the cached-agent reuse path unconditionally closed that handle when the parent got a new turn / server-side wakeup — so every still-running child then failed `append_message` with `'NoneType' object has no attribute 'execute'` (observed 600+ times under one parent). The reuse path now keeps a still-open handle (closing only the unused fresh one, preserving the #1421 FD-leak fix) and only replaces a dead/missing one. The credential self-heal path was hardened the same way: if rebuilding the handle fails it degrades cleanly to a lazy reinit instead of reusing a closed handle. Thanks @carlotestor. (#6143) + +- **The CLI/cron session-list projection cache is now bounded.** `_CLI_SESSIONS_CACHE` was an unbounded plain dict of deep-copied session-list projections whose fingerprint-folded key advances on every streamed message, so orphaned heavy copies accumulated on a long-lived server until the next structural clear. It's now an `OrderedDict` capped at 8 entries with drop-oldest LRU eviction (recency refresh on hit), mirroring the existing `_CLAUDE_CODE_PARSE_CACHE` / `_SIDECAR_METADATA_CACHE` pattern; TTL stays the freshness control. Thanks @rh-id. (#6140) + - **Models selected from a list-shaped custom-provider allowlist now route to the right provider.** A `custom_providers[].models` allowlist can be written as either a mapping or a list, and the model picker accepted both — but runtime routing only read mapping keys, so a model chosen from a *list*-shaped allowlist could silently stay on the active/default provider instead of its own. The supported-model-ID extraction is now centralized in one helper used by both catalog construction and routing, so the picker and runtime can no longer drift. Existing mapping-shaped and list-of-strings configs are unaffected. Thanks @franksong2702. (#6127, #6121) - **Stale sidebar approval attention badge now clears once its gateway approval has drained.** A parent session could keep showing the red approval "attention" dot in the sidebar after the underlying gateway approval was already resolved/gone — the live gateway queue drained but the sidebar summary kept the mirrored approval alive. `_session_attention_summary()` now reconciles the gateway pending-mirror before counting attention (the same self-healing repair the approval-pending route already does), so the badge clears correctly. Live approvals stay visible and actionable and the clarify fallback is unchanged. This is the WebUI read-path mitigation; the child-session context-propagation root cause stays tracked in #6100. Thanks @rodboev. (#6117, #6100)