fix(auth): make hermes auth reset actually clear a binding cooldown

Port onto the split layout: persist_pool_entries now forwards
status_cleared_ids to BOTH writers — write_credential_pool (hermes_cli/
auth.py) and the root-store row merge for single-use-refresh providers
(_update_root_pool_rows), which the original fix predates. Both skip the
disk recency merge for deliberately-cleared entries; the kwarg is only
sent when non-empty so upstream fakes with the old signature keep
working. reset_statuses clears failure_reason (lives in extra, replace()
cannot reach it) alongside the status fields.

tests/agent/test_credential_pool.py: 62 passed; refresh-race suite: 3
passed (mutation: the reset/binding tests fail with the fix stashed).
The dashboard-auth-gate / user-providers failures are pre-existing on
pristine upstream/main (reproduced with changes stashed).

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:

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

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