refactor(auth): relocate single-consumer helpers next to their users; _decode_jwt_claims to constants leaf

hermes_cli/auth.py

@@ -14,7 +14,6 @@ import os
 import shutil
 import shlex
 import stat
-import hashlib
 import threading
 import time
 import uuid
@@ -52,6 +51,7 @@ from hermes_cli.auth_model_picker import (  # noqa: F401  (re-exported; callers/
     _save_model_choice,
 )
 from hermes_cli.auth_device_flow import (  # noqa: F401  (re-exported; callers/tests use hermes_cli.auth.<name>)
+    _CONSOLE_BROWSER_NAMES,
     _can_open_graphical_browser,
     _default_verify,
     _is_remote_session,
@@ -92,7 +92,9 @@ from hermes_cli.auth_nous import (  # noqa: F401  (re-exported; callers/tests us
     NOUS_SHARED_STORE_FILENAME,
     _ALLOWED_NOUS_INFERENCE_HOSTS,
     _NOUS_EFFECTIVE_STATE_IGNORED_KEYS,
+    _NOUS_EMPTY_AGENT_KEY_FIELDS,
     _NOUS_SHARED_STATE_KEYS,
+    _NOUS_STALE_PORTAL_HOSTS,
     _NousStatePersister,
     _OAUTH_GRANT_DEAD_CODES,
     _TERMINAL_REFRESH_ERROR_CODES,
@@ -101,11 +103,14 @@ from hermes_cli.auth_nous import (  # noqa: F401  (re-exported; callers/tests us
     _assert_nous_inference_jwt_usable,
     _clear_shared_nous_state,
     _compute_nous_auth_status,
-    _decode_jwt_claims,
     _empty_nous_auth_status,
     _format_nous_entitlement_auth_error,
     _healed_nous_inference_url,
+    _is_terminal_codex_oauth_refresh_error,
+    _is_terminal_nous_refresh_error,
     _is_terminal_refresh_error,
+    _is_terminal_xai_oauth_refresh_error,
+    _iso_after,
     _log_nous_invoke_jwt_selected,
     _login_nous,
     _merge_shared_nous_oauth_state,
@@ -125,6 +130,8 @@ from hermes_cli.auth_nous import (  # noqa: F401  (re-exported; callers/tests us
     _nous_shared_store_lock,
     _nous_shared_store_path,
     _nous_status_from_state,
+    _oauth_trace,
+    _oauth_trace_enabled,
     _offer_shared_nous_import,
     _pick_nous_model_after_login,
     _pool_first_oauth_status,
@@ -138,6 +145,7 @@ from hermes_cli.auth_nous import (  # noqa: F401  (re-exported; callers/tests us
     _set_nous_agent_key_from_invoke_jwt,
     _snapshot_nous_pool_status,
     _sync_nous_pool_from_auth_store,
+    _token_fingerprint,
     _try_import_shared_nous_state,
     _validate_nous_inference_url_from_network,
     _write_shared_nous_state,
@@ -194,6 +202,8 @@ from hermes_cli.auth_xai import (  # noqa: F401  (re-exported; callers/tests use
 )
 from hermes_cli.auth_codex import (  # noqa: F401  (re-exported; callers/tests use hermes_cli.auth.<name>)
     CODEX_QUOTA_PROBE_MIN_INTERVAL_SECONDS,
+    _clear_pool_entry_status,
+    _codex_access_token_is_expiring,
     _codex_base_url,
     _codex_device_code_login,
     _codex_exchange_authorization_code,
@@ -212,6 +222,7 @@ from hermes_cli.auth_codex import (  # noqa: F401  (re-exported; callers/tests u
     _is_codex_rate_limit_shaped,
     _load_auth_store_maybe_locked,
     _login_openai_codex,
+    _parse_retry_after_seconds,
     _pool_codex_access_token,
     _pool_entries,
     _probe_codex_quota_restored,
@@ -259,6 +270,7 @@ from hermes_cli.auth_qwen import (  # noqa: F401  (re-exported; callers/tests us
     resolve_qwen_runtime_credentials,
 )
 from hermes_cli.auth_constants import (  # noqa: F401  (re-exported; callers/tests use hermes_cli.auth.<name>)
+    _decode_jwt_claims,
     AUTH_STORE_VERSION,
     AUTH_LOCK_TIMEOUT_SECONDS,
     DEFAULT_NOUS_PORTAL_URL,
@@ -837,14 +849,6 @@ def is_rate_limited_auth_error(error: Exception) -> bool:
     )
 
 
-def _parse_retry_after_seconds(headers: Any) -> Optional[int]:
-    """Best-effort parse of a ``Retry-After`` header into whole seconds."""
-    from agent.retry_utils import parse_retry_after_seconds
-
-    seconds = parse_retry_after_seconds(headers)
-    return None if seconds is None else int(seconds)
-
-
 def format_auth_error(error: Exception) -> str:
     """Map auth failures to concise user-facing guidance."""
     if not isinstance(error, AuthError):
@@ -886,31 +890,6 @@ def _nonempty_str(value: Any) -> bool:
     return isinstance(value, str) and bool(value.strip())
 
 
-def _token_fingerprint(token: Any) -> Optional[str]:
-    """Return a short hash fingerprint for telemetry without leaking token bytes."""
-    if not isinstance(token, str):
-        return None
-    cleaned = token.strip()
-    if not cleaned:
-        return None
-    return hashlib.sha256(cleaned.encode("utf-8")).hexdigest()[:12]
-
-
-def _oauth_trace_enabled() -> bool:
-    raw = os.getenv("HERMES_OAUTH_TRACE", "").strip().lower()
-    return raw in {"1", "true", "yes", "on"}
-
-
-def _oauth_trace(event: str, *, sequence_id: Optional[str] = None, **fields: Any) -> None:
-    if not _oauth_trace_enabled():
-        return
-    payload: Dict[str, Any] = {"event": event}
-    if sequence_id:
-        payload["sequence_id"] = sequence_id
-    payload.update(fields)
-    logger.info("oauth_trace %s", json.dumps(payload, sort_keys=True, ensure_ascii=False))
-
-
 # =============================================================================
 # Auth Store — persistence layer for ~/.hermes/auth.json
 # =============================================================================
@@ -1532,12 +1511,6 @@ _POOL_STATUS_FIELDS = (
 )
 
 
-def _clear_pool_entry_status(entry: Dict[str, Any]) -> None:
-    """Reset a pool entry's cooldown / last-error metadata to healthy."""
-    for status_field in _POOL_STATUS_FIELDS:
-        entry[status_field] = None
-
-
 def _merge_disk_cooldown_state(
     entry: Dict[str, Any],
     disk_entry: Optional[Dict[str, Any]],
@@ -2266,11 +2239,6 @@ def _is_expiring(expires_at_iso: Any, skew_seconds: int) -> bool:
     return expires_epoch <= (time.time() + skew_seconds)
 
 
-def _iso_after(now: datetime, ttl_seconds: int) -> str:
-    """ISO timestamp *ttl_seconds* after *now* (UTC)."""
-    return datetime.fromtimestamp(now.timestamp() + ttl_seconds, tz=timezone.utc).isoformat()
-
-
 def _tls_state_from_verify(verify: Any) -> Dict[str, Any]:
     """Persistable ``tls`` block derived from an httpx ``verify`` value."""
     return {
@@ -2298,16 +2266,6 @@ def _last_auth_error_marker(
 
 
 _FLAT_OAUTH_TOKEN_KEYS = ("access_token", "refresh_token", "expires_at", "expires_in", "obtained_at")
-# Nous agent-key slots; a fresh login persists them as None, quarantine strips them.
-_NOUS_EMPTY_AGENT_KEY_FIELDS: Dict[str, Any] = {
-    "agent_key": None,
-    "agent_key_id": None,
-    "agent_key_expires_at": None,
-    "agent_key_expires_in": None,
-    "agent_key_reused": None,
-    "agent_key_obtained_at": None,
-}
-
 
 def _quarantine_flat_oauth_state(state: Dict[str, Any], provider: str, exc: "AuthError") -> None:
     """Strip dead tokens from a flat OAuth state after a terminal runtime refresh failure.
@@ -2337,10 +2295,6 @@ def _optional_base_url(value: Any) -> Optional[str]:
     return cleaned if cleaned else None
 
 
-_NOUS_STALE_PORTAL_HOSTS: FrozenSet[str] = frozenset({
-    "api.nousresearch.com",
-})
-
 # Allowlist of valid Nous Portal hosts. A portal_base_url outside this
 # set is treated as a misconfiguration and falls back to the default.
 # "localhost" / "127.0.0.1" are valid for local development and testing.
@@ -2351,14 +2305,6 @@ _NOUS_PORTAL_ALLOWED_HOSTS: FrozenSet[str] = frozenset({
 })
 
 
-def _codex_access_token_is_expiring(access_token: Any, skew_seconds: int) -> bool:
-    claims = _decode_jwt_claims(access_token)
-    exp = claims.get("exp")
-    if not isinstance(exp, (int, float)):
-        return False
-    return float(exp) <= (time.time() + max(0, int(skew_seconds)))
-
-
 # =============================================================================
 # Spotify auth — PKCE tokens stored in ~/.hermes/auth.json
 # =============================================================================
@@ -2369,26 +2315,6 @@ def _codex_access_token_is_expiring(access_token: Any, skew_seconds: int) -> boo
 # =============================================================================
 
 
-# Console/text-mode browsers that ``webbrowser`` will happily launch INSIDE
-# the terminal.  Opening one of these is worse than not opening anything —
-# it hijacks the user's TTY with an unusable text browser (the xAI OAuth
-# "Account Management" page rendered in w3m, reported May 2026) instead of
-# letting them copy the URL to a real browser.  When the resolved browser is
-# one of these we refuse to auto-open and fall back to the print-the-URL
-# path, same as a remote session.
-_CONSOLE_BROWSER_NAMES: FrozenSet[str] = frozenset(
-    {
-        "w3m",
-        "lynx",
-        "links",
-        "links2",
-        "elinks",
-        "www-browser",
-        "browsh",  # TUI browser — still hijacks the terminal
-    }
-)
-
-
 # =============================================================================
 # OpenAI Codex auth — tokens stored in ~/.hermes/auth.json (not ~/.codex/)
 #
@@ -2437,18 +2363,6 @@ _CONSOLE_BROWSER_NAMES: FrozenSet[str] = frozenset(
 # -----------------------------------------------------------------------------
 
 
-def _is_terminal_nous_refresh_error(exc: Exception) -> bool:
-    return _is_terminal_refresh_error(exc, "nous")
-
-
-def _is_terminal_xai_oauth_refresh_error(exc: Exception) -> bool:
-    return _is_terminal_refresh_error(exc, "xai-oauth")
-
-
-def _is_terminal_codex_oauth_refresh_error(exc: Exception) -> bool:
-    return _is_terminal_refresh_error(exc, "openai-codex")
-
-
 # Per-process memo for resolve_nous_access_token. Startup runs
 # check_tool_availability once per managed-tool check_fn (browser, image_gen,
 # etc.), and each one independently triggers a ~15s blocking token-refresh

hermes_cli/auth_codex.py

@@ -18,6 +18,7 @@ from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 from hermes_cli.auth_constants import (
+    _decode_jwt_claims,
     AUTH_LOCK_TIMEOUT_SECONDS,
     AuthError,
     CODEX_ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
@@ -40,6 +41,29 @@ if TYPE_CHECKING:  # annotation-only; the runtime import would be a cycle
 logger = logging.getLogger("hermes_cli.auth")
 
 
+def _parse_retry_after_seconds(headers: Any) -> Optional[int]:
+    """Best-effort parse of a ``Retry-After`` header into whole seconds."""
+    from agent.retry_utils import parse_retry_after_seconds
+
+    seconds = parse_retry_after_seconds(headers)
+    return None if seconds is None else int(seconds)
+
+
+def _clear_pool_entry_status(entry: Dict[str, Any]) -> None:
+    """Reset a pool entry's cooldown / last-error metadata to healthy."""
+    from hermes_cli.auth import _POOL_STATUS_FIELDS
+    for status_field in _POOL_STATUS_FIELDS:
+        entry[status_field] = None
+
+
+def _codex_access_token_is_expiring(access_token: Any, skew_seconds: int) -> bool:
+    claims = _decode_jwt_claims(access_token)
+    exp = claims.get("exp")
+    if not isinstance(exp, (int, float)):
+        return False
+    return float(exp) <= (time.time() + max(0, int(skew_seconds)))
+
+
 def _codex_base_url() -> str:
     return os.getenv("HERMES_CODEX_BASE_URL", "").strip().rstrip("/") or DEFAULT_CODEX_BASE_URL
 
@@ -115,7 +139,6 @@ def _sync_codex_pool_entries(
     credentials (an explicit API key, a different ChatGPT account, etc.) and must not be overwritten
     by a single re-auth.
     """
-    from hermes_cli.auth import _clear_pool_entry_status
     access_token = tokens.get("access_token")
     if not access_token:
         return
@@ -345,7 +368,7 @@ def refresh_codex_oauth_pure(
     timeout_seconds: float = 20.0,
 ) -> Dict[str, Any]:
     """Refresh Codex OAuth tokens without mutating Hermes auth state."""
-    from hermes_cli.auth import _nonempty_str, _parse_retry_after_seconds, _utc_now_z
+    from hermes_cli.auth import _nonempty_str, _utc_now_z
     del access_token  # Access token is only used by callers to decide whether to refresh.
     if not _nonempty_str(refresh_token):
         raise _codex_err(
@@ -634,7 +657,7 @@ def _probe_codex_quota_restored(
     Probes are throttled per access token (module-local cache) so the hot selection path can fire
     this freely.
     """
-    from hermes_cli.auth import _codex_quota_probe_cache, _decode_jwt_claims, _nonempty_str
+    from hermes_cli.auth import _codex_quota_probe_cache, _nonempty_str
     token = str(access_token or "").strip()
     if not token:
         return None
@@ -704,7 +727,7 @@ def clear_codex_pool_quota_cooldowns(access_token: Optional[str] = None) -> int:
     entry clears (a redeemed banked reset restores the whole account, and any entry that is
     genuinely still exhausted just re-freezes with fresh metadata on its next 429).
     """
-    from hermes_cli.auth import _auth_store_lock, _clear_pool_entry_status, _load_auth_store, _save_auth_store
+    from hermes_cli.auth import _auth_store_lock, _load_auth_store, _save_auth_store
     cleared = 0
     try:
         with _auth_store_lock():
@@ -894,7 +917,6 @@ def _login_openai_codex(
 
 def _codex_login_rate_limited_error(response: "httpx.Response", *, during: str = "") -> AuthError:
     """AuthError for a 429 from OpenAI's device-auth endpoints (a throttle, not a credential fault)."""
-    from hermes_cli.auth import _parse_retry_after_seconds
     retry_after = _parse_retry_after_seconds(getattr(response, "headers", None))
     wait_hint = (
         f" Try again in about {retry_after}s."
@@ -911,7 +933,6 @@ def _codex_login_rate_limited_error(response: "httpx.Response", *, during: str =
 
 def _codex_request_device_code(issuer: str, client_id: str) -> Dict[str, Any]:
     """Step 1 of the Codex device flow: request a user code, retrying capped on HTTP 429."""
-    from hermes_cli.auth import _parse_retry_after_seconds
     # OpenAI's auth endpoint rate-limits this request (HTTP 429) when login is
     # attempted too often from the same IP/account — retry with capped backoff
     # (honoring ``Retry-After``) before surfacing a clear, actionable message.

hermes_cli/auth_constants.py

@@ -6,7 +6,9 @@ Pure leaf: imports nothing from ``hermes_cli.auth`` so the per-provider modules
 
 from __future__ import annotations
 
-from typing import Callable, Dict, Optional
+import base64
+import json
+from typing import Any, Callable, Dict, Optional
 
 # httpx is imported lazily: it costs ~30ms at import time and hermes_cli.auth
 # is on the interactive-CLI startup path via credential_pool → auxiliary_client
@@ -181,3 +183,16 @@ _codex_err = _provider_error_factory("openai-codex")
 _spotify_err = _provider_error_factory("spotify")
 _qwen_err = _provider_error_factory("qwen-oauth")
 _minimax_err = _provider_error_factory("minimax-oauth")
+
+
+def _decode_jwt_claims(token: Any) -> Dict[str, Any]:
+    if not isinstance(token, str) or token.count(".") != 2:
+        return {}
+    payload = token.split(".")[1]
+    payload += "=" * ((4 - len(payload) % 4) % 4)
+    try:
+        raw = base64.urlsafe_b64decode(payload.encode("utf-8"))
+        claims = json.loads(raw.decode("utf-8"))
+    except Exception:
+        return {}
+    return claims if isinstance(claims, dict) else {}

hermes_cli/auth_device_flow.py

@@ -9,6 +9,7 @@ helpers are imported lazily inside each function (no import cycle; patches on
 from __future__ import annotations
 
 import logging
+from typing import FrozenSet
 import os
 import ssl
 import sys
@@ -31,6 +32,26 @@ from utils import is_truthy_value
 logger = logging.getLogger("hermes_cli.auth")
 
 
+# Console/text-mode browsers that ``webbrowser`` will happily launch INSIDE
+# the terminal.  Opening one of these is worse than not opening anything —
+# it hijacks the user's TTY with an unusable text browser (the xAI OAuth
+# "Account Management" page rendered in w3m, reported May 2026) instead of
+# letting them copy the URL to a real browser.  When the resolved browser is
+# one of these we refuse to auto-open and fall back to the print-the-URL
+# path, same as a remote session.
+_CONSOLE_BROWSER_NAMES: FrozenSet[str] = frozenset(
+    {
+        "w3m",
+        "lynx",
+        "links",
+        "links2",
+        "elinks",
+        "www-browser",
+        "browsh",  # TUI browser — still hijacks the terminal
+    }
+)
+
+
 def _is_remote_session() -> bool:
     """Detect environments where loopback OAuth can't reach the local browser.
 
@@ -66,7 +87,6 @@ def _can_open_graphical_browser() -> bool:
     require a display server (``$DISPLAY`` / ``$WAYLAND_DISPLAY``) unless ``$BROWSER`` points at
     something graphical; no display server almost always means no GUI browser.
     """
-    from hermes_cli.auth import _CONSOLE_BROWSER_NAMES
     import webbrowser as _webbrowser
 
     def _names_console_browser(value: str) -> bool:

hermes_cli/auth_nous.py

@@ -9,7 +9,7 @@ helpers are imported lazily inside each function (no import cycle; patches on
 from __future__ import annotations
 
 import logging
-import base64
+import hashlib
 import json
 import os
 import threading
@@ -22,6 +22,7 @@ from typing import Any, Callable, Dict, FrozenSet, List, Optional
 from urllib.parse import urlparse
 from hermes_cli.auth_codex import _pool_entries
 from hermes_cli.auth_constants import (
+    _decode_jwt_claims,
     AUTH_LOCK_TIMEOUT_SECONDS,
     AuthError,
     DEFAULT_NOUS_CLIENT_ID,
@@ -46,6 +47,64 @@ if TYPE_CHECKING:  # annotation-only; the runtime import would be a cycle
 logger = logging.getLogger("hermes_cli.auth")
 
 
+def _token_fingerprint(token: Any) -> Optional[str]:
+    """Return a short hash fingerprint for telemetry without leaking token bytes."""
+    if not isinstance(token, str):
+        return None
+    cleaned = token.strip()
+    if not cleaned:
+        return None
+    return hashlib.sha256(cleaned.encode("utf-8")).hexdigest()[:12]
+
+
+def _oauth_trace_enabled() -> bool:
+    raw = os.getenv("HERMES_OAUTH_TRACE", "").strip().lower()
+    return raw in {"1", "true", "yes", "on"}
+
+
+def _oauth_trace(event: str, *, sequence_id: Optional[str] = None, **fields: Any) -> None:
+    if not _oauth_trace_enabled():
+        return
+    payload: Dict[str, Any] = {"event": event}
+    if sequence_id:
+        payload["sequence_id"] = sequence_id
+    payload.update(fields)
+    logger.info("oauth_trace %s", json.dumps(payload, sort_keys=True, ensure_ascii=False))
+
+
+def _iso_after(now: datetime, ttl_seconds: int) -> str:
+    """ISO timestamp *ttl_seconds* after *now* (UTC)."""
+    return datetime.fromtimestamp(now.timestamp() + ttl_seconds, tz=timezone.utc).isoformat()
+
+
+# Nous agent-key slots; a fresh login persists them as None, quarantine strips them.
+_NOUS_EMPTY_AGENT_KEY_FIELDS: Dict[str, Any] = {
+    "agent_key": None,
+    "agent_key_id": None,
+    "agent_key_expires_at": None,
+    "agent_key_expires_in": None,
+    "agent_key_reused": None,
+    "agent_key_obtained_at": None,
+}
+
+
+_NOUS_STALE_PORTAL_HOSTS: FrozenSet[str] = frozenset({
+    "api.nousresearch.com",
+})
+
+
+def _is_terminal_nous_refresh_error(exc: Exception) -> bool:
+    return _is_terminal_refresh_error(exc, "nous")
+
+
+def _is_terminal_xai_oauth_refresh_error(exc: Exception) -> bool:
+    return _is_terminal_refresh_error(exc, "xai-oauth")
+
+
+def _is_terminal_codex_oauth_refresh_error(exc: Exception) -> bool:
+    return _is_terminal_refresh_error(exc, "openai-codex")
+
+
 def _format_nous_entitlement_auth_error(error: AuthError) -> str:
     try:
         from hermes_cli.nous_account import (
@@ -66,7 +125,6 @@ def _format_nous_entitlement_auth_error(error: AuthError) -> str:
 
 
 def _migrate_stale_nous_portal_url(providers: Dict[str, Any]) -> None:
-    from hermes_cli.auth import _NOUS_STALE_PORTAL_HOSTS
     nous = providers.get("nous")
     if not isinstance(nous, dict):
         return
@@ -151,19 +209,6 @@ def _nous_portal_env_override() -> Optional[str]:
     )
 
 
-def _decode_jwt_claims(token: Any) -> Dict[str, Any]:
-    if not isinstance(token, str) or token.count(".") != 2:
-        return {}
-    payload = token.split(".")[1]
-    payload += "=" * ((4 - len(payload) % 4) % 4)
-    try:
-        raw = base64.urlsafe_b64decode(payload.encode("utf-8"))
-        claims = json.loads(raw.decode("utf-8"))
-    except Exception:
-        return {}
-    return claims if isinstance(claims, dict) else {}
-
-
 def _scope_values(raw_scope: Any) -> set[str]:
     # OAuth token responses normally return a space-separated string. Keep
     # collection support for JWT ``scp`` claims and older stored test fixtures.
@@ -255,7 +300,6 @@ def _log_nous_invoke_jwt_selected(
     access_token: Any,
     sequence_id: Optional[str] = None,
 ) -> None:
-    from hermes_cli.auth import _oauth_trace, _token_fingerprint
     logger.debug("Nous inference auth: using NAS invoke JWT")
     _oauth_trace(
         "nous_invoke_jwt_selected",
@@ -477,7 +521,7 @@ def _write_shared_nous_state(state: Dict[str, Any]) -> None:
     Best-effort: any failure is swallowed after logging. The shared store is a convenience layer;
     the per-profile auth.json remains the source of truth.
     """
-    from hermes_cli.auth import _nonempty_str, _oauth_trace, _token_fingerprint, _write_private_file_atomic
+    from hermes_cli.auth import _nonempty_str, _write_private_file_atomic
     refresh_token = state.get("refresh_token")
     access_token = state.get("access_token")
     # No refresh_token = nothing worth sharing across profiles
@@ -532,7 +576,6 @@ def _read_shared_nous_state() -> Optional[Dict[str, Any]]:
 
 def _clear_shared_nous_state(reason: str) -> None:
     """Remove the shared Nous OAuth store after a terminal token failure."""
-    from hermes_cli.auth import _oauth_trace
     try:
         with _nous_shared_store_lock():
             path = _nous_shared_store_path()
@@ -577,7 +620,7 @@ def _quarantine_nous_oauth_state(
     reason: str,
 ) -> None:
     """Keep routing metadata but remove dead OAuth material so it is not replayed."""
-    from hermes_cli.auth import _FLAT_OAUTH_TOKEN_KEYS, _NOUS_EMPTY_AGENT_KEY_FIELDS, _auth_file_path, _last_auth_error_marker, _token_fingerprint, invalidate_nous_auth_status_cache
+    from hermes_cli.auth import _FLAT_OAUTH_TOKEN_KEYS, _auth_file_path, _last_auth_error_marker, invalidate_nous_auth_status_cache
     # Forensic logging BEFORE we clear the token material. A hosted agent
     # can take a terminal invalid_grant and get quarantined here silently: the
     # only downstream signal is a "No access token found" WARNING once the pool
@@ -644,7 +687,6 @@ def _quarantine_nous_pool_entries(
     reason: str,
 ) -> bool:
     """Remove singleton-seeded Nous pool entries that contain dead OAuth state."""
-    from hermes_cli.auth import _oauth_trace
     entries = _pool_entries(auth_store, "nous")
     if entries is None:
         return False
@@ -680,7 +722,7 @@ def _try_import_shared_nous_state(
     Returns ``None`` on any failure (expired token, portal unreachable) so the caller falls
     through to the normal device-code flow.
     """
-    from hermes_cli.auth import _is_terminal_nous_refresh_error, _oauth_trace, _read_shared_nous_state, _write_shared_nous_state, refresh_nous_oauth_from_state
+    from hermes_cli.auth import _read_shared_nous_state, _write_shared_nous_state, refresh_nous_oauth_from_state
     try:
         with _nous_shared_store_lock(timeout_seconds=max(timeout_seconds + 5.0, AUTH_LOCK_TIMEOUT_SECONDS)):
             shared = _read_shared_nous_state()
@@ -796,7 +838,7 @@ def _refresh_nous_or_quarantine(
     persist: Callable[[], None],
 ) -> Dict[str, Any]:
     """Redeem the Nous refresh token; on a terminal failure quarantine state + pool, persist, re-raise."""
-    from hermes_cli.auth import _is_terminal_nous_refresh_error, _refresh_access_token
+    from hermes_cli.auth import _refresh_access_token
     try:
         return _refresh_access_token(
             client=client,
@@ -824,7 +866,7 @@ def _apply_nous_refreshed_tokens(
     *inference_base_url*, when given, is the healed network-provenance URL to persist alongside
     the rotated tokens (key order in auth.json is preserved from the original login shape).
     """
-    from hermes_cli.auth import _coerce_ttl_seconds, _iso_after
+    from hermes_cli.auth import _coerce_ttl_seconds
     now = datetime.now(timezone.utc)
     access_ttl = _coerce_ttl_seconds(refreshed.get("expires_in"))
     state["access_token"] = refreshed["access_token"]
@@ -1106,7 +1148,7 @@ class _NousStatePersister:
         self.persisted_any = False
 
     def persist(self, reason: str) -> None:
-        from hermes_cli.auth import _oauth_trace, _save_provider_state_to_source, _token_fingerprint, _write_shared_nous_state
+        from hermes_cli.auth import _save_provider_state_to_source, _write_shared_nous_state
         state = self._state
         if (
             _nous_effective_provider_state(state)
@@ -1217,7 +1259,7 @@ def resolve_nous_runtime_credentials(
     of rotating the shared grant again (otherwise N concurrent processes at the
     same expiry issue N refreshes, each invalidating a sibling's fresh token).
     """
-    from hermes_cli.auth import _assert_nous_inference_jwt_usable, _auth_file_path, _coerce_ttl_seconds, _nous_invoke_jwt_status, _oauth_trace, _parse_iso_timestamp, _provider_state_transaction, _resolve_verify, _select_nous_invoke_jwt, _sync_nous_pool_from_auth_store, _tls_state_from_verify, _token_fingerprint
+    from hermes_cli.auth import _assert_nous_inference_jwt_usable, _auth_file_path, _coerce_ttl_seconds, _nous_invoke_jwt_status, _parse_iso_timestamp, _provider_state_transaction, _resolve_verify, _select_nous_invoke_jwt, _sync_nous_pool_from_auth_store, _tls_state_from_verify
     sequence_id = uuid.uuid4().hex[:12]
 
     with _provider_state_transaction("nous") as (
@@ -1744,7 +1786,7 @@ def _nous_device_code_login(
     on_verification: Optional[Callable[[str, str], None]] = None,
 ) -> Dict[str, Any]:
     """Run the Nous device-code flow and return full OAuth state without persisting."""
-    from hermes_cli.auth import PROVIDER_REGISTRY, _NOUS_EMPTY_AGENT_KEY_FIELDS, _coerce_ttl_seconds, _is_remote_session, _optional_base_url, _poll_for_token, _print_device_code_instructions, _request_device_code, _tls_state_from_verify, format_auth_error, refresh_nous_oauth_from_state
+    from hermes_cli.auth import PROVIDER_REGISTRY, _coerce_ttl_seconds, _is_remote_session, _optional_base_url, _poll_for_token, _print_device_code_instructions, _request_device_code, _tls_state_from_verify, format_auth_error, refresh_nous_oauth_from_state
     pconfig = PROVIDER_REGISTRY["nous"]
     portal_base_url = (
         portal_base_url

hermes_cli/auth_oauth_grants.py

@@ -13,7 +13,7 @@ import json
 import os
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
-from hermes_cli.auth_nous import _decode_jwt_claims
+from hermes_cli.auth_constants import _decode_jwt_claims
 
 # Log-record parity with the origin module (caplog tests pin "hermes_cli.auth").
 logger = logging.getLogger("hermes_cli.auth")