From 581acb8bf9268acc4d1f5668d94f2ba663625913 Mon Sep 17 00:00:00 2001
From: Vladimir Dimitrov <vladimir.ddimitrov@experian.com>
Date: Sat, 23 May 2026 22:16:50 +0300
Subject: [PATCH 1/2] feat: add provider rotation cooldowns

Persist provider/model cooldowns across turns and sessions so exhausted
providers (rate-limit, billing/quota) are skipped at turn start until
the cooldown window expires.

Changes:
- agent/provider_rotation.py: ProviderRotationState (JSON persistence),
  is_unavailable(), mark_unavailable(), reset(), filter_available_entries(),
  is_rotation_enabled(), cooldown_for_reason()
- hermes_cli/rotation_cmd.py: 'hermes rotation list/reset/clear' commands
- agent/chat_completion_helpers.py: mark current provider on rate-limit/billing
  failure, skip cooled fallback entries when rotation enabled
- agent/agent_runtime_helpers.py: turn-start check skips cooled primary and
  advances to next available provider in fallback chain
- hermes_cli/config.py: provider_rotation config schema with defaults
  (enabled: false, cooldown 6h, billing cooldown 24h)
- hermes_cli/main.py: wire 'rotation' subparser and cmd_rotation handler
- website/docs/user-guide/features/fallback-providers.md: clarify fallback
  vs rotation, document provider_rotation config and CLI commands
- tests: 31 new tests across test_provider_rotation, test_rotation_cmd,
  and test_provider_fallback (all pass)
---
 agent/agent_runtime_helpers.py                |  21 +++
 agent/chat_completion_helpers.py              |  49 ++++++
 agent/provider_rotation.py                    | 160 ++++++++++++++++++
 hermes_cli/config.py                          |  13 +-
 hermes_cli/main.py                            |  34 +++-
 hermes_cli/rotation_cmd.py                    |  97 +++++++++++
 tests/agent/test_provider_rotation.py         |  75 ++++++++
 tests/hermes_cli/test_rotation_cmd.py         |  86 ++++++++++
 tests/run_agent/test_provider_fallback.py     | 105 ++++++++++++
 .../user-guide/features/fallback-providers.md |  39 ++++-
 10 files changed, 676 insertions(+), 3 deletions(-)
 create mode 100644 agent/provider_rotation.py
 create mode 100644 hermes_cli/rotation_cmd.py
 create mode 100644 tests/agent/test_provider_rotation.py
 create mode 100644 tests/hermes_cli/test_rotation_cmd.py

diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 4175f3e1898bb..48b7e58c01dbc 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -860,6 +860,27 @@ def restore_primary_runtime(agent) -> bool:
         # entirely, stranding the index and silently blocking all future
         # fallback attempts for the session.  Fixes #20465.
         agent._fallback_index = 0
+        try:
+            from hermes_cli.config import load_config
+            from agent.provider_rotation import ProviderRotationState, is_rotation_enabled
+
+            rotation_config = load_config()
+            primary_provider = ((agent._primary_runtime or {}).get("provider") or getattr(agent, "provider", "") or "").strip()
+            primary_model = ((agent._primary_runtime or {}).get("model") or getattr(agent, "model", "") or "").strip()
+            if (
+                is_rotation_enabled(rotation_config)
+                and primary_provider
+                and primary_model
+                and ProviderRotationState.load().is_unavailable(primary_provider, primary_model)
+            ):
+                logging.info(
+                    "Provider rotation: primary %s (%s) is cooling down; trying fallback",
+                    primary_model,
+                    primary_provider,
+                )
+                return bool(agent._try_activate_fallback())
+        except Exception:
+            logging.debug("Provider rotation turn-start check skipped", exc_info=True)
         return False
 
     if getattr(agent, "_rate_limited_until", 0) > time.monotonic():
diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py
index 59e7752a625ed..00ceb68a2be50 100644
--- a/agent/chat_completion_helpers.py
+++ b/agent/chat_completion_helpers.py
@@ -729,6 +729,8 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool
     auth resolution and client construction — no duplicated provider→key
     mappings.
     """
+    rotation_enabled = False
+    rotation_config = {}
     if reason in {FailoverReason.rate_limit, FailoverReason.billing}:
         # Only start cooldown when leaving the primary provider.  If we're
         # already on a fallback and chain-switching, the primary wasn't the
@@ -738,11 +740,58 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool
         primary_provider = ((agent._primary_runtime or {}).get("provider") or "").strip().lower()
         if (not fallback_already_active) or (primary_provider and current_provider == primary_provider):
             agent._rate_limited_until = time.monotonic() + 60
+    try:
+        from hermes_cli.config import load_config
+        from agent.provider_rotation import (
+            ProviderRotationState,
+            cooldown_for_reason,
+            is_rotation_enabled,
+        )
+
+        rotation_config = load_config()
+        rotation_enabled = is_rotation_enabled(rotation_config)
+        if rotation_enabled and reason in {FailoverReason.rate_limit, FailoverReason.billing}:
+            current_provider_for_state = (getattr(agent, "provider", "") or "").strip()
+            current_model_for_state = (getattr(agent, "model", "") or "").strip()
+            if current_provider_for_state and current_model_for_state:
+                ProviderRotationState.load().mark_unavailable(
+                    provider=current_provider_for_state,
+                    model=current_model_for_state,
+                    reason=getattr(reason, "value", str(reason)),
+                    cooldown_seconds=cooldown_for_reason(
+                        rotation_config,
+                        getattr(reason, "value", str(reason)),
+                    ),
+                )
+    except Exception:
+        logger.debug("Provider rotation state update skipped", exc_info=True)
+
     if agent._fallback_index >= len(agent._fallback_chain):
         return False
 
     fb = agent._fallback_chain[agent._fallback_index]
     agent._fallback_index += 1
+    if rotation_enabled:
+        try:
+            from agent.provider_rotation import ProviderRotationState
+
+            while (
+                isinstance(fb, dict)
+                and ProviderRotationState.load().is_unavailable(
+                    fb.get("provider") or "",
+                    fb.get("model") or "",
+                )
+                and agent._fallback_index < len(agent._fallback_chain)
+            ):
+                fb = agent._fallback_chain[agent._fallback_index]
+                agent._fallback_index += 1
+            if isinstance(fb, dict) and ProviderRotationState.load().is_unavailable(
+                fb.get("provider") or "",
+                fb.get("model") or "",
+            ):
+                return False
+        except Exception:
+            logger.debug("Provider rotation filtering skipped", exc_info=True)
     fb_provider = (fb.get("provider") or "").strip().lower()
     fb_model = (fb.get("model") or "").strip()
     if not fb_provider or not fb_model:
diff --git a/agent/provider_rotation.py b/agent/provider_rotation.py
new file mode 100644
index 0000000000000..cf10740ecb034
--- /dev/null
+++ b/agent/provider_rotation.py
@@ -0,0 +1,160 @@
+"""Persistent provider rotation cooldown state.
+
+This module intentionally keeps the first version small: providers that expose
+quota APIs can add proactive probes later, while every provider benefits from
+reactive cooldown after capacity errors.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Iterable
+
+from hermes_constants import get_hermes_home
+
+STATE_VERSION = 1
+STATE_FILE = "provider_rotation_state.json"
+DEFAULT_COOLDOWN_SECONDS = 6 * 60 * 60
+
+
+def _norm(value: str | None) -> str:
+    return (value or "").strip().lower()
+
+
+def provider_key(provider: str | None, model: str | None = None) -> str:
+    """Return stable key for provider/model rotation state."""
+    provider_part = _norm(provider)
+    model_part = (model or "").strip()
+    return f"{provider_part}:{model_part}" if model_part else provider_part
+
+
+def state_path() -> Path:
+    return get_hermes_home() / STATE_FILE
+
+
+@dataclass
+class ProviderRotationState:
+    """Durable cooldown records for provider rotation."""
+
+    unavailable: dict[str, dict[str, Any]] = field(default_factory=dict)
+    version: int = STATE_VERSION
+
+    @classmethod
+    def load(cls) -> "ProviderRotationState":
+        path = state_path()
+        try:
+            raw = json.loads(path.read_text(encoding="utf-8"))
+        except (FileNotFoundError, json.JSONDecodeError, OSError):
+            return cls()
+        unavailable = raw.get("unavailable", {})
+        if not isinstance(unavailable, dict):
+            unavailable = {}
+        return cls(unavailable=unavailable, version=int(raw.get("version", STATE_VERSION) or STATE_VERSION))
+
+    def save(self) -> None:
+        path = state_path()
+        path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = path.with_suffix(path.suffix + ".tmp")
+        payload = {"version": self.version, "unavailable": self.unavailable}
+        tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        tmp.replace(path)
+
+    def mark_unavailable(
+        self,
+        *,
+        provider: str,
+        model: str,
+        reason: str,
+        cooldown_seconds: int | float = DEFAULT_COOLDOWN_SECONDS,
+        now: float | None = None,
+        message: str | None = None,
+    ) -> None:
+        timestamp = time.time() if now is None else float(now)
+        cooldown = max(0.0, float(cooldown_seconds or 0))
+        key = provider_key(provider, model)
+        self.unavailable[key] = {
+            "provider": (provider or "").strip(),
+            "model": (model or "").strip(),
+            "reason": (reason or "unknown").strip() or "unknown",
+            "message": (message or "").strip(),
+            "unavailable_at": timestamp,
+            "retry_after": timestamp + cooldown,
+        }
+        self.save()
+
+    def is_unavailable(self, provider: str, model: str, *, now: float | None = None) -> bool:
+        timestamp = time.time() if now is None else float(now)
+        record = self.unavailable.get(provider_key(provider, model))
+        if not isinstance(record, dict):
+            return False
+        retry_after = float(record.get("retry_after") or 0)
+        if retry_after <= timestamp:
+            self.unavailable.pop(provider_key(provider, model), None)
+            self.save()
+            return False
+        return True
+
+    def reset(self, provider: str | None = None, model: str | None = None) -> int:
+        """Remove matching cooldown records. Returns count removed."""
+        if not provider:
+            count = len(self.unavailable)
+            self.unavailable.clear()
+            self.save()
+            return count
+        provider_norm = _norm(provider)
+        model_text = (model or "").strip()
+        removed = 0
+        for key, record in list(self.unavailable.items()):
+            rec_provider = _norm(record.get("provider") if isinstance(record, dict) else key.split(":", 1)[0])
+            rec_model = (record.get("model") if isinstance(record, dict) else "") or ""
+            if rec_provider != provider_norm:
+                continue
+            if model_text and rec_model != model_text:
+                continue
+            self.unavailable.pop(key, None)
+            removed += 1
+        if removed:
+            self.save()
+        return removed
+
+
+def filter_available_entries(entries: Iterable[dict[str, Any]], *, now: float | None = None) -> list[dict[str, Any]]:
+    """Return entries not currently cooled down, preserving original order."""
+    state = ProviderRotationState.load()
+    available: list[dict[str, Any]] = []
+    for entry in entries:
+        if not isinstance(entry, dict):
+            continue
+        provider = entry.get("provider") or ""
+        model = entry.get("model") or ""
+        if not provider or not model:
+            continue
+        if state.is_unavailable(provider, model, now=now):
+            continue
+        available.append(entry)
+    return available
+
+
+def is_rotation_enabled(config: dict[str, Any] | None) -> bool:
+    section = (config or {}).get("provider_rotation", {})
+    return isinstance(section, dict) and bool(section.get("enabled", False))
+
+
+def cooldown_for_reason(config: dict[str, Any] | None, reason: str | None = None) -> int:
+    section = (config or {}).get("provider_rotation", {}) if isinstance(config, dict) else {}
+    if not isinstance(section, dict):
+        return DEFAULT_COOLDOWN_SECONDS
+    by_reason = section.get("cooldown_seconds_by_reason")
+    reason_key = (reason or "").strip().lower()
+    if isinstance(by_reason, dict) and reason_key in by_reason:
+        try:
+            return int(by_reason[reason_key])
+        except (TypeError, ValueError):
+            pass
+    try:
+        return int(section.get("cooldown_seconds", DEFAULT_COOLDOWN_SECONDS))
+    except (TypeError, ValueError):
+        return DEFAULT_COOLDOWN_SECONDS
diff --git a/hermes_cli/config.py b/hermes_cli/config.py
index 715fd7eb76ff3..283a4b05d001e 100644
--- a/hermes_cli/config.py
+++ b/hermes_cli/config.py
@@ -504,6 +504,17 @@ def _ensure_hermes_home_managed(home: Path):
     "model": "",
     "providers": {},
     "fallback_providers": [],
+    "provider_rotation": {
+        "enabled": False,
+        # Persist cooldowns after provider capacity errors so new prompts skip
+        # exhausted providers until the window expires. Reactive by default:
+        # providers without quota APIs still participate after first failure.
+        "cooldown_seconds": 21600,
+        "cooldown_seconds_by_reason": {
+            "rate_limit": 21600,
+            "billing": 86400,
+        },
+    },
     "credential_pool_strategies": {},
     "toolsets": ["hermes-cli"],
     "agent": {
@@ -3298,7 +3309,7 @@ def check_config_version() -> Tuple[int, int]:
 # Fields that are valid at root level of config.yaml
 _KNOWN_ROOT_KEYS = {
     "_config_version", "model", "providers", "fallback_model",
-    "fallback_providers", "credential_pool_strategies", "toolsets",
+    "fallback_providers", "provider_rotation", "credential_pool_strategies", "toolsets",
     "agent", "terminal", "display", "compression", "delegation",
     "auxiliary", "custom_providers", "context", "memory", "gateway",
     "sessions",
diff --git a/hermes_cli/main.py b/hermes_cli/main.py
index 27d24f7eb630f..7fc5582241578 100644
--- a/hermes_cli/main.py
+++ b/hermes_cli/main.py
@@ -10655,7 +10655,7 @@ def _build_provider_choices() -> list[str]:
         "dump", "fallback", "gateway", "hooks", "import", "insights",
         "kanban", "login", "logout", "logs", "lsp", "mcp", "memory", "migrate",
         "model", "pairing", "plugins", "portal", "postinstall", "profile", "proxy",
-        "send", "sessions", "setup",
+        "rotation", "send", "sessions", "setup",
         "skills", "slack", "status", "tools", "uninstall", "update",
         "version", "webhook", "whatsapp", "chat", "secrets",
         # Help-ish invocations — plugin commands not being listed in
@@ -11043,6 +11043,38 @@ def main():
     )
     fallback_parser.set_defaults(func=cmd_fallback)
 
+    # =========================================================================
+    # rotation command — inspect/reset provider rotation cooldowns
+    # =========================================================================
+    from hermes_cli.rotation_cmd import cmd_rotation
+
+    rotation_parser = subparsers.add_parser(
+        "rotation",
+        help="Inspect or reset provider rotation cooldowns",
+        description=(
+            "Inspect and reset provider rotation cooldowns written when "
+            "provider_rotation.enabled is true and a provider hits rate-limit "
+            "or billing/quota exhaustion."
+        ),
+    )
+    rotation_subparsers = rotation_parser.add_subparsers(dest="rotation_command")
+    rotation_subparsers.add_parser(
+        "list",
+        aliases=["ls"],
+        help="Show active provider rotation cooldowns (default when no subcommand)",
+    )
+    rotation_reset = rotation_subparsers.add_parser(
+        "reset",
+        help="Reset cooldowns for a provider, optionally one model",
+    )
+    rotation_reset.add_argument("provider", help="Provider name to reset")
+    rotation_reset.add_argument("--model", help="Specific model to reset")
+    rotation_subparsers.add_parser(
+        "clear",
+        help="Clear all provider rotation cooldowns",
+    )
+    rotation_parser.set_defaults(func=cmd_rotation)
+
     # =========================================================================
     # secrets command — external secret managers (currently: Bitwarden)
     # =========================================================================
diff --git a/hermes_cli/rotation_cmd.py b/hermes_cli/rotation_cmd.py
new file mode 100644
index 0000000000000..f93413b8228f5
--- /dev/null
+++ b/hermes_cli/rotation_cmd.py
@@ -0,0 +1,97 @@
+"""hermes rotation — inspect/reset provider rotation cooldown state."""
+
+from __future__ import annotations
+
+import time
+from typing import Any
+
+
+def _fmt_seconds(seconds: float) -> str:
+    seconds = max(0, int(seconds))
+    hours, rem = divmod(seconds, 3600)
+    minutes, secs = divmod(rem, 60)
+    if hours:
+        return f"{hours}h {minutes}m"
+    if minutes:
+        return f"{minutes}m {secs}s"
+    return f"{secs}s"
+
+
+def _records(now: float | None = None) -> list[dict[str, Any]]:
+    from agent.provider_rotation import ProviderRotationState
+
+    timestamp = time.time() if now is None else float(now)
+    state = ProviderRotationState.load()
+    rows: list[dict[str, Any]] = []
+    for key, record in sorted(state.unavailable.items()):
+        if not isinstance(record, dict):
+            continue
+        retry_after = float(record.get("retry_after") or 0)
+        if retry_after <= timestamp:
+            continue
+        rows.append({"key": key, **record, "remaining": retry_after - timestamp})
+    return rows
+
+
+def cmd_rotation_list(args) -> None:
+    """Print provider/model cooldown records."""
+    now = getattr(args, "now", None)
+    rows = _records(now=now)
+    print()
+    if not rows:
+        print("  No provider rotation cooldowns active.")
+        print()
+        return
+    print(f"  Provider rotation cooldowns ({len(rows)} {'entry' if len(rows) == 1 else 'entries'}):")
+    for idx, row in enumerate(rows, 1):
+        provider = row.get("provider") or row.get("key") or "?"
+        model = row.get("model") or "?"
+        reason = row.get("reason") or "unknown"
+        remaining = _fmt_seconds(float(row.get("remaining") or 0))
+        print(f"    {idx}. {model} (via {provider}) — cooling down {remaining} [{reason}]")
+    print()
+    print("  Reset one with: hermes rotation reset PROVIDER [--model MODEL]")
+    print("  Clear all with:  hermes rotation clear")
+    print()
+
+
+def cmd_rotation_reset(args) -> None:
+    """Reset cooldowns for one provider, optionally one model."""
+    from agent.provider_rotation import ProviderRotationState
+
+    provider = getattr(args, "provider", None)
+    model = getattr(args, "model", None)
+    if not provider:
+        raise SystemExit("provider is required")
+    count = ProviderRotationState.load().reset(provider, model)
+    suffix = f" for {provider}"
+    if model:
+        suffix += f"/{model}"
+    print()
+    print(f"  Reset {count} provider rotation cooldown {'entry' if count == 1 else 'entries'}{suffix}.")
+    print()
+
+
+def cmd_rotation_clear(args) -> None:  # noqa: ARG001
+    """Clear all provider rotation cooldown state."""
+    from agent.provider_rotation import ProviderRotationState
+
+    count = ProviderRotationState.load().reset()
+    print()
+    print(f"  Cleared {count} provider rotation cooldown {'entry' if count == 1 else 'entries'}.")
+    print()
+
+
+def cmd_rotation(args) -> None:
+    """Top-level dispatcher for ``hermes rotation [subcommand]``."""
+    sub = getattr(args, "rotation_command", None)
+    if sub in {None, "", "list", "ls"}:
+        cmd_rotation_list(args)
+    elif sub == "reset":
+        cmd_rotation_reset(args)
+    elif sub == "clear":
+        cmd_rotation_clear(args)
+    else:
+        print(f"Unknown rotation subcommand: {sub}")
+        print("Use one of: list, reset, clear")
+        raise SystemExit(2)
diff --git a/tests/agent/test_provider_rotation.py b/tests/agent/test_provider_rotation.py
new file mode 100644
index 0000000000000..ad5a87c62d844
--- /dev/null
+++ b/tests/agent/test_provider_rotation.py
@@ -0,0 +1,75 @@
+"""Tests for priority provider rotation state and cooldown helpers."""
+
+from __future__ import annotations
+
+import time
+
+from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+
+class TestProviderRotationState:
+    def test_marks_provider_exhausted_and_skips_until_cooldown_expires(self, tmp_path):
+        """A capacity failure should persist cooldown state for matching provider/model."""
+        from agent.provider_rotation import ProviderRotationState
+
+        token = set_hermes_home_override(tmp_path)
+        try:
+            state = ProviderRotationState.load()
+            state.mark_unavailable(
+                provider="openai-codex",
+                model="gpt-5.3-codex",
+                reason="rate_limit",
+                cooldown_seconds=60,
+                now=1000.0,
+            )
+
+            reloaded = ProviderRotationState.load()
+            assert reloaded.is_unavailable("openai-codex", "gpt-5.3-codex", now=1010.0)
+            assert not reloaded.is_unavailable("openai-codex", "gpt-5.3-codex", now=1061.0)
+        finally:
+            reset_hermes_home_override(token)
+
+    def test_filters_unavailable_entries_but_keeps_available_order(self, tmp_path):
+        """Rotation should preserve user priority while removing cooled-down entries."""
+        from agent.provider_rotation import ProviderRotationState, filter_available_entries
+
+        token = set_hermes_home_override(tmp_path)
+        try:
+            state = ProviderRotationState.load()
+            state.mark_unavailable(
+                provider="openai-codex",
+                model="gpt-5.3-codex",
+                reason="billing",
+                cooldown_seconds=3600,
+                now=1000.0,
+            )
+            chain = [
+                {"provider": "openai-codex", "model": "gpt-5.3-codex"},
+                {"provider": "anthropic", "model": "claude-sonnet-4-6"},
+                {"provider": "google-gemini-cli", "model": "gemini-3-pro-preview"},
+            ]
+
+            assert filter_available_entries(chain, now=1200.0) == chain[1:]
+        finally:
+            reset_hermes_home_override(token)
+
+    def test_reset_removes_provider_state(self, tmp_path):
+        """Manual reset should make a provider immediately eligible again."""
+        from agent.provider_rotation import ProviderRotationState
+
+        token = set_hermes_home_override(tmp_path)
+        try:
+            state = ProviderRotationState.load()
+            state.mark_unavailable(
+                provider="anthropic",
+                model="claude-sonnet-4-6",
+                reason="rate_limit",
+                cooldown_seconds=3600,
+                now=time.time(),
+            )
+            assert state.reset("anthropic", "claude-sonnet-4-6") == 1
+            assert not ProviderRotationState.load().is_unavailable(
+                "anthropic", "claude-sonnet-4-6", now=time.time()
+            )
+        finally:
+            reset_hermes_home_override(token)
diff --git a/tests/hermes_cli/test_rotation_cmd.py b/tests/hermes_cli/test_rotation_cmd.py
new file mode 100644
index 0000000000000..7c82480aabfbc
--- /dev/null
+++ b/tests/hermes_cli/test_rotation_cmd.py
@@ -0,0 +1,86 @@
+"""Tests for `hermes rotation` cooldown visibility and reset commands."""
+
+from __future__ import annotations
+
+import types
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture()
+def isolated_home(tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    home = tmp_path / ".hermes"
+    home.mkdir(exist_ok=True)
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    return tmp_path
+
+
+class TestRotationCommand:
+    def test_list_shows_cooling_down_provider(self, isolated_home, capsys):
+        from agent.provider_rotation import ProviderRotationState
+        from hermes_cli.rotation_cmd import cmd_rotation_list
+
+        ProviderRotationState.load().mark_unavailable(
+            provider="openai-codex",
+            model="gpt-5.3-codex",
+            reason="rate_limit",
+            cooldown_seconds=3600,
+            now=1000.0,
+        )
+
+        cmd_rotation_list(types.SimpleNamespace(now=1200.0))
+
+        out = capsys.readouterr().out
+        assert "openai-codex" in out
+        assert "gpt-5.3-codex" in out
+        assert "rate_limit" in out
+        assert "cooling down" in out
+
+    def test_reset_provider_removes_matching_state(self, isolated_home, capsys):
+        from agent.provider_rotation import ProviderRotationState
+        from hermes_cli.rotation_cmd import cmd_rotation_reset
+
+        ProviderRotationState.load().mark_unavailable(
+            provider="anthropic",
+            model="claude-sonnet-4-6",
+            reason="billing",
+            cooldown_seconds=3600,
+            now=1000.0,
+        )
+
+        cmd_rotation_reset(types.SimpleNamespace(provider="anthropic", model=None))
+
+        out = capsys.readouterr().out
+        assert "Reset 1" in out
+        assert not ProviderRotationState.load().is_unavailable(
+            "anthropic", "claude-sonnet-4-6", now=1200.0
+        )
+
+    def test_clear_removes_all_state(self, isolated_home, capsys):
+        from agent.provider_rotation import ProviderRotationState
+        from hermes_cli.rotation_cmd import cmd_rotation_clear
+
+        state = ProviderRotationState.load()
+        state.mark_unavailable(
+            provider="openai-codex",
+            model="gpt-5.3-codex",
+            reason="rate_limit",
+            cooldown_seconds=3600,
+            now=1000.0,
+        )
+        state = ProviderRotationState.load()
+        state.mark_unavailable(
+            provider="anthropic",
+            model="claude-sonnet-4-6",
+            reason="billing",
+            cooldown_seconds=3600,
+            now=1000.0,
+        )
+
+        cmd_rotation_clear(types.SimpleNamespace())
+
+        out = capsys.readouterr().out
+        assert "Cleared 2" in out
+        assert ProviderRotationState.load().unavailable == {}
diff --git a/tests/run_agent/test_provider_fallback.py b/tests/run_agent/test_provider_fallback.py
index b179cc341cc50..ab738ed5873af 100644
--- a/tests/run_agent/test_provider_fallback.py
+++ b/tests/run_agent/test_provider_fallback.py
@@ -7,6 +7,7 @@
 
 from unittest.mock import MagicMock, patch
 
+from hermes_constants import reset_hermes_home_override, set_hermes_home_override
 from run_agent import AIAgent, _pool_may_recover_from_rate_limit
 
 
@@ -182,6 +183,110 @@ def test_resolves_key_env_for_fallback_provider(self):
             assert agent._try_activate_fallback() is True
             assert mock_rpc.call_args.kwargs["explicit_api_key"] == "env-secret"
 
+    def test_provider_rotation_skips_cooled_down_fallback_entry(self, tmp_path):
+        """Rotation mode should preserve priority while skipping exhausted entries."""
+        fbs = [
+            {"provider": "anthropic", "model": "claude-sonnet-4-6"},
+            {"provider": "google-gemini-cli", "model": "gemini-3-pro-preview"},
+        ]
+        token = set_hermes_home_override(tmp_path)
+        try:
+            from agent.provider_rotation import ProviderRotationState
+
+            ProviderRotationState.load().mark_unavailable(
+                provider="anthropic",
+                model="claude-sonnet-4-6",
+                reason="rate_limit",
+                cooldown_seconds=3600,
+                now=1000.0,
+            )
+            agent = _make_agent(fallback_model=fbs)
+            called = []
+
+            def _resolve(provider, model=None, raw_codex=False, **kwargs):
+                called.append((provider, model))
+                return _mock_client(), model
+
+            with (
+                patch("hermes_cli.config.load_config", return_value={"provider_rotation": {"enabled": True}}),
+                patch("time.time", return_value=1200.0),
+                patch("agent.auxiliary_client.resolve_provider_client", side_effect=_resolve),
+                patch("hermes_cli.model_normalize.normalize_model_for_provider", side_effect=lambda m, p: m),
+            ):
+                assert agent._try_activate_fallback() is True
+
+            assert called == [("google-gemini-cli", "gemini-3-pro-preview")]
+            assert agent.model == "gemini-3-pro-preview"
+        finally:
+            reset_hermes_home_override(token)
+
+    def test_provider_rotation_marks_failed_provider_unavailable(self, tmp_path):
+        """Capacity failover should persist cooldown state for next turns/sessions."""
+        from agent.error_classifier import FailoverReason
+        from agent.provider_rotation import ProviderRotationState
+
+        token = set_hermes_home_override(tmp_path)
+        try:
+            fbs = [{"provider": "anthropic", "model": "claude-sonnet-4-6"}]
+            agent = _make_agent(fallback_model=fbs)
+            agent.provider = "openai-codex"
+            agent.model = "gpt-5.3-codex"
+
+            with (
+                patch(
+                    "hermes_cli.config.load_config",
+                    return_value={
+                        "provider_rotation": {
+                            "enabled": True,
+                            "cooldown_seconds_by_reason": {"rate_limit": 7200},
+                        }
+                    },
+                ),
+                patch("time.time", return_value=2000.0),
+                patch("time.monotonic", return_value=2000.0),
+                patch("agent.auxiliary_client.resolve_provider_client", return_value=(_mock_client(), "claude-sonnet-4-6")),
+            ):
+                assert agent._try_activate_fallback(FailoverReason.rate_limit) is True
+
+            state = ProviderRotationState.load()
+            assert state.is_unavailable("openai-codex", "gpt-5.3-codex", now=9000.0)
+            assert not state.is_unavailable("openai-codex", "gpt-5.3-codex", now=9201.0)
+        finally:
+            reset_hermes_home_override(token)
+
+    def test_provider_rotation_uses_fallback_at_turn_start_when_primary_is_cooling_down(self, tmp_path):
+        """A new prompt should skip primary while persisted cooldown is active."""
+        from agent.provider_rotation import ProviderRotationState
+
+        token = set_hermes_home_override(tmp_path)
+        try:
+            fbs = [{"provider": "anthropic", "model": "claude-sonnet-4-6"}]
+            agent = _make_agent(fallback_model=fbs)
+            agent.provider = "openai-codex"
+            agent.model = "gpt-5.3-codex"
+            agent._primary_runtime["provider"] = "openai-codex"
+            agent._primary_runtime["model"] = "gpt-5.3-codex"
+            ProviderRotationState.load().mark_unavailable(
+                provider="openai-codex",
+                model="gpt-5.3-codex",
+                reason="rate_limit",
+                cooldown_seconds=3600,
+                now=1000.0,
+            )
+
+            with (
+                patch("hermes_cli.config.load_config", return_value={"provider_rotation": {"enabled": True}}),
+                patch("time.time", return_value=1200.0),
+                patch("agent.auxiliary_client.resolve_provider_client", return_value=(_mock_client(), "claude-sonnet-4-6")),
+            ):
+                assert agent._restore_primary_runtime() is True
+
+            assert agent.provider == "anthropic"
+            assert agent.model == "claude-sonnet-4-6"
+            assert agent._fallback_activated is True
+        finally:
+            reset_hermes_home_override(token)
+
 
 # ── Pool-rotation vs fallback gating (#11314) ────────────────────────────
 
diff --git a/website/docs/user-guide/features/fallback-providers.md b/website/docs/user-guide/features/fallback-providers.md
index 977416b171e4a..a67c8c7140933 100644
--- a/website/docs/user-guide/features/fallback-providers.md
+++ b/website/docs/user-guide/features/fallback-providers.md
@@ -118,9 +118,46 @@ When triggered, Hermes:
 The switch is seamless — your conversation history, tool calls, and context are preserved. The agent continues from exactly where it left off, just using a different model.
 
 :::info Per-Turn, Not Per-Session
-Fallback is **turn-scoped**: each new user message starts with the primary model restored. If the primary fails mid-turn, fallback activates for that turn only. On the next message, Hermes tries the primary again. Within a single turn, fallback activates at most once — if the fallback also fails, normal error handling takes over (retries, then error message). This prevents cascading failover loops within a turn while giving the primary model a fresh chance every turn.
+Fallback is **turn-scoped** by default: each new user message starts with the primary model restored. If the primary fails mid-turn, fallback activates for that turn only. On the next message, Hermes tries the primary again. Within a single turn, fallback walks the ordered fallback chain; if every fallback also fails, normal error handling returns the final error.
 :::
 
+### Optional: Provider Rotation Cooldowns
+
+If you want Hermes to use your subscriptions like a priority queue — for example ChatGPT/Codex first, then Claude, then Gemini, then Grok — enable provider rotation on top of `fallback_providers`:
+
+```yaml
+model:
+  provider: openai-codex
+  default: gpt-5.3-codex
+
+fallback_providers:
+  - provider: anthropic
+    model: claude-sonnet-4-6
+  - provider: google-gemini-cli
+    model: gemini-3-pro-preview
+  - provider: xai-oauth
+    model: grok-4
+
+provider_rotation:
+  enabled: true
+  cooldown_seconds: 21600              # default: 6 hours
+  cooldown_seconds_by_reason:
+    rate_limit: 21600                  # subscription/rate-limit capacity errors
+    billing: 86400                     # billing/quota exhaustion
+```
+
+When rotation is enabled, Hermes persists provider/model cooldowns in `~/.hermes/provider_rotation_state.json`. On a capacity failure (rate-limit or billing/quota exhaustion), Hermes marks the current provider unavailable for the configured cooldown and moves to the next fallback. Future prompts skip cooled-down providers until their cooldown expires, even across new Hermes processes.
+
+Inspect or override cooldown state with:
+
+```bash
+hermes rotation list
+hermes rotation reset openai-codex --model gpt-5.3-codex
+hermes rotation clear
+```
+
+This first implementation is intentionally **reactive**: it learns from provider errors. Some subscription-backed providers do not expose reliable public usage APIs, so Hermes cannot always preflight exact remaining quota before a request. Providers that do expose quota endpoints can add proactive probes later without changing the user-facing rotation config.
+
 ### Examples
 
 **OpenRouter as fallback for Anthropic native:**

From a35bf750de195590717d1379182d6621ac070731 Mon Sep 17 00:00:00 2001
From: Vladimir Dimitrov <vladimir.ddimitrov@experian.com>
Date: Tue, 25 Aug 2026 19:21:44 +0300
Subject: [PATCH 2/2] fix: address PR review feedback

---
 agent/agent_runtime_helpers.py            |  39 +++-
 agent/chat_completion_helpers.py          |  43 +++--
 agent/conversation_loop.py                |  11 +-
 agent/provider_rotation.py                | 213 ++++++++++++++++++----
 run_agent.py                              |  15 +-
 tests/agent/test_provider_rotation.py     |  60 ++++++
 tests/run_agent/test_provider_fallback.py | 200 +++++++++++++++++---
 7 files changed, 495 insertions(+), 86 deletions(-)

diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 48b7e58c01dbc..132550b4aa7c0 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -867,11 +867,20 @@ def restore_primary_runtime(agent) -> bool:
             rotation_config = load_config()
             primary_provider = ((agent._primary_runtime or {}).get("provider") or getattr(agent, "provider", "") or "").strip()
             primary_model = ((agent._primary_runtime or {}).get("model") or getattr(agent, "model", "") or "").strip()
+            primary_base_url = str(
+                (agent._primary_runtime or {}).get("base_url")
+                or getattr(agent, "base_url", "")
+                or ""
+            ).strip()
             if (
                 is_rotation_enabled(rotation_config)
                 and primary_provider
                 and primary_model
-                and ProviderRotationState.load().is_unavailable(primary_provider, primary_model)
+                and ProviderRotationState.load().is_unavailable(
+                    primary_provider,
+                    primary_model,
+                    base_url=primary_base_url,
+                )
             ):
                 logging.info(
                     "Provider rotation: primary %s (%s) is cooling down; trying fallback",
@@ -886,6 +895,34 @@ def restore_primary_runtime(agent) -> bool:
     if getattr(agent, "_rate_limited_until", 0) > time.monotonic():
         return False  # primary still in rate-limit cooldown, stay on fallback
 
+    try:
+        from hermes_cli.config import load_config
+        from agent.provider_rotation import ProviderRotationState, is_rotation_enabled
+
+        rotation_config = load_config()
+        rt = agent._primary_runtime
+        primary_provider = (rt.get("provider") or "").strip()
+        primary_model = (rt.get("model") or "").strip()
+        primary_base_url = str(rt.get("base_url") or "").strip()
+        if (
+            is_rotation_enabled(rotation_config)
+            and primary_provider
+            and primary_model
+            and ProviderRotationState.load().is_unavailable(
+                primary_provider,
+                primary_model,
+                base_url=primary_base_url,
+            )
+        ):
+            logging.info(
+                "Provider rotation: primary %s (%s) still cooling down after transient gate; staying on fallback",
+                primary_model,
+                primary_provider,
+            )
+            return False
+    except Exception:
+        logging.debug("Provider rotation restore-path check skipped", exc_info=True)
+
     rt = agent._primary_runtime
     try:
         # ── Core runtime state ──
diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py
index 00ceb68a2be50..19036a0d5767f 100644
--- a/agent/chat_completion_helpers.py
+++ b/agent/chat_completion_helpers.py
@@ -717,7 +717,13 @@ def build_assistant_message(agent, assistant_message, finish_reason: str) -> dic
 
 
 
-def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool:
+def try_activate_fallback(
+    agent,
+    reason: "FailoverReason | None" = None,
+    *,
+    rate_limit_headers: Any = None,
+    error_context: dict[str, Any] | None = None,
+) -> bool:
     """Switch to the next fallback model/provider in the chain.
 
     Called when the current model is failing after retries.  Swaps the
@@ -745,24 +751,33 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool
         from agent.provider_rotation import (
             ProviderRotationState,
             cooldown_for_reason,
+            has_durable_rate_limit_evidence,
             is_rotation_enabled,
         )
 
         rotation_config = load_config()
         rotation_enabled = is_rotation_enabled(rotation_config)
         if rotation_enabled and reason in {FailoverReason.rate_limit, FailoverReason.billing}:
-            current_provider_for_state = (getattr(agent, "provider", "") or "").strip()
-            current_model_for_state = (getattr(agent, "model", "") or "").strip()
-            if current_provider_for_state and current_model_for_state:
-                ProviderRotationState.load().mark_unavailable(
-                    provider=current_provider_for_state,
-                    model=current_model_for_state,
-                    reason=getattr(reason, "value", str(reason)),
-                    cooldown_seconds=cooldown_for_reason(
-                        rotation_config,
-                        getattr(reason, "value", str(reason)),
-                    ),
-                )
+            should_persist_cooldown = reason == FailoverReason.billing or has_durable_rate_limit_evidence(
+                headers=rate_limit_headers,
+                last_known_state=getattr(agent, "_rate_limit_state", None),
+                error_context=error_context,
+            )
+            if should_persist_cooldown:
+                current_provider_for_state = (getattr(agent, "provider", "") or "").strip()
+                current_model_for_state = (getattr(agent, "model", "") or "").strip()
+                current_base_url_for_state = str(getattr(agent, "base_url", "") or "").strip()
+                if current_provider_for_state and current_model_for_state:
+                    ProviderRotationState.load().mark_unavailable(
+                        provider=current_provider_for_state,
+                        model=current_model_for_state,
+                        base_url=current_base_url_for_state,
+                        reason=getattr(reason, "value", str(reason)),
+                        cooldown_seconds=cooldown_for_reason(
+                            rotation_config,
+                            getattr(reason, "value", str(reason)),
+                        ),
+                    )
     except Exception:
         logger.debug("Provider rotation state update skipped", exc_info=True)
 
@@ -780,6 +795,7 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool
                 and ProviderRotationState.load().is_unavailable(
                     fb.get("provider") or "",
                     fb.get("model") or "",
+                    base_url=fb.get("base_url") or "",
                 )
                 and agent._fallback_index < len(agent._fallback_chain)
             ):
@@ -788,6 +804,7 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool
             if isinstance(fb, dict) and ProviderRotationState.load().is_unavailable(
                 fb.get("provider") or "",
                 fb.get("model") or "",
+                base_url=fb.get("base_url") or "",
             ):
                 return False
         except Exception:
diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py
index fdf65c0755877..36ceb3ad59856 100644
--- a/agent/conversation_loop.py
+++ b/agent/conversation_loop.py
@@ -2430,7 +2430,16 @@ def _stop_spinner():
                     )
                     if not pool_may_recover:
                         agent._emit_status("⚠️ Rate limited — switching to fallback provider...")
-                        if agent._try_activate_fallback(reason=classified.reason):
+                        _err_resp = getattr(api_error, "response", None)
+                        _err_hdrs = (
+                            getattr(_err_resp, "headers", None)
+                            if _err_resp else None
+                        )
+                        if agent._try_activate_fallback(
+                            reason=classified.reason,
+                            rate_limit_headers=_err_hdrs,
+                            error_context=error_context,
+                        ):
                             retry_count = 0
                             compression_attempts = 0
                             primary_recovery_attempted = False
diff --git a/agent/provider_rotation.py b/agent/provider_rotation.py
index cf10740ecb034..89b23f578d3de 100644
--- a/agent/provider_rotation.py
+++ b/agent/provider_rotation.py
@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import copy
 import json
 import time
 from dataclasses import dataclass, field
@@ -14,27 +15,123 @@
 from typing import Any, Iterable
 
 from hermes_constants import get_hermes_home
+from utils import atomic_json_write
+
+try:
+    import fcntl
+except ImportError:  # pragma: no cover - Windows fallback
+    fcntl = None
 
 STATE_VERSION = 1
 STATE_FILE = "provider_rotation_state.json"
 DEFAULT_COOLDOWN_SECONDS = 6 * 60 * 60
+_MIN_DURABLE_RESET_SECONDS = 60.0
 
 
 def _norm(value: str | None) -> str:
     return (value or "").strip().lower()
 
 
-def provider_key(provider: str | None, model: str | None = None) -> str:
-    """Return stable key for provider/model rotation state."""
+def _normalize_base_url(base_url: str | None) -> str:
+    return (base_url or "").strip().rstrip("/").lower()
+
+
+def provider_key(
+    provider: str | None,
+    model: str | None = None,
+    base_url: str | None = None,
+) -> str:
+    """Return stable key for provider/model/base_url rotation state."""
     provider_part = _norm(provider)
     model_part = (model or "").strip()
-    return f"{provider_part}:{model_part}" if model_part else provider_part
+    base_part = _normalize_base_url(base_url)
+    key = f"{provider_part}:{model_part}" if model_part else provider_part
+    return f"{key}@{base_part}" if base_part else key
 
 
 def state_path() -> Path:
     return get_hermes_home() / STATE_FILE
 
 
+def _state_lock_path() -> Path:
+    path = state_path()
+    return path.with_name(path.name + ".lock")
+
+
+def _atomic_save_state(*, version: int, unavailable: dict[str, dict[str, Any]]) -> None:
+    path = state_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    atomic_json_write(
+        path,
+        {"version": version, "unavailable": unavailable},
+        sort_keys=True,
+    )
+
+
+def _bucket_has_durable_exhaustion(bucket: Any) -> bool:
+    if bucket is None:
+        return False
+    remaining = getattr(bucket, "remaining", None)
+    reset_seconds = getattr(bucket, "remaining_seconds_now", None)
+    if reset_seconds is None:
+        reset_seconds = getattr(bucket, "reset_seconds", None)
+    if remaining is None or reset_seconds is None:
+        return False
+    try:
+        remaining_val = int(remaining)
+        reset_val = float(reset_seconds)
+    except (TypeError, ValueError):
+        return False
+    return remaining_val <= 0 and reset_val >= _MIN_DURABLE_RESET_SECONDS
+
+
+def has_durable_rate_limit_evidence(
+    *,
+    headers: Any = None,
+    last_known_state: Any = None,
+    error_context: dict[str, Any] | None = None,
+) -> bool:
+    """Return True when a 429 looks like durable quota exhaustion.
+
+    Durable cooldowns should only be written when the provider gives evidence
+    that the caller's own bucket is actually exhausted. Short-lived upstream
+    capacity 429s must stay session-local.
+    """
+    try:
+        from agent.rate_limit_tracker import parse_rate_limit_headers
+
+        if headers:
+            parsed = parse_rate_limit_headers(headers, provider="")
+            if parsed is not None:
+                for bucket in (
+                    parsed.requests_min,
+                    parsed.requests_hour,
+                    parsed.tokens_min,
+                    parsed.tokens_hour,
+                ):
+                    if _bucket_has_durable_exhaustion(bucket):
+                        return True
+    except Exception:
+        pass
+
+    if last_known_state is not None:
+        for name in ("requests_min", "requests_hour", "tokens_min", "tokens_hour"):
+            if _bucket_has_durable_exhaustion(getattr(last_known_state, name, None)):
+                return True
+
+    if isinstance(error_context, dict):
+        reset_at = error_context.get("reset_at")
+        if reset_at not in {None, ""}:
+            try:
+                reset_at_val = float(reset_at if reset_at is not None else 0.0)
+                if reset_at_val > time.time() + _MIN_DURABLE_RESET_SECONDS:
+                    return True
+            except (TypeError, ValueError):
+                pass
+
+    return False
+
+
 @dataclass
 class ProviderRotationState:
     """Durable cooldown records for provider rotation."""
@@ -52,21 +149,40 @@ def load(cls) -> "ProviderRotationState":
         unavailable = raw.get("unavailable", {})
         if not isinstance(unavailable, dict):
             unavailable = {}
-        return cls(unavailable=unavailable, version=int(raw.get("version", STATE_VERSION) or STATE_VERSION))
+        return cls(
+            unavailable=unavailable,
+            version=int(raw.get("version", STATE_VERSION) or STATE_VERSION),
+        )
 
     def save(self) -> None:
-        path = state_path()
-        path.parent.mkdir(parents=True, exist_ok=True)
-        tmp = path.with_suffix(path.suffix + ".tmp")
-        payload = {"version": self.version, "unavailable": self.unavailable}
-        tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
-        tmp.replace(path)
+        _atomic_save_state(version=self.version, unavailable=self.unavailable)
+
+    def _locked_update(self, updater) -> Any:
+        lock_path = _state_lock_path()
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+        with lock_path.open("a+", encoding="utf-8") as lock_handle:
+            if fcntl is not None:
+                fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX)
+            try:
+                current = type(self).load()
+                result = updater(current)
+                _atomic_save_state(
+                    version=current.version,
+                    unavailable=current.unavailable,
+                )
+                self.unavailable = copy.deepcopy(current.unavailable)
+                self.version = current.version
+                return result
+            finally:
+                if fcntl is not None:
+                    fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN)
 
     def mark_unavailable(
         self,
         *,
         provider: str,
         model: str,
+        base_url: str | None = None,
         reason: str,
         cooldown_seconds: int | float = DEFAULT_COOLDOWN_SECONDS,
         now: float | None = None,
@@ -74,26 +190,38 @@ def mark_unavailable(
     ) -> None:
         timestamp = time.time() if now is None else float(now)
         cooldown = max(0.0, float(cooldown_seconds or 0))
-        key = provider_key(provider, model)
-        self.unavailable[key] = {
-            "provider": (provider or "").strip(),
-            "model": (model or "").strip(),
-            "reason": (reason or "unknown").strip() or "unknown",
-            "message": (message or "").strip(),
-            "unavailable_at": timestamp,
-            "retry_after": timestamp + cooldown,
-        }
-        self.save()
-
-    def is_unavailable(self, provider: str, model: str, *, now: float | None = None) -> bool:
+        normalized_base_url = _normalize_base_url(base_url)
+
+        def _update(current: ProviderRotationState) -> None:
+            key = provider_key(provider, model, normalized_base_url)
+            current.unavailable[key] = {
+                "provider": (provider or "").strip(),
+                "model": (model or "").strip(),
+                "base_url": normalized_base_url,
+                "reason": (reason or "unknown").strip() or "unknown",
+                "message": (message or "").strip(),
+                "unavailable_at": timestamp,
+                "retry_after": timestamp + cooldown,
+            }
+
+        self._locked_update(_update)
+
+    def is_unavailable(
+        self,
+        provider: str,
+        model: str,
+        *,
+        base_url: str | None = None,
+        now: float | None = None,
+    ) -> bool:
         timestamp = time.time() if now is None else float(now)
-        record = self.unavailable.get(provider_key(provider, model))
+        key = provider_key(provider, model, base_url)
+        record = self.unavailable.get(key)
         if not isinstance(record, dict):
             return False
         retry_after = float(record.get("retry_after") or 0)
         if retry_after <= timestamp:
-            self.unavailable.pop(provider_key(provider, model), None)
-            self.save()
+            self._locked_update(lambda current: current.unavailable.pop(key, None))
             return False
         return True
 
@@ -104,21 +232,27 @@ def reset(self, provider: str | None = None, model: str | None = None) -> int:
             self.unavailable.clear()
             self.save()
             return count
+
         provider_norm = _norm(provider)
         model_text = (model or "").strip()
-        removed = 0
-        for key, record in list(self.unavailable.items()):
-            rec_provider = _norm(record.get("provider") if isinstance(record, dict) else key.split(":", 1)[0])
-            rec_model = (record.get("model") if isinstance(record, dict) else "") or ""
-            if rec_provider != provider_norm:
-                continue
-            if model_text and rec_model != model_text:
-                continue
-            self.unavailable.pop(key, None)
-            removed += 1
-        if removed:
-            self.save()
-        return removed
+
+        def _update(current: ProviderRotationState) -> int:
+            removed_local = 0
+            for key, record in list(current.unavailable.items()):
+                rec_provider = _norm(
+                    record.get("provider") if isinstance(record, dict) else key.split(":", 1)[0]
+                )
+                rec_model = (record.get("model") if isinstance(record, dict) else "") or ""
+                if rec_provider != provider_norm:
+                    continue
+                if model_text and rec_model != model_text:
+                    continue
+                current.unavailable.pop(key, None)
+                removed_local += 1
+            return removed_local
+
+        removed = self._locked_update(_update)
+        return int(removed or 0)
 
 
 def filter_available_entries(entries: Iterable[dict[str, Any]], *, now: float | None = None) -> list[dict[str, Any]]:
@@ -130,9 +264,10 @@ def filter_available_entries(entries: Iterable[dict[str, Any]], *, now: float |
             continue
         provider = entry.get("provider") or ""
         model = entry.get("model") or ""
+        base_url = entry.get("base_url") or ""
         if not provider or not model:
             continue
-        if state.is_unavailable(provider, model, now=now):
+        if state.is_unavailable(provider, model, base_url=base_url, now=now):
             continue
         available.append(entry)
     return available
diff --git a/run_agent.py b/run_agent.py
index b364127c27805..939ad1538044d 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -3148,10 +3148,21 @@ def _interruptible_streaming_api_call(
         from agent.chat_completion_helpers import interruptible_streaming_api_call
         return interruptible_streaming_api_call(self, api_kwargs, on_first_delta=on_first_delta)
 
-    def _try_activate_fallback(self, reason: "FailoverReason | None" = None) -> bool:
+    def _try_activate_fallback(
+        self,
+        reason: "FailoverReason | None" = None,
+        *,
+        rate_limit_headers: Any = None,
+        error_context: dict[str, Any] | None = None,
+    ) -> bool:
         """Forwarder — see ``agent.chat_completion_helpers.try_activate_fallback``."""
         from agent.chat_completion_helpers import try_activate_fallback
-        return try_activate_fallback(self, reason)
+        return try_activate_fallback(
+            self,
+            reason,
+            rate_limit_headers=rate_limit_headers,
+            error_context=error_context,
+        )
 
     # ── Per-turn primary restoration ─────────────────────────────────────
 
diff --git a/tests/agent/test_provider_rotation.py b/tests/agent/test_provider_rotation.py
index ad5a87c62d844..376b49f3eb253 100644
--- a/tests/agent/test_provider_rotation.py
+++ b/tests/agent/test_provider_rotation.py
@@ -73,3 +73,63 @@ def test_reset_removes_provider_state(self, tmp_path):
             )
         finally:
             reset_hermes_home_override(token)
+
+    def test_same_provider_model_different_base_urls_do_not_share_cooldown(self, tmp_path):
+        """Endpoint identity is part of rotation state for custom providers."""
+        from agent.provider_rotation import ProviderRotationState
+
+        token = set_hermes_home_override(tmp_path)
+        try:
+            state = ProviderRotationState.load()
+            state.mark_unavailable(
+                provider="custom",
+                model="claude-opus-4-7",
+                base_url="https://proxy-one.example/v1",
+                reason="rate_limit",
+                cooldown_seconds=3600,
+                now=1000.0,
+            )
+            reloaded = ProviderRotationState.load()
+            assert reloaded.is_unavailable(
+                "custom",
+                "claude-opus-4-7",
+                base_url="https://proxy-one.example/v1",
+                now=1200.0,
+            )
+            assert not reloaded.is_unavailable(
+                "custom",
+                "claude-opus-4-7",
+                base_url="https://proxy-two.example/v1",
+                now=1200.0,
+            )
+        finally:
+            reset_hermes_home_override(token)
+
+    def test_stale_instance_merge_keeps_concurrent_records(self, tmp_path):
+        """Two sessions loaded before either write must not clobber each other."""
+        from agent.provider_rotation import ProviderRotationState
+
+        token = set_hermes_home_override(tmp_path)
+        try:
+            state_a = ProviderRotationState.load()
+            state_b = ProviderRotationState.load()
+            state_a.mark_unavailable(
+                provider="openai-codex",
+                model="gpt-5.3-codex",
+                reason="rate_limit",
+                cooldown_seconds=3600,
+                now=1000.0,
+            )
+            state_b.mark_unavailable(
+                provider="anthropic",
+                model="claude-sonnet-4-6",
+                reason="billing",
+                cooldown_seconds=3600,
+                now=1001.0,
+            )
+
+            reloaded = ProviderRotationState.load()
+            assert reloaded.is_unavailable("openai-codex", "gpt-5.3-codex", now=1200.0)
+            assert reloaded.is_unavailable("anthropic", "claude-sonnet-4-6", now=1200.0)
+        finally:
+            reset_hermes_home_override(token)
diff --git a/tests/run_agent/test_provider_fallback.py b/tests/run_agent/test_provider_fallback.py
index ab738ed5873af..6dae25bb9ba5b 100644
--- a/tests/run_agent/test_provider_fallback.py
+++ b/tests/run_agent/test_provider_fallback.py
@@ -97,8 +97,10 @@ def test_advances_index(self):
             {"provider": "zai", "model": "glm-4.7"},
         ]
         agent = _make_agent(fallback_model=fbs)
-        with patch("agent.auxiliary_client.resolve_provider_client",
-                    return_value=(_mock_client(), "gpt-4o")):
+        with patch(
+            "agent.auxiliary_client.resolve_provider_client",
+            return_value=(_mock_client(), "gpt-4o"),
+        ):
             assert agent._try_activate_fallback() is True
             assert agent._fallback_index == 1
             assert agent.model == "gpt-4o"
@@ -110,8 +112,10 @@ def test_second_fallback_works(self):
             {"provider": "zai", "model": "glm-4.7"},
         ]
         agent = _make_agent(fallback_model=fbs)
-        with patch("agent.auxiliary_client.resolve_provider_client",
-                    return_value=(_mock_client(), "resolved")):
+        with patch(
+            "agent.auxiliary_client.resolve_provider_client",
+            return_value=(_mock_client(), "resolved"),
+        ):
             assert agent._try_activate_fallback() is True
             assert agent.model == "gpt-4o"
             assert agent._try_activate_fallback() is True
@@ -121,8 +125,10 @@ def test_second_fallback_works(self):
     def test_all_exhausted_returns_false(self):
         fbs = [{"provider": "openai", "model": "gpt-4o"}]
         agent = _make_agent(fallback_model=fbs)
-        with patch("agent.auxiliary_client.resolve_provider_client",
-                    return_value=(_mock_client(), "gpt-4o")):
+        with patch(
+            "agent.auxiliary_client.resolve_provider_client",
+            return_value=(_mock_client(), "gpt-4o"),
+        ):
             assert agent._try_activate_fallback() is True
             assert agent._try_activate_fallback() is False
 
@@ -135,8 +141,8 @@ def test_skips_unconfigured_provider_to_next(self):
         agent = _make_agent(fallback_model=fbs)
         with patch("agent.auxiliary_client.resolve_provider_client") as mock_rpc:
             mock_rpc.side_effect = [
-                (None, None),                    # broken provider
-                (_mock_client(), "gpt-4o"),       # fallback succeeds
+                (None, None),
+                (_mock_client(), "gpt-4o"),
             ]
             assert agent._try_activate_fallback() is True
             assert agent.model == "gpt-4o"
@@ -208,10 +214,19 @@ def _resolve(provider, model=None, raw_codex=False, **kwargs):
                 return _mock_client(), model
 
             with (
-                patch("hermes_cli.config.load_config", return_value={"provider_rotation": {"enabled": True}}),
+                patch(
+                    "hermes_cli.config.load_config",
+                    return_value={"provider_rotation": {"enabled": True}},
+                ),
                 patch("time.time", return_value=1200.0),
-                patch("agent.auxiliary_client.resolve_provider_client", side_effect=_resolve),
-                patch("hermes_cli.model_normalize.normalize_model_for_provider", side_effect=lambda m, p: m),
+                patch(
+                    "agent.auxiliary_client.resolve_provider_client",
+                    side_effect=_resolve,
+                ),
+                patch(
+                    "hermes_cli.model_normalize.normalize_model_for_provider",
+                    side_effect=lambda m, p: m,
+                ),
             ):
                 assert agent._try_activate_fallback() is True
 
@@ -221,7 +236,7 @@ def _resolve(provider, model=None, raw_codex=False, **kwargs):
             reset_hermes_home_override(token)
 
     def test_provider_rotation_marks_failed_provider_unavailable(self, tmp_path):
-        """Capacity failover should persist cooldown state for next turns/sessions."""
+        """Durable quota evidence should persist cooldown state across sessions."""
         from agent.error_classifier import FailoverReason
         from agent.provider_rotation import ProviderRotationState
 
@@ -231,6 +246,11 @@ def test_provider_rotation_marks_failed_provider_unavailable(self, tmp_path):
             agent = _make_agent(fallback_model=fbs)
             agent.provider = "openai-codex"
             agent.model = "gpt-5.3-codex"
+            rate_limit_headers = {
+                "x-ratelimit-limit-requests-1h": "800",
+                "x-ratelimit-remaining-requests-1h": "0",
+                "x-ratelimit-reset-requests-1h": "7200",
+            }
 
             with (
                 patch(
@@ -244,13 +264,81 @@ def test_provider_rotation_marks_failed_provider_unavailable(self, tmp_path):
                 ),
                 patch("time.time", return_value=2000.0),
                 patch("time.monotonic", return_value=2000.0),
-                patch("agent.auxiliary_client.resolve_provider_client", return_value=(_mock_client(), "claude-sonnet-4-6")),
+                patch(
+                    "agent.auxiliary_client.resolve_provider_client",
+                    return_value=(_mock_client(), "claude-sonnet-4-6"),
+                ),
             ):
-                assert agent._try_activate_fallback(FailoverReason.rate_limit) is True
+                assert (
+                    agent._try_activate_fallback(
+                        FailoverReason.rate_limit,
+                        rate_limit_headers=rate_limit_headers,
+                    )
+                    is True
+                )
 
             state = ProviderRotationState.load()
-            assert state.is_unavailable("openai-codex", "gpt-5.3-codex", now=9000.0)
-            assert not state.is_unavailable("openai-codex", "gpt-5.3-codex", now=9201.0)
+            assert state.is_unavailable(
+                "openai-codex",
+                "gpt-5.3-codex",
+                base_url="https://openrouter.ai/api/v1",
+                now=9000.0,
+            )
+            assert not state.is_unavailable(
+                "openai-codex",
+                "gpt-5.3-codex",
+                base_url="https://openrouter.ai/api/v1",
+                now=9201.0,
+            )
+        finally:
+            reset_hermes_home_override(token)
+
+    def test_provider_rotation_skips_persist_for_transient_rate_limit(self, tmp_path):
+        """Healthy buckets on a 429 should not write cross-session cooldown state."""
+        from agent.error_classifier import FailoverReason
+        from agent.provider_rotation import ProviderRotationState
+
+        token = set_hermes_home_override(tmp_path)
+        try:
+            fbs = [{"provider": "anthropic", "model": "claude-sonnet-4-6"}]
+            agent = _make_agent(fallback_model=fbs)
+            agent.provider = "openai-codex"
+            agent.model = "gpt-5.3-codex"
+            rate_limit_headers = {
+                "x-ratelimit-limit-requests": "200",
+                "x-ratelimit-remaining-requests": "198",
+                "x-ratelimit-reset-requests": "20",
+                "x-ratelimit-limit-requests-1h": "800",
+                "x-ratelimit-remaining-requests-1h": "790",
+                "x-ratelimit-reset-requests-1h": "1800",
+            }
+
+            with (
+                patch(
+                    "hermes_cli.config.load_config",
+                    return_value={"provider_rotation": {"enabled": True}},
+                ),
+                patch("time.time", return_value=2000.0),
+                patch("time.monotonic", return_value=2000.0),
+                patch(
+                    "agent.auxiliary_client.resolve_provider_client",
+                    return_value=(_mock_client(), "claude-sonnet-4-6"),
+                ),
+            ):
+                assert (
+                    agent._try_activate_fallback(
+                        FailoverReason.rate_limit,
+                        rate_limit_headers=rate_limit_headers,
+                    )
+                    is True
+                )
+
+            state = ProviderRotationState.load()
+            assert not state.is_unavailable(
+                "openai-codex",
+                "gpt-5.3-codex",
+                now=2001.0,
+            )
         finally:
             reset_hermes_home_override(token)
 
@@ -266,18 +354,26 @@ def test_provider_rotation_uses_fallback_at_turn_start_when_primary_is_cooling_d
             agent.model = "gpt-5.3-codex"
             agent._primary_runtime["provider"] = "openai-codex"
             agent._primary_runtime["model"] = "gpt-5.3-codex"
+            agent._primary_runtime["base_url"] = "https://openrouter.ai/api/v1"
             ProviderRotationState.load().mark_unavailable(
                 provider="openai-codex",
                 model="gpt-5.3-codex",
+                base_url="https://openrouter.ai/api/v1",
                 reason="rate_limit",
                 cooldown_seconds=3600,
                 now=1000.0,
             )
 
             with (
-                patch("hermes_cli.config.load_config", return_value={"provider_rotation": {"enabled": True}}),
+                patch(
+                    "hermes_cli.config.load_config",
+                    return_value={"provider_rotation": {"enabled": True}},
+                ),
                 patch("time.time", return_value=1200.0),
-                patch("agent.auxiliary_client.resolve_provider_client", return_value=(_mock_client(), "claude-sonnet-4-6")),
+                patch(
+                    "agent.auxiliary_client.resolve_provider_client",
+                    return_value=(_mock_client(), "claude-sonnet-4-6"),
+                ),
             ):
                 assert agent._restore_primary_runtime() is True
 
@@ -287,6 +383,45 @@ def test_provider_rotation_uses_fallback_at_turn_start_when_primary_is_cooling_d
         finally:
             reset_hermes_home_override(token)
 
+    def test_restore_path_respects_persisted_primary_cooldown_after_60s_gate(self, tmp_path):
+        """Expired transient gate must not restore a primary still on durable cooldown."""
+        from agent.provider_rotation import ProviderRotationState
+
+        token = set_hermes_home_override(tmp_path)
+        try:
+            agent = _make_agent(fallback_model=[{"provider": "anthropic", "model": "claude-sonnet-4-6"}])
+            agent.provider = "anthropic"
+            agent.model = "claude-sonnet-4-6"
+            agent.base_url = "https://api.anthropic.com/v1"
+            agent._fallback_activated = True
+            agent._rate_limited_until = 0
+            agent._primary_runtime["provider"] = "openai-codex"
+            agent._primary_runtime["model"] = "gpt-5.3-codex"
+            agent._primary_runtime["base_url"] = "https://openrouter.ai/api/v1"
+            ProviderRotationState.load().mark_unavailable(
+                provider="openai-codex",
+                model="gpt-5.3-codex",
+                base_url="https://openrouter.ai/api/v1",
+                reason="rate_limit",
+                cooldown_seconds=3600,
+                now=1000.0,
+            )
+
+            with (
+                patch(
+                    "hermes_cli.config.load_config",
+                    return_value={"provider_rotation": {"enabled": True}},
+                ),
+                patch("time.time", return_value=1200.0),
+            ):
+                assert agent._restore_primary_runtime() is False
+
+            assert agent.provider == "anthropic"
+            assert agent.model == "claude-sonnet-4-6"
+            assert agent._fallback_activated is True
+        finally:
+            reset_hermes_home_override(token)
+
 
 # ── Pool-rotation vs fallback gating (#11314) ────────────────────────────
 
@@ -340,9 +475,7 @@ def test_skips_entry_matching_current_provider_and_model(self):
         """Chain has [same-as-current, real-fallback]; activate must skip
         the first and use the second."""
         fbs = [
-            # First entry == current state. Should be skipped.
             {"provider": "openrouter", "model": "z-ai/glm-4.7"},
-            # Second entry: real fallback.
             {"provider": "zai", "model": "glm-4.7"},
         ]
         agent = _make_agent(fallback_model=fbs)
@@ -350,18 +483,20 @@ def test_skips_entry_matching_current_provider_and_model(self):
         agent.model = "z-ai/glm-4.7"
         agent.base_url = "https://openrouter.ai/api/v1"
 
-        # Stub out resolve_provider_client so we can assert which entry was
-        # actually used — return a MagicMock client tagged with the provider.
         called = []
+
         def _resolve(provider, model=None, raw_codex=False, **kwargs):
             called.append((provider, model))
             return _mock_client(), model
+
         with patch("agent.auxiliary_client.resolve_provider_client", side_effect=_resolve):
-            with patch("hermes_cli.model_normalize.normalize_model_for_provider", side_effect=lambda m, p: m):
+            with patch(
+                "hermes_cli.model_normalize.normalize_model_for_provider",
+                side_effect=lambda m, p: m,
+            ):
                 ok = agent._try_activate_fallback()
 
         assert ok is True
-        # The first entry was skipped — only the second reached resolve.
         assert called == [("zai", "glm-4.7")], (
             f"expected fallback to skip same-state entry, got call order: {called}"
         )
@@ -370,10 +505,11 @@ def test_skips_entry_matching_current_base_url_and_model(self):
         """Two custom_providers entries pointing at the same shim URL
         with the same model should dedup even if their provider names differ."""
         fbs = [
-            # Different provider name but same shim URL + model — same backend.
-            {"provider": "claude-cli-alt", "model": "claude-opus-4.7",
-             "base_url": "http://127.0.0.1:7891/v1"},
-            # Real different fallback.
+            {
+                "provider": "claude-cli-alt",
+                "model": "claude-opus-4.7",
+                "base_url": "http://127.0.0.1:7891/v1",
+            },
             {"provider": "openrouter", "model": "anthropic/claude-opus-4.7"},
         ]
         agent = _make_agent(fallback_model=fbs)
@@ -382,15 +518,19 @@ def test_skips_entry_matching_current_base_url_and_model(self):
         agent.base_url = "http://127.0.0.1:7891/v1"
 
         called = []
+
         def _resolve(provider, model=None, raw_codex=False, **kwargs):
             called.append((provider, model))
             return _mock_client(), model
+
         with patch("agent.auxiliary_client.resolve_provider_client", side_effect=_resolve):
-            with patch("hermes_cli.model_normalize.normalize_model_for_provider", side_effect=lambda m, p: m):
+            with patch(
+                "hermes_cli.model_normalize.normalize_model_for_provider",
+                side_effect=lambda m, p: m,
+            ):
                 ok = agent._try_activate_fallback()
 
         assert ok is True
-        # Same shim/base_url+model entry skipped, second one used.
         assert called == [("openrouter", "anthropic/claude-opus-4.7")], (
             f"expected base_url-aware dedup, got call order: {called}"
         )