feat(api): backend-acknowledged session model lock with runtime routing

Add a persisted, backend-confirmed provider/model lock for Hermes
Browser and other session API clients. A confirmed lock is an
execution contract rather than response metadata:

- POST /api/sessions/{session_id}/model validates and persists a
  confirmed browser_model_lock (advertised in /v1/capabilities)
- session chat + chat/stream consume the persisted lock on body-only
  follow-up turns; a confirmed lock wins over an older gateway session
  /model override and the session-persisted model
- a later successful session /model switch explicitly clears and
  replaces the lock while preserving lineage markers (_branched_from)
  and invalidating cached system-prompt model/provider metadata
- ordinary one-off request overrides never replace a confirmed lock
- provider-resolution failure fails closed as a typed provider-auth
  error (controlled response, never global-credential reuse)
- confirmed locks disable the global fallback model chain
- the completed agent's actual provider/model must match the locked
  route or the turn fails with a runtime-mismatch error
- responses carry sanitized runtime metadata reporting actual vs
  requested provider/model and lock state

Rebased onto the provider-aware request routing (#70853) and
session-model parity (#70931) that landed since the original branch;
the lock now slots into that precedence chain as the top rung.

Salvaged from PR #61236 by @abundantbeing.

gateway/platforms/api_server.py

@@ -1772,6 +1772,7 @@ class APIServerAdapter(BasePlatformAdapter):
             ("POST", "/api/sessions/{session_id}/fork", self._handle_fork_session),
             ("POST", "/api/sessions/{session_id}/chat", self._handle_session_chat),
             ("POST", "/api/sessions/{session_id}/chat/stream", self._handle_session_chat_stream),
+            ("POST", "/api/sessions/{session_id}/model", self._handle_session_model_lock),
             ("POST", "/v1/chat/completions", self._handle_chat_completions),
             ("POST", "/v1/responses", self._handle_responses),
             ("GET", "/v1/responses/{response_id}", self._handle_get_response),
@@ -2000,6 +2001,232 @@ class APIServerAdapter(BasePlatformAdapter):
             return None
         return self._model_routes.get(model_alias)
 
+    @staticmethod
+    def _clean_runtime_id(value: Any, *, max_len: int = 200) -> str:
+        if value is None:
+            return ""
+        text = str(value).strip()
+        if not text or len(text) > max_len:
+            return ""
+        if re.search(r"[\r\n\x00]", text):
+            return ""
+        return text
+
+    @classmethod
+    def _split_provider_prefixed_model(cls, model: str) -> tuple[str, str]:
+        text = cls._clean_runtime_id(model)
+        if "::" in text:
+            provider, raw = text.split("::", 1)
+            if re.match(r"^[a-zA-Z0-9_.-]{2,64}$", provider) and raw.strip():
+                return provider, raw.strip()
+        return "", text
+
+    @classmethod
+    def _runtime_options_from_model_options(cls, model_options: Any) -> Dict[str, Any]:
+        if not isinstance(model_options, dict):
+            return {}
+        runtime_options: Dict[str, Any] = {}
+        reasoning = model_options.get("reasoning")
+        if isinstance(reasoning, dict):
+            enabled = reasoning.get("enabled")
+            effort = cls._clean_runtime_id(reasoning.get("effort"), max_len=32)
+            if enabled is False:
+                runtime_options["reasoning_config"] = {"enabled": False}
+            elif effort:
+                runtime_options["reasoning_config"] = {"enabled": True, "effort": effort}
+            elif enabled is True:
+                runtime_options["reasoning_config"] = {"enabled": True}
+        service_tier = cls._clean_runtime_id(model_options.get("service_tier"), max_len=32)
+        if service_tier:
+            runtime_options["service_tier"] = service_tier
+        elif _coerce_request_bool(model_options.get("fast"), default=False):
+            runtime_options["service_tier"] = "priority"
+        return runtime_options
+
+    def _session_runtime_request_from_body(self, body: Dict[str, Any]) -> Dict[str, Any]:
+        raw_model = self._clean_runtime_id(body.get("model") or body.get("model_id"))
+        raw_provider = self._clean_runtime_id(body.get("provider") or body.get("provider_id"), max_len=80)
+        prefixed_provider, split_model = self._split_provider_prefixed_model(raw_model)
+        provider = raw_provider or prefixed_provider
+        model = split_model or raw_model
+        alias_route = self._resolve_route(raw_model) or self._resolve_route(model)
+        route = dict(alias_route) if isinstance(alias_route, dict) else None
+        route_source = "model_routes" if route else "global"
+        if not route and model and model != self._model_name:
+            route = {"model": model}
+            if provider:
+                route["provider"] = provider
+            route_source = "raw_request"
+        elif not route and provider and model:
+            route = {"model": model, "provider": provider}
+            route_source = "raw_request"
+        runtime_options = self._runtime_options_from_model_options(body.get("model_options"))
+        requested = {"provider": provider, "model": model, "raw_model": raw_model}
+        return {
+            "requested": requested,
+            "route": route,
+            "route_source": route_source,
+            "runtime_options": runtime_options,
+            "require_model_lock": _coerce_request_bool(body.get("require_model_lock"), default=False),
+            "model_options": body.get("model_options") if isinstance(body.get("model_options"), dict) else {},
+        }
+
+    def _runtime_lock_error(self, runtime_request: Dict[str, Any]) -> Optional["web.Response"]:
+        if not runtime_request.get("require_model_lock"):
+            return None
+        requested = runtime_request.get("requested") or {}
+        model = self._clean_runtime_id(requested.get("model"))
+        provider = self._clean_runtime_id(requested.get("provider"), max_len=80)
+        route = runtime_request.get("route")
+        if not model and not provider:
+            return web.json_response(
+                _openai_error("require_model_lock was set but no model/provider was provided", code="missing_model"),
+                status=400,
+            )
+        if not route or runtime_request.get("route_source") == "global":
+            return web.json_response(
+                _openai_error("Requested Browser model lock cannot be routed; refusing silent global fallback", code="model_lock_unavailable"),
+                status=409,
+            )
+        return None
+
+    def _persist_session_runtime_lock(self, session_id: str, runtime_request: Dict[str, Any]) -> bool:
+        # Persist only a newly confirmed lock. Reusing a stored lock should not
+        # rewrite its timestamp/prompt state on every turn, and an ordinary
+        # one-off request override must not erase a previously confirmed lock.
+        if runtime_request.get("persisted_lock") or not runtime_request.get("require_model_lock"):
+            return True
+        requested = runtime_request.get("requested") or {}
+        model = self._clean_runtime_id(requested.get("model"))
+        provider = self._clean_runtime_id(requested.get("provider"), max_len=80)
+        if not model and not provider:
+            return False
+        db = self._ensure_session_db()
+        if db is None:
+            return False
+        try:
+            db.update_session_runtime_lock(
+                session_id,
+                model=model or None,
+                provider=provider or None,
+                model_options=runtime_request.get("model_options") or {},
+                route_source=runtime_request.get("route_source") or "",
+                confirmed=bool(runtime_request.get("require_model_lock")),
+            )
+            return True
+        except Exception:
+            logger.warning("[%s] failed to persist session runtime lock for %s", self.name, session_id, exc_info=True)
+            return False
+
+    @staticmethod
+    def _parse_session_model_config(raw: Any) -> Dict[str, Any]:
+        if isinstance(raw, dict):
+            return dict(raw)
+        if isinstance(raw, str) and raw.strip():
+            try:
+                parsed = json.loads(raw)
+            except Exception:
+                return {}
+            if isinstance(parsed, dict):
+                return parsed
+        return {}
+
+    def _runtime_request_from_persisted_session_lock(
+        self,
+        session: Optional[Dict[str, Any]],
+        body: Dict[str, Any],
+    ) -> Optional[Dict[str, Any]]:
+        if not isinstance(session, dict):
+            return None
+        model_config = self._parse_session_model_config(session.get("model_config"))
+        lock = model_config.get("browser_model_lock")
+        if not isinstance(lock, dict) or not _coerce_request_bool(lock.get("confirmed"), default=False):
+            return None
+        model = self._clean_runtime_id(lock.get("model"))
+        provider = self._clean_runtime_id(lock.get("provider"), max_len=80)
+        if not model and not provider:
+            return None
+        persisted_route_source = self._clean_runtime_id(
+            lock.get("route_source"),
+            max_len=64,
+        ).lower()
+        route: Optional[Dict[str, Any]] = None
+        if persisted_route_source == "model_routes":
+            route = self._resolve_route(model) if model else None
+        else:
+            route = {"model": model} if model else {}
+            if provider:
+                route["provider"] = provider
+        model_options = (
+            body.get("model_options")
+            if isinstance(body.get("model_options"), dict)
+            else lock.get("model_options")
+        )
+        return {
+            "requested": {
+                "provider": provider,
+                "model": model,
+                "raw_model": model,
+            },
+            "route": route or None,
+            "route_source": "session_model_lock",
+            "runtime_options": self._runtime_options_from_model_options(model_options),
+            "require_model_lock": True,
+            "model_options": model_options if isinstance(model_options, dict) else {},
+            "persisted_lock": True,
+        }
+
+    def _effective_session_runtime_request(
+        self,
+        *,
+        session: Optional[Dict[str, Any]],
+        body: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        runtime_request = self._session_runtime_request_from_body(body)
+        requested = runtime_request.get("requested") or {}
+        if requested.get("model") or requested.get("provider"):
+            return runtime_request
+        persisted = self._runtime_request_from_persisted_session_lock(session, body)
+        return persisted or runtime_request
+
+    @classmethod
+    def _sanitize_runtime_metadata(
+        cls,
+        *,
+        runtime: Optional[Dict[str, Any]] = None,
+        requested_runtime: Optional[Dict[str, Any]] = None,
+        route_source: str = "global",
+        model_lock: str = "",
+    ) -> Dict[str, Any]:
+        payload = dict(runtime or {})
+        provider = cls._clean_runtime_id(
+            payload.get("provider") or payload.get("provider_id") or payload.get("effective_provider"),
+            max_len=80,
+        )
+        model = cls._clean_runtime_id(payload.get("model") or payload.get("model_id") or payload.get("effective_model"))
+        result: Dict[str, Any] = {
+            "provider": provider,
+            "model": model,
+            "route_source": cls._clean_runtime_id(payload.get("route_source") or route_source, max_len=64) or "global",
+        }
+        if requested_runtime or payload.get("requested"):
+            req = requested_runtime or payload.get("requested") or {}
+            result["requested"] = {
+                "provider": cls._clean_runtime_id(req.get("provider"), max_len=80),
+                "model": cls._clean_runtime_id(req.get("model")),
+            }
+        if model_lock or payload.get("model_lock"):
+            result["model_lock"] = cls._clean_runtime_id(model_lock or payload.get("model_lock"), max_len=32)
+        return result
+
+    @staticmethod
+    def _normalize_session_source(value: Any) -> str:
+        text = str(value or "").strip().lower()
+        allowed = {"api_server", "hermes_browser", "browser", "cli", "telegram", "discord", "slack", "desktop", "dashboard"}
+        if text in allowed:
+            return "hermes_browser" if text == "browser" else text
+        return "api_server"
+
     def _session_model_override_for(self, session_key: Optional[str]) -> Optional[Dict[str, Any]]:
         """Return the gateway's session ``/model`` override for *session_key*, if any.
 
@@ -2080,6 +2307,7 @@ class APIServerAdapter(BasePlatformAdapter):
         model_options: Optional[Dict[str, Any]] = None,
         route: Optional[Dict[str, Any]] = None,
         session_model: Optional[str] = None,
+        confirmed_runtime_lock: bool = False,
     ) -> Any:
         """
         Create an AIAgent instance using the gateway's runtime config.
@@ -2107,6 +2335,12 @@ class APIServerAdapter(BasePlatformAdapter):
         handlers pass either ``route`` (alias hit) or ``session_model`` (raw
         model), never both.  Precedence: session ``/model`` override →
         ``session_model`` → route alias / per-request selection → global.
+
+        ``confirmed_runtime_lock`` marks a backend-acknowledged Browser model
+        lock (POST /api/sessions/{id}/model).  A confirmed lock beats the
+        session ``/model`` override, disables the global fallback model
+        chain, and fails closed if the locked provider's credentials cannot
+        be resolved.
         """
         from run_agent import AIAgent
         from gateway.run import (
@@ -2172,7 +2406,7 @@ class APIServerAdapter(BasePlatformAdapter):
                     provider_name,
                     target_model=target_model or None,
                 )
-            except Exception:
+            except Exception as exc:
                 try:
                     from gateway.run import _resolve_runtime_agent_kwargs_for_provider
 
@@ -2180,7 +2414,10 @@ class APIServerAdapter(BasePlatformAdapter):
                 except Exception:
                     pass
                 if required:
-                    raise
+                    # Surface as the typed provider-auth failure so
+                    # _run_agent()/_handle_runs() return the controlled
+                    # response shape instead of a raw 500.
+                    raise _ProviderAuthResolutionError(str(exc)) from exc
                 logger.debug(
                     "api_server provider-runtime refresh failed for provider=%s model=%s",
                     provider_name,
@@ -2190,14 +2427,19 @@ class APIServerAdapter(BasePlatformAdapter):
                 return None
 
         # Final precedence mirrors the gateway contract:
-        # session /model override → session-persisted model (POST
-        # /api/sessions {"model": ...}) → model_routes mapping selected by
-        # the request model alias → direct per-request provider/model →
-        # global defaults.  model_options stay request-scoped regardless
-        # of which selection wins.
+        # confirmed Browser model lock → session /model override →
+        # session-persisted model (POST /api/sessions {"model": ...}) →
+        # model_routes mapping selected by the request model alias → direct
+        # per-request provider/model → global defaults.  model_options stay
+        # request-scoped regardless of which selection wins.  A confirmed
+        # lock is an execution contract: it bypasses the session /model
+        # override and fails closed (never reuses global credentials) if
+        # its provider cannot be resolved.
         session_key = gateway_session_key or session_id
         session_row_model = _clean_request_string(session_model)
-        session_override = self._session_model_override_for(session_key)
+        session_override = None
+        if not confirmed_runtime_lock:
+            session_override = self._session_model_override_for(session_key)
         if session_override:
             override_model = _clean_request_string(session_override.get("model")) or model
             session_provider = _clean_request_string(session_override.get("provider"))
@@ -2216,7 +2458,7 @@ class APIServerAdapter(BasePlatformAdapter):
                     "api_server request selection skipped: session /model override wins for %s",
                     session_key or "",
                 )
-        elif session_row_model:
+        elif session_row_model and not confirmed_runtime_lock:
             # Session-persisted model (raw string that resolved to no route
             # alias).  Pins this session's turns ahead of per-request body
             # values — a session's chosen model is a standing selection,
@@ -2253,7 +2495,10 @@ class APIServerAdapter(BasePlatformAdapter):
                 provider_runtime = _resolve_provider_runtime(
                     effective_provider,
                     target_model=effective_model,
-                    required=bool(request_provider),
+                    # A confirmed Browser lock fails closed: if the locked
+                    # provider cannot be resolved, never fall through to
+                    # the previous global provider's credentials.
+                    required=bool(request_provider) or confirmed_runtime_lock,
                 )
             if provider_runtime:
                 _apply_runtime_agent_overrides(runtime_kwargs, provider_runtime)
@@ -2331,7 +2576,11 @@ class APIServerAdapter(BasePlatformAdapter):
 
         # Load fallback provider chain so the API server platform has the
         # same fallback behaviour as Telegram/Discord/Slack (fixes #4954).
-        fallback_model = GatewayRunner._load_fallback_model()
+        fallback_model = (
+            None
+            if confirmed_runtime_lock
+            else GatewayRunner._load_fallback_model()
+        )
 
         agent_kwargs = {
             "model": model,
@@ -2357,6 +2606,19 @@ class APIServerAdapter(BasePlatformAdapter):
             agent_kwargs["service_tier"] = request_service_tier
 
         agent = AIAgent(**agent_kwargs)
+        agent._hermes_api_runtime = {
+            "provider": runtime_kwargs.get("provider") or getattr(agent, "provider", "") or "",
+            "model": getattr(agent, "model", None) or model,
+            "route_source": (
+                "session_model_lock"
+                if confirmed_runtime_lock
+                else "session_model_override"
+                if session_override
+                else "raw_request"
+                if route or request_model or request_provider
+                else "global"
+            ),
+        }
         return agent
 
     # ------------------------------------------------------------------
@@ -2559,6 +2821,7 @@ class APIServerAdapter(BasePlatformAdapter):
                 "session_chat": True,
                 "session_chat_streaming": True,
                 "session_fork": True,
+                "session_model_lock": True,
                 "admin_config_rw": False,
                 "jobs_admin": False,
                 "memory_write_api": False,
@@ -2592,6 +2855,7 @@ class APIServerAdapter(BasePlatformAdapter):
                 "session_fork": {"method": "POST", "path": "/api/sessions/{session_id}/fork"},
                 "session_chat": {"method": "POST", "path": "/api/sessions/{session_id}/chat"},
                 "session_chat_stream": {"method": "POST", "path": "/api/sessions/{session_id}/chat/stream"},
+                "session_model_lock": {"method": "POST", "path": "/api/sessions/{session_id}/model"},
             },
         })
 
@@ -2816,6 +3080,25 @@ class APIServerAdapter(BasePlatformAdapter):
         system_prompt = body.get("system_prompt")
         if system_prompt is not None and not isinstance(system_prompt, str):
             return web.json_response(_openai_error("system_prompt must be a string", code="invalid_system_prompt"), status=400)
+        source = self._normalize_session_source(body.get("source") or "api_server")
+        runtime_request = self._session_runtime_request_from_body(body)
+        lock_error = self._runtime_lock_error(runtime_request)
+        if lock_error is not None:
+            return lock_error
+        requested = runtime_request.get("requested") or {}
+        model_name = self._clean_runtime_id(requested.get("model")) or (str(model) if model else None)
+        model_config = None
+        if requested.get("model") or requested.get("provider"):
+            model_config = {
+                "browser_model_lock": {
+                    "provider": requested.get("provider") or "",
+                    "model": requested.get("model") or "",
+                    "model_options": runtime_request.get("model_options") or {},
+                    "route_source": runtime_request.get("route_source") or "",
+                    "confirmed": bool(runtime_request.get("require_model_lock")),
+                    "updated_at": time.time(),
+                }
+            }
         title = body.get("title")
 
         # Run the entire check-insert-title sequence inside a single
@@ -2833,12 +3116,13 @@ class APIServerAdapter(BasePlatformAdapter):
                 import time as _time
                 conn.execute(
                     """INSERT INTO sessions (
-                       id, source, model, system_prompt, started_at
-                    ) VALUES (?, ?, ?, ?, ?)""",
+                       id, source, model, model_config, system_prompt, started_at
+                    ) VALUES (?, ?, ?, ?, ?, ?)""",
                     (
                         session_id,
-                        "api_server",
-                        str(model) if model else None,
+                        source,
+                        model_name,
+                        json.dumps(model_config) if model_config else None,
                         system_prompt,
                         _time.time(),
                     ),
@@ -2863,8 +3147,8 @@ class APIServerAdapter(BasePlatformAdapter):
                     "SELECT * FROM sessions WHERE id = ?", (session_id,)
                 ).fetchone()
                 return (dict(session_row) if session_row else {
-                    "id": session_id, "source": "api_server",
-                    "model": model, "title": title,
+                    "id": session_id, "source": source,
+                    "model": model_name, "title": title,
                 }), None
             return db._execute_write(_atomic)
 
@@ -3010,26 +3294,57 @@ class APIServerAdapter(BasePlatformAdapter):
         system_prompt = body.get("system_message") or body.get("instructions")
         if system_prompt is not None and not isinstance(system_prompt, str):
             return web.json_response(_openai_error("system_message must be a string", code="invalid_system_message"), status=400)
-        # Session-persisted model (POST /api/sessions {"model": ...}) —
-        # previously fetched and discarded here, so a session's chosen model
-        # silently had no effect on any chat turn. If the stored value is a
-        # model_routes alias, apply it through the route path so route
-        # provider/credentials come along; a raw model string threads
-        # through as session_model.
-        stored_model = session.get("model") if isinstance(session, dict) else None
-        stored_route = self._resolve_route(stored_model)
-        route = stored_route or self._resolve_route(body.get("model"))
-        session_model = stored_model if (stored_model and stored_route is None) else None
-        agent_overrides = _request_agent_overrides(body, virtual_model=self._model_name)
-        selection_error = self._request_route_conflict_error(
-            session_id=session_id,
-            gateway_session_key=gateway_session_key,
-            requested_model=agent_overrides.get("requested_model"),
-            requested_provider=agent_overrides.get("requested_provider"),
-            route=route,
+        # Runtime selection. A backend-acknowledged Browser model lock
+        # (require_model_lock in the body, or a previously confirmed lock
+        # persisted on the session row) is an execution contract and wins.
+        # Otherwise: session-persisted model (POST /api/sessions
+        # {"model": ...}) — previously fetched and discarded here — routes
+        # through model_routes when it is an alias (route
+        # provider/credentials come along) or threads through as
+        # session_model when it is a raw string; per-request body values
+        # come after that.
+        runtime_request = self._effective_session_runtime_request(
+            session=session,
+            body=body,
         )
-        if selection_error:
-            return web.json_response(_openai_error(selection_error), status=400)
+        lock_error = self._runtime_lock_error(runtime_request)
+        if lock_error is not None:
+            return lock_error
+        if not self._persist_session_runtime_lock(session_id, runtime_request):
+            return web.json_response(
+                _openai_error(
+                    "Could not persist the requested session model lock",
+                    code="model_lock_persistence_failed",
+                ),
+                status=500,
+            )
+        lock_active = bool(runtime_request.get("require_model_lock"))
+        if lock_active:
+            route = runtime_request.get("route")
+            session_model = None
+            requested = runtime_request.get("requested") or {}
+            agent_overrides: Dict[str, Any] = {}
+            if requested.get("model"):
+                agent_overrides["requested_model"] = requested["model"]
+            if requested.get("provider"):
+                agent_overrides["requested_provider"] = requested["provider"]
+            if runtime_request.get("model_options"):
+                agent_overrides["model_options"] = runtime_request["model_options"]
+        else:
+            stored_model = session.get("model") if isinstance(session, dict) else None
+            stored_route = self._resolve_route(stored_model)
+            route = stored_route or self._resolve_route(body.get("model"))
+            session_model = stored_model if (stored_model and stored_route is None) else None
+            agent_overrides = _request_agent_overrides(body, virtual_model=self._model_name)
+            selection_error = self._request_route_conflict_error(
+                session_id=session_id,
+                gateway_session_key=gateway_session_key,
+                requested_model=agent_overrides.get("requested_model"),
+                requested_provider=agent_overrides.get("requested_provider"),
+                route=route,
+            )
+            if selection_error:
+                return web.json_response(_openai_error(selection_error), status=400)
         history = await self._conversation_history_for_session(session_id)
         result, usage = await self._run_agent(
             user_message=user_message,
@@ -3039,6 +3354,9 @@ class APIServerAdapter(BasePlatformAdapter):
             gateway_session_key=gateway_session_key,
             route=route,
             session_model=session_model,
+            requested_runtime=runtime_request.get("requested") or {},
+            route_source=runtime_request.get("route_source") or "global",
+            confirmed_runtime_lock=lock_active,
             **agent_overrides,
         )
         effective_session_id = result.get("session_id") if isinstance(result, dict) else session_id
@@ -3046,12 +3364,30 @@ class APIServerAdapter(BasePlatformAdapter):
         headers = {"X-Hermes-Session-Id": effective_session_id or session_id}
         if gateway_session_key:
             headers["X-Hermes-Session-Key"] = gateway_session_key
+        runtime = {}
+        if isinstance(result, dict):
+            runtime = result.get("runtime") or {}
+        if not runtime and isinstance(usage, dict):
+            runtime = usage.get("runtime") or {}
+        runtime = self._sanitize_runtime_metadata(
+            runtime=runtime,
+            requested_runtime=runtime_request.get("requested"),
+            route_source=runtime_request.get("route_source") or "global",
+            model_lock=(
+                "confirmed"
+                if runtime and runtime_request.get("require_model_lock")
+                else "accepted"
+                if runtime_request.get("require_model_lock")
+                else ""
+            ),
+        )
         return web.json_response(
             {
                 "object": "hermes.session.chat.completion",
                 "session_id": effective_session_id or session_id,
                 "message": {"role": "assistant", "content": final_response},
                 "usage": usage,
+                "runtime": runtime,
             },
             headers=headers,
         )
@@ -3075,22 +3411,55 @@ class APIServerAdapter(BasePlatformAdapter):
         system_prompt = body.get("system_message") or body.get("instructions")
         if system_prompt is not None and not isinstance(system_prompt, str):
             return web.json_response(_openai_error("system_message must be a string", code="invalid_system_message"), status=400)
-        # Session-persisted model — see _handle_session_chat for the
-        # non-streaming twin of this resolution.
-        stored_model = session.get("model") if isinstance(session, dict) else None
-        stored_route = self._resolve_route(stored_model)
-        route = stored_route or self._resolve_route(body.get("model"))
-        session_model = stored_model if (stored_model and stored_route is None) else None
-        agent_overrides = _request_agent_overrides(body, virtual_model=self._model_name)
-        selection_error = self._request_route_conflict_error(
-            session_id=session_id,
-            gateway_session_key=gateway_session_key,
-            requested_model=agent_overrides.get("requested_model"),
-            requested_provider=agent_overrides.get("requested_provider"),
-            route=route,
+        # Runtime selection — mirrors _handle_session_chat (lock wins,
+        # otherwise session-persisted model then per-request values).
+        runtime_request = self._effective_session_runtime_request(
+            session=session,
+            body=body,
+        )
+        lock_error = self._runtime_lock_error(runtime_request)
+        if lock_error is not None:
+            return lock_error
+        if not self._persist_session_runtime_lock(session_id, runtime_request):
+            return web.json_response(
+                _openai_error(
+                    "Could not persist the requested session model lock",
+                    code="model_lock_persistence_failed",
+                ),
+                status=500,
+            )
+        lock_active = bool(runtime_request.get("require_model_lock"))
+        if lock_active:
+            route = runtime_request.get("route")
+            session_model = None
+            requested = runtime_request.get("requested") or {}
+            agent_overrides: Dict[str, Any] = {}
+            if requested.get("model"):
+                agent_overrides["requested_model"] = requested["model"]
+            if requested.get("provider"):
+                agent_overrides["requested_provider"] = requested["provider"]
+            if runtime_request.get("model_options"):
+                agent_overrides["model_options"] = runtime_request["model_options"]
+        else:
+            stored_model = session.get("model") if isinstance(session, dict) else None
+            stored_route = self._resolve_route(stored_model)
+            route = stored_route or self._resolve_route(body.get("model"))
+            session_model = stored_model if (stored_model and stored_route is None) else None
+            agent_overrides = _request_agent_overrides(body, virtual_model=self._model_name)
+            selection_error = self._request_route_conflict_error(
+                session_id=session_id,
+                gateway_session_key=gateway_session_key,
+                requested_model=agent_overrides.get("requested_model"),
+                requested_provider=agent_overrides.get("requested_provider"),
+                route=route,
+            )
+            if selection_error:
+                return web.json_response(_openai_error(selection_error), status=400)
+        runtime_meta = self._sanitize_runtime_metadata(
+            requested_runtime=runtime_request.get("requested"),
+            route_source=runtime_request.get("route_source") or "global",
+            model_lock=("accepted" if lock_active else ""),
         )
-        if selection_error:
-            return web.json_response(_openai_error(selection_error), status=400)
 
         loop = asyncio.get_running_loop()
         queue: "asyncio.Queue[Optional[tuple[str, Dict[str, Any]]]]" = asyncio.Queue()
@@ -3134,7 +3503,10 @@ class APIServerAdapter(BasePlatformAdapter):
 
         async def _run_and_signal() -> None:
             try:
-                await queue.put(_event_payload("run.started", {"user_message": {"role": "user", "content": user_message}}))
+                await queue.put(_event_payload("run.started", {
+                    "user_message": {"role": "user", "content": user_message},
+                    "runtime": runtime_meta,
+                }))
                 await queue.put(_event_payload("message.started", {"message": {"id": message_id, "role": "assistant"}}))
                 history = await self._conversation_history_for_session(session_id)
                 result, usage = await self._run_agent(
@@ -3147,11 +3519,31 @@ class APIServerAdapter(BasePlatformAdapter):
                     gateway_session_key=gateway_session_key,
                     route=route,
                     session_model=session_model,
+                    requested_runtime=runtime_request.get("requested") or {},
+                    route_source=runtime_request.get("route_source") or "global",
+                    confirmed_runtime_lock=lock_active,
                     **agent_overrides,
                 )
                 final_response = _resolve_media_to_data_urls(result.get("final_response", "") if isinstance(result, dict) else "")
                 effective_session_id = result.get("session_id", session_id) if isinstance(result, dict) else session_id
                 turn_messages = self._turn_transcript_messages(history, user_message, result) if isinstance(result, dict) else []
+                effective_runtime = {}
+                if isinstance(result, dict):
+                    effective_runtime = result.get("runtime") or {}
+                if not effective_runtime and isinstance(usage, dict):
+                    effective_runtime = usage.get("runtime") or {}
+                effective_runtime = self._sanitize_runtime_metadata(
+                    runtime=effective_runtime,
+                    requested_runtime=runtime_request.get("requested"),
+                    route_source=runtime_request.get("route_source") or "global",
+                    model_lock=(
+                        "confirmed"
+                        if effective_runtime and runtime_request.get("require_model_lock")
+                        else "accepted"
+                        if runtime_request.get("require_model_lock")
+                        else ""
+                    ),
+                )
                 await queue.put(_event_payload("assistant.completed", {
                     "session_id": effective_session_id,
                     "message_id": message_id,
@@ -3159,6 +3551,7 @@ class APIServerAdapter(BasePlatformAdapter):
                     "completed": True,
                     "partial": False,
                     "interrupted": False,
+                    "runtime": effective_runtime,
                 }))
                 await queue.put(_event_payload("run.completed", {
                     "session_id": effective_session_id,
@@ -3166,6 +3559,7 @@ class APIServerAdapter(BasePlatformAdapter):
                     "completed": True,
                     "messages": turn_messages,
                     "usage": usage,
+                    "runtime": effective_runtime,
                 }))
             except Exception as exc:
                 logger.exception("[api_server] session chat stream failed")
@@ -3214,6 +3608,48 @@ class APIServerAdapter(BasePlatformAdapter):
             logger.debug("[api_server] session SSE stream error: %s", exc)
         return response
 
+    async def _handle_session_model_lock(self, request: "web.Request") -> "web.Response":
+        """POST /api/sessions/{session_id}/model — backend-ack a Browser model lock."""
+        auth_err = self._check_auth(request)
+        if auth_err:
+            return auth_err
+        session_id = request.match_info["session_id"]
+        _, err = await self._get_existing_session_or_404(session_id)
+        if err:
+            return err
+        body, err = await self._read_json_body(request)
+        if err:
+            return err
+        runtime_request = self._session_runtime_request_from_body(body)
+        runtime_request["require_model_lock"] = True
+        lock_error = self._runtime_lock_error(runtime_request)
+        if lock_error is not None:
+            return lock_error
+        if not self._persist_session_runtime_lock(session_id, runtime_request):
+            return web.json_response(
+                _openai_error(
+                    "Could not persist the requested session model lock",
+                    code="model_lock_persistence_failed",
+                ),
+                status=500,
+            )
+        requested = runtime_request.get("requested") or {}
+        route = runtime_request.get("route") or {}
+        runtime = self._sanitize_runtime_metadata(
+            runtime={
+                "provider": route.get("provider") or requested.get("provider") or "",
+                "model": route.get("model") or requested.get("model") or "",
+                "route_source": runtime_request.get("route_source") or "raw_request",
+            },
+            requested_runtime=requested,
+            route_source=runtime_request.get("route_source") or "raw_request",
+            model_lock="accepted",
+        )
+        return web.json_response({
+            "object": "hermes.session.model_lock",
+            "session_id": session_id,
+            "runtime": runtime,
+        })
     @_admit_api_agent_request
     async def _handle_chat_completions(self, request: "web.Request") -> "web.Response":
         """POST /v1/chat/completions — OpenAI Chat Completions format."""
@@ -5281,6 +5717,9 @@ class APIServerAdapter(BasePlatformAdapter):
         model_options: Optional[Dict[str, Any]] = None,
         route: Optional[Dict[str, Any]] = None,
         session_model: Optional[str] = None,
+        requested_runtime: Optional[Dict[str, Any]] = None,
+        route_source: str = "global",
+        confirmed_runtime_lock: bool = False,
     ) -> tuple:
         """
         Create an agent and run a conversation in a thread executor.
@@ -5296,6 +5735,12 @@ class APIServerAdapter(BasePlatformAdapter):
         row.  It is used only when the persisted value did not resolve to a
         ``model_routes`` alias — see ``_create_agent`` for precedence.
 
+        *requested_runtime* / *route_source* / *confirmed_runtime_lock*
+        carry the Browser model-lock contract: when a confirmed lock is
+        active the completed agent's actual provider/model must match the
+        locked selection or the turn fails, and the response carries
+        sanitized ``runtime`` metadata reporting actual vs requested.
+
         If *agent_ref* is a one-element list, the AIAgent instance is stored
         at ``agent_ref[0]`` before ``run_conversation`` begins.  This allows
         callers (e.g. the SSE writer) to call ``agent.interrupt()`` from
@@ -5330,6 +5775,7 @@ class APIServerAdapter(BasePlatformAdapter):
                         model_options=model_options,
                         route=route,
                         session_model=session_model,
+                        confirmed_runtime_lock=confirmed_runtime_lock,
                     )
                     if agent_ref is not None:
                         agent_ref[0] = agent
@@ -5362,6 +5808,71 @@ class APIServerAdapter(BasePlatformAdapter):
                     )
                     if _compacted_in_place or _session_rotated:
                         result["_compressed"] = True
+                    include_runtime = bool(
+                        requested_runtime
+                        or route
+                        or confirmed_runtime_lock
+                        or (route_source and route_source != "global")
+                    )
+                    if include_runtime:
+                        runtime = dict(getattr(agent, "_hermes_api_runtime", {}) or {})
+                        raw_provider = getattr(agent, "provider", "")
+                        raw_model = getattr(agent, "model", "")
+                        actual_provider = (
+                            self._clean_runtime_id(raw_provider, max_len=80)
+                            if isinstance(raw_provider, str)
+                            else ""
+                        )
+                        actual_model = (
+                            self._clean_runtime_id(raw_model)
+                            if isinstance(raw_model, str)
+                            else ""
+                        )
+                        if actual_provider:
+                            runtime["provider"] = actual_provider
+                        else:
+                            runtime.setdefault("provider", "")
+                        if actual_model:
+                            runtime["model"] = actual_model
+                        else:
+                            runtime.setdefault("model", "")
+                        if confirmed_runtime_lock:
+                            expected_provider = self._clean_runtime_id(
+                                (route or {}).get("provider")
+                                or (requested_runtime or {}).get("provider"),
+                                max_len=80,
+                            )
+                            expected_model = self._clean_runtime_id(
+                                (route or {}).get("model")
+                                or (requested_runtime or {}).get("model")
+                            )
+                            mismatched = (
+                                (expected_provider and actual_provider != expected_provider)
+                                or (expected_model and actual_model != expected_model)
+                            )
+                            if mismatched:
+                                raise RuntimeError(
+                                    "confirmed model lock runtime mismatch: "
+                                    f"expected provider={expected_provider or '<unspecified>'} "
+                                    f"model={expected_model or '<unspecified>'}; "
+                                    f"actual provider={actual_provider or '<unknown>'} "
+                                    f"model={actual_model or '<unknown>'}"
+                                )
+                        if requested_runtime:
+                            runtime["requested"] = {
+                                "provider": self._clean_runtime_id((requested_runtime or {}).get("provider"), max_len=80),
+                                "model": self._clean_runtime_id((requested_runtime or {}).get("model")),
+                            }
+                        runtime["route_source"] = route_source or runtime.get("route_source") or "global"
+                        runtime = self._sanitize_runtime_metadata(
+                            runtime=runtime,
+                            requested_runtime=requested_runtime,
+                            route_source=route_source or "global",
+                            model_lock=("confirmed" if confirmed_runtime_lock else ""),
+                        )
+                        if isinstance(result, dict):
+                            result["runtime"] = runtime
+                        usage["runtime"] = runtime
                     return result, usage
                 except _ProviderAuthResolutionError as exc:
                     # Only _ProviderAuthResolutionError — raised exclusively

hermes_state.py

@@ -4366,14 +4366,81 @@ class SessionDB:
         Unlike ``update_token_counts`` which uses ``COALESCE(model, ?)``
         (only filling in NULL), this unconditionally sets the model column
         so that the dashboard reflects the user's latest /model choice.
+        Also nulls ``system_prompt`` so stale ``Model:`` / ``Provider:``
+        footer metadata is rebuilt on the next turn. A successful /model
+        switch explicitly replaces any confirmed Browser runtime lock while
+        preserving unrelated lineage markers in ``model_config``.
         """
         def _do(conn):
             conn.execute(
-                "UPDATE sessions SET model = ? WHERE id = ?",
+                """UPDATE sessions SET
+                   model = ?,
+                   model_config = CASE
+                       WHEN model_config IS NULL THEN NULL
+                       WHEN json_valid(model_config)
+                           THEN json_remove(model_config, '$.browser_model_lock')
+                       ELSE model_config
+                   END,
+                   system_prompt = NULL
+                   WHERE id = ?""",
                 (model, session_id),
             )
         self._execute_write(_do)
 
+    def update_session_runtime_lock(
+        self,
+        session_id: str,
+        *,
+        model: Optional[str] = None,
+        provider: Optional[str] = None,
+        model_options: Optional[Dict[str, Any]] = None,
+        route_source: Optional[str] = None,
+        confirmed: bool = False,
+    ) -> None:
+        """Persist a Browser / API client runtime lock without clobbering lineage markers.
+
+        Merges ``browser_model_lock`` into the existing ``model_config`` JSON so
+        ``_branched_from`` / ``_delegate_from`` survive. Nulls ``system_prompt``
+        so cached ``Model:`` / ``Provider:`` footers cannot lie after a switch.
+        """
+        lock = {
+            "provider": provider or "",
+            "model": model or "",
+            "model_options": model_options or {},
+            "route_source": route_source or "",
+            "confirmed": bool(confirmed),
+            "updated_at": time.time(),
+        }
+
+        def _do(conn):
+            row = conn.execute(
+                "SELECT model_config FROM sessions WHERE id = ?",
+                (session_id,),
+            ).fetchone()
+            if row is None:
+                return
+            raw = row["model_config"] if isinstance(row, sqlite3.Row) else row[0]
+            config: Dict[str, Any] = {}
+            if isinstance(raw, str) and raw.strip():
+                try:
+                    parsed = json.loads(raw)
+                    if isinstance(parsed, dict):
+                        config = parsed
+                except Exception:
+                    config = {}
+            elif isinstance(raw, dict):
+                config = dict(raw)
+            config["browser_model_lock"] = lock
+            conn.execute(
+                """UPDATE sessions SET
+                   model_config = ?,
+                   model = COALESCE(?, model),
+                   system_prompt = NULL
+                   WHERE id = ?""",
+                (json.dumps(config), model, session_id),
+            )
+        self._execute_write(_do)
+
     def update_session_billing_route(
         self,
         session_id: str,

tests/gateway/test_multiplex_api_server_routing.py

@@ -65,11 +65,13 @@ class TestApiServerRouteTable:
         assert "/v1/models" in paths
         assert "/api/model/options" in paths
         assert "/v1/chat/completions" in paths
+        assert "/api/sessions/{session_id}/model" in paths
         # connect() mirrors every native path under /p/{profile}/…
         mirrored = {f"/p/{{profile}}{path}" for path in paths}
         assert "/p/{profile}/v1/models" in mirrored
         assert "/p/{profile}/api/model/options" in mirrored
         assert "/p/{profile}/v1/chat/completions" in mirrored
+        assert "/p/{profile}/api/sessions/{session_id}/model" in mirrored
 
 
 class TestApiServerModelsUnderProfile:

tests/gateway/test_session_api.py

@@ -113,7 +113,10 @@ async def test_run_agent_binds_api_session_context_for_tool_env(adapter, monkeyp
     )
 
     assert result["session_id"] == "request-session"
-    assert usage == {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
+    assert usage["input_tokens"] == 0
+    assert usage["output_tokens"] == 0
+    assert usage["total_tokens"] == 0
+    assert "runtime" not in usage
     assert observed == {
         "task_id": "request-session",
         "context_session_id": "request-session",
@@ -617,3 +620,578 @@ async def test_session_chat_surfaces_controlled_response_on_provider_auth_failur
         payload = await resp.json()
 
     assert payload["message"]["content"] == "⚠️ Provider authentication failed: Auth failed: token expired"
+def _register_session_model_route(app, adapter):
+    app.router.add_post("/api/sessions/{session_id}/model", adapter._handle_session_model_lock)
+
+
+def _patch_api_server_runtime(monkeypatch):
+    monkeypatch.setattr(
+        "gateway.run._resolve_runtime_agent_kwargs",
+        lambda: {
+            "provider": "openrouter",
+            "api_key": "sk-global",
+            "base_url": "https://openrouter.example/v1",
+            "api_mode": "chat_completions",
+        },
+    )
+    monkeypatch.setattr("gateway.run._resolve_gateway_model", lambda: "global/model")
+    monkeypatch.setattr("gateway.run._load_gateway_config", lambda: {})
+    monkeypatch.setattr(
+        "gateway.run.GatewayRunner._load_reasoning_config",
+        staticmethod(lambda model="": {}),
+    )
+    monkeypatch.setattr(
+        "gateway.run.GatewayRunner._load_fallback_model",
+        staticmethod(lambda: None),
+    )
+    monkeypatch.setattr("gateway.run._current_max_iterations", lambda: 90)
+    monkeypatch.setattr("hermes_cli.tools_config._get_platform_tools", lambda *_: set())
+    monkeypatch.setattr(
+        "gateway.run._resolve_runtime_agent_kwargs_for_provider",
+        lambda provider: {
+            "provider": provider,
+            "api_key": f"sk-{provider}",
+            "base_url": f"https://{provider}.example/v1",
+            "api_mode": "chat_completions",
+        },
+    )
+
+
+@pytest.mark.asyncio
+async def test_session_chat_builds_raw_provider_model_route_when_alias_missing(adapter, session_db):
+    session_id = session_db.create_session("route-session", "api_server")
+    mock_run = AsyncMock(
+        return_value=(
+            {
+                "final_response": "ok",
+                "session_id": session_id,
+                "runtime": {"provider": "nous", "model": "x-ai/grok-4.5", "route_source": "raw_request"},
+            },
+            {"total_tokens": 2, "runtime": {"provider": "nous", "model": "x-ai/grok-4.5"}},
+        )
+    )
+    app = _create_session_app(adapter)
+    with patch.object(adapter, "_resolve_route", return_value=None), patch.object(adapter, "_run_agent", mock_run):
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                f"/api/sessions/{session_id}/chat",
+                json={
+                    "message": "hello",
+                    "provider": "nous",
+                    "model": "x-ai/grok-4.5",
+                    "require_model_lock": True,
+                },
+            )
+            assert resp.status == 200, await resp.text()
+            payload = await resp.json()
+
+    kwargs = mock_run.call_args.kwargs
+    assert kwargs["route"] == {"provider": "nous", "model": "x-ai/grok-4.5"}
+    assert payload["runtime"]["provider"] == "nous"
+    assert payload["runtime"]["model"] == "x-ai/grok-4.5"
+    assert payload["runtime"]["requested"]["model"] == "x-ai/grok-4.5"
+
+
+@pytest.mark.asyncio
+async def test_session_chat_passes_runtime_options_to_run_agent(adapter, session_db):
+    session_id = session_db.create_session("options-session", "api_server")
+    mock_run = AsyncMock(return_value=({"final_response": "ok", "session_id": session_id}, {}))
+    app = _create_session_app(adapter)
+    with patch.object(adapter, "_resolve_route", return_value=None), patch.object(adapter, "_run_agent", mock_run):
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                f"/api/sessions/{session_id}/chat",
+                json={
+                    "message": "hello",
+                    "provider": "nous",
+                    "model": "x-ai/grok-4.5",
+                    "model_options": {
+                        "reasoning": {"enabled": True, "effort": "xhigh"},
+                        "service_tier": "priority",
+                        "fast": True,
+                    },
+                },
+            )
+            assert resp.status == 200, await resp.text()
+
+    kwargs = mock_run.call_args.kwargs
+    # In the merged design model_options travel raw to _create_agent, which
+    # parses reasoning/service-tier itself (see _request_reasoning_config /
+    # _request_service_tier) — there is no separate runtime_options kwarg.
+    assert kwargs["model_options"] == {
+        "reasoning": {"enabled": True, "effort": "xhigh"},
+        "service_tier": "priority",
+        "fast": True,
+    }
+
+
+@pytest.mark.asyncio
+async def test_session_chat_stream_uses_same_runtime_lock(adapter, session_db):
+    session_id = session_db.create_session("stream-lock-session", "api_server")
+    captured = {}
+
+    async def fake_run(**kwargs):
+        captured.update(kwargs)
+        kwargs["stream_delta_callback"]("hi")
+        return (
+            {
+                "final_response": "hi",
+                "session_id": session_id,
+                "runtime": {
+                    "provider": "nous",
+                    "model": "x-ai/grok-4.5",
+                    "requested": {"provider": "nous", "model": "x-ai/grok-4.5"},
+                    "route_source": "raw_request",
+                },
+            },
+            {
+                "total_tokens": 1,
+                "runtime": {
+                    "provider": "nous",
+                    "model": "x-ai/grok-4.5",
+                    "requested": {"provider": "nous", "model": "x-ai/grok-4.5"},
+                },
+            },
+        )
+
+    app = _create_session_app(adapter)
+    with patch.object(adapter, "_resolve_route", return_value=None), patch.object(adapter, "_run_agent", side_effect=fake_run):
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                f"/api/sessions/{session_id}/chat/stream",
+                json={
+                    "message": "stream",
+                    "provider": "nous",
+                    "model": "x-ai/grok-4.5",
+                    "model_options": {"reasoning": {"enabled": False}},
+                    "require_model_lock": True,
+                },
+            )
+            assert resp.status == 200, await resp.text()
+            body = await resp.text()
+
+    assert captured["route"] == {"provider": "nous", "model": "x-ai/grok-4.5"}
+    assert captured["model_options"] == {"reasoning": {"enabled": False}}
+    assert captured["confirmed_runtime_lock"] is True
+    assert "x-ai/grok-4.5" in body
+    assert "run.started" in body or "event: run.started" in body
+
+
+@pytest.mark.asyncio
+async def test_create_session_respects_browser_source_and_model_lock(adapter, session_db):
+    app = _create_session_app(adapter)
+    async with TestClient(TestServer(app)) as cli:
+        resp = await cli.post(
+            "/api/sessions",
+            json={
+                "id": "browser-lock-session",
+                "source": "hermes_browser",
+                "provider": "nous",
+                "model": "x-ai/grok-4.5",
+                "require_model_lock": True,
+                "title": "Browser lock",
+                "system_prompt": "browser prompt",
+            },
+        )
+        assert resp.status == 201, await resp.text()
+        payload = await resp.json()
+
+    assert payload["session"]["source"] == "hermes_browser"
+    assert payload["session"]["model"] == "x-ai/grok-4.5"
+    row = session_db.get_session("browser-lock-session")
+    assert row["source"] == "hermes_browser"
+    assert row["model"] == "x-ai/grok-4.5"
+    import json as _json
+    model_config = row.get("model_config")
+    if isinstance(model_config, str):
+        model_config = _json.loads(model_config)
+    assert model_config["browser_model_lock"]["provider"] == "nous"
+    assert model_config["browser_model_lock"]["model"] == "x-ai/grok-4.5"
+    assert model_config["browser_model_lock"]["confirmed"] is True
+
+
+@pytest.mark.asyncio
+async def test_session_model_lock_endpoint_persists_and_invalidates_prompt(adapter, session_db):
+    session_id = session_db.create_session(
+        "lock-endpoint-session",
+        "api_server",
+        model="gpt-5.5",
+        model_config={"_branched_from": "parent-session"},
+        system_prompt="Conversation started:\nModel: gpt-5.5\nProvider: openai-codex\n",
+    )
+    app = _create_session_app(adapter)
+    _register_session_model_route(app, adapter)
+    with patch.object(adapter, "_resolve_route", return_value=None):
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                f"/api/sessions/{session_id}/model",
+                json={
+                    "provider": "nous",
+                    "model": "x-ai/grok-4.5",
+                    "model_options": {"reasoning": {"enabled": True, "effort": "high"}},
+                    "require_model_lock": True,
+                },
+            )
+            assert resp.status == 200, await resp.text()
+            payload = await resp.json()
+
+    assert payload["object"] == "hermes.session.model_lock"
+    assert payload["runtime"]["requested"]["provider"] == "nous"
+    assert payload["runtime"]["model"] == "x-ai/grok-4.5"
+    assert payload["runtime"]["model_lock"] in {"accepted", "confirmed"}
+    row = session_db.get_session(session_id)
+    assert row["model"] == "x-ai/grok-4.5"
+    assert row["system_prompt"] is None
+    import json as _json
+    model_config = row.get("model_config")
+    if isinstance(model_config, str):
+        model_config = _json.loads(model_config)
+    assert model_config["_branched_from"] == "parent-session"
+    assert model_config["browser_model_lock"]["provider"] == "nous"
+
+
+@pytest.mark.asyncio
+async def test_session_model_lock_endpoint_then_chat_reuses_persisted_lock_and_provider_credentials(
+    adapter,
+    session_db,
+    monkeypatch,
+):
+    session_id = session_db.create_session(
+        "endpoint-lock-chat",
+        "api_server",
+        model="gpt-5.5",
+        system_prompt="Conversation started:\nModel: gpt-5.5\nProvider: openai-codex\n",
+    )
+    captured = {}
+
+    class FakeAgent:
+        session_prompt_tokens = 0
+        session_completion_tokens = 0
+        session_total_tokens = 0
+
+        def __init__(self, **kwargs):
+            captured.update(kwargs)
+            self.session_id = kwargs["session_id"]
+            self.provider = kwargs.get("provider") or ""
+            self.model = kwargs.get("model") or ""
+
+        def run_conversation(self, user_message, conversation_history, task_id):
+            return {"final_response": "locked", "session_id": self.session_id}
+
+    _patch_api_server_runtime(monkeypatch)
+    monkeypatch.setattr("run_agent.AIAgent", FakeAgent)
+    monkeypatch.setattr(
+        adapter,
+        "_session_model_override_for",
+        lambda *_: {
+            "model": "session/override-model",
+            "provider": "openai-codex",
+            "api_key": "sk-session-override",
+            "base_url": "https://override.example/v1",
+            "api_mode": "codex_responses",
+        },
+    )
+
+    app = _create_session_app(adapter)
+    _register_session_model_route(app, adapter)
+    with patch.object(adapter, "_resolve_route", return_value=None):
+        async with TestClient(TestServer(app)) as cli:
+            lock_resp = await cli.post(
+                f"/api/sessions/{session_id}/model",
+                json={
+                    "provider": "nous",
+                    "model": "x-ai/grok-4.5",
+                    "require_model_lock": True,
+                },
+            )
+            assert lock_resp.status == 200, await lock_resp.text()
+
+            resp = await cli.post(
+                f"/api/sessions/{session_id}/chat",
+                json={"message": "use the stored lock"},
+            )
+            assert resp.status == 200, await resp.text()
+            payload = await resp.json()
+
+    assert captured["provider"] == "nous"
+    assert captured["model"] == "x-ai/grok-4.5"
+    assert captured["api_key"] == "sk-nous"
+    assert captured["base_url"] == "https://nous.example/v1"
+    assert payload["runtime"]["provider"] == "nous"
+    assert payload["runtime"]["model"] == "x-ai/grok-4.5"
+    assert payload["runtime"]["requested"] == {
+        "provider": "nous",
+        "model": "x-ai/grok-4.5",
+    }
+    assert payload["runtime"]["route_source"] == "session_model_lock"
+
+
+@pytest.mark.asyncio
+async def test_session_model_lock_endpoint_then_chat_stream_reuses_persisted_lock(
+    adapter,
+    session_db,
+):
+    session_id = session_db.create_session("endpoint-lock-stream", "api_server")
+    captured = {}
+
+    async def fake_run(**kwargs):
+        captured.update(kwargs)
+        kwargs["stream_delta_callback"]("hi")
+        return (
+            {
+                "final_response": "hi",
+                "session_id": session_id,
+                "runtime": {
+                    "provider": "nous",
+                    "model": "x-ai/grok-4.5",
+                    "requested": {"provider": "nous", "model": "x-ai/grok-4.5"},
+                    "route_source": "session_model_lock",
+                },
+            },
+            {
+                "total_tokens": 1,
+                "runtime": {
+                    "provider": "nous",
+                    "model": "x-ai/grok-4.5",
+                    "requested": {"provider": "nous", "model": "x-ai/grok-4.5"},
+                    "route_source": "session_model_lock",
+                },
+            },
+        )
+
+    app = _create_session_app(adapter)
+    _register_session_model_route(app, adapter)
+    with patch.object(adapter, "_resolve_route", return_value=None), patch.object(
+        adapter,
+        "_run_agent",
+        side_effect=fake_run,
+    ):
+        async with TestClient(TestServer(app)) as cli:
+            lock_resp = await cli.post(
+                f"/api/sessions/{session_id}/model",
+                json={
+                    "provider": "nous",
+                    "model": "x-ai/grok-4.5",
+                    "require_model_lock": True,
+                },
+            )
+            assert lock_resp.status == 200, await lock_resp.text()
+
+            resp = await cli.post(
+                f"/api/sessions/{session_id}/chat/stream",
+                json={"message": "stream with stored lock"},
+            )
+            assert resp.status == 200, await resp.text()
+            body = await resp.text()
+
+    assert captured["route"] == {"provider": "nous", "model": "x-ai/grok-4.5"}
+    assert captured["requested_runtime"]["provider"] == "nous"
+    assert captured["requested_runtime"]["model"] == "x-ai/grok-4.5"
+    assert captured["route_source"] == "session_model_lock"
+    assert "x-ai/grok-4.5" in body
+
+
+@pytest.mark.asyncio
+async def test_run_agent_reports_actual_agent_runtime_not_requested_metadata(adapter, monkeypatch):
+    class FakeAgent:
+        session_prompt_tokens = 0
+        session_completion_tokens = 0
+        session_total_tokens = 0
+
+        def __init__(self):
+            self.session_id = "runtime-session"
+            self.provider = "actual-provider"
+            self.model = "actual-model"
+            self._hermes_api_runtime = {
+                "provider": "requested-provider",
+                "model": "requested-model",
+                "route_source": "raw_request",
+            }
+
+        def run_conversation(self, user_message, conversation_history, task_id):
+            return {"final_response": "ok", "session_id": self.session_id}
+
+    monkeypatch.setattr(adapter, "_create_agent", lambda **kwargs: FakeAgent())
+
+    result, usage = await adapter._run_agent(
+        user_message="hello",
+        conversation_history=[],
+        session_id="runtime-session",
+        route={"provider": "requested-provider", "model": "requested-model"},
+        requested_runtime={
+            "provider": "requested-provider",
+            "model": "requested-model",
+        },
+        route_source="session_model_lock",
+    )
+
+    assert result["runtime"]["provider"] == "actual-provider"
+    assert result["runtime"]["model"] == "actual-model"
+    assert result["runtime"]["requested"] == {
+        "provider": "requested-provider",
+        "model": "requested-model",
+    }
+    assert usage["runtime"]["provider"] == "actual-provider"
+    assert usage["runtime"]["model"] == "actual-model"
+
+
+@pytest.mark.asyncio
+async def test_confirmed_runtime_lock_rejects_actual_runtime_mismatch(adapter, monkeypatch):
+    class FakeAgent:
+        session_prompt_tokens = 0
+        session_completion_tokens = 0
+        session_total_tokens = 0
+        session_id = "mismatch-session"
+        provider = "fallback-provider"
+        model = "fallback-model"
+
+        def run_conversation(self, user_message, conversation_history, task_id):
+            return {"final_response": "wrong runtime", "session_id": self.session_id}
+
+    monkeypatch.setattr(adapter, "_create_agent", lambda **kwargs: FakeAgent())
+
+    with pytest.raises(RuntimeError, match="confirmed model lock runtime mismatch"):
+        await adapter._run_agent(
+            user_message="hello",
+            conversation_history=[],
+            session_id="mismatch-session",
+            route={"provider": "nous", "model": "x-ai/grok-4.5"},
+            requested_runtime={"provider": "nous", "model": "x-ai/grok-4.5"},
+            route_source="session_model_lock",
+            confirmed_runtime_lock=True,
+        )
+
+
+def test_confirmed_runtime_lock_fails_closed_on_provider_resolution_error(adapter, monkeypatch):
+    _patch_api_server_runtime(monkeypatch)
+    # Break BOTH resolution paths (primary picker-based resolver + the
+    # gateway fallback) — a confirmed lock must propagate the failure
+    # instead of constructing an agent on the previous global credentials.
+    monkeypatch.setattr(
+        "hermes_cli.runtime_provider.resolve_runtime_provider",
+        lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("provider unavailable")),
+    )
+    monkeypatch.setattr(
+        "gateway.run._resolve_runtime_agent_kwargs_for_provider",
+        lambda provider: (_ for _ in ()).throw(RuntimeError("provider unavailable")),
+    )
+    agent_ctor = patch("run_agent.AIAgent")
+    with agent_ctor as mocked_agent:
+        with pytest.raises(RuntimeError, match="provider unavailable"):
+            adapter._create_agent(
+                session_id="locked-session",
+                route={"provider": "nous", "model": "x-ai/grok-4.5"},
+                confirmed_runtime_lock=True,
+            )
+    mocked_agent.assert_not_called()
+
+
+def test_confirmed_runtime_lock_disables_global_fallback_model(adapter, monkeypatch):
+    _patch_api_server_runtime(monkeypatch)
+    monkeypatch.setattr(
+        "gateway.run.GatewayRunner._load_fallback_model",
+        staticmethod(lambda: "openrouter/fallback-model"),
+    )
+    captured = {}
+
+    class FakeAgent:
+        provider = "nous"
+        model = "x-ai/grok-4.5"
+
+        def __init__(self, **kwargs):
+            captured.update(kwargs)
+
+    monkeypatch.setattr("run_agent.AIAgent", FakeAgent)
+
+    adapter._create_agent(
+        session_id="locked-session",
+        route={"provider": "nous", "model": "x-ai/grok-4.5"},
+        confirmed_runtime_lock=True,
+    )
+
+    assert captured["fallback_model"] is None
+
+
+@pytest.mark.asyncio
+async def test_unconfirmed_request_does_not_replace_confirmed_session_lock(adapter, session_db):
+    session_id = session_db.create_session("one-off-override", "api_server")
+    session_db.update_session_runtime_lock(
+        session_id,
+        provider="nous",
+        model="x-ai/grok-4.5",
+        route_source="raw_request",
+        confirmed=True,
+    )
+    mock_run = AsyncMock(
+        return_value=(
+            {
+                "final_response": "ok",
+                "session_id": session_id,
+                "runtime": {"provider": "openrouter", "model": "anthropic/claude-sonnet"},
+            },
+            {"total_tokens": 1},
+        )
+    )
+    app = _create_session_app(adapter)
+    with patch.object(adapter, "_resolve_route", return_value=None), patch.object(
+        adapter,
+        "_run_agent",
+        mock_run,
+    ):
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                f"/api/sessions/{session_id}/chat",
+                json={
+                    "message": "one turn only",
+                    "provider": "openrouter",
+                    "model": "anthropic/claude-sonnet",
+                },
+            )
+            assert resp.status == 200, await resp.text()
+
+    import json as _json
+
+    row = session_db.get_session(session_id)
+    config = row["model_config"]
+    if isinstance(config, str):
+        config = _json.loads(config)
+    assert config["browser_model_lock"]["provider"] == "nous"
+    assert config["browser_model_lock"]["model"] == "x-ai/grok-4.5"
+    assert config["browser_model_lock"]["confirmed"] is True
+
+
+@pytest.mark.asyncio
+async def test_require_model_lock_hard_fails_when_global_default_would_be_used(adapter, session_db, monkeypatch):
+    session_id = session_db.create_session("lock-fail-session", "api_server")
+    monkeypatch.setattr(adapter, "_model_name", "gpt-5.5")
+    app = _create_session_app(adapter)
+    with patch.object(adapter, "_resolve_route", return_value=None), patch.object(adapter, "_run_agent", new_callable=AsyncMock) as mock_run:
+        async with TestClient(TestServer(app)) as cli:
+            # empty model + require_model_lock must not silently fall through
+            resp = await cli.post(
+                f"/api/sessions/{session_id}/chat",
+                json={
+                    "message": "hello",
+                    "provider": "nous",
+                    "model": "",
+                    "require_model_lock": True,
+                },
+            )
+            assert resp.status in (400, 409), await resp.text()
+            body = await resp.json()
+            assert body["error"]["code"] in {"model_lock_unavailable", "invalid_model_lock", "missing_model"}
+    mock_run.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_capabilities_advertises_session_model_lock(adapter):
+    app = _create_session_app(adapter)
+    async with TestClient(TestServer(app)) as cli:
+        resp = await cli.get("/v1/capabilities")
+        assert resp.status == 200
+        data = await resp.json()
+    assert data["features"]["session_model_lock"] is True
+    assert data["endpoints"]["session_model_lock"] == {
+        "method": "POST",
+        "path": "/api/sessions/{session_id}/model",
+    }

tests/test_hermes_state.py

@@ -543,6 +543,30 @@ class TestSessionLifecycle:
                                model="xiaomi/mimo-v2.5-pro")
         assert db.get_session("s1")["model"] == "xiaomi/mimo-v2.5"
 
+    def test_update_session_model_clears_browser_lock_and_preserves_lineage(self, db):
+        """A later /model switch must replace, not compete with, a Browser lock."""
+        db.create_session(
+            session_id="s1",
+            source="hermes_browser",
+            model="x-ai/grok-4.5",
+            model_config={
+                "_branched_from": "parent-session",
+                "browser_model_lock": {
+                    "provider": "nous",
+                    "model": "x-ai/grok-4.5",
+                    "confirmed": True,
+                },
+            },
+        )
+
+        db.update_session_model("s1", "anthropic/claude-opus-4.8")
+
+        session = db.get_session("s1")
+        model_config = json.loads(session["model_config"])
+        assert session["model"] == "anthropic/claude-opus-4.8"
+        assert "browser_model_lock" not in model_config
+        assert model_config["_branched_from"] == "parent-session"
+
     def test_update_session_billing_route_overwrites_after_switch(self, db):
         """A mid-session provider switch must overwrite the billing route.