diff --git a/api/agent_runtime.py b/api/agent_runtime.py
new file mode 100644
index 00000000..865a2fd5
--- /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/routes.py b/api/routes.py
index 2a654ab5..58b93f6e 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 5e222a5f..7ef3dd99 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
 
 
@@ -7296,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
diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md
index 71d63d73..5d0e7ea6 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/tests/test_agent_runtime_revision_guard.py b/tests/test_agent_runtime_revision_guard.py
new file mode 100644
index 00000000..9712f711
--- /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 9f56f5fb..05267306 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_goal_command_webui.py b/tests/test_goal_command_webui.py
index 2389c08b..b16f3892 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 39121ebc..2a970bac 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 82725429..8d38e449 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 272f8668..9ca5bed6 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 0ab7a051..0da5ad41 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