fix(agent): scope auxiliary client eviction to the calling profile and stop closing in-flight clients
Two review findings on the re-armed _evict_cached_clients: 1. It matched on the provider slot only and ignored the leading hermes_home_key() slot, so an OAuth rotation/refresh in profile A also dropped (and closed) profile B's cached client for the same provider in a multiplexing gateway, although B's per-profile credentials did not change. Callers all run inside the rotating profile's scope, so filter on key[0] too. 2. It closed the shared client inline. A concurrent caller mid-request on the same cached client (the normal cache-hit case for parallel compressions) got ReadError, and any caller still holding it got "Cannot send a request, as the client has been closed" - the exact hazard the #113022 reporter cited for NOT enabling eviction. Pop the entry instead and let the dropped client retire via GC, as the FIFO overflow path in _get_cached_client already does; the next call builds a fresh client from the rotated credentials. Part of #113022
This commit is contained in:
@@ -3297,14 +3297,20 @@ def _should_skip_same_provider_retry(task: Optional[str], exc: Exception) -> boo
|
||||
|
||||
|
||||
def _evict_cached_clients(provider: str) -> None:
|
||||
"""Drop cached auxiliary clients for a provider so fresh creds are used."""
|
||||
"""Drop this profile's cached auxiliary clients for a provider so fresh creds are used.
|
||||
|
||||
Scoped to the calling profile (``hermes_home_key()`` is the first key slot): a rotation in
|
||||
one profile must not drop another profile's client for the same provider in a multiplexing
|
||||
gateway, since that profile's credentials did not change. Entries are popped, not closed:
|
||||
a concurrent caller may be mid-request on the shared client (closing it raises ReadError /
|
||||
"client has been closed" for them); the dropped client is retired by GC like the FIFO
|
||||
overflow path in ``_get_cached_client``.
|
||||
"""
|
||||
normalized = _normalize_aux_provider(provider)
|
||||
home = hermes_home_key()
|
||||
with _client_cache_lock:
|
||||
# Cache keys begin with the profile home; the provider is the second component.
|
||||
for key in [key for key in _client_cache if _normalize_aux_provider(str(key[1])) == normalized]:
|
||||
client = _client_cache.get(key, (None, None, None))[0]
|
||||
if client is not None:
|
||||
_close_cached_client(client)
|
||||
for key in [key for key in _client_cache
|
||||
if key[0] == home and _normalize_aux_provider(str(key[1])) == normalized]:
|
||||
_client_cache.pop(key, None)
|
||||
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import agent.auxiliary_client as aux
|
||||
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
|
||||
|
||||
|
||||
def _cache_entry() -> tuple[MagicMock, str, None]:
|
||||
@@ -29,11 +30,38 @@ def test_evict_cached_clients_matches_provider_after_profile_home_key(monkeypatc
|
||||
assert sync_key not in aux._client_cache
|
||||
assert async_key not in aux._client_cache
|
||||
assert other_key in aux._client_cache
|
||||
anthropic_sync[0].close.assert_called_once()
|
||||
anthropic_async[0].close.assert_called_once()
|
||||
other_provider[0].close.assert_not_called()
|
||||
|
||||
|
||||
def test_evict_cached_clients_is_scoped_to_the_calling_profile(monkeypatch, tmp_path):
|
||||
"""A rotation in profile A must not drop profile B's client for the same provider."""
|
||||
key_a = aux._client_cache_key("anthropic", async_mode=False)
|
||||
token = set_hermes_home_override(tmp_path / "profiles" / "b")
|
||||
try:
|
||||
key_b = aux._client_cache_key("anthropic", async_mode=False)
|
||||
finally:
|
||||
reset_hermes_home_override(token)
|
||||
assert key_a[0] != key_b[0]
|
||||
monkeypatch.setattr(aux, "_client_cache", {key_a: _cache_entry(), key_b: _cache_entry()})
|
||||
|
||||
aux._evict_cached_clients("anthropic")
|
||||
|
||||
assert key_a not in aux._client_cache
|
||||
assert key_b in aux._client_cache
|
||||
|
||||
|
||||
def test_evict_cached_clients_does_not_close_possibly_in_flight_client(monkeypatch):
|
||||
"""Eviction pops the entry; a concurrent caller mid-request must not get a closed client."""
|
||||
entry = _cache_entry()
|
||||
key = aux._client_cache_key("anthropic", async_mode=False)
|
||||
monkeypatch.setattr(aux, "_client_cache", {key: entry})
|
||||
|
||||
aux._evict_cached_clients("anthropic")
|
||||
|
||||
assert key not in aux._client_cache
|
||||
entry[0].close.assert_not_called()
|
||||
|
||||
|
||||
def test_pool_rotation_evicts_client_built_with_revoked_credential(monkeypatch):
|
||||
"""A 401 pool rotation drops the old provider client before retry/fallback."""
|
||||
stale_entry = _cache_entry()
|
||||
@@ -51,5 +79,4 @@ def test_pool_rotation_evicts_client_built_with_revoked_credential(monkeypatch):
|
||||
assert aux._recover_provider_pool("anthropic", auth_error, failed_api_key="revoked-token") is True
|
||||
|
||||
assert stale_key not in aux._client_cache
|
||||
stale_entry[0].close.assert_called_once()
|
||||
pool.mark_exhausted_and_rotate.assert_called_once()
|
||||
|
||||
Reference in New Issue
Block a user