feat(auth): add credential pool switch command

agent/credential_pool.py

@@ -1515,6 +1515,29 @@ def remove_index(self, index: int) -> Optional[PooledCredential]:
             self._current_id = None
         return removed
 
+    def activate_index(self, index: int) -> Optional[PooledCredential]:
+        """Make the 1-based entry at *index* the first credential in the pool.
+
+        Credential-pool selection is intentionally priority/order based for the
+        default ``fill_first`` strategy and as the stable tie-breaker for lease
+        selection. Persisting the selected credential at priority 0 gives the
+        CLI a durable, profile-local way to choose the next credential without
+        introducing a second "active credential" state that could drift from
+        the on-disk pool order.
+        """
+        with self._lock:
+            if index < 1 or index > len(self._entries):
+                return None
+            selected = self._entries[index - 1]
+            ordered = [selected, *self._entries[: index - 1], *self._entries[index:]]
+            self._entries = [
+                replace(entry, priority=new_priority)
+                for new_priority, entry in enumerate(ordered)
+            ]
+            self._current_id = selected.id
+            self._persist()
+            return self._entries[0]
+
     def resolve_target(self, target: Any) -> Tuple[Optional[int], Optional[PooledCredential], Optional[str]]:
         raw = str(target or "").strip()
         if not raw:

hermes_cli/_parser.py

@@ -51,6 +51,7 @@ def _inherited_flag(parser, *args, **kwargs):
     hermes auth add <provider>    Add a pooled credential
     hermes auth list              List pooled credentials
     hermes auth remove <p> <t>    Remove pooled credential by index, id, or label
+    hermes auth switch <p> <t>    Move pooled credential to first by index, id, or label
     hermes auth reset <provider>  Clear exhaustion status for a provider
     hermes model                  Select default model
     hermes fallback [list]        Show fallback provider chain

hermes_cli/auth_commands.py

@@ -5,6 +5,7 @@
 import math
 import sys
 import time
+from dataclasses import replace
 from types import SimpleNamespace
 import uuid
 
@@ -14,6 +15,7 @@
     CUSTOM_POOL_PREFIX,
     SOURCE_MANUAL,
     SOURCE_MANUAL_DEVICE_CODE,
+    STATUS_DEAD,
     STATUS_EXHAUSTED,
     STRATEGY_FILL_FIRST,
     STRATEGY_ROUND_ROBIN,
@@ -499,6 +501,126 @@ def auth_remove_command(args) -> None:
         print(line)
 
 
+def _switch_codex_pool_and_singleton(
+    selected_id: str,
+) -> PooledCredential | None:
+    """Atomically promote a Codex OAuth pool entry and mirror singleton state.
+
+    Codex still has singleton-auth paths for token refresh and usage/status
+    checks.  If the user switches to an independent manual Codex account, the
+    selected entry must become the singleton-backed ``device_code`` entry rather
+    than merely copying its tokens into ``providers.openai-codex``. Otherwise
+    the next ``load_pool('openai-codex')`` would seed a duplicate device_code
+    entry and leave the old account mislabeled.
+    """
+    with auth_mod._auth_store_lock():
+        auth_store = auth_mod._load_auth_store()
+        pool_store = auth_store.get("credential_pool")
+        if not isinstance(pool_store, dict):
+            return None
+        payloads = pool_store.get("openai-codex")
+        if not isinstance(payloads, list):
+            return None
+        entries = [
+            PooledCredential.from_dict("openai-codex", payload)
+            for payload in payloads
+            if isinstance(payload, dict)
+        ]
+        selected = next((entry for entry in entries if entry.id == selected_id), None)
+        if selected is None or selected.auth_type != AUTH_TYPE_OAUTH:
+            return None
+        access_token = (selected.access_token or "").strip()
+        refresh_token = (selected.refresh_token or "").strip()
+        if not access_token:
+            return None
+        if not refresh_token:
+            raise SystemExit(
+                "Selected openai-codex OAuth credential is missing refresh_token. "
+                "Re-authenticate it with `hermes auth add openai-codex` before switching."
+            )
+
+        promoted_entries: list[PooledCredential] = []
+        for entry in entries:
+            updated = entry
+            if entry.id == selected_id:
+                updated = replace(entry, source="device_code")
+            elif entry.source == "device_code":
+                updated = replace(entry, source=SOURCE_MANUAL_DEVICE_CODE)
+            promoted_entries.append(updated)
+
+        selected_promoted = next(entry for entry in promoted_entries if entry.id == selected_id)
+        ordered = [
+            selected_promoted,
+            *[entry for entry in promoted_entries if entry.id != selected_id],
+        ]
+        ordered = [
+            replace(entry, priority=new_priority)
+            for new_priority, entry in enumerate(ordered)
+        ]
+        selected_promoted = ordered[0]
+        pool_store["openai-codex"] = [entry.to_dict() for entry in ordered]
+
+        state = auth_mod._load_provider_state(auth_store, "openai-codex") or {}
+        if not isinstance(state, dict):
+            state = {}
+        tokens = state.get("tokens")
+        if not isinstance(tokens, dict):
+            tokens = {}
+        tokens["access_token"] = access_token
+        tokens["refresh_token"] = refresh_token
+        state["tokens"] = tokens
+        state["auth_mode"] = state.get("auth_mode") or "chatgpt"
+        if selected_promoted.last_refresh:
+            state["last_refresh"] = selected_promoted.last_refresh
+        else:
+            state.pop("last_refresh", None)
+        if selected_promoted.label:
+            state["label"] = selected_promoted.label
+        auth_mod._store_provider_state(
+            auth_store,
+            "openai-codex",
+            state,
+            set_active=False,
+        )
+        auth_mod._save_auth_store(auth_store)
+    return selected_promoted
+
+
+def auth_switch_command(args) -> None:
+    provider = _normalize_provider(getattr(args, "provider", ""))
+    target = getattr(args, "target", None)
+    pool = load_pool(provider)
+    index, matched, error = pool.resolve_target(target)
+    if matched is None or index is None:
+        raise SystemExit(f"{error} Provider: {provider}.")
+
+    selected = None
+    synced_singleton = False
+    if provider == "openai-codex" and matched.auth_type == AUTH_TYPE_OAUTH:
+        selected = _switch_codex_pool_and_singleton(matched.id)
+        synced_singleton = selected is not None
+    else:
+        selected = pool.activate_index(index)
+    if selected is None:
+        raise SystemExit(f'No credential matching "{target}" for provider {provider}.')
+
+    print(f"Switched {provider} to credential \"{selected.label}\" (now #1)")
+    strategy = get_pool_strategy(provider)
+    if strategy in {STRATEGY_ROUND_ROBIN, STRATEGY_RANDOM, STRATEGY_LEAST_USED}:
+        print(
+            f"Note: {provider} uses {strategy}; later requests may select "
+            "another credential according to that strategy."
+        )
+    if selected.last_status in {STATUS_EXHAUSTED, STATUS_DEAD}:
+        print(
+            "Note: selected credential is currently marked "
+            f"{selected.last_status}; run `hermes auth reset {provider}` "
+            "if you intentionally want to clear cooldown state."
+        )
+    if synced_singleton:
+        print("Synchronized openai-codex singleton auth state with the selected credential.")
+
+
 def auth_reset_command(args) -> None:
     provider = _normalize_provider(getattr(args, "provider", ""))
     pool = load_pool(provider)
@@ -626,6 +748,7 @@ def _interactive_auth() -> None:
     choices = [
         "Add a credential",
         "Remove a credential",
+        "Switch active credential for a provider",
         "Reset cooldowns for a provider",
         "Set rotation strategy for a provider",
         "Exit",
@@ -647,8 +770,10 @@ def _interactive_auth() -> None:
     elif raw == "2":
         _interactive_remove()
     elif raw == "3":
-        _interactive_reset()
+        _interactive_switch()
     elif raw == "4":
+        _interactive_reset()
+    elif raw == "5":
         _interactive_strategy()
 
 
@@ -727,6 +852,29 @@ def _interactive_remove() -> None:
     auth_remove_command(SimpleNamespace(provider=provider, target=raw))
 
 
+def _interactive_switch() -> None:
+    provider = _pick_provider("Provider to switch credential for")
+    pool = load_pool(provider)
+    if not pool.has_credentials():
+        print(f"No credentials for {provider}.")
+        return
+
+    current = pool.peek()
+    for i, e in enumerate(pool.entries(), 1):
+        marker = " ←" if current is not None and e.id == current.id else ""
+        exhausted = _format_exhausted_status(e)
+        print(f"  #{i}  {e.label:25s} {e.auth_type:10s} {e.source}{exhausted} [id:{e.id}]{marker}")
+
+    try:
+        raw = input("Switch to #, id, or label (blank to cancel): ").strip()
+    except (EOFError, KeyboardInterrupt):
+        return
+    if not raw:
+        return
+
+    auth_switch_command(SimpleNamespace(provider=provider, target=raw))
+
+
 def _interactive_reset() -> None:
     provider = _pick_provider("Provider to reset cooldowns for")
 
@@ -786,6 +934,9 @@ def auth_command(args) -> None:
     if action == "remove":
         auth_remove_command(args)
         return
+    if action == "switch":
+        auth_switch_command(args)
+        return
     if action == "reset":
         auth_reset_command(args)
         return

hermes_cli/subcommands/auth.py

@@ -69,6 +69,13 @@ def build_auth_parser(subparsers, *, cmd_auth: Callable) -> None:
     auth_remove.add_argument(
         "target", help="Credential index, entry id, or exact label"
     )
+    auth_switch = auth_subparsers.add_parser(
+        "switch", help="Move a pooled credential to #1 for a provider"
+    )
+    auth_switch.add_argument("provider", help="Provider id")
+    auth_switch.add_argument(
+        "target", help="Credential index, entry id, or exact label"
+    )
     auth_reset = auth_subparsers.add_parser(
         "reset", help="Clear exhaustion status for all credentials for a provider"
     )