Image routing ignores session model overrides

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #57649 fix(vision): custom-provider routing — stripped custom:<name> override, session model, canonical slug

  merged to main

  files: agent/image_routing.py, gateway/run.py, hermes_cli/model_setup_flows.py, tests/agent/test_image_routing.py, tests/gateway/test_image_input_routing_runtime.py, tests/gateway/test_native_image_buffer_isolation.py, tests/gateway/test_queued_native_image_session_key.py, tests/hermes_cli/test_model_provider_persistence.py, tests/hermes_cli/test_runtime_provider_resolution.py

  Show the diff

  • agent/image_routing.py

    @@ -185,7 +185,8 @@ def _supports_vision_override(
           2. ``providers.<provider>.models.<model>.supports_vision``
              (named custom providers — ``provider`` may be the runtime-resolved
              value ``"custom"`` and/or the user-declared name under
    -         ``model.provider``; both are tried)
    +         ``model.provider``; both are tried. For ``custom:<name>`` syntax,
    +         the stripped ``<name>`` is also tried as a provider key.)
     
         Returns None when no override is set, so the caller falls through to
         models.dev. Returns False explicitly only when the user wrote a
    @@ -205,11 +206,16 @@ def _supports_vision_override(
         # get rewritten to provider="custom" at runtime
         # (hermes_cli/runtime_provider.py:_resolve_named_custom_runtime), so the
         # config still holds the user-declared name under model.provider. Try
    -    # both as candidate provider keys.
    +    # both as candidate provider keys, plus the stripped suffix from
    +    # "custom:<name>" (where <name> is the key under providers:).
         config_provider = str(model_cfg.get("provider") or "").strip()
    +    # Extract the stripped name from "custom:<name>" if present
    +    stripped_suffix = ""
    +    if config_provider.startswith("custom:"):
    +        stripped_suffix = config_provider[len("custom:"):]
         providers_raw = cfg.get("providers")
         providers_cfg: Dict[str, Any] = providers_raw if isinstance(providers_raw, dict) else {}
    -    for p in dict.fromkeys(filter(None, (provider, config_provider))):
    +    for p in dict.fromkeys(filter(None, (provider, config_provider, stripped_suffix))):
             entry_raw = providers_cfg.get(p)
             entry: Dict[str, Any] = entry_raw if isinstance(entry_raw, dict) else {}
             models_raw = entry.get("models")

  • gateway/run.py

    @@ -10035,7 +10035,10 @@ async def _prepare_inbound_message_text(
                 if image_paths:
                     # Decide routing: native (attach pixels) vs text (vision_analyze
                     # pre-run + prepend description).  See agent/image_routing.py.
    -                _img_mode = self._decide_image_input_mode()
    +                _img_mode = self._decide_image_input_mode(
    +                    source=source,
    +                    session_key=session_key,
    +                )
                     if _img_mode == "native":
                         # Defer attachment to the run_conversation call site.
                         pending_native = getattr(self, "_pending_native_image_paths_by_session", None)
    @@ -14381,25 +14384,62 @@ def _shutdown_executor(self) -> None:
             except TypeError:
                 executor.shutdown(wait=False)
     
    -    def _decide_image_input_mode(self) -> str:
    -        """Resolve the image-input routing for the currently active model.
    +    def _decide_image_input_mode(
    +        self,
    +        *,
    +        source: Optional[SessionSource] = None,
    +        session_key: Optional[str] = None,
    +        user_config: Optional[dict] = None,
    +        provider: Optional[str] = None,
    +        model: Optional[str] = None,
    +    ) -> str:
    +        """Resolve image-input routing for the effective model this turn.
     
             Returns ``"native"`` (attach pixels on the user turn) or ``"text"``
             (pre-analyze with vision_analyze and prepend the description). See
             agent/image_routing.py for the full decision table.
     
    -        The active provider/model are read from config.yaml so the decision
    -        tracks ``/model`` switches automatically on the next message.
    +        Gateway sessions can have /model overrides that live outside
    +        config.yaml. Image preprocessing runs before AIAgent sets the
    +        auxiliary_client runtime globals, so resolve the same per-session
    +        runtime bundle the upcoming agent turn will use instead of consulting
    +        only the persisted default model.
             """
             try:
                 from agent.image_routing import decide_image_input_mode
                 from agent.auxiliary_client import _read_main_model, _read_main_provider
                 from hermes_cli.config import load_config
     
    -            cfg = load_config()
    -            provider = _read_main_provider()
    -            model = _read_main_model()
    -            return decide_image_input_mode(provider, model, cfg)
    +            cfg = user_config if isinstance(user_config, dict) else load_config()
    +            resolved_provider = (provider or "").strip()
    +            resolved_model = (model or "").strip()
    +
    +            needs_session_runtime = not resolved_provider or not resolved_model
    +            has_session_identity = source is not None or session_key
    +            if needs_session_runtime and has_session_identity:
    +                try:
    +                    turn_model, runtime_kwargs = self._resolve_session_agent_runtime(
    +                        source=source,
    +                        session_key=session_key,
    +                        user_config=cfg,
    +                    )
    +                    if not resolved_model and isinstance(turn_model, str):
    +                        resolved_model = turn_model.strip()
    +                    runtime_provider = runtime_kwargs.get("provider") if isinstance(runtime_kwargs, dict) else None
    +                    if not resolved_provider and isinstance(runtime_provider, str):
    +                        resolved_provider = runtime_provider.strip()
    +                except Exception as exc:
    +                    logger.debug(
    +                        "image_routing: session runtime resolution failed, falling back to config — %s",
    +                        exc,
    +                    )
    +
    +            if not resolved_provider:
    +                resolved_provider = _read_main_provider()
    +            if not resolved_model:
    +                resolved_model = _read_main_model()
    +
    +            return decide_image_input_mode(resolved_provider, resolved_model, cfg)
             except Exception as exc:
                 logger.debug("image_routing: decision failed, falling back to text — %s", exc)
                 return "text"

  • hermes_cli/model_setup_flows.py

    @@ -1446,7 +1446,7 @@ def _model_flow_named_custom(config, provider_info):
             model = {"default": model} if model else {}
             cfg["model"] = model
         if provider_key:
    -        model["provider"] = provider_key
    +        model["provider"] = "custom:" + provider_key.strip().lower().replace(" ", "-")
             model.pop("base_url", None)
             model.pop("api_key", None)
         else:

  • tests/agent/test_image_routing.py

    @@ -224,6 +224,37 @@ def test_malformed_sections_are_ignored(self):
             cfg = {"model": "some-string", "providers": ["not-a-dict"]}
             assert _supports_vision_override(cfg, "custom", "my-llava") is None
     
    +    def test_custom_colon_name_stripped_suffix_lookup(self):
    +        # model.provider: custom:my-proxy → should resolve stripped key "my-proxy"
    +        cfg = {
    +            "model": {"provider": "custom:my-proxy"},
    +            "providers": {
    +                "my-proxy": {"models": {"gpt-5.5": {"supports_vision": True}}},
    +            },
    +        }
    +        assert _supports_vision_override(cfg, "custom", "gpt-5.5") is True
    +
    +    def test_custom_colon_name_stripped_suffix_false(self):
    +        # Explicitly disabled vision on the stripped key.
    +        cfg = {
    +            "model": {"provider": "custom:my-proxy"},
    +            "providers": {
    +                "my-proxy": {"models": {"gpt-5.5": {"supports_vision": False}}},
    +            },
    +        }
    +        assert _supports_vision_override(cfg, "custom", "gpt-5.5") is False
    +
    +    def test_custom_colon_name_no_stripped_key_falls_through(self):
    +        # custom:my-proxy but providers only has "custom" — stripped key
    +        # doesn't match, but "custom" does via runtime provider.
    +        cfg = {
    +            "model": {"provider": "custom:my-proxy"},
    +            "providers": {
    +                "custom": {"models": {"gpt-5.5": {"supports_vision": True}}},
    +            },
    +        }
    +        assert _supports_vision_override(cfg, "custom", "gpt-5.5") is True
    +
     
     # ─── _lookup_supports_vision (override-aware) ────────────────────────────────
     

  • tests/gateway/test_image_input_routing_runtime.py

    @@ -0,0 +1,140 @@
    +import pytest
    +
    +from gateway.config import GatewayConfig, Platform, PlatformConfig
    +from gateway.platforms.base import MessageEvent, MessageType
    +from gateway.run import GatewayRunner
    +from gateway.session import SessionSource
    +
    +
    +def _make_runner() -> GatewayRunner:
    +    runner = object.__new__(GatewayRunner)
    +    runner.config = GatewayConfig(
    +        platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="fake")}
    +    )
    +    runner.adapters = {}
    +    runner._pending_native_image_paths_by_session = {}
    +    runner._session_model_overrides = {}
    +    runner._session_reasoning_overrides = {}
    +    return runner
    +
    +
    +def _source() -> SessionSource:
    +    return SessionSource(
    +        platform=Platform.TELEGRAM,
    +        chat_id="273403055",
    +        chat_type="dm",
    +        user_id="42",
    +        user_name="Maxim",
    +    )
    +
    +
    +def _image_event(text: str = "look") -> MessageEvent:
    +    return MessageEvent(
    +        text=text,
    +        message_type=MessageType.PHOTO,
    +        source=_source(),
    +        media_urls=["/tmp/cashback.png"],
    +        media_types=["image/png"],
    +    )
    +
    +
    +def _auto_config() -> dict:
    +    return {
    +        "agent": {"image_input_mode": "auto"},
    +        "auxiliary": {"vision": {"provider": "auto", "model": "", "base_url": ""}},
    +        "model": {"provider": "xiaomi", "default": "mimo-v2.5-pro"},
    +    }
    +
    +
    [redacted:email]
    +async def test_prepare_image_routing_uses_session_vision_model_override(monkeypatch):
    +    """Telegram /model overrides must affect native-vs-text image routing.
    +
    +    Regression: _prepare_inbound_message_text used config.yaml's default model
    +    before the per-session model override was installed on auxiliary_client's
    +    runtime globals. A Telegram session switched to a vision model still had
    +    screenshots pre-analyzed as text when config.default was text-only.
    +    """
    +    runner = _make_runner()
    +    source = _source()
    +    event = _image_event()
    +    cfg = _auto_config()
    +
    +    monkeypatch.setattr("gateway.run._load_gateway_config", lambda: cfg)
    +    monkeypatch.setattr("hermes_cli.config.load_config", lambda: cfg)
    +    monkeypatch.setattr("agent.auxiliary_client._read_main_provider", lambda: "xiaomi")
    +    monkeypatch.setattr("agent.auxiliary_client._read_main_model", lambda: "mimo-v2.5-pro")
    +    monkeypatch.setattr(
    +        runner,
    +        "_resolve_session_agent_runtime",
    +        lambda **_: ("gpt-5.5", {"provider": "openai-codex"}),
    +    )
    +
    +    def fake_supports(provider, model, config):
    +        return provider == "openai-codex" and model == "gpt-5.5"
    +
    +    monkeypatch.setattr("agent.image_routing._lookup_supports_vision", fake_supports)
    +
    +    async def fail_enrich(*_args, **_kwargs):
    +        pytest.fail("vision-capable session override should use native image routing")
    +
    +    monkeypatch.setattr(runner, "_enrich_message_with_vision", fail_enrich)
    +
    +    result = await runner._prepare_inbound_message_text(
    +        event=event,
    +        source=source,
    +        history=[],
    +    )
    +
    +    session_key = runner._session_key_for_source(source)
    +    assert result == "look"
    +    assert runner._pending_native_image_paths_by_session[session_key] == [
    +        "/tmp/cashback.png"
    +    ]
    +
    +
    [redacted:email]
    +async def test_prepare_image_routing_falls_back_to_text_for_text_only_session_override(monkeypatch):
    +    """A text-only session override should get vision_analyze text fallback.
    +
    +    Regression mirror case: if config.default is a vision model but the current
    +    Telegram session is switched to a text-only provider (for example Mimo),
    +    auto routing must not attach pixels natively to the text-only model.
    +    """
    +    runner = _make_runner()
    +    source = _source()
    +    event = _image_event()
    +    cfg = _auto_config()
    +    cfg["model"] = {"provider": "openai-codex", "default": "gpt-5.5"}
    +
    +    monkeypatch.setattr("gateway.run._load_gateway_config", lambda: cfg)
    +    monkeypatch.setattr("hermes_cli.config.load_config", lambda: cfg)
    +    monkeypatch.setattr("agent.auxiliary_client._read_main_provider", lambda: "openai-codex")
    +    monkeypatch.setattr("agent.auxiliary_client._read_main_model", lambda: "gpt-5.5")
    +    monkeypatch.setattr(
    +        runner,
    +        "_resolve_session_agent_runtime",
    +        lambda **_: ("mimo-v2.5-pro", {"provider": "xiaomi"}),
    +    )
    +
    +    def fake_supports(provider, model, config):
    +        return provider == "openai-codex" and model == "gpt-5.5"
    +
    +    monkeypatch.setattr("agent.image_routing._lookup_supports_vision", fake_supports)
    +
    +    async def fake_enrich(user_text, image_paths):
    +        assert user_text == "look"
    +        assert image_paths == ["/tmp/cashback.png"]
    +        return "[vision summary]\n\nlook"
    +
    +    monkeypatch.setattr(runner, "_enrich_message_with_vision", fake_enrich)
    +
    +    result = await runner._prepare_inbound_message_text(
    +        event=event,
    +        source=source,
    +        history=[],
    +    )
    +
    +    session_key = runner._session_key_for_source(source)
    +    assert result == "[vision summary]\n\nlook"
    +    assert runner._pending_native_image_paths_by_session.get(session_key) is None

  • tests/gateway/test_native_image_buffer_isolation.py

    @@ -14,7 +14,7 @@ def _make_runner() -> GatewayRunner:
         runner.adapters = {}
         runner._model = "openai/gpt-4.1-mini"
         runner._base_url = None
    -    runner._decide_image_input_mode = lambda: "native"
    +    runner._decide_image_input_mode = lambda **_: "native"
         return runner
     
     

  • tests/gateway/test_queued_native_image_session_key.py

    @@ -86,7 +86,7 @@ def _make_runner(adapter):
         )
         runner._model = "openai/gpt-4.1-mini"
         runner._base_url = None
    -    runner._decide_image_input_mode = lambda: "native"
    +    runner._decide_image_input_mode = lambda **_kw: "native"
         return runner
     
     

  • tests/hermes_cli/test_model_provider_persistence.py

    @@ -207,6 +207,51 @@ def test_named_custom_provider_preserves_explicit_api_mode(self, config_home):
             assert model.get("base_url") == "https://packy.example.com/v1"
             assert model.get("api_mode") == "codex_responses"
     
    +    def test_named_custom_provider_with_builtin_slug_persists_custom_prefix(
    +        self, config_home, monkeypatch
    +    ):
    +        """providers.<builtin-slug> must persist as a named custom provider."""
    +        import yaml
    +
    +        from hermes_cli.main import _model_flow_named_custom
    +
    +        config_path = config_home / "config.yaml"
    +        config_path.write_text(
    +            "providers:\n"
    +            "  minimax-cn:\n"
    +            "    name: MiniMax CN Proxy\n"
    +            "    api: https://mimimax.cn/v1\n"
    +            "    key_env: MINIMAX_CN_PROXY_KEY\n"
    +            "    transport: chat_completions\n"
    +            "    model: MiniMax-M3\n"
    +            "    default_model: MiniMax-M3\n"
    +        )
    +        monkeypatch.setenv("MINIMAX_CN_PROXY_KEY", "proxy-secret")
    +
    +        provider_info = {
    +            "name": "MiniMax CN Proxy",
    +            "base_url": "https://mimimax.cn/v1",
    +            "api_key": "",
    +            "key_env": "MINIMAX_CN_PROXY_KEY",
    +            "model": "MiniMax-M3",
    +            "api_mode": "chat_completions",
    +            "provider_key": "minimax-cn",
    +        }
    +
    +        with patch("hermes_cli.auth._save_model_choice"), \
    +             patch("hermes_cli.auth.deactivate_provider"), \
    +             patch("hermes_cli.models.fetch_api_models", return_value=["MiniMax-M3"]), \
    +             patch("hermes_cli.curses_ui.curses_radiolist", side_effect=OSError("no tty in test")), \
    +             patch("builtins.input", return_value="1"):
    +            _model_flow_named_custom({}, provider_info)
    +
    +        config = yaml.safe_load(config_path.read_text()) or {}
    +        model = config.get("model")
    +        assert isinstance(model, dict)
    +        assert model.get("provider") == "custom:minimax-cn"
    +        assert "base_url" not in model
    +        assert "api_key" not in model
    +
         def test_copilot_acp_provider_saved_when_selected(self, config_home):
             """_model_flow_copilot_acp should persist provider/base_url/model together."""
             from hermes_cli.main import _model_flow_copilot_acp
    @@ -555,4 +600,3 @@ def fake_choice(choices, *, default=0, title=""):
                 _select_zai_endpoint(custom_url)
     
             assert captured["default"] == expected_default
    -

  • tests/hermes_cli/test_runtime_provider_resolution.py

    @@ -1272,6 +1272,33 @@ def test_resolve_requested_provider_precedence(monkeypatch):
         assert rp.resolve_requested_provider() == "auto"
     
     
    +def test_resolve_runtime_provider_named_custom_with_builtin_slug(monkeypatch):
    +    monkeypatch.setenv("MINIMAX_CN_PROXY_KEY", "proxy-secret")
    +    monkeypatch.setattr(
    +        rp,
    +        "load_config",
    +        lambda: {
    +            "model": {"provider": "custom:minimax-cn"},
    +            "providers": {
    +                "minimax-cn": {
    +                    "name": "MiniMax CN Proxy",
    +                    "api": "https://mimimax.cn/v1",
    +                    "key_env": "MINIMAX_CN_PROXY_KEY",
    +                    "transport": "chat_completions",
    +                    "default_model": "MiniMax-M3",
    +                }
    +            },
    +        },
    +    )
    +
    +    resolved = rp.resolve_runtime_provider()
    +
    +    assert resolved["provider"] == "custom"
    +    assert resolved["base_url"] == "https://mimimax.cn/v1"
    +    assert resolved["api_key"] == "proxy-secret"
    +    assert resolved["api_mode"] == "chat_completions"
    +
    +
     # ── api_mode config override tests ──────────────────────────────────────
     
     

• #63521 fix(auxiliary): isolate model-scoped runtime state

  merged to main

  files: agent/auxiliary_client.py, agent/context_compressor.py, agent/image_routing.py, agent/turn_context.py, gateway/run.py, run_agent.py, scripts/release.py, tests/agent/test_auxiliary_runtime_cache_key.py, tests/agent/test_context_compressor.py, tests/agent/test_turn_context.py, tests/gateway/test_image_input_routing_runtime.py, tests/run_agent/test_compression_feasibility.py, tests/tui_gateway/test_image_routing_stale_model.py, tui_gateway/server.py

  Show the diff

  • agent/auxiliary_client.py

    @@ -41,6 +41,9 @@
     """
     
     import contextlib
    +import contextvars
    +import hashlib
    +import inspect
     import json
     import logging
     import os
    @@ -2207,7 +2210,7 @@ def _read_main_model() -> str:
         that gate on "the active main model" (e.g. ``vision_analyze``'s native
         fast path) see the live runtime, not the persisted config default.
         """
    -    override = _RUNTIME_MAIN_MODEL
    +    override = _runtime_main_value("model")
         if isinstance(override, str) and override.strip():
             return override.strip()
         try:
    @@ -2234,7 +2237,7 @@ def _read_main_provider() -> str:
         Runtime override: see ``_read_main_model`` — same mechanism for the
         provider half of the runtime tuple.
         """
    -    override = _RUNTIME_MAIN_PROVIDER
    +    override = _runtime_main_value("provider")
         if isinstance(override, str) and override.strip():
             return override.strip().lower()
         try:
    @@ -2263,7 +2266,7 @@ def _read_main_api_key() -> str:
         the main model's credentials instead of falling to ``no-key-required``
         (issue #9318).
         """
    -    override = _RUNTIME_MAIN_API_KEY
    +    override = _runtime_main_value("api_key")
         if isinstance(override, str) and override.strip():
             return override.strip()
         try:
    @@ -2284,7 +2287,7 @@ def _read_main_base_url() -> str:
     
         Same override-then-config pattern as ``_read_main_api_key``.
         """
    -    override = _RUNTIME_MAIN_BASE_URL
    +    override = _runtime_main_value("base_url")
         if isinstance(override, str) and override.strip():
             return override.strip()
         try:
    @@ -2320,13 +2323,55 @@ def _read_main_api_key_if_same_host(aux_base_url: str) -> str:
         return _read_main_api_key()
     
     
    -# Process-local override set by AIAgent at session/turn start. Single-threaded
    -# per turn — no lock needed. Cleared by ``clear_runtime_main()``.
    +# Compatibility mirrors for older readers/tests. The authoritative value is
    +# the ContextVar below: gateway sessions can overlap in one process, so a
    +# process-global tuple is not safe as routing or cache-key input.
     _RUNTIME_MAIN_PROVIDER: str = ""
     _RUNTIME_MAIN_MODEL: str = ""
     _RUNTIME_MAIN_BASE_URL: str = ""
    -_RUNTIME_MAIN_API_KEY: str = ""
    +_RUNTIME_MAIN_API_KEY: Any = ""
     _RUNTIME_MAIN_API_MODE: str = ""
    +_RUNTIME_MAIN_AUTH_MODE: str = ""
    +_RUNTIME_MAIN_CONTEXT: contextvars.ContextVar[Optional[Dict[str, Any]]] = (
    +    contextvars.ContextVar("auxiliary_runtime_main", default=None)
    +)
    +_RUNTIME_MAIN_COMPAT_SNAPSHOT: Tuple[Any, ...] = ("", "", "", "", "", "")
    +_RUNTIME_MAIN_COMPAT_LOCK = threading.Lock()
    +
    +
    +def _compat_runtime_main() -> Optional[Dict[str, Any]]:
    +    """Expose deliberately patched legacy globals in a single main context.
    +
    +    ``set_runtime_main`` mirrors values into the old module attributes for
    +    introspection, but those mirrors must never become runtime inputs. A direct
    +    patch is recognized only when it differs from the mirrored snapshot and
    +    only on the main thread, keeping concurrent session workers isolated.
    +    """
    +    if threading.current_thread() is not threading.main_thread():
    +        return None
    +    values = (
    +        _RUNTIME_MAIN_PROVIDER,
    +        _RUNTIME_MAIN_MODEL,
    +        _RUNTIME_MAIN_BASE_URL,
    +        _RUNTIME_MAIN_API_KEY,
    +        _RUNTIME_MAIN_API_MODE,
    +        _RUNTIME_MAIN_AUTH_MODE,
    +    )
    +    if values == _RUNTIME_MAIN_COMPAT_SNAPSHOT:
    +        return None
    +    return dict(zip(_MAIN_RUNTIME_FIELDS, values))
    +
    +
    +def _runtime_main_value(field: str) -> Any:
    +    """Read one runtime field through context-local/controlled legacy state."""
    +    runtime = _RUNTIME_MAIN_CONTEXT.get()
    +    if runtime is None:
    +        runtime = _compat_runtime_main()
    +    if isinstance(runtime, dict):
    +        value = runtime.get(field)
    +        if value:
    +            return value
    +    return ""
     
     
     def set_runtime_main(
    @@ -2334,38 +2379,85 @@ def set_runtime_main(
         model: str,
         *,
         base_url: str = "",
    -    api_key: str = "",
    +    api_key: Any = "",
         api_mode: str = "",
    -) -> None:
    -    """Record the live runtime provider/model/credentials for the current AIAgent.
    -
    -    Called by ``run_agent.AIAgent._sync_runtime_main_for_aux_routing`` (or
    -    equivalent setter) at the top of each turn so that
    -    ``_read_main_provider`` / ``_read_main_model`` reflect CLI/gateway
    -    overrides instead of the stale config.yaml default.
    +    auth_mode: str = "",
    +) -> contextvars.Token:
    +    """Record the current context's live main runtime for auxiliary routing.
     
    -    For ``custom:`` providers, ``base_url`` and ``api_key`` must also be
    -    recorded so that ``_resolve_auto`` can construct a valid client in
    -    Step 1 instead of falling through to the aggregator chain.
    +    Context-local state prevents concurrent gateway sessions from overwriting
    +    one another while retaining compatibility mirrors for legacy readers.
         """
         global _RUNTIME_MAIN_PROVIDER, _RUNTIME_MAIN_MODEL
         global _RUNTIME_MAIN_BASE_URL, _RUNTIME_MAIN_API_KEY, _RUNTIME_MAIN_API_MODE
    -    _RUNTIME_MAIN_PROVIDER = (provider or "").strip().lower()
    -    _RUNTIME_MAIN_MODEL = (model or "").strip()
    -    _RUNTIME_MAIN_BASE_URL = (base_url or "").strip()
    -    _RUNTIME_MAIN_API_KEY = [redacted:secret-assignment] if isinstance(api_key, str) else ""
    -    _RUNTIME_MAIN_API_MODE = (api_mode or "").strip()
    +    global _RUNTIME_MAIN_AUTH_MODE, _RUNTIME_MAIN_COMPAT_SNAPSHOT
    +    runtime = {
    +        "provider": (provider or "").strip().lower(),
    +        "model": (model or "").strip(),
    +        "base_url": (base_url or "").strip(),
    +        "api_key": (
    +            api_key.strip()
    +            if isinstance(api_key, str)
    +            else api_key if callable(api_key) else ""
    +        ),
    +        "api_mode": (api_mode or "").strip(),
    +        "auth_mode": (auth_mode or "").strip().lower(),
    +    }
    +    # Publish authoritative context before updating locked compatibility
    +    # mirrors; concurrent sessions never read those mirrors at runtime.
    +    token = [redacted:secret-assignment]
    +    with _RUNTIME_MAIN_COMPAT_LOCK:
    +        (
    +            _RUNTIME_MAIN_PROVIDER,
    +            _RUNTIME_MAIN_MODEL,
    +            _RUNTIME_MAIN_BASE_URL,
    +            _RUNTIME_MAIN_API_KEY,
    +            _RUNTIME_MAIN_API_MODE,
    +            _RUNTIME_MAIN_AUTH_MODE,
    +        ) = (runtime[field] for field in _MAIN_RUNTIME_FIELDS)
    +        _RUNTIME_MAIN_COMPAT_SNAPSHOT = tuple(
    +            runtime[field] for field in _MAIN_RUNTIME_FIELDS
    +        )
    +    return token
    +
    +
    +def reset_runtime_main(token: [redacted:secret-assignment] -> None:
    +    """Restore the runtime binding that preceded one scoped turn."""
    +    if token is None:
    +        return
    +    try:
    +        _RUNTIME_MAIN_CONTEXT.reset(token)
    +    except (RuntimeError, ValueError):
    +        # A token cannot be reset from another copied Context. Background
    +        # workers inherit values, not ownership of the parent's token.
    +        pass
    +
    +
    [redacted:email]
    +def scoped_runtime_main(main_runtime: Optional[Dict[str, Any]]):
    +    """Temporarily bind an explicit runtime without touching legacy mirrors."""
    +    runtime = _normalize_main_runtime(main_runtime)
    +    token = [redacted:secret-assignment] or None)
    +    try:
    +        yield runtime
    +    finally:
    +        _RUNTIME_MAIN_CONTEXT.reset(token)
     
     
     def clear_runtime_main() -> None:
    -    """Clear the runtime override (e.g. on session end)."""
    +    """Clear the runtime override in the current context."""
         global _RUNTIME_MAIN_PROVIDER, _RUNTIME_MAIN_MODEL
         global _RUNTIME_MAIN_BASE_URL, _RUNTIME_MAIN_API_KEY, _RUNTIME_MAIN_API_MODE
    -    _RUNTIME_MAIN_PROVIDER = ""
    -    _RUNTIME_MAIN_MODEL = ""
    -    _RUNTIME_MAIN_BASE_URL = ""
    -    _RUNTIME_MAIN_API_KEY = ""
    -    _RUNTIME_MAIN_API_MODE = ""
    +    global _RUNTIME_MAIN_AUTH_MODE, _RUNTIME_MAIN_COMPAT_SNAPSHOT
    +    _RUNTIME_MAIN_CONTEXT.set(None)
    +    with _RUNTIME_MAIN_COMPAT_LOCK:
    +        _RUNTIME_MAIN_PROVIDER = ""
    +        _RUNTIME_MAIN_MODEL = ""
    +        _RUNTIME_MAIN_BASE_URL = ""
    +        _RUNTIME_MAIN_API_KEY = ""
    +        _RUNTIME_MAIN_API_MODE = ""
    +        _RUNTIME_MAIN_AUTH_MODE = ""
    +        _RUNTIME_MAIN_COMPAT_SNAPSHOT = ("", "", "", "", "", "")
     
     
     def _resolve_custom_runtime() -> Tuple[Optional[str], Optional[str], Optional[str]]:
    @@ -2784,6 +2876,14 @@ def _normalize_main_runtime(main_runtime: Optional[Dict[str, Any]]) -> Dict[str,
         surface as the main agent. The OpenAI SDK accepts ``Callable[[], str]``
         for ``api_key`` and calls it before every request.
         """
    +    if main_runtime is None:
    +        # Context-local state is inherited by tool worker wrappers while
    +        # remaining isolated across concurrent gateway sessions. Never fall
    +        # back to compatibility mirrors here: another session may have written
    +        # them most recently, which would leak its endpoint/key into this call.
    +        main_runtime = _RUNTIME_MAIN_CONTEXT.get()
    +        if main_runtime is None:
    +            main_runtime = _compat_runtime_main()
         if not isinstance(main_runtime, dict):
             return {}
         normalized: Dict[str, Any] = {}
    @@ -3311,13 +3411,7 @@ def _evict_cached_clients(provider: str) -> None:
             for key in stale_keys:
                 client = _client_cache.get(key, (None, None, None))[0]
                 if client is not None:
    -                _force_close_async_httpx(client)
    -                try:
    -                    close_fn = getattr(client, "close", None)
    -                    if callable(close_fn):
    -                        close_fn()
    -                except Exception:
    -                    pass
    +                _close_cached_client(client)
                 _client_cache.pop(key, None)
     
     
    @@ -4308,17 +4402,6 @@ def _resolve_auto(
         runtime_api_key = [redacted:secret-assignment]"api_key", "")
         runtime_api_mode = str(runtime.get("api_mode") or "")
     
    -    # Fall back to process-local globals when main_runtime dict was not
    -    # provided or was incomplete.  ``set_runtime_main()`` now records
    -    # base_url/api_key/api_mode alongside provider/model, so custom:
    -    # providers get the full credential surface in Step 1 of the
    -    # auto-detect chain.
    -    if not runtime_base_url and _RUNTIME_MAIN_BASE_URL:
    -        runtime_base_url = _RUNTIME_MAIN_BASE_URL
    -    if not runtime_api_key and _RUNTIME_MAIN_API_KEY:
    -        runtime_api_key = [redacted:secret-assignment]
    -    if not runtime_api_mode and _RUNTIME_MAIN_API_MODE:
    -        runtime_api_mode = _RUNTIME_MAIN_API_MODE
     
         # ── Warn once if OPENAI_BASE_URL is set but config.yaml uses a named
         #    provider (not 'custom').  This catches the common "env poisoning"
    @@ -4632,12 +4715,14 @@ def resolve_provider_client(
         # Normalise aliases
         provider = _normalize_aux_provider(provider)
     
    -    # Universal model-resolution fallback chain.  Callers (notably title
    -    # generation, vision, session search, and other auxiliary tasks) can
    -    # reach this function without an explicit model — the user picked their
    -    # main provider, didn't bother configuring a per-task ``auxiliary.<task>.model``,
    -    # and just expects "use my main model for side tasks too."  Resolve in
    -    # this order, stopping at the first non-empty answer:
    +    # Universal model-resolution fallback for concrete providers. ``auto`` is
    +    # intentionally excluded: `_resolve_auto(main_runtime=...)` returns the
    +    # model paired with the provider it actually selected. Pre-filling an auto
    +    # call from `_read_main_model()` can leak a stale process-global runtime
    +    # into a different provider (for example Claude model slug on Codex OAuth)
    +    # and override that correctly resolved model.
    +    #
    +    # Concrete provider resolution order:
         #
         #   1. ``model`` argument (caller knew what they wanted)
         #   2. Provider's catalog default — cheap/fast model the provider
    @@ -5482,6 +5567,7 @@ def resolve_vision_provider_client(
         base_url: Optional[str] = None,
         api_key: [redacted:secret-assignment] = None,
         async_mode: bool = False,
    +    main_runtime: Optional[Dict[str, Any]] = None,
     ) -> Tuple[Optional[str], Optional[Any], Optional[str]]:
         """Resolve the client actually used for vision tasks.
     
    @@ -5490,6 +5576,7 @@ def resolve_vision_provider_client(
         backends, so users can intentionally force experimental providers. Auto mode
         stays conservative and only tries vision backends known to work today.
         """
    +    runtime = _normalize_main_runtime(main_runtime)
         requested, resolved_model, resolved_base_url, resolved_api_key, resolved_api_mode = _resolve_task_provider_model(
             "vision", provider, model, base_url, api_key
         )
    @@ -5515,6 +5602,7 @@ def _finalize(resolved_provider: str, sync_client: Any, default_model: Optional[
                 explicit_base_url=resolved_base_url,
                 explicit_api_key=[redacted:secret-assignment]
                 api_mode=resolved_api_mode,
    +            main_runtime=runtime,
             )
             if client is None:
                 return provider_for_base_override, None, None
    @@ -5538,8 +5626,8 @@ def _finalize(resolved_provider: str, sync_client: Any, default_model: Optional[
             #                   live from the catalog — tried when
             #                   DEEPINFRA_API_KEY is set)
             #   5. Stop
    -        main_provider = _read_main_provider()
    -        main_model = _read_main_model()
    +        main_provider = str(runtime.get("provider") or _read_main_provider())
    +        main_model = str(runtime.get("model") or _read_main_model())
             if main_provider and main_provider not in {"auto", ""}:
                 # A provider-specific vision default wins over the user's chat model:
                 # static overrides (xiaomi/zai) and catalog-backed discovery (the
    @@ -5602,10 +5690,15 @@ def _finalize(resolved_provider: str, sync_client: Any, default_model: Optional[
                     rpc_api_key = None
                     rpc_api_mode = resolved_api_mode
                     if main_provider == "custom" or main_provider.startswith("custom:"):
    -                    if _RUNTIME_MAIN_BASE_URL:
    -                        rpc_base_url = _RUNTIME_MAIN_BASE_URL
    -                        rpc_api_key = [redacted:secret-assignment] or None
    -                        rpc_api_mode = resolved_api_mode or _RUNTIME_MAIN_API_MODE or None
    +                    runtime_base_url = runtime.get("base_url")
    +                    if runtime_base_url:
    +                        rpc_base_url = runtime_base_url
    +                        rpc_api_key = [redacted:secret-assignment]"api_key") or None
    +                        rpc_api_mode = (
    +                            resolved_api_mode
    +                            or runtime.get("api_mode")
    +                            or None
    +                        )
                         else:
                             # No live runtime recorded (non-gateway caller): fall
                             # back to resolving the configured custom endpoint.
    @@ -5619,6 +5712,7 @@ def _finalize(resolved_provider: str, sync_client: Any, default_model: Optional[
                         api_mode=rpc_api_mode,
                         explicit_base_url=rpc_base_url,
                         explicit_api_key=[redacted:secret-assignment]
    +                    main_runtime=runtime,
                         is_vision=True)
                     if rpc_client is not None:
                         logger.info(
    @@ -5661,6 +5755,7 @@ def _finalize(resolved_provider: str, sync_client: Any, default_model: Optional[
                     base_url=_zai_url,
                     api_key=[redacted:secret-assignment] or None,
                     api_mode="chat_completions",
    +                main_runtime=runtime,
                     is_vision=True,
                 )
                 if client is not None:
    @@ -5668,6 +5763,7 @@ def _finalize(resolved_provider: str, sync_client: Any, default_model: Optional[
             # Fallback: try without explicit base_url (old behavior)
             client, final_model = _get_cached_client(requested, resolved_model, async_mode,
                                                      api_mode=resolved_api_mode,
    +                                                 main_runtime=runtime,
                                                      is_vision=True)
             if client is None:
                 return requested, None, None
    @@ -5675,6 +5771,7 @@ def _finalize(resolved_provider: str, sync_client: Any, default_model: Optional[
     
         client, final_model = _get_cached_client(requested, resolved_model, async_mode,
                                                  api_mode=resolved_api_mode,
    +                                             main_runtime=runtime,
                                                  is_vision=True)
         if client is None:
             return requested, None, None
    @@ -5742,6 +5839,38 @@ def auxiliary_max_tokens_param(value: int, *, model: Optional[str] = None) -> di
     _CLIENT_CACHE_MAX_SIZE = 64  # safety belt — evict oldest when exceeded
     
     
    +class _CallableCacheDiscriminator:
    +    """Hash a credential callback by identity without exposing its state."""
    +
    +    __slots__ = ("_callback",)
    +
    +    def __init__(self, callback: Any) -> None:
    +        # Retain the callback so its id cannot be reused while cached.
    +        self._callback = callback
    +
    +    def __hash__(self) -> int:
    +        return id(self._callback)
    +
    +    def __eq__(self, other: object) -> bool:
    +        return (
    +            isinstance(other, _CallableCacheDiscriminator)
    +            and self._callback is other._callback
    +        )
    +
    +    def __repr__(self) -> str:
    +        return "<callable-api-key>"
    +
    +
    +def _runtime_cache_discriminator(field: str, value: Any) -> Any:
    +    """Return a hashable, secret-safe runtime cache-key component."""
    +    if field == "api_key" and callable(value):
    +        return _CallableCacheDiscriminator(value)
    +    if field == "api_key" and isinstance(value, str) and value:
    +        digest = hashlib.blake2b(value.encode("utf-8"), digest_size=16).digest()
    +        return ("api-key-digest", digest)
    +    return value
    +
    +
     def _client_cache_key(
         provider: str,
         *,
    @@ -5755,7 +5884,10 @@ def _client_cache_key(
         model: Optional[str] = None,
     ) -> tuple:
         runtime = _normalize_main_runtime(main_runtime)
    -    runtime_key = tuple(runtime.get(field, "") for field in _MAIN_RUNTIME_FIELDS) if provider == "auto" else ()
    +    runtime_key = tuple(
    +        _runtime_cache_discriminator(field, runtime.get(field, ""))
    +        for field in _MAIN_RUNTIME_FIELDS
    +    ) if provider == "auto" else ()
         # `auto` can now resolve through task-specific or main fallback policy,
         # so the task participates in the cache key. Non-auto providers keep the
         # old cache shape because the explicit provider/model tuple is sufficient.
    @@ -5770,21 +5902,16 @@ def _client_cache_key(
         # APIConnectionError that fails the sibling advisor (root cause of the run2
         # double-advisor "Connection error" collapse). Keying on model gives each
         # model its own client, so concurrent fan-out calls never cross-close.
    -    model_key = model or ""
    -    return (provider, async_mode, base_url or "", api_key or "", api_mode or "", runtime_key, is_vision, task_key, pool_hint, model_key)
    +    model_key = model or runtime.get("model", "")
    +    api_key_key = [redacted:secret-assignment]"api_key", api_key or "")
    +    return (provider, async_mode, base_url or "", api_key_key, api_mode or "", runtime_key, is_vision, task_key, pool_hint, model_key)
     
     
     def _store_cached_client(cache_key: tuple, client: Any, default_model: Optional[str], *, bound_loop: Any = None) -> None:
         with _client_cache_lock:
             old_entry = _client_cache.get(cache_key)
             if old_entry is not None and old_entry[0] is not client:
    -            _force_close_async_httpx(old_entry[0])
    -            try:
    -                close_fn = getattr(old_entry[0], "close", None)
    -                if callable(close_fn):
    -                    close_fn()
    -            except Exception:
    -                pass
    +            _close_cached_client(old_entry[0])
             _client_cache[cache_key] = (client, default_model, bound_loop)
     
     
    @@ -5886,30 +6013,31 @@ def _force_close_async_httpx(client: Any) -> None:
             pass
     
     
    +def _close_cached_client(client: Any) -> None:
    +    """Apply the canonical best-effort close policy to one cached client."""
    +    if client is None:
    +        return
    +    _force_close_async_httpx(client)
    +    try:
    +        close_fn = getattr(client, "close", None)
    +        if callable(close_fn) and not inspect.iscoroutinefunction(close_fn):
    +            close_fn()
    +    except Exception:
    +        pass
    +
    +
     def shutdown_cached_clients() -> None:
         """Close all cached clients (sync and async) to prevent event-loop errors.
     
         Call this during CLI shutdown, *before* the event loop is closed, to
         avoid ``AsyncHttpxClientWrapper.__del__`` raising on a dead loop.
         """
    -    import inspect
    -
         with _client_cache_lock:
             for key, entry in list(_client_cache.items()):
                 client = entry[0]
                 if client is None:
                     continue
    -            # Mark any async httpx transport as closed first (prevents __del__
    -            # from scheduling aclose() on a dead event loop).
    -            _force_close_async_httpx(client)
    -            # Sync clients: close the httpx connection pool cleanly.
    -            # Async clients: skip — we already neutered __del__ above.
    -            try:
    -                close_fn = getattr(client, "close", None)
    -                if close_fn and not inspect.iscoroutinefunction(close_fn):
    -                    close_fn()
    -            except Exception:
    -                pass
    +            _close_cached_client(client)
             _client_cache.clear()
     
     
    @@ -6059,13 +6187,20 @@ def _get_cached_client(
                 if cache_key not in _client_cache:
                     # Safety belt: if the cache has grown beyond the max, evict
                     # the oldest entries (FIFO — dict preserves insertion order).
    +                # Do not close an evicted client here: another caller may be
    +                # mid-request with the object it obtained from this cache.
    +                # Dropping the cache reference lets normal refcount/GC cleanup
    +                # happen after in-flight users release it.
                     while len(_client_cache) >= _CLIENT_CACHE_MAX_SIZE:
    -                    evict_key, evict_entry = next(iter(_client_cache.items()))
    -                    _force_close_async_httpx(evict_entry[0])
    +                    evict_key = next(iter(_client_cache))
                         del _client_cache[evict_key]
                     _client_cache[cache_key] = (client, default_model, bound_loop)
                 else:
    +                built_client = client
                     client, default_model, _ = _client_cache[cache_key]
    +                # This concurrently built loser was never exposed to a caller,
    +                # so it is safe to close immediately.
    +                _close_cached_client(built_client)
         return client, model or default_model
     
     
    @@ -6818,6 +6953,11 @@ def call_llm(
         Raises:
             RuntimeError: If no provider is configured.
         """
    +    # Capture one immutable runtime snapshot for keying, resolution, retries,
    +    # and fallbacks. Reading ambient state independently in each phase lets a
    +    # concurrent /model switch produce a key for one runtime and a client for
    +    # another.
    +    main_runtime = _normalize_main_runtime(main_runtime)
         resolved_provider, resolved_model, resolved_base_url, resolved_api_key, resolved_api_mode = _resolve_task_provider_model(
             task, provider, model, base_url, api_key)
         if api_mode:
    @@ -6832,6 +6972,7 @@ def call_llm(
                 base_url=resolved_base_url or base_url,
                 api_key=[redacted:secret-assignment] or api_key,
                 async_mode=False,
    +            main_runtime=main_runtime,
             )
             if client is None and resolved_provider != "auto" and not resolved_base_url:
                 logger.warning(
    @@ -6842,6 +6983,7 @@ def call_llm(
                     provider="auto",
                     model=resolved_model,
                     async_mode=False,
    +                main_runtime=main_runtime,
                 )
             if client is None:
                 raise RuntimeError(
    @@ -7444,6 +7586,9 @@ async def async_call_llm(
     
         Same as call_llm() but async. See call_llm() for full documentation.
         """
    +    # Keep every async phase on the same runtime identity, even if another
    +    # session switches models while this task is awaiting network I/O.
    +    main_runtime = _normalize_main_runtime(main_runtime)
         resolved_provider, resolved_model, resolved_base_url, resolved_api_key, resolved_api_mode = _resolve_task_provider_model(
             task, provider, model, base_url, api_key)
         effective_extra_body = _get_task_extra_body(task)
    @@ -7456,6 +7601,7 @@ async def async_call_llm(
                 base_url=resolved_base_url or base_url,
                 api_key=[redacted:secret-assignment] or api_key,
                 async_mode=True,
    +            main_runtime=main_runtime,
             )
             if client is None and resolved_provider != "auto" and not resolved_base_url:
                 logger.warning(
    @@ -7466,6 +7612,7 @@ async def async_call_llm(
                     provider="auto",
                     model=resolved_model,
                     async_mode=True,
    +                main_runtime=main_runtime,
                 )
             if client is None:
                 raise RuntimeError(
    @@ -7481,6 +7628,7 @@ async def async_call_llm(
                 base_url=resolved_base_url,
                 api_key=[redacted:secret-assignment]
                 api_mode=resolved_api_mode,
    +            main_runtime=main_runtime,
             )
             if client is None:
                 _explicit = (resolved_provider or "").strip().lower()

  • agent/context_compressor.py

    @@ -1175,6 +1175,9 @@ def update_model(
             if runtime_changed:
                 self._fallback_compression_streak = 0
                 self._persist_fallback_compression_streak()
    +            # Failure cooldowns are scoped to the model/provider that failed.
    +            # A switch must give the new runtime an immediate summary attempt.
    +            self._clear_compression_failure_cooldown()
             self._verify_compaction_cleared_threshold = False
             self._last_compression_made_progress = False
     
    @@ -1260,7 +1263,6 @@ def _compute_threshold_tokens(
                 return max(1, min(int(effective_window * ContextCompressor._MIN_CTX_TRIGGER_RATIO),
                                   effective_window - 1))
             return floored
    -
         def __init__(
             self,
             model: str,

  • agent/image_routing.py

    @@ -267,10 +267,12 @@ def _resolve_inference_base_url(
     ) -> str:
         """Best-effort base URL for the active inference provider."""
         try:
    -        from agent.auxiliary_client import _RUNTIME_MAIN_BASE_URL
    +        from agent.auxiliary_client import _runtime_main_value
     
    -        runtime = str(_RUNTIME_MAIN_BASE_URL or "").strip()
    -        if runtime:
    +        runtime = str(_runtime_main_value("base_url") or "").strip()
    +        runtime_provider = str(_runtime_main_value("provider") or "").strip().lower()
    +        requested_provider = str(provider or "").strip().lower()
    +        if runtime and (not requested_provider or requested_provider == runtime_provider):
                 return runtime
         except Exception:
             pass

  • agent/turn_context.py

    @@ -151,7 +151,17 @@ def build_turn_context(
         # null; rebuilding from scratch" warning and a needless first-turn prefix
         # cache miss. (Issue #45499.)
     
    -    # Tell auxiliary_client what the live main provider/model are for this turn.
    +    # Tag log records on this thread with the session ID for ``hermes logs``.
    +    set_session_context(agent.session_id)
    +
    +    # Bind the skill write-origin ContextVar for this thread.
    +    set_current_write_origin(getattr(agent, "_memory_write_origin", "assistant_tool"))
    +
    +    # Restore the primary runtime if the previous turn activated fallback.
    +    agent._restore_primary_runtime()
    +
    +    # Tell auxiliary_client what the live main provider/model are for this turn
    +    # after primary restoration has settled the runtime.
         try:
             from agent.auxiliary_client import set_runtime_main
             set_runtime_main(
    @@ -160,19 +170,11 @@ def build_turn_context(
                 base_url=getattr(agent, "base_url", "") or "",
                 api_key=[redacted:secret-assignment] "api_key", "") or "",
                 api_mode=getattr(agent, "api_mode", "") or "",
    +            auth_mode=getattr(agent, "auth_mode", "") or "",
             )
         except Exception:
             pass
     
    -    # Tag log records on this thread with the session ID for ``hermes logs``.
    -    set_session_context(agent.session_id)
    -
    -    # Bind the skill write-origin ContextVar for this thread.
    -    set_current_write_origin(getattr(agent, "_memory_write_origin", "assistant_tool"))
    -
    -    # Restore the primary runtime if the previous turn activated fallback.
    -    agent._restore_primary_runtime()
    -
         # Between-turns MCP refresh: an MCP server that finished connecting since
         # the previous turn (slow HTTP/OAuth servers routinely take 2-6s on a cold
         # connect, missing the bounded startup wait) lands in THIS turn's tool

  • gateway/run.py

    @@ -10860,10 +10860,30 @@ async def _prepare_inbound_message_text(
                             "Image routing: text (mode=%s). Pre-analyzing %d image(s) via vision_analyze.",
                             _img_mode, len(image_paths),
                         )
    -                    message_text = await self._enrich_message_with_vision(
    -                        message_text,
    -                        image_paths,
    -                    )
    +                    # Vision enrichment runs before AIAgent.run_conversation(),
    +                    # so bind this session's resolved runtime explicitly rather
    +                    # than consulting process-global compatibility mirrors.
    +                    vision_runtime = None
    +                    try:
    +                        turn_model, runtime_kwargs = self._resolve_session_agent_runtime(
    +                            source=source,
    +                            session_key=session_key,
    +                        )
    +                        vision_runtime = dict(runtime_kwargs or {})
    +                        vision_runtime["model"] = turn_model
    +                    except Exception:
    +                        logger.debug(
    +                            "vision enrichment: session runtime resolution failed",
    +                            exc_info=True,
    +                        )
    +
    +                    from agent.auxiliary_client import scoped_runtime_main
    +
    +                    with scoped_runtime_main(vision_runtime):
    +                        message_text = await self._enrich_message_with_vision(
    +                            message_text,
    +                            image_paths,
    +                        )
     
                 if audio_paths:
                     message_text, _successful_transcripts = await self._enrich_message_with_transcription(

  • run_agent.py

    @@ -1199,6 +1199,7 @@ def _current_main_runtime(self) -> Dict[str, str]:
                 "base_url": getattr(self, "base_url", "") or "",
                 "api_key": getattr(self, "api_key", "") or "",
                 "api_mode": getattr(self, "api_mode", "") or "",
    +            "auth_mode": getattr(self, "auth_mode", "") or "",
             }
     
         def _check_compression_model_feasibility(self) -> None:
    @@ -6201,21 +6202,28 @@ def run_conversation(
             acct_token = [redacted:secret-assignment]
                 getattr(self, "_session_db", None), getattr(self, "session_id", None)
             )
    -        try:
    -            return run_conversation(
    -                self,
    -                user_message,
    -                system_message,
    -                conversation_history,
    -                task_id,
    -                stream_callback,
    -                persist_user_message,
    -                persist_user_timestamp=persist_user_timestamp,
    -                moa_config=moa_config,
    -            )
    -        finally:
    -            reset_accounting_context(acct_token)
    -            reset_conversation_context(token)
    +        from agent.auxiliary_client import scoped_runtime_main
    +
    +        # The outer token restores the caller's Context even though turn setup
    +        # replaces the value with the live runtime after fallback restoration.
    +        # Keep the scope local instead of storing ContextVar tokens on the agent,
    +        # which may be observed from another thread.
    +        with scoped_runtime_main({}):
    +            try:
    +                return run_conversation(
    +                    self,
    +                    user_message,
    +                    system_message,
    +                    conversation_history,
    +                    task_id,
    +                    stream_callback,
    +                    persist_user_message,
    +                    persist_user_timestamp=persist_user_timestamp,
    +                    moa_config=moa_config,
    +                )
    +            finally:
    +                reset_accounting_context(acct_token)
    +                reset_conversation_context(token)
     
         def chat(self, message: str, stream_callback: Optional[callable] = None) -> str:
             """

  • scripts/release.py

    @@ -371,6 +371,8 @@
         "[redacted:email]": "synapsesx",
         "[redacted:email]": "itsflownium",
         "[redacted:email]": "dirtyren",
    +    "[redacted:email]": "krowd3v",
    +    "[redacted:email]": "dfein38347g",
         "[redacted:email]": "s96919",
         "[redacted:email]": "rasitakyol",
         "[redacted:email]": "Thatgfsj",

  • tests/agent/test_auxiliary_runtime_cache_key.py

    @@ -0,0 +1,365 @@
    +"""Regression coverage for implicit live-runtime auxiliary cache keys.
    +
    +#49151/#49156 is specifically the ``provider='auto'`` path where callers omit
    +``main_runtime`` after a mid-session model switch.  This is distinct from
    +#56889, which isolates callers that pass different explicit ``model=`` values.
    +"""
    +
    +import asyncio
    +from concurrent.futures import ThreadPoolExecutor
    +from threading import Barrier
    +from types import SimpleNamespace
    +from unittest.mock import AsyncMock, MagicMock, patch
    +
    +import pytest
    +
    +import agent.auxiliary_client as aux
    +
    +
    [redacted:email](autouse=True)
    +def _clean_aux_state():
    +    aux.shutdown_cached_clients()
    +    aux.clear_runtime_main()
    +    yield
    +    aux.shutdown_cached_clients()
    +    aux.clear_runtime_main()
    +
    +
    +def _runtime(model: str, *, provider: str = "custom:llama-swap") -> dict:
    +    return {
    +        "provider": provider,
    +        "model": model,
    +        "base_url": "http://llama-swap.test/v1",
    +        "api_key": "local-key",
    +        "api_mode": "chat_completions",
    +        "auth_mode": "api_key",
    +    }
    +
    +
    +def test_implicit_auto_cache_rebuilds_after_runtime_model_switch():
    +    """A /model switch must not reuse the old implicit-auto cache entry."""
    +    built = []
    +
    +    def fake_resolve(_provider, _model, _async_mode, *, main_runtime, **_kwargs):
    +        client = MagicMock(name=f"client-{main_runtime['model']}")
    +        built.append((client, dict(main_runtime)))
    +        return client, main_runtime["model"]
    +
    +    with patch.object(aux, "resolve_provider_client", side_effect=fake_resolve):
    +        aux.set_runtime_main(**_runtime("qwen35b-code"))
    +        first_client, first_model = aux._get_cached_client("auto")
    +
    +        aux.set_runtime_main(**_runtime("qwen27b-code"))
    +        second_client, second_model = aux._get_cached_client("auto")
    +
    +    assert first_model == "qwen35b-code"
    +    assert second_model == "qwen27b-code"
    +    assert second_client is not first_client
    +    assert [runtime["model"] for _, runtime in built] == [
    +        "qwen35b-code",
    +        "qwen27b-code",
    +    ]
    +
    +
    +def test_implicit_runtime_cache_key_covers_full_connection_and_auth_surface():
    +    """Provider/endpoint/credential/wire/auth changes all isolate auto clients."""
    +    base = _runtime("same-model")
    +    variants = [
    +        {**base, "provider": "custom:other"},
    +        {**base, "base_url": "https://other.test/v1"},
    +        {**base, "api_key": "other-key"},
    +        {**base, "api_mode": "codex_responses"},
    +        {**base, "auth_mode": "entra_id", "api_key": lambda: "token"},
    +    ]
    +
    +    aux.set_runtime_main(**base)
    +    baseline = aux._client_cache_key("auto", async_mode=False)
    +    keys = []
    +    for variant in variants:
    +        aux.set_runtime_main(**variant)
    +        keys.append(aux._client_cache_key("auto", async_mode=False))
    +
    +    assert all(key != baseline for key in keys)
    +    assert len(set(keys)) == len(keys)
    +
    +
    +def test_implicit_runtime_is_isolated_between_concurrent_session_contexts():
    +    """Concurrent gateway sessions must not read each other's live runtime."""
    +    barrier = Barrier(2)
    +
    +    def session(model: str):
    +        aux.set_runtime_main(**_runtime(model))
    +        barrier.wait()
    +        normalized = aux._normalize_main_runtime(None)
    +        return normalized["model"], aux._client_cache_key("auto", async_mode=False)
    +
    +    with ThreadPoolExecutor(max_workers=2) as pool:
    +        first = pool.submit(session, "session-a-model")
    +        second = pool.submit(session, "session-b-model")
    +        model_a, key_a = first.result()
    +        model_b, key_b = second.result()
    +
    +    assert model_a == "session-a-model"
    +    assert model_b == "session-b-model"
    +    assert key_a != key_b
    +
    +
    +def test_context_without_runtime_does_not_fall_back_to_other_session_globals():
    +    """A fresh context must not inherit another session's compatibility mirrors."""
    +    aux.set_runtime_main(**_runtime("other-session-model"))
    +
    +    def fresh_context():
    +        return aux._normalize_main_runtime(None)
    +
    +    import contextvars
    +
    +    assert contextvars.Context().run(fresh_context) == {}
    +
    +
    +def test_runtime_context_token_restores_previous_value_after_turn():
    +    """Turn-scoped runtime binding must not leak into later work in the same context."""
    +    token = [redacted:secret-assignment]"turn-model"))
    +    assert aux._normalize_main_runtime(None)["model"] == "turn-model"
    +
    +    aux.reset_runtime_main(token)
    +
    +    assert aux._normalize_main_runtime(None) == {}
    +
    +
    +def test_aiagent_wrapper_resets_runtime_context_after_turn():
    +    """Every production run_conversation exit restores the caller's Context."""
    +    from run_agent import AIAgent
    +
    +    agent = SimpleNamespace(
    +        _conversation_root_id=lambda: "root-session",
    +        _session_db=None,
    +        session_id="session-id",
    +    )
    +
    +    def fake_turn(*_args, **_kwargs):
    +        aux.set_runtime_main(**_runtime("wrapped-turn"))
    +        return {"final_response": "ok"}
    +
    +    with patch("agent.conversation_loop.run_conversation", side_effect=fake_turn):
    +        result = AIAgent.run_conversation(agent, "hello")
    +
    +    assert result["final_response"] == "ok"
    +    assert aux._normalize_main_runtime(None) == {}
    +
    +
    +def test_legacy_patched_globals_are_visible_only_without_an_active_runtime():
    +    """Direct legacy patches work, but never override context-local session state."""
    +    with patch.object(aux, "_RUNTIME_MAIN_PROVIDER", "custom:legacy"), patch.object(
    +        aux, "_RUNTIME_MAIN_MODEL", "legacy-model"
    +    ), patch.object(
    +        aux, "_RUNTIME_MAIN_BASE_URL", "https://legacy.test/v1"
    +    ):
    +        assert aux._normalize_main_runtime(None)["model"] == "legacy-model"
    +
    +        aux.set_runtime_main(**_runtime("active-session-model"))
    +        runtime = aux._normalize_main_runtime(None)
    +
    +    assert runtime["model"] == "active-session-model"
    +    assert runtime["base_url"] == "http://llama-swap.test/v1"
    +
    +
    +def test_concurrent_vision_probes_use_each_sessions_endpoint_and_model():
    +    """Vision auto-routing must not mix custom endpoints across sessions."""
    +    barrier = Barrier(2)
    +
    +    def fake_resolve(provider, model, **kwargs):
    +        barrier.wait()
    +        client = MagicMock()
    +        client.probed_base_url = kwargs.get("explicit_base_url")
    +        return client, model
    +
    +    def probe(model: str, base_url: str):
    +        runtime = _runtime(model)
    +        runtime["base_url"] = base_url
    +        aux.set_runtime_main(**runtime)
    +        provider, client, resolved_model = aux.resolve_vision_provider_client()
    +        assert client is not None
    +        return provider, resolved_model, client.probed_base_url
    +
    +    with patch.object(
    +        aux, "_resolve_task_provider_model", return_value=("auto", None, None, None, None)
    +    ), patch.object(aux, "_main_model_supports_vision", return_value=True), patch.object(
    +        aux, "resolve_provider_client", side_effect=fake_resolve
    +    ):
    +        with ThreadPoolExecutor(max_workers=2) as pool:
    +            first = pool.submit(probe, "vision-a", "https://a.test/v1")
    +            second = pool.submit(probe, "vision-b", "https://b.test/v1")
    +            result_a = first.result()
    +            result_b = second.result()
    +
    +    assert result_a == ("custom:llama-swap", "vision-a", "https://a.test/v1")
    +    assert result_b == ("custom:llama-swap", "vision-b", "https://b.test/v1")
    +
    +
    +def test_explicit_model_cache_isolation_remains_independent_of_runtime_key():
    +    """#56889 remains covered: explicit model values isolate non-auto clients."""
    +    first = aux._client_cache_key(
    +        "openrouter", async_mode=False, model="anthropic/claude-opus-4.8"
    +    )
    +    second = aux._client_cache_key(
    +        "openrouter", async_mode=False, model="openai/gpt-5.5"
    +    )
    +
    +    assert first != second
    +
    +
    +def test_pinned_provider_without_model_inherits_live_runtime_model_in_cache_key():
    +    """A pinned provider with model=auto must follow the switched main model."""
    +    first = aux._client_cache_key(
    +        "openrouter",
    +        async_mode=False,
    +        main_runtime=_runtime("old-model", provider="openrouter"),
    +    )
    +    second = aux._client_cache_key(
    +        "openrouter",
    +        async_mode=False,
    +        main_runtime=_runtime("new-model", provider="openrouter"),
    +    )
    +
    +    assert first != second
    +
    +
    +def test_explicit_vision_runtime_wins_over_stale_ambient_runtime():
    +    """Vision resolution must use the immutable runtime supplied by its caller."""
    +    aux.set_runtime_main(**_runtime("ambient-old"))
    +    explicit = _runtime("explicit-new")
    +    captured = {}
    +
    +    def fake_resolve(provider, model, **kwargs):
    +        captured.update(provider=provider, model=model, **kwargs)
    +        return MagicMock(), model
    +
    +    with patch.object(
    +        aux, "_resolve_task_provider_model", return_value=("auto", None, None, None, None)
    +    ), patch.object(aux, "_main_model_supports_vision", return_value=True), patch.object(
    +        aux, "resolve_provider_client", side_effect=fake_resolve
    +    ):
    +        provider, _client, model = aux.resolve_vision_provider_client(
    +            main_runtime=explicit
    +        )
    +
    +    assert provider == "custom:llama-swap"
    +    assert model == "explicit-new"
    +    assert captured["explicit_base_url"] == "http://llama-swap.test/v1"
    +
    +
    +def test_image_routing_does_not_borrow_base_url_from_different_provider():
    +    """An explicit provider must not inherit another runtime's custom endpoint."""
    +    from agent.image_routing import _resolve_inference_base_url
    +
    +    aux.set_runtime_main(**_runtime("custom-model"))
    +    cfg = {
    +        "model": {
    +            "provider": "openrouter",
    +            "base_url": "https://openrouter.ai/api/v1",
    +        }
    +    }
    +
    +    assert (
    +        _resolve_inference_base_url(cfg, "openrouter")
    +        == "https://openrouter.ai/api/v1"
    +    )
    +
    +
    +def test_async_initial_cache_lookup_receives_explicit_runtime_snapshot():
    +    """The first async lookup must not drop main_runtime and only pass it on fallback."""
    +    runtime = _runtime("async-new")
    +    response = MagicMock()
    +    response.choices = [MagicMock(message=MagicMock(content="ok"))]
    +    client = MagicMock()
    +    client.chat.completions.create = AsyncMock(return_value=response)
    +
    +    with patch.object(
    +        aux,
    +        "_resolve_task_provider_model",
    +        return_value=("openrouter", None, None, None, None),
    +    ), patch.object(aux, "_get_cached_client", return_value=(client, "async-new")) as get_client:
    +        asyncio.run(
    +            aux.async_call_llm(
    +                task="approval",
    +                main_runtime=runtime,
    +                messages=[{"role": "user", "content": "approve?"}],
    +            )
    +        )
    +
    +    assert get_client.call_args.kwargs["main_runtime"] == aux._normalize_main_runtime(runtime)
    +
    +
    +def test_unhashable_callable_runtime_api_keys_are_safe_secret_free_discriminators():
    +    """Callable token providers remain cacheable without leaking returned tokens."""
    +
    +    class TokenProvider(list):
    +        def __init__(self, token: str):
    +            super().__init__()
    +            self.token = token
    +
    +        def __call__(self) -> str:
    +            return self.token
    +
    +    first_provider = TokenProvider("first-super-secret-token")
    +    second_provider = TokenProvider("second-super-secret-token")
    +
    +    first = aux._client_cache_key(
    +        "auto", async_mode=False, main_runtime={**_runtime("same"), "api_key": first_provider}
    +    )
    +    second = aux._client_cache_key(
    +        "auto", async_mode=False, main_runtime={**_runtime("same"), "api_key": second_provider}
    +    )
    +
    +    hash(first)
    +    hash(second)
    +    assert first != second
    +    rendered = repr((first, second))
    +    assert "first-super-secret-token" not in rendered
    +    assert "second-super-secret-token" not in rendered
    +
    +
    +def test_string_api_keys_are_not_retained_in_cache_key_repr():
    +    """String credentials discriminate clients without living in cache-key memory."""
    +    first_secret = "[redacted:secret-assignment]"
    +    second_secret = "[redacted:secret-assignment]"
    +    first = aux._client_cache_key(
    +        "auto",
    +        async_mode=False,
    +        api_key=[redacted:secret-assignment]
    +        main_runtime={**_runtime("same"), "api_key": first_secret},
    +    )
    +    second = aux._client_cache_key(
    +        "auto",
    +        async_mode=False,
    +        api_key=[redacted:secret-assignment]
    +        main_runtime={**_runtime("same"), "api_key": second_secret},
    +    )
    +
    +    assert first != second
    +    rendered = repr((first, second))
    +    assert first_secret not in rendered
    +    assert second_secret not in rendered
    +
    +
    +def test_fifo_eviction_does_not_close_client_that_may_have_an_inflight_call():
    +    """A bounded-cache eviction must not invalidate another caller's client."""
    +    clients = []
    +
    +    def fake_resolve(_provider, model, _async_mode, **_kwargs):
    +        client = MagicMock(name=f"client-{model}")
    +        clients.append(client)
    +        return client, model
    +
    +    with patch.object(aux, "resolve_provider_client", side_effect=fake_resolve):
    +        for index in range(65):
    +            aux._get_cached_client("custom", model=f"model-{index}")
    +
    +    assert len(aux._client_cache) == 64
    +    for client in clients:
    +        client.close.assert_not_called()
    +
    +    aux.shutdown_cached_clients()
    +    clients[0].close.assert_not_called()
    +    for client in clients[1:]:
    +        client.close.assert_called_once_with()

  • tests/agent/test_context_compressor.py

    @@ -3039,6 +3039,31 @@ def test_defer_no_longer_suppresses_after_switch(self):
             assert comp.should_defer_preflight_to_real_usage(comp.threshold_tokens + 5_000) is False
     
     
    +    def test_summary_failure_cooldown_cleared(self):
    +        """Stale summary-failure cooldown from the old model must not block
    +        the new model from generating summaries after a switch."""
    +        import time
    +        comp = self._comp()
    +        # Simulate a 600-second cooldown set because the old model had no
    +        # provider configured for summarization.
    +        comp._summary_failure_cooldown_until = time.monotonic() + 600
    +
    +        comp.update_model("new-model", context_length=128_000)
    +
    +        assert comp._summary_failure_cooldown_until == 0.0
    +
    +    def test_summary_failure_cooldown_survives_same_runtime_refresh(self):
    +        """Refreshing metadata for the same runtime must not defeat backoff."""
    +        import time
    +        comp = self._comp()
    +        cooldown_until = time.monotonic() + 600
    +        comp._summary_failure_cooldown_until = cooldown_until
    +
    +        comp.update_model("big-model", context_length=128_000)
    +
    +        assert comp._summary_failure_cooldown_until == cooldown_until
    +
    +
     class TestTruncateToolCallArgsJson:
         """Regression tests for #11762.
     

  • tests/agent/test_turn_context.py

    @@ -259,6 +259,40 @@ def test_pending_cli_message_uses_clean_override_for_api_local_note():
         assert agent._pending_cli_user_message is None
     
     
    +def test_runtime_main_sync_happens_after_restore():
    +    agent = _FakeAgent()
    +    agent.model = "stale-fallback-model"
    +    agent.provider = "openai-codex"
    +    agent.base_url = "https://chatgpt.com/backend-api/codex"
    +    agent.api_key = "[redacted:secret-assignment]"
    +    agent.api_mode = "codex_responses"
    +
    +    def restore_primary():
    +        agent.model = "primary-model"
    +        agent.provider = "anthropic"
    +        agent.base_url = "https://api.anthropic.com"
    +        agent.api_key = "[redacted:secret-assignment]"
    +        agent.api_mode = "anthropic_messages"
    +
    +    agent._restore_primary_runtime = restore_primary
    +    calls = []
    +    with patch(
    +        "agent.auxiliary_client.set_runtime_main",
    +        side_effect=lambda *args, **kwargs: calls.append((args, kwargs)),
    +    ):
    +        _build(agent)
    +
    +    assert calls == [(
    +        ("anthropic", "primary-model"),
    +        {
    +            "base_url": "https://api.anthropic.com",
    +            "api_key": "primary-key",
    +            "api_mode": "anthropic_messages",
    +            "auth_mode": "",
    +        },
    +    )]
    +
    +
     def test_memory_nudge_fires_at_interval():
         agent = _FakeAgent()
         agent._memory_nudge_interval = 1

  • tests/gateway/test_image_input_routing_runtime.py

    @@ -123,8 +123,13 @@ def fake_supports(provider, model, config):
         monkeypatch.setattr("agent.image_routing._lookup_supports_vision", fake_supports)
     
         async def fake_enrich(user_text, image_paths):
    +        from agent import auxiliary_client as aux
    +
             assert user_text == "look"
             assert image_paths == ["/tmp/cashback.png"]
    +        runtime = aux._normalize_main_runtime(None)
    +        assert runtime["provider"] == "xiaomi"
    +        assert runtime["model"] == "mimo-v2.5-pro"
             return "[vision summary]\n\nlook"
     
         monkeypatch.setattr(runner, "_enrich_message_with_vision", fake_enrich)

  • tests/run_agent/test_compression_feasibility.py

    @@ -167,6 +167,7 @@ def test_feasibility_check_passes_live_main_runtime():
                 "base_url": "https://chatgpt.com/backend-api/codex",
                 "api_key": "codex-token",
                 "api_mode": "codex_responses",
    +            "auth_mode": "",
             },
         )
     

  • tests/tui_gateway/test_image_routing_stale_model.py

    @@ -0,0 +1,34 @@
    +"""Regression coverage for live TUI image-routing identity."""
    +
    +from types import SimpleNamespace
    +from unittest.mock import patch
    +
    +from tui_gateway.server import _active_image_routing_identity
    +
    +
    +def test_live_agent_identity_wins_after_model_switch():
    +    agent = SimpleNamespace(provider="alibaba", model="qwen3.7-plus")
    +
    +    with patch(
    +        "agent.auxiliary_client._read_main_provider",
    +        side_effect=AssertionError("stale provider fallback used"),
    +    ), patch(
    +        "agent.auxiliary_client._read_main_model",
    +        side_effect=AssertionError("stale model fallback used"),
    +    ):
    +        identity = _active_image_routing_identity(agent)
    +
    +    assert identity == ("alibaba", "qwen3.7-plus")
    +
    +
    +def test_missing_agent_identity_uses_runtime_fallback():
    +    agent = SimpleNamespace(provider="", model=None)
    +
    +    with patch(
    +        "agent.auxiliary_client._read_main_provider", return_value="openai-codex"
    +    ), patch(
    +        "agent.auxiliary_client._read_main_model", return_value="gpt-5.5-codex"
    +    ):
    +        identity = _active_image_routing_identity(agent)
    +
    +    assert identity == ("openai-codex", "gpt-5.5-codex")

  • tui_gateway/server.py

    @@ -5091,6 +5091,16 @@ def _resolve_checkpoint_hash(mgr, cwd: str, ref: str) -> str:
         raise ValueError(f"Invalid checkpoint number. Use 1-{len(checkpoints)}.")
     
     
    +def _active_image_routing_identity(agent: Any) -> tuple[str, str]:
    +    """Return the live provider/model, falling back before agent startup."""
    +    from agent.auxiliary_client import _read_main_model, _read_main_provider
    +
    +    return (
    +        getattr(agent, "provider", "") or _read_main_provider(),
    +        getattr(agent, "model", "") or _read_main_model(),
    +    )
    +
    +
     def _enrich_with_attached_images(user_text: str, image_paths: list[str]) -> str:
         """Pre-analyze attached images via vision and prepend descriptions to user text."""
         import asyncio, json as _json
    @@ -9423,16 +9433,13 @@ def run():
                             decide_image_input_mode,
                             build_native_content_parts,
                         )
    -                    from agent.auxiliary_client import (
    -                        _read_main_model,
    -                        _read_main_provider,
    -                    )
                         from hermes_cli.config import load_config as _tui_load_config
     
                         _cfg = _tui_load_config()
    +                    _provider, _model = _active_image_routing_identity(agent)
                         _mode = decide_image_input_mode(
    -                        _read_main_provider(),
    -                        _read_main_model(),
    +                        _provider,
    +                        _model,
                             _cfg,
                         )
                         if getattr(agent, "api_mode", "") == "codex_app_server":