diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bec8aa28..a7c98a350 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,8 +3,16 @@ ## [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 +- **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) diff --git a/api/agent_runtime.py b/api/agent_runtime.py new file mode 100644 index 000000000..865a2fd50 --- /dev/null +++ b/api/agent_runtime.py @@ -0,0 +1,163 @@ +"""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 sys +import subprocess +import threading + +# 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. " + "Restart Hermes WebUI before retrying this action." +) + + +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 + + 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: + worktree_result = subprocess.run( + ["git", "-C", str(agent_dir), "rev-parse", "--show-toplevel"], + check=False, + capture_output=True, + text=True, + timeout=2, + ) + 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 = revision_result.stdout.strip() + return revision if revision_result.returncode == 0 and revision else None + + +_AGENT_SOURCE_DIR: Path | None = None +_AGENT_MODULE_PATH: Path | None = None +_AGENT_REVISION: str | None = None +_AIAgent = None +_RUNTIME_LOCK = threading.Lock() + + +class AgentRuntimeChangedError(RuntimeError): + """Raised when the loaded Agent runtime no longer matches its source tree.""" + + +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: + 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_MODULE_PATH, _AGENT_REVISION + + if _AGENT_REVISION is not None: + 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 + + +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, module_path=_AGENT_MODULE_PATH) + != _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 + + _capture_loaded_agent_revision() + 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 + return _AIAgent 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/api/routes.py b/api/routes.py index 2a654ab5a..58b93f6e7 100644 --- a/api/routes.py +++ b/api/routes.py @@ -34,6 +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 ( + AgentRuntimeChangedError, + ensure_agent_runtime_current, + require_ai_agent_class, +) from api.agent_sessions import ( MESSAGING_SOURCES, _looks_like_default_cli_title, @@ -16052,6 +16057,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 +16078,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, @@ -20425,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: @@ -20484,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: @@ -20770,6 +20782,31 @@ def _active_run_stream_for_session(session_id: str | None) -> str | None: return 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 + + 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, *, @@ -20783,8 +20820,18 @@ def _start_chat_stream_for_session( goal_related: bool = False, source: str = "webui", moa_config=None, + 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( + external_runtime_owned=backend_is_gateway, + ) + 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 @@ -20903,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: @@ -21074,6 +21120,7 @@ def _start_run( diag=diag, source=source, moa_config=moa_config, + external_runtime_owned=webui_gateway_chat_enabled(get_config()), ) @@ -21167,6 +21214,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) @@ -21643,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) @@ -21661,6 +21713,12 @@ 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. + 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) @@ -21956,6 +22014,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"]) @@ -21991,7 +22052,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 +22683,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 +22700,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, @@ -22684,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) @@ -22715,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) @@ -23739,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" @@ -23773,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, } ) @@ -23784,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: @@ -23820,6 +23915,34 @@ def _handle_session_compress_start(handler, body): 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. + try: + ensure_agent_runtime_current() + except AgentRuntimeChangedError as exc: + return j( + handler, + { + "error": str(exc), + "type": "agent_runtime_stale", + "retryable": True, + }, + status=409, + ) + + # 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) @@ -24035,10 +24158,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 +24205,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, @@ -24176,6 +24300,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)}") @@ -24685,10 +24815,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 +24875,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, @@ -24824,6 +24955,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/api/streaming.py b/api/streaming.py index c733a8f9a..7ef3dd998 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 @@ -6265,6 +6266,65 @@ 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: + # 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: + 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 + + def _build_session_db_for_stream(state_db_path): """Build a per-request SessionDB handle for WebUI session search. @@ -6280,12 +6340,36 @@ 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: + # 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: + 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() @@ -7213,6 +7297,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 @@ -8295,15 +8380,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/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/docs/troubleshooting.md b/docs/troubleshooting.md index 71d63d73a..5d0e7ea6a 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 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. + +**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/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; diff --git a/tests/test_agent_runtime_revision_guard.py b/tests/test_agent_runtime_revision_guard.py new file mode 100644 index 000000000..9712f711f --- /dev/null +++ b/tests/test_agent_runtime_revision_guard.py @@ -0,0 +1,687 @@ +"""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 +import types + +import pytest + + +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 + + +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, **_kwargs: None, + ) + + 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_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") + _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_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 + + +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_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") + + +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, **_kwargs: None, + ) + + with pytest.raises(agent_runtime.AgentRuntimeChangedError): + 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_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, **_kwargs: next(revisions), + ) + + with pytest.raises(agent_runtime.AgentRuntimeChangedError): + agent_runtime.require_ai_agent_class() + + 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 + + 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 + + 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, + }, + } + + +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 + + +@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 + + 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, + } 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_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) 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_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_sprint46.py b/tests/test_sprint46.py index 272f86689..9ca5bed66 100644 --- a/tests/test_sprint46.py +++ b/tests/test_sprint46.py @@ -129,6 +129,178 @@ 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_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_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 + + 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 = [ 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 diff --git a/tests/test_v050259_sessiondb_fd_leak.py b/tests/test_v050259_sessiondb_fd_leak.py index 3b80ac974..12bfd90e4 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,56 @@ 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" + ) + # 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 ──────── def test_lru_eviction_closes_evicted_agent_session_db(): @@ -66,27 +91,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 +123,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 +133,123 @@ 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 production helpers. + + 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. + """ + try: + from api.streaming import ( # type: ignore + _adopt_session_db_for_cached_agent, + _session_db_is_open, + ) + 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(): + """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 +260,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,8 +276,59 @@ 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" 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"