diff --git a/agent/credential_pool.py b/agent/credential_pool.py index 4286903c0e..027d484c62 100644 --- a/agent/credential_pool.py +++ b/agent/credential_pool.py @@ -791,7 +791,10 @@ def _borrowed_single_use_pool_root() -> Optional[Path]: return None -def _update_root_pool_rows(provider: str, payloads: List[Dict[str, Any]], global_path: Path) -> None: +def _update_root_pool_rows( + provider: str, payloads: List[Dict[str, Any]], global_path: Path, + *, status_cleared_ids: Optional[Iterable[str]] = None, +) -> None: """UPDATE-ONLY merge of *payloads* into the root store's rows for *provider*. A borrower may refresh the root's rows (rotation, cooldown state) but @@ -816,7 +819,13 @@ def _update_root_pool_rows(provider: str, payloads: List[Dict[str, Any]], global if incoming is None: merged.append(disk_entry) continue - updated = auth_mod._merge_disk_cooldown_state(incoming, disk_entry, provider) + cleared = {cid for cid in (status_cleared_ids or ()) if cid} + if did in cleared: + # The caller deliberately cleared this entry's status; do not + # let the disk recency merge restore the cooldown. + updated = incoming + else: + updated = auth_mod._merge_disk_cooldown_state(incoming, disk_entry, provider) if updated != disk_entry: changed = True merged.append(updated) @@ -830,6 +839,7 @@ def persist_pool_entries( payloads: List[Dict[str, Any]], *, removed_ids: Optional[Iterable[str]] = None, + status_cleared_ids: Optional[Iterable[str]] = None, ) -> None: """Persist a provider's pool rows to the store that OWNS them. @@ -845,7 +855,10 @@ def persist_pool_entries( global_path = _borrowed_single_use_pool_root() if global_path is not None: try: - _update_root_pool_rows(provider, payloads, global_path) + _update_root_pool_rows( + provider, payloads, global_path, + status_cleared_ids=status_cleared_ids, + ) except Exception as exc: # Fail closed on the FORK, not on the save: never fall back to # writing a local copy (that IS the bug). The in-memory pool @@ -856,7 +869,11 @@ def persist_pool_entries( provider, exc, ) return - write_credential_pool(provider, payloads, removed_ids=removed_ids) + write_credential_pool( + provider, payloads, + removed_ids=removed_ids, + **({"status_cleared_ids": status_cleared_ids} if status_cleared_ids else {}), + ) # --- Per-provider singleton refresh plumbing ------------------------------- @@ -1018,13 +1035,19 @@ class CredentialPool: self._entries[idx] = new return - def _persist(self, *, removed_ids: Optional[List[str]] = None) -> None: + def _persist( + self, + *, + removed_ids: Optional[List[str]] = None, + status_cleared_ids: Optional[List[str]] = None, + ) -> None: # Self-locking: snapshotting self._entries must not race a rotation. with self._lock: persist_pool_entries( self.provider, [entry.to_dict() for entry in self._entries], removed_ids=removed_ids, + status_cleared_ids=status_cleared_ids, ) def _adopt(self, entry: PooledCredential, *, persist: bool = True, **updates: Any) -> PooledCredential: @@ -2139,15 +2162,63 @@ class CredentialPool: return refreshed def reset_statuses(self) -> int: + """Clear exhaustion state on every entry. Returns how many were cleared. + + Two details are load-bearing, and both were missing: + + ``failure_reason`` is cleared alongside the status fields. It lives in + ``extra`` rather than as a dataclass field, so ``replace()`` cannot reach + it and it survived every reset — leaving an entry with no status but + still classified ``billing``, a contradiction ``hermes auth list`` + renders as if it were a finding. + + The persist declares the clear as DELIBERATE. ``write_credential_pool`` + merges on-disk status over the caller's snapshot whenever the disk copy + is more recent and still binding, so a process cannot resurrect a key + another one has just rate-limited. Clearing sets ``last_status_at`` to + None, which compares as epoch 0 — older than any real timestamp — so + that merge read an operator reset as exactly the stale snapshot it + exists to reject and copied the cooldown straight back. The command + reported success and changed nothing, and only while the cooldown was + still binding: once expired the merge bails out early and the reset + appeared to work. Broken precisely when it is needed. + """ with self._lock: - stale = [e for e in self._entries if e.last_status or e.last_status_at or e.last_error_code] - if stale: - stale_ids = {e.id for e in stale} - self._entries = [ - replace(e, **_CLEAR_STATUS) if e.id in stale_ids else e for e in self._entries - ] - self._persist() - return len(stale) + count = 0 + cleared_ids: List[str] = [] + new_entries = [] + for entry in self._entries: + if ( + entry.last_status + or entry.last_status_at + or entry.last_error_code + or getattr(entry, "failure_reason", None) + ): + extra = { + key: value + for key, value in (entry.extra or {}).items() + if key != "failure_reason" + } + new_entries.append( + replace( + entry, + last_status=None, + last_status_at=None, + last_error_code=None, + last_error_reason=None, + last_error_message=None, + last_error_reset_at=None, + extra=extra, + ) + ) + cleared_ids.append(entry.id) + count += 1 + else: + new_entries.append(entry) + if count: + self._entries = new_entries + self._persist(status_cleared_ids=cleared_ids) + return count def remove_index(self, index: int) -> Optional[PooledCredential]: with self._lock: diff --git a/hermes_cli/auth.py b/hermes_cli/auth.py index 9983ae6141..7790178c29 100644 --- a/hermes_cli/auth.py +++ b/hermes_cli/auth.py @@ -924,13 +924,23 @@ def _entry_ids(entries: Iterable[Any]) -> Dict[str, Dict[str, Any]]: def write_credential_pool( - provider_id: str, entries: List[Dict[str, Any]], *, removed_ids: Optional[Iterable[str]] = None, + provider_id: str, entries: List[Dict[str, Any]], *, + removed_ids: Optional[Iterable[str]] = None, + status_cleared_ids: Optional[Iterable[str]] = None, ) -> Path: """Persist one provider's credential pool under auth.json. Final disk-boundary sanitizer for borrowed credentials (callers may pass raw dicts). Entries on disk but missing from *entries* (added concurrently) are merged back unless in *removed_ids*, - so a rotation/exhaustion rewrite never drops a concurrent credential.""" + so a rotation/exhaustion rewrite never drops a concurrent credential. + + Pass ``status_cleared_ids`` for entries whose status the caller intentionally + cleared — the same problem one step further in. The recency merge cannot tell + an operator's ``hermes auth reset`` from a stale snapshot: a clear sets + ``last_status_at`` to None, which compares as epoch 0 and so always loses to + the on-disk timestamp, and the cooldown was copied straight back. Declaring + the intent is what separates "I have not seen the newer status" from "I have + seen it and I am dropping it".""" removed = {rid for rid in (removed_ids or ()) if rid} with _auth_store_lock(): auth_store = _load_auth_store() @@ -942,9 +952,15 @@ def write_credential_pool( existing_list = existing_list if isinstance(existing_list, list) else [] existing_by_id = _entry_ids(existing_list) new_ids = set(_entry_ids(sanitized)) + status_cleared = {cid for cid in (status_cleared_ids or ()) if cid} merged: List[Dict[str, Any]] = [ - _merge_disk_cooldown_state(e, existing_by_id.get(e.get("id")), provider_id) - if isinstance(e, dict) else e + e + if isinstance(e, dict) and e.get("id") in status_cleared + else ( + _merge_disk_cooldown_state(e, existing_by_id.get(e.get("id")), provider_id) + if isinstance(e, dict) + else e + ) for e in sanitized] for disk_entry in existing_list: disk_id = disk_entry.get("id") if isinstance(disk_entry, dict) else None diff --git a/tests/agent/test_credential_pool.py b/tests/agent/test_credential_pool.py index edc370a776..4213bb611f 100644 --- a/tests/agent/test_credential_pool.py +++ b/tests/agent/test_credential_pool.py @@ -2079,3 +2079,133 @@ class TestCredentialPoolQueryLocking: inner.release() assert done.wait(timeout=2.0), f"{method}() did not complete after lock release" + + +def _exhausted_billing_store(tmp_path, *, age_seconds: float): + """An auth store with one deepseek entry benched for a billing failure.""" + _write_auth_store( + tmp_path, + { + "version": 1, + "credential_pool": { + "deepseek": [ + { + "id": "cred-1", + "label": "api-key-1", + "auth_type": "api_key", + "priority": 0, + "source": "manual", + "access_token": "sk-test", + "last_status": "exhausted", + "last_status_at": time.time() - age_seconds, + "last_error_code": 402, + "last_error_reason": "invalid_request_error", + "last_error_message": "Insufficient Balance", + "failure_reason": "billing", + } + ] + }, + }, + ) + + +def _disk_entry(tmp_path) -> dict: + """The deepseek entry as it actually reached disk.""" + store = json.loads((tmp_path / "hermes" / "auth.json").read_text()) + entries = store["credential_pool"]["deepseek"] + assert len(entries) == 1, entries + return entries[0] + + +def test_reset_statuses_clears_a_cooldown_that_is_still_binding(tmp_path, monkeypatch): + """An operator reset has to survive the disk-recency merge. + + ``write_credential_pool`` keeps a NEWER on-disk cooldown over the caller's + snapshot so one process cannot resurrect a key another has just benched. + ``reset_statuses`` clears ``last_status_at`` to None, which that merge reads + as epoch 0 — older than any real timestamp — so the reset always lost and + the cooldown was copied straight back. ``hermes auth reset`` printed "Reset + status on 1 credentials" and changed nothing on disk. + + The cooldown here is deliberately RECENT. Once a cooldown has expired the + merge bails out early, so the same assertions pass with or without the fix: + a test written against an expired cooldown proves nothing. Verified by + reverting the source change with the tests kept: this one and the + failure_reason test fail, and the guard test below keeps passing, which is + how it is known to pin pre-existing behaviour rather than the new flag. + """ + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + _exhausted_billing_store(tmp_path, age_seconds=5) + + from agent.credential_pool import load_pool + + assert load_pool("deepseek").reset_statuses() == 1 + + entry = _disk_entry(tmp_path) + assert entry["last_status"] is None + assert entry["last_status_at"] is None + assert entry["last_error_code"] is None + # And a fresh load agrees, which is what the next process will see. The + # operational symptom of the bug was the CLI refusing the provider outright + # with "No usable credentials found", so availability is the property that + # matters here, not any single field. + assert load_pool("deepseek").has_available() is True + + +def test_reset_statuses_clears_the_classified_failure_reason(tmp_path, monkeypatch): + """``failure_reason`` is part of the exhaustion state, so a reset clears it. + + It lives in ``extra`` rather than as a dataclass field, so ``replace()`` + could not reach it and it outlived every reset — leaving an entry with no + status and no error code but still classified ``billing``. ``hermes auth + list`` renders that leftover as though it were a current finding. + """ + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + _exhausted_billing_store(tmp_path, age_seconds=5) + + from agent.credential_pool import load_pool + + assert load_pool("deepseek").reset_statuses() == 1 + + entry = _disk_entry(tmp_path) + assert entry.get("failure_reason") is None + + +def test_a_persist_without_declared_intent_still_cannot_erase_a_cooldown( + tmp_path, monkeypatch +): + """The concurrency guard the fix threads through must still hold. + + This is the property ``status_cleared_ids`` is scoped against: a writer that + has NOT declared a deliberate clear is presumed to be holding a stale + snapshot, and a binding on-disk cooldown outranks it. Without this test the + fix could have been "skip the merge always", which would let one process + resurrect a key another had just rate-limited — the exact lost update the + merge exists to prevent. + """ + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + _exhausted_billing_store(tmp_path, age_seconds=5) + + from hermes_cli.auth import write_credential_pool + + # A stale snapshot: same id, status cleared, intent NOT declared. + write_credential_pool( + "deepseek", + [ + { + "id": "cred-1", + "label": "api-key-1", + "auth_type": "api_key", + "priority": 0, + "source": "manual", + "access_token": "sk-test", + "last_status": None, + "last_status_at": None, + "last_error_code": None, + } + ], + ) + + entry = _disk_entry(tmp_path) + assert entry["last_status"] == "exhausted" + assert entry["last_error_code"] == 402