From 3d47257e10b90977000ee1253f9d606051be8db4 Mon Sep 17 00:00:00 2001 From: Anthony Ruiz Date: Thu, 20 Aug 2026 23:14:10 +0000 Subject: [PATCH] fix: reset auth cooldowns across profiles --- hermes_cli/auth.py | 39 +++++ hermes_cli/auth_commands.py | 112 +++++++++++- hermes_cli/subcommands/auth.py | 11 ++ tests/hermes_cli/test_auth_reset_profiles.py | 174 +++++++++++++++++++ 4 files changed, 333 insertions(+), 3 deletions(-) create mode 100644 tests/hermes_cli/test_auth_reset_profiles.py diff --git a/hermes_cli/auth.py b/hermes_cli/auth.py index 2b1d4e4a4543..a290a338615a 100644 --- a/hermes_cli/auth.py +++ b/hermes_cli/auth.py @@ -1665,6 +1665,45 @@ def read_credential_pool(provider_id: Optional[str] = None) -> Dict[str, Any]: ) +def reset_credential_pool_statuses( + provider_id: str, + *, + auth_file: Optional[Path] = None, +) -> int: + """Clear one provider's cooldown fields in exactly one local auth store. + + ``read_credential_pool`` may return entries inherited from the global-root + fallback, while ordinary pool writes remain profile-local. Resetting a + loaded :class:`~agent.credential_pool.CredentialPool` would therefore risk + materialising inherited credentials in a named profile. Work directly on + the requested store instead so callers can reset several profile stores + without copying credentials between them. + """ + target_path = auth_file if auth_file is not None else _auth_file_path() + with _auth_store_lock(target_path=target_path): + auth_store = _load_auth_store(target_path) + pool = auth_store.get("credential_pool") + if not isinstance(pool, dict): + return 0 + entries = pool.get(provider_id) + if not isinstance(entries, list): + return 0 + + count = 0 + for entry in entries: + if not isinstance(entry, dict): + continue + if not any(entry.get(field) is not None for field in _POOL_STATUS_FIELDS): + continue + for field in _POOL_STATUS_FIELDS: + entry[field] = None + count += 1 + + if count: + _save_auth_store(auth_store, target_path=target_path) + return count + + def _merge_disk_cooldown_state( entry: Dict[str, Any], disk_entry: Optional[Dict[str, Any]], diff --git a/hermes_cli/auth_commands.py b/hermes_cli/auth_commands.py index a23c0f2a9cba..8d3e78c380c0 100644 --- a/hermes_cli/auth_commands.py +++ b/hermes_cli/auth_commands.py @@ -6,6 +6,7 @@ import math import sys import time +from pathlib import Path from types import SimpleNamespace import uuid @@ -500,11 +501,116 @@ def auth_remove_command(args) -> None: print(line) +def _same_path(left: Path, right: Path) -> bool: + try: + return left.resolve(strict=False) == right.resolve(strict=False) + except Exception: + return left == right + + +def _iter_known_profile_homes() -> list[tuple[str, Path]]: + """Return the default and valid named profile homes without metadata scans.""" + from hermes_constants import get_default_hermes_root + + default_home = get_default_hermes_root() + targets: list[tuple[str, Path]] = [("default", default_home)] + profiles_root = default_home / "profiles" + try: + entries = sorted(profiles_root.iterdir()) if profiles_root.is_dir() else () + except OSError: + return targets + + from hermes_cli.profiles import normalize_profile_name, validate_profile_name + + for entry in entries: + if not entry.is_dir() or entry.name == "default": + continue + try: + name = normalize_profile_name(entry.name) + validate_profile_name(name) + except ValueError: + continue + targets.append((name, entry)) + return targets + + +def _label_for_profile_home( + home: Path, + known_homes: list[tuple[str, Path]], +) -> str: + for name, known_home in known_homes: + if _same_path(home, known_home): + return name + return "current" + + +def _auth_reset_targets( + *, + include_all_profiles: bool, + current_profile_only: bool, +) -> list[tuple[str, Path]]: + from hermes_constants import get_default_hermes_root, get_hermes_home + + current_home = get_hermes_home() + default_home = get_default_hermes_root() + known_homes = _iter_known_profile_homes() + candidates = [ + (_label_for_profile_home(current_home, known_homes), current_home), + ] + if not current_profile_only: + # Named profiles can read a provider from the root fallback. Reset that + # store too, but never copy its credentials into the profile-local file. + candidates.append(("default", default_home)) + if include_all_profiles: + candidates.extend(known_homes) + + deduped: list[tuple[str, Path]] = [] + seen: set[str] = set() + for name, home in candidates: + try: + key = str(home.resolve(strict=False)) + except Exception: + key = str(home) + if key in seen: + continue + seen.add(key) + deduped.append((name, home)) + return deduped + + def auth_reset_command(args) -> None: provider = _normalize_provider(getattr(args, "provider", "")) - pool = load_pool(provider) - count = pool.reset_statuses() - print(f"Reset status on {count} {provider} credentials") + from hermes_constants import get_default_hermes_root, get_hermes_home + + current_profile_only = bool(getattr(args, "current_profile_only", False)) + include_all_profiles = bool(getattr(args, "all_profiles", False)) or ( + _same_path(get_hermes_home(), get_default_hermes_root()) + and not current_profile_only + ) + + results: list[tuple[str, int]] = [] + total = 0 + for name, home in _auth_reset_targets( + include_all_profiles=include_all_profiles, + current_profile_only=current_profile_only, + ): + count = auth_mod.reset_credential_pool_statuses( + provider, + auth_file=home / "auth.json", + ) + results.append((name, count)) + total += count + + if len(results) <= 1: + print(f"Reset status on {total} {provider} credentials") + return + + touched = ", ".join(f"{name}:{count}" for name, count in results if count) + suffix = f" ({touched})" if touched else "" + print( + f"Reset status on {total} {provider} credentials " + f"across {len(results)} profiles{suffix}" + ) def auth_status_command(args) -> None: diff --git a/hermes_cli/subcommands/auth.py b/hermes_cli/subcommands/auth.py index e81fcea8c100..413597073595 100644 --- a/hermes_cli/subcommands/auth.py +++ b/hermes_cli/subcommands/auth.py @@ -62,6 +62,17 @@ def build_auth_parser(subparsers, *, cmd_auth: Callable) -> None: "reset", help="Clear exhaustion status for all credentials for a provider" ) auth_reset.add_argument("provider", help="Provider id") + auth_reset_scope = auth_reset.add_mutually_exclusive_group() + auth_reset_scope.add_argument( + "--all-profiles", + action="store_true", + help="Clear matching credential status in the default and every named profile", + ) + auth_reset_scope.add_argument( + "--current-profile-only", + action="store_true", + help="Only clear the active profile's local auth store", + ) auth_status = auth_subparsers.add_parser( "status", help="Show auth status for a provider" ) diff --git a/tests/hermes_cli/test_auth_reset_profiles.py b/tests/hermes_cli/test_auth_reset_profiles.py new file mode 100644 index 000000000000..6ee2b1460e75 --- /dev/null +++ b/tests/hermes_cli/test_auth_reset_profiles.py @@ -0,0 +1,174 @@ +"""Behavior contracts for profile-scoped ``hermes auth reset``.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest + + +PROVIDER = "openai-codex" +STATUS_FIELDS = ( + "last_status", + "last_status_at", + "last_error_code", + "last_error_reason", + "last_error_message", + "last_error_reset_at", +) + + +def _exhausted_entry(entry_id: str) -> dict: + """Return status-only fixture data: no credential or token material.""" + return { + "id": entry_id, + "label": entry_id, + "priority": 0, + "marker": f"preserve-{entry_id}", + "last_status": "exhausted", + "last_status_at": 1_711_230_000.0, + "last_error_code": 429, + "last_error_reason": "usage_limit_reached", + "last_error_message": "The usage limit has been reached", + "last_error_reset_at": 1_711_233_600.0, + } + + +def _write_store(home: Path, entries: list[dict] | None) -> None: + home.mkdir(parents=True, exist_ok=True) + pool = {} if entries is None else {PROVIDER: entries} + (home / "auth.json").write_text( + json.dumps({"version": 1, "providers": {}, "credential_pool": pool}), + encoding="utf-8", + ) + + +def _read_store(home: Path) -> dict: + return json.loads((home / "auth.json").read_text(encoding="utf-8")) + + +def _assert_reset(home: Path) -> None: + entry = _read_store(home)["credential_pool"][PROVIDER][0] + assert all(entry[field] is None for field in STATUS_FIELDS) + assert entry["marker"] == f"preserve-{entry['id']}" + + +@pytest.fixture() +def profile_tree(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Path]: + """Build an isolated real root/profiles layout; never touch user auth.""" + fake_home = tmp_path / "home" + root = fake_home / ".hermes" + alpha = root / "profiles" / "alpha" + beta = root / "profiles" / "beta" + fake_home.mkdir() + monkeypatch.setenv("HOME", str(fake_home)) + monkeypatch.setattr(Path, "home", lambda: fake_home) + monkeypatch.setenv("HERMES_HOME", str(root)) + return {"root": root, "alpha": alpha, "beta": beta} + + +def test_default_reset_reaches_each_named_profile( + profile_tree: dict[str, Path], + capsys: pytest.CaptureFixture[str], +) -> None: + from hermes_cli.auth_commands import auth_reset_command + + for name, home in profile_tree.items(): + _write_store(home, [_exhausted_entry(name)]) + + auth_reset_command(SimpleNamespace(provider=PROVIDER)) + + assert "across 3 profiles" in capsys.readouterr().out + for home in profile_tree.values(): + _assert_reset(home) + + +def test_named_reset_updates_root_fallback_without_materializing_it( + profile_tree: dict[str, Path], + monkeypatch: pytest.MonkeyPatch, +) -> None: + from hermes_cli.auth_commands import auth_reset_command + + root = profile_tree["root"] + current = profile_tree["alpha"] + _write_store(root, [_exhausted_entry("root")]) + _write_store(current, None) + monkeypatch.setenv("HERMES_HOME", str(current)) + current_before = (current / "auth.json").read_bytes() + + auth_reset_command(SimpleNamespace(provider=PROVIDER)) + + _assert_reset(root) + assert _read_store(current)["credential_pool"] == {} + assert (current / "auth.json").read_bytes() == current_before + + +def test_current_profile_only_does_not_modify_fallback_root_store( + profile_tree: dict[str, Path], + monkeypatch: pytest.MonkeyPatch, +) -> None: + from hermes_cli.auth_commands import auth_reset_command + + root = profile_tree["root"] + current = profile_tree["alpha"] + _write_store(root, [_exhausted_entry("root")]) + _write_store(current, [_exhausted_entry("alpha")]) + monkeypatch.setenv("HERMES_HOME", str(current)) + root_before = (root / "auth.json").read_bytes() + + auth_reset_command( + SimpleNamespace( + provider=PROVIDER, + all_profiles=False, + current_profile_only=True, + ) + ) + + _assert_reset(current) + assert (root / "auth.json").read_bytes() == root_before + + +def test_all_profiles_from_named_profile_reaches_every_profile( + profile_tree: dict[str, Path], + monkeypatch: pytest.MonkeyPatch, +) -> None: + from hermes_cli.auth_commands import auth_reset_command + + for name, home in profile_tree.items(): + _write_store(home, [_exhausted_entry(name)]) + monkeypatch.setenv("HERMES_HOME", str(profile_tree["alpha"])) + + auth_reset_command( + SimpleNamespace( + provider=PROVIDER, + all_profiles=True, + current_profile_only=False, + ) + ) + + for home in profile_tree.values(): + _assert_reset(home) + + +def test_auth_reset_parser_rejects_conflicting_profile_scopes() -> None: + from hermes_cli.subcommands.auth import build_auth_parser + + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers(dest="command") + build_auth_parser(subparsers, cmd_auth=lambda _args: None) + + with pytest.raises(SystemExit) as exc_info: + parser.parse_args( + [ + "auth", + "reset", + PROVIDER, + "--all-profiles", + "--current-profile-only", + ] + ) + + assert exc_info.value.code == 2 \ No newline at end of file