fix(auth): key sibling per-process credential memos by profile home under multiplex

Same class as the resolve_nous_access_token memo: three more process-wide
memos carried a credential resolved under one profile's HERMES_HOME override
into another profile's turn for their TTL.

- hermes_cli/nous_billing.py::_token_cache (30s (token, base) memo for the
  charge poll loop) was a single unkeyed slot -> dict keyed by
  hermes_home_key(); invalidate_cached_token() clears the dict.
- agent/moa_loop.py::_runtime_cache carried api_key/base_url/api_mode keyed
  only (provider, model) for 5 min -> (hermes_home_key(), provider, model).
- agent/auxiliary_client.py::_client_cache_key had no profile component, so
  callers that omit api_key (pool / Nous auth.json paths) could be handed a
  client built with another profile's bearer -> hermes_home_key() leads the key.

WHY hermes_home_key(): it reads the per-turn HERMES_HOME override the
multiplex gateway sets (falling back to the env var), and it is
symlink-stable, so the memo key is exactly the credential home the
resolution itself read from. Profiles stay independent islands; the
default-profile process env never leaks into a secondary's turn.

Tests: one invariant per site, proven red on origin/main.

agent/auxiliary_client.py

@@ -113,7 +113,7 @@ from agent.model_metadata import (
 )
 from hermes_cli.config import get_hermes_home
 from agent.auxiliary_health import _custom_health_base_url, _unhealthy_cache_key
-from hermes_constants import OPENROUTER_BASE_URL
+from hermes_constants import OPENROUTER_BASE_URL, hermes_home_key
 from utils import base_url_host_matches, base_url_hostname, base_url_origin, env_float, is_truthy_value, model_forces_max_completion_tokens, normalize_proxy_env_vars
 
 logger = logging.getLogger(__name__)
@@ -5314,7 +5314,9 @@ def _client_cache_key(
     # share an entry, and the second builder's _store_cached_client would close the first's client.
     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)
+    # Profile home leads the key: callers that omit api_key (pool / Nous auth.json paths) would
+    # otherwise share one client across multiplex profiles holding different credentials.
+    return (hermes_home_key(), provider, async_mode, base_url or "", api_key_key, api_mode or "", runtime_key, is_vision, task_key, pool_hint, model_key)
 
 
 def _current_event_loop() -> Any:

agent/moa_loop.py

@@ -142,7 +142,7 @@ def _resolve_preset_cached(preset_name: str) -> tuple[dict[str, Any], Any]:
 
 
 _runtime_cache_lock = threading.Lock()
-_runtime_cache: dict[tuple[str, str], tuple[float, dict[str, Any]]] = {}
+_runtime_cache: dict[tuple[str, str, str], tuple[float, dict[str, Any]]] = {}
 
 # Short TTL so rotated keys / base_url edits are picked up within 5 minutes.
 _RUNTIME_CACHE_TTL_SECONDS = 300.0
@@ -245,12 +245,15 @@ def _aggregator_reasoning_config(aggregator: dict[str, Any]) -> dict[str, Any] |
 def _slot_runtime(slot: dict[str, Any]) -> dict[str, Any]:
     """Slot → ``call_llm`` kwargs with the provider's real api_mode/base_url/api_key.
 
-    Cached per (provider, model) with a short TTL. Falls back to bare provider/model
+    Cached per (profile home, provider, model) with a short TTL. Falls back to bare provider/model
     on error — never cached, or a transient error would pin bare kwargs for a TTL.
     """
     provider = str(slot.get("provider") or "").strip()
     model = str(slot.get("model") or "").strip()
-    cache_key = (provider, model)
+    # hermes_home_key() in the key: the resolved api_key/base_url are per-profile, and under a
+    # multiplex gateway two profiles can share (provider, model) with different accounts.
+    from hermes_constants import hermes_home_key
+    cache_key = (hermes_home_key(), provider, model)
     now = time.monotonic()
     with _runtime_cache_lock:
         entry = _runtime_cache.get(cache_key)

hermes_cli/nous_billing.py

@@ -121,8 +121,9 @@ def _absolutize_portal_url(portal_url: Optional[str]) -> Optional[str]:
 # cross-process file locks + reads two files per call, wasteful for the 2s charge poll loop
 # (~150 calls per purchase). The resolver only returns tokens with >=120s of life (its refresh
 # skew), so a 30s cache can never hand back an about-to-expire token; a 401 still surfaces.
+# Keyed by hermes_home_key() so a multiplex profile never bills through a sibling's token.
 _TOKEN_CACHE_TTL_SECONDS = 30.0
-_token_cache: tuple[float, str, str] | None = None  # (cached_at, token, base)
+_token_cache: dict[str, tuple[float, str, str]] = {}  # home key -> (cached_at, token, base)
 
 
 def invalidate_cached_token() -> None:
@@ -131,8 +132,7 @@ def invalidate_cached_token() -> None:
     ``_request`` only self-busts on a 401, not on a 403 scope denial — after a step-up grant the
     cache would otherwise still hold the pre-grant unscoped token and the replay would 403 again.
     """
-    global _token_cache
-    _token_cache = None
+    _token_cache.clear()
 
 
 def _billing_not_logged_in(exc: Optional[BaseException] = None) -> "BillingAuthError":
@@ -145,9 +145,12 @@ def _billing_not_logged_in(exc: Optional[BaseException] = None) -> "BillingAuthE
 
 def _resolve_token_and_base(*, use_cache: bool = True) -> tuple[str, str]:
     """``(access_token, portal_base_url)``, cached for ``_TOKEN_CACHE_TTL_SECONDS`` unless ``use_cache=False``."""
-    global _token_cache
-    if use_cache and _token_cache is not None:
-        cached_at, token, base = _token_cache
+    from hermes_constants import hermes_home_key
+
+    cache_key = hermes_home_key()
+    cached = _token_cache.get(cache_key) if use_cache else None
+    if cached is not None:
+        cached_at, token, base = cached
         if (time.time() - cached_at) < _TOKEN_CACHE_TTL_SECONDS:
             return token, base
     try:
@@ -170,7 +173,7 @@ def _resolve_token_and_base(*, use_cache: bool = True) -> tuple[str, str]:
         except AuthError as exc:
             raise _billing_not_logged_in(exc) from exc
     resolved = (token.strip(), base)
-    _token_cache = (time.time(), *resolved)
+    _token_cache[cache_key] = (time.time(), *resolved)
     return resolved
 
 

tests/agent/test_auxiliary_runtime_cache_key.py

@@ -152,3 +152,20 @@ def test_string_api_keys_are_not_retained_in_cache_key_repr():
     assert second_secret not in rendered
 
 
+
+
+def test_client_cache_key_is_scoped_per_profile_home(tmp_path):
+    """Callers that omit api_key (pool / Nous auth.json paths) must not share a client across
+    multiplex profiles: the per-turn HERMES_HOME override has to participate in the key."""
+    import hermes_constants
+
+    a, b = tmp_path / "a", tmp_path / "b"
+    a.mkdir(); b.mkdir()
+    keys = []
+    for home in (a, b):
+        tok = hermes_constants.set_hermes_home_override(str(home))
+        try:
+            keys.append(aux._client_cache_key("nous", async_mode=False, base_url="https://inf.example", model="m"))
+        finally:
+            hermes_constants.reset_hermes_home_override(tok)
+    assert keys[0] != keys[1]

tests/agent/test_moa_cold_start_cache_66793.py

@@ -14,6 +14,8 @@ import types  # noqa: F401  (used by _fake_response)
 
 import pytest
 
+from hermes_constants import hermes_home_key
+
 
 def _make_preset_config() -> dict:
     return {
@@ -181,7 +183,7 @@ def test_slot_runtime_cache_expires_after_ttl(monkeypatch):
     assert calls["n"] == 1
 
     # Age the entry past the TTL and confirm re-resolution.
-    key = ("openai", "gpt-5")
+    key = (hermes_home_key(), "openai", "gpt-5")
     stamped_at, cached = moa._runtime_cache[key]
     moa._runtime_cache[key] = (
         stamped_at - moa._RUNTIME_CACHE_TTL_SECONDS - 1, cached
@@ -224,3 +226,33 @@ def _fake_response():
     ns = types.SimpleNamespace()
     ns.usage = None
     return ns
+
+
+def test_slot_runtime_cache_is_scoped_per_profile_home(monkeypatch, tmp_path):
+    """Under a multiplex gateway two profiles can share (provider, model) with different
+    accounts; a cached api_key/base_url must never cross the per-turn HERMES_HOME override."""
+    import agent.moa_loop as moa
+    import hermes_constants
+
+    moa._runtime_cache.clear()
+    a, b = tmp_path / "a", tmp_path / "b"
+    a.mkdir(); b.mkdir()
+
+    import hermes_cli.runtime_provider as rt_mod
+    monkeypatch.setattr(
+        rt_mod, "resolve_runtime_provider",
+        lambda **kw: {"provider": "openai", "model": "gpt-5",
+                      "api_key": f"key-{hermes_constants.get_hermes_home().name}",
+                      "base_url": "https://x", "api_mode": None})
+
+    slot = {"provider": "openai", "model": "gpt-5"}
+    tok = hermes_constants.set_hermes_home_override(str(a))
+    try:
+        assert moa._slot_runtime(slot)["api_key"] == "key-a"
+    finally:
+        hermes_constants.reset_hermes_home_override(tok)
+    tok = hermes_constants.set_hermes_home_override(str(b))
+    try:
+        assert moa._slot_runtime(slot)["api_key"] == "key-b"
+    finally:
+        hermes_constants.reset_hermes_home_override(tok)

tests/hermes_cli/test_nous_billing_request.py

@@ -48,7 +48,7 @@ def _http_error(status: int, body: bytes | dict[str, object] = b"{}", headers=No
 def _sequence(monkeypatch, *outcomes, resolver=None):
     """Stub urlopen with ordered outcomes and record each Request."""
     seen: list[dict[str, object]] = []
-    monkeypatch.setattr(nb, "_token_cache", None, raising=False)
+    monkeypatch.setattr(nb, "_token_cache", {}, raising=False)
     monkeypatch.setattr(
         nb,
         "_resolve_token_and_base",
@@ -77,7 +77,7 @@ def _sequence(monkeypatch, *outcomes, resolver=None):
 def _stub(monkeypatch, body: bytes, status: int = 200):
     # Bypass auth/token resolution entirely — we only exercise response parsing.
     monkeypatch.setattr(nb, "_resolve_token_and_base", lambda **kw: ("tok", "https://portal.example"))
-    monkeypatch.setattr(nb, "_token_cache", None, raising=False)
+    monkeypatch.setattr(nb, "_token_cache", {}, raising=False)
     monkeypatch.setattr(nb.urllib.request, "urlopen", lambda req, timeout=None: _FakeResp(body, status))
     yield
 
@@ -212,3 +212,26 @@ def test_404_get_charge_status_maps_to_generic_billing_error(monkeypatch):
 
 
 
+
+
+def test_billing_token_cache_is_scoped_per_profile_home(monkeypatch, tmp_path):
+    """The 30s (token, base) memo must not hand profile A's Portal bearer to profile B under a
+    multiplex gateway, where the per-turn HERMES_HOME override selects the auth.json."""
+    import hermes_constants
+    import hermes_cli.auth as auth
+
+    monkeypatch.setattr(nb, "_token_cache", {}, raising=False)
+    monkeypatch.setattr(auth, "get_provider_auth_state", lambda provider: {})
+    monkeypatch.setattr(
+        auth, "resolve_nous_access_token",
+        lambda **kw: f"tok-{hermes_constants.get_hermes_home().name}")
+    a, b = tmp_path / "a", tmp_path / "b"
+    a.mkdir(); b.mkdir()
+    seen = []
+    for home in (a, b):
+        tok = hermes_constants.set_hermes_home_override(str(home))
+        try:
+            seen.append(nb._resolve_token_and_base()[0])
+        finally:
+            hermes_constants.reset_hermes_home_override(tok)
+    assert seen == ["tok-a", "tok-b"]