fix: reset auth cooldowns across profiles

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]],

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:

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"
     )

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