Auxiliary client cache isolation and invalidation fixes

Fixes on record

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

• #56889 fix(auxiliary): retry transient blips harder + isolate client cache per model

  merged to main

  files: agent/auxiliary_client.py, hermes_cli/config.py, tests/agent/test_auxiliary_client.py, tests/agent/test_auxiliary_transient_retry.py

  Show the diff

  • agent/auxiliary_client.py

    @@ -2743,7 +2743,7 @@ def _is_connection_error(exc: Exception) -> bool:
     
     
     def _is_transient_transport_error(exc: Exception) -> bool:
    -    """Return True for a one-off transport blip worth retrying ONCE on the
    +    """Return True for a one-off transport blip worth retrying ON the
         same provider before any provider/model fallback.
     
         Covers connection/streaming-close errors (via the canonical
    @@ -2761,6 +2761,34 @@ def _is_transient_transport_error(exc: Exception) -> bool:
         return isinstance(status, int) and (status == 408 or 500 <= status < 600)
     
     
    +_DEFAULT_TRANSIENT_RETRIES = 2
    +# Base for exponential backoff between transient retries (seconds). Overridable
    +# so tests can zero it out and not sleep real wall-clock time.
    +_TRANSIENT_RETRY_BACKOFF_BASE = 1.0
    +
    +
    +def _transient_retry_count() -> int:
    +    """Number of same-provider retries for a transient transport blip.
    +
    +    Read from ``auxiliary.transient_retries`` in config.yaml (default 2 →
    +    3 total attempts). Clamped to [0, 6] to bound worst-case wall time. A
    +    connection blip to a pinned auxiliary target (e.g. a MoA reference
    +    advisor) has no meaningful provider fallback, so a couple of retries with
    +    backoff is the difference between recovering and silently losing the call.
    +    Best-effort: any config-read failure falls back to the default.
    +    """
    +    try:
    +        from hermes_cli.config import cfg_get, load_config
    +
    +        val = cfg_get(load_config(), "auxiliary", "transient_retries")
    +        if val is None:
    +            return _DEFAULT_TRANSIENT_RETRIES
    +        n = int(val)
    +        return max(0, min(n, 6))
    +    except Exception:
    +        return _DEFAULT_TRANSIENT_RETRIES
    +
    +
     def _is_auth_error(exc: Exception) -> bool:
         """Detect auth failures that should trigger provider-specific refresh."""
         status = getattr(exc, "status_code", None)
    @@ -5036,6 +5064,7 @@ def _client_cache_key(
         main_runtime: Optional[Dict[str, Any]] = None,
         is_vision: bool = False,
         task: Optional[str] = None,
    +    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 ()
    @@ -5044,7 +5073,17 @@ def _client_cache_key(
         # old cache shape because the explicit provider/model tuple is sufficient.
         task_key = (task or "") if provider == "auto" else ""
         pool_hint = _pool_cache_hint(provider, main_runtime=main_runtime)
    -    return (provider, async_mode, base_url or "", api_key or "", api_mode or "", runtime_key, is_vision, task_key, pool_hint)
    +    # The model MUST participate in the key. Two concurrent auxiliary calls to
    +    # the SAME provider/base_url/key but DIFFERENT models (e.g. a MoA reference
    +    # fan-out running opus + gpt-5.5 in parallel threads) would otherwise share
    +    # one cache entry. On a cache MISS both build a client for the same key; the
    +    # second's _store_cached_client sees the first as the "old" entry and CLOSES
    +    # it — while the first call is still mid-request on it — yielding a spurious
    +    # 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)
     
     
     def _store_cached_client(cache_key: tuple, client: Any, default_model: Optional[str], *, bound_loop: Any = None) -> None:
    @@ -5100,6 +5139,7 @@ def _refresh_nous_auxiliary_client(
             api_mode=api_mode,
             main_runtime=main_runtime,
             is_vision=is_vision,
    +        model=final_model,
         )
         _store_cached_client(cache_key, client, final_model, bound_loop=current_loop)
         return client, final_model
    @@ -5276,6 +5316,7 @@ def _get_cached_client(
             main_runtime=main_runtime,
             is_vision=is_vision,
             task=task,
    +        model=model,
         )
         with _client_cache_lock:
             if cache_key in _client_cache:
    @@ -6005,15 +6046,22 @@ def call_llm(
         # Handle unsupported temperature, max_tokens vs max_completion_tokens retry,
         # then payment fallback.
         try:
    -        # Retry ONCE on the same provider for a one-off transient transport
    -        # blip (streaming-close / incomplete chunked read / 5xx / 408) before
    -        # the except-chain below escalates to provider/model fallback. A
    -        # single dropped connection shouldn't abandon an otherwise-healthy
    -        # provider. A second failure (or any non-transient error) falls
    -        # through to ``first_err`` and the existing fallback handling
    -        # unchanged. This is the unified home for the transient retry that
    -        # every auxiliary task (compression, memory flush, title-gen,
    -        # session-search, vision) shares. (PR #16587)
    +        # Retry on the same provider for a transient transport blip
    +        # (connection reset / streaming-close / incomplete chunked read / 5xx /
    +        # 408) before the except-chain below escalates to provider/model
    +        # fallback. A dropped connection shouldn't abandon an otherwise-healthy
    +        # provider — this especially matters for pinned auxiliary calls like MoA
    +        # reference advisors, where "fallback to another provider" is not a
    +        # meaningful recovery (the advisor is a specific model), so a transient
    +        # blip that isn't retried simply loses that advisor for the turn (root
    +        # of the run2 double-advisor "Connection error" collapse — a genuine
    +        # upstream blip hitting both parallel advisors at once).
    +        #
    +        # Attempts are bounded and use exponential backoff. Count is configurable
    +        # via auxiliary.transient_retries (default 2 retries → 3 total attempts);
    +        # a second/third failure or any non-transient error falls through to
    +        # ``first_err`` and the existing fallback handling unchanged. Unified home
    +        # for the transient retry every auxiliary task shares. (PR #16587)
             try:
                 return _validate_llm_response(
                     client.chat.completions.create(**kwargs), task)
    @@ -6035,13 +6083,26 @@ def call_llm(
                         transient_err,
                     )
                     raise
    -            logger.info(
    -                "Auxiliary %s: transient transport error; retrying once on "
    -                "the same provider before fallback: %s",
    -                task or "call", transient_err,
    -            )
    -            return _validate_llm_response(
    -                client.chat.completions.create(**kwargs), task)
    +            _max_transient_retries = _transient_retry_count()
    +            _last_transient = transient_err
    +            for _attempt in range(1, _max_transient_retries + 1):
    +                _backoff = min(_TRANSIENT_RETRY_BACKOFF_BASE * (2.0 ** (_attempt - 1)), 8.0)
    +                logger.info(
    +                    "Auxiliary %s: transient transport error (attempt %d/%d); "
    +                    "retrying same provider after %.1fs before fallback: %s",
    +                    task or "call", _attempt, _max_transient_retries, _backoff,
    +                    _last_transient,
    +                )
    +                time.sleep(_backoff)
    +                try:
    +                    return _validate_llm_response(
    +                        client.chat.completions.create(**kwargs), task)
    +                except Exception as retry_transient:
    +                    if not _is_transient_transport_error(retry_transient):
    +                        raise
    +                    _last_transient = retry_transient
    +            # Retries exhausted — fall through to first_err fallback handling.
    +            raise _last_transient
         except Exception as first_err:
             if "temperature" in kwargs and _is_unsupported_temperature_error(first_err):
                 retry_kwargs = dict(kwargs)

  • hermes_cli/config.py

    @@ -1465,6 +1465,13 @@ def _ensure_hermes_home_managed(home: Path):
         # Each aux task is independent — main-agent provider_routing and
         # openrouter.min_coding_score do NOT propagate to aux calls by design.
         "auxiliary": {
    +        # Same-provider retries for a transient transport blip (connection
    +        # reset / timeout / 5xx / 408) on ANY auxiliary call before falling
    +        # back. Default 2 (→ 3 total attempts), clamped [0,6]. Matters most for
    +        # pinned calls like MoA reference advisors, where provider fallback is
    +        # not a meaningful recovery, so an unretried blip silently loses the
    +        # call.
    +        "transient_retries": 2,
             "vision": {
                 "provider": "auto",    # auto | openrouter | nous | codex | custom
                 "model": "",           # e.g. "google/gemini-2.5-flash", "gpt-4o"

  • tests/agent/test_auxiliary_client.py

    @@ -1523,7 +1523,13 @@ def test_cached_gmi_client_keeps_explicit_slash_model_override(self):
     
             assert client is fake_client
             assert model == "openai/gpt-5.4-mini"
    -        assert mock_resolve.call_count == 1
    +        # A DIFFERENT model resolves its own client (model participates in the
    +        # cache key). This isolation is what stops two concurrent advisors on
    +        # the same provider/base_url/key (e.g. a MoA fan-out) from sharing — and
    +        # racing the lifecycle of — one cached client. Same-model reuse is still
    +        # a single resolve (verified elsewhere); distinct models => distinct
    +        # resolves.
    +        assert mock_resolve.call_count == 2
     
     
     # ── Payment / credit exhaustion fallback ─────────────────────────────────
    @@ -2581,6 +2587,8 @@ def test_second_transient_failure_escalates_to_fallback(self):
             p1, p2, p3 = self._patches(primary)
             with (
                 p1, p2, p3,
    +            patch("agent.auxiliary_client._transient_retry_count", return_value=1),
    +            patch("agent.auxiliary_client._TRANSIENT_RETRY_BACKOFF_BASE", 0.0),
                 patch(
                     "agent.auxiliary_client._try_configured_fallback_chain",
                     return_value=(None, None, ""),
    @@ -2592,7 +2600,7 @@ def test_second_transient_failure_escalates_to_fallback(self):
             ):
                 result = call_llm(task="compression", messages=[{"role": "user", "content": "hi"}])
             assert result == {"fallback": True}
    -        # Primary tried twice (initial + same-target retry), then fallback.
    +        # Primary tried twice (initial + one same-target retry), then fallback.
             assert primary.chat.completions.create.call_count == 2
             assert fb_client.chat.completions.create.call_count == 1
     

  • tests/agent/test_auxiliary_transient_retry.py

    @@ -0,0 +1,82 @@
    +"""Transient-transport retry count + per-model client-cache isolation.
    +
    +Two related hardening behaviors for auxiliary calls (which include MoA
    +reference advisors, a pinned-model path where provider fallback is not a
    +meaningful recovery):
    +
    +1. A transient transport blip (connection reset / timeout / 5xx) is retried
    +   on the SAME provider several times with backoff before giving up — a single
    +   upstream blip should not silently lose a pinned auxiliary call (root of the
    +   run2 double-advisor "Connection error" collapse).
    +2. Two auxiliary calls to the same provider/base_url/key but DIFFERENT models
    +   get DISTINCT client-cache keys, so a concurrent fan-out (e.g. opus + gpt-5.5
    +   advisors) never shares one client entry.
    +"""
    +
    +from __future__ import annotations
    +
    +import os
    +import types
    +from unittest.mock import patch
    +
    +import pytest
    +
    +
    +class _ConnErr(Exception):
    +    """Stand-in that the transient detector recognizes as a connection blip."""
    +
    +
    +def test_transient_retry_count_default(monkeypatch):
    +    from agent import auxiliary_client as ac
    +
    +    # No config value -> default.
    +    monkeypatch.setattr(ac, "load_config", lambda: {}, raising=False)
    +    with patch("hermes_cli.config.load_config", return_value={}), \
    +         patch("hermes_cli.config.cfg_get", return_value=None):
    +        assert ac._transient_retry_count() == ac._DEFAULT_TRANSIENT_RETRIES
    +
    +
    +def test_transient_retry_count_configurable_and_clamped():
    +    from agent import auxiliary_client as ac
    +
    +    with patch("hermes_cli.config.cfg_get", return_value=4):
    +        assert ac._transient_retry_count() == 4
    +    with patch("hermes_cli.config.cfg_get", return_value=100):
    +        assert ac._transient_retry_count() == 6  # clamped high
    +    with patch("hermes_cli.config.cfg_get", return_value=-3):
    +        assert ac._transient_retry_count() == 0  # clamped low
    +    with patch("hermes_cli.config.cfg_get", side_effect=RuntimeError):
    +        assert ac._transient_retry_count() == ac._DEFAULT_TRANSIENT_RETRIES
    +
    +
    +def test_model_participates_in_client_cache_key():
    +    """Same provider/base_url/key, different model -> different cache key.
    +
    +    This is what stops two concurrent advisors from sharing (and racing on)
    +    one cached client entry."""
    +    from agent.auxiliary_client import _client_cache_key
    +
    +    k_opus = _client_cache_key(
    +        "openrouter", async_mode=False, base_url="https://openrouter.ai/api/v1",
    +        api_key="K", model="anthropic/claude-opus-4.8",
    +    )
    +    k_gpt = _client_cache_key(
    +        "openrouter", async_mode=False, base_url="https://openrouter.ai/api/v1",
    +        api_key="K", model="openai/gpt-5.5",
    +    )
    +    assert k_opus != k_gpt
    +    # Same model still collides (cache still works for reuse).
    +    k_opus2 = _client_cache_key(
    +        "openrouter", async_mode=False, base_url="https://openrouter.ai/api/v1",
    +        api_key="K", model="anthropic/claude-opus-4.8",
    +    )
    +    assert k_opus == k_opus2
    +
    +
    +def test_missing_model_key_is_stable():
    +    """Omitting model (legacy callers) is still a valid, stable key."""
    +    from agent.auxiliary_client import _client_cache_key
    +
    +    a = _client_cache_key("openrouter", async_mode=False, base_url="u", api_key="k")
    +    b = _client_cache_key("openrouter", async_mode=False, base_url="u", api_key="k")
    +    assert a == b

• #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 = api_key.strip() 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 = runtime.get("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 = runtime.get("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 = _runtime_cache_discriminator("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": [redacted:secret-assignment] "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 = set_accounting_context(
                 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 @@
         "[email protected]": "synapsesx",
         "[email protected]": "itsflownium",
         "[email protected]": "dirtyren",
    +    "[email protected]": "krowd3v",
    +    "[email protected]": "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": "[redacted:secret-assignment]",
    +        "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": "[redacted:secret-assignment]"},
    +        {**base, "api_mode": "codex_responses"},
    +        {**base, "auth_mode": "entra_id", "api_key": [redacted:secret-assignment] "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": [redacted:secret-assignment]
    +    )
    +    second = aux._client_cache_key(
    +        "auto", async_mode=False, main_runtime={**_runtime("same"), "api_key": [redacted:secret-assignment]
    +    )
    +
    +    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": [redacted:secret-assignment]
    +    )
    +    second = aux._client_cache_key(
    +        "auto",
    +        async_mode=False,
    +        api_key=[redacted:secret-assignment]
    +        main_runtime={**_runtime("same"), "api_key": [redacted:secret-assignment]
    +    )
    +
    +    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": "[redacted:secret-assignment]",
    +            "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": "[redacted:secret-assignment]",
                 "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":