refactor(hermes_cli): spotify — inline single-use exchange/scope helpers; minimax — simpler bounded body reader; drop post-import blanks

hermes_cli/approval_mode.py

@@ -27,7 +27,6 @@ class ApprovalModeResult:
 def _effective_mode() -> str:
     """Return the exact mode enforced by the terminal approval guard."""
     from tools.approval import _get_approval_mode
-
     return _get_approval_mode()
 
 
@@ -46,7 +45,6 @@ def run_approval_mode_command(requested_mode: Optional[str]) -> ApprovalModeResu
     # unparseable config.yaml; capture both for slash-command output instead of terminating the
     # interactive worker.
     from hermes_cli.config import set_config_value
-
     output = StringIO()
     try:
         with redirect_stdout(output), redirect_stderr(output):

hermes_cli/approvals_suggest.py

@@ -112,7 +112,6 @@ class Proposal:
 
 def default_db_path() -> Path:
     from hermes_constants import get_hermes_home
-
     return get_hermes_home() / "state.db"
 
 
@@ -175,7 +174,6 @@ def scan_approval_history(db_path: Optional[Path] = None, days: int = 90) -> lis
     that actually executed (i.e. carried an implied user approval).
     """
     from tools.approval import detect_dangerous_command, detect_hardline_command
-
     path = Path(db_path) if db_path else default_db_path()
     if not path.exists():
         return []
@@ -208,7 +206,6 @@ def scan_approval_history(db_path: Optional[Path] = None, days: int = 90) -> lis
 def normalize_command(command: str) -> str:
     """Fold user/hermes home prefixes and collapse whitespace."""
     from tools.approval import _rewrite_resolved_hermes_home, _rewrite_resolved_user_home
-
     return " ".join(_rewrite_resolved_user_home(_rewrite_resolved_hermes_home(command)).split())
 
 
@@ -229,7 +226,6 @@ def derive_glob(normalized: str) -> Optional[str]:
     those anyway) and for commands anchored on an unsafe root binary.
     """
     from tools.approval import _has_allowlist_shell_operator
-
     tokens = normalized.split()
     if _has_allowlist_shell_operator(normalized) or not tokens or _unsafe_root_binary(tokens[0]):
         return None
@@ -298,7 +294,6 @@ def parse_apply_indices(spec: str, total: int) -> list[int]:
 def apply_proposals(proposals: list[Proposal], indices: list[int]) -> set:
     """Merge chosen proposal patterns into command_allowlist and persist."""
     import tools.approval as approval_module
-
     merged = set(approval_module.load_permanent_allowlist()) | {proposals[idx].pattern for idx in indices}
     approval_module.save_permanent_allowlist(merged)
     # Keep the in-process allowlist consistent so a long-lived process sees the new entries
@@ -341,7 +336,6 @@ def suggest_command(args) -> int:
         return 1
 
     import tools.approval as approval_module
-
     existing = set(approval_module.load_permanent_allowlist())
     proposals = build_proposals(
         scan_approval_history(db_path, days=days), existing=existing,

hermes_cli/approvals_test.py

@@ -35,7 +35,6 @@ def evaluate_command(command: str, env_type: str = "local") -> dict:
     de-obfuscated forms the detectors actually evaluated).
     """
     import tools.approval as approval
-
     # Sync config-persisted "always" patterns so the allowlist check below sees what the runtime
     # would see (load is read-only).
     try:

hermes_cli/auth_minimax.py

@@ -23,7 +23,6 @@ from hermes_cli.auth_constants import (
 
 if TYPE_CHECKING:  # annotation-only; the runtime import would be a cycle
     from hermes_cli.auth import ProviderConfig
-
 # Log-record parity with the origin module (caplog tests pin "hermes_cli.auth").
 logger = logging.getLogger("hermes_cli.auth")
 
@@ -33,34 +32,23 @@ _MINIMAX_OAUTH_ERROR_BODY_LIMIT = 16 * 1024
 def _minimax_response_error_text(response: httpx.Response, *, limit: int = _MINIMAX_OAUTH_ERROR_BODY_LIMIT) -> str:
     """Return a bounded error body from a streamed MiniMax OAuth response."""
     limit = max(0, int(limit))
-    chunks: list[bytes] = []
-    total = 0
-    truncated = False
     try:
         if getattr(response, "is_stream_consumed", False):
             text = response.text
             return text[:limit] + ("...[truncated]" if len(text) > limit else "")
-
+        # Read at most limit+1 bytes so truncation can be detected without buffering the whole body.
+        chunks: list[bytes] = []
+        total = 0
         for chunk in response.iter_bytes():
             if not chunk:
                 continue
-            remaining = limit + 1 - total
-            if remaining <= 0:
-                truncated = True
+            chunks.append(chunk[: limit + 1 - total])
+            total += len(chunks[-1])
+            if total > limit:
                 break
-            if len(chunk) > remaining:
-                chunks.append(chunk[:remaining])
-                total += remaining
-                truncated = True
-                break
-            chunks.append(chunk)
-            total += len(chunk)
         raw = b"".join(chunks)
-        if len(raw) > limit:
-            raw = raw[:limit]
-            truncated = True
-        text = raw.decode(response.encoding or "utf-8", errors="replace")
-        return text + ("...[truncated]" if truncated else "")
+        text = raw[:limit].decode(response.encoding or "utf-8", errors="replace")
+        return text + ("...[truncated]" if len(raw) > limit else "")
     finally:
         response.close()
 

hermes_cli/auth_model_picker.py

@@ -31,7 +31,6 @@ def _confirm_selection_guards(
     """
     try:
         from hermes_cli.model_selection_guards import combined_message, selection_warnings
-
         warnings = selection_warnings(
             model_id, provider=provider, base_url=base_url, api_key=api_key, include_kinds=include_kinds,
         )
@@ -64,7 +63,6 @@ class _ModelPickerRows:
         current_model: str, sale_chrome: bool,
     ) -> None:
         from hermes_cli.models import _format_price_per_mtok, compute_sale_discount
-
         self.current_model = current_model
         self.has_pricing = bool(pricing and any(pricing.get(m) for m in all_models))
         # Leave room for a leading "★ " on sale rows (Nous only).
@@ -165,7 +163,6 @@ def _prompt_model_selection(
     *unavailable_models* render grayed out and unselectable with an upgrade link to *portal_url*.
     """
     from hermes_cli.cli_output import line_input
-
     _unavailable = unavailable_models or []
     # Sale chrome (★ / -N% / was) is Nous Portal-only — never for OpenRouter or other providers
     # even if pricing.original is somehow present.
@@ -206,7 +203,6 @@ def _prompt_model_selection(
     # Try arrow-key menu first, fall back to number input.
     try:
         from hermes_cli.curses_ui import curses_radiolist
-
         choices = [rows.segments(mid) for mid in ordered] + [_CUSTOM_LABEL, _SKIP_LABEL]
 
         unavailable_footer = unavailable_message.strip()
@@ -229,7 +225,6 @@ def _prompt_model_selection(
         # ids (e.g. Kimi Coding `k3` ↔ query "kimi"). model_search_text always starts with the
         # wire id; only append when aliases add tokens beyond the bare id already in the label.
         from hermes_cli.model_search import model_search_text
-
         model_search_labels = []
         for mid in ordered:
             label, haystack = rows.label(mid), model_search_text(mid)
@@ -259,7 +254,6 @@ def _prompt_model_selection(
     # Fallback: numbered list (ANSI colors for sale chrome)
     from hermes_cli.curses_ui import format_radio_item_ansi
     from hermes_cli.colors import Colors, color
-
     for line in menu_title.splitlines():
         print(line.replace("★", color("★", Colors.YELLOW), 1) if "★" in line else line)
     num_width = len(str(n + 2))
@@ -301,7 +295,6 @@ def _prompt_model_selection(
 def _save_model_choice(model_id: str) -> None:
     """Save the selected model to config.yaml only — NOT .env, which would stomp in multi-agent setups."""
     from hermes_cli.config import save_config, load_config
-
     config = load_config()
     # Always use dict format so provider/base_url can be stored alongside
     if isinstance(config.get("model"), dict):

hermes_cli/auth_spotify.py

@@ -18,7 +18,7 @@ import uuid
 import webbrowser
 from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, HTTPServer
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, Optional, Tuple
 from urllib.parse import parse_qs, urlencode, urlparse
 from hermes_cli.auth_constants import (
     AuthError, DEFAULT_SPOTIFY_ACCOUNTS_BASE_URL, DEFAULT_SPOTIFY_API_BASE_URL, DEFAULT_SPOTIFY_REDIRECT_URI,
@@ -36,12 +36,9 @@ def _clean(value: Any) -> str:
     return str(value or "").strip()
 
 
-def _spotify_scope_list(raw_scope: Optional[str] = None) -> List[str]:
-    return list(dict.fromkeys((raw_scope or DEFAULT_SPOTIFY_SCOPE).split()))
-
-
 def _spotify_scope_string(raw_scope: Optional[str] = None) -> str:
-    return " ".join(_spotify_scope_list(raw_scope))
+    """Requested scope, whitespace-normalized and de-duplicated (order kept)."""
+    return " ".join(dict.fromkeys((raw_scope or DEFAULT_SPOTIFY_SCOPE).split()))
 
 
 def _spotify_setting(
@@ -50,7 +47,6 @@ def _spotify_setting(
 ) -> str:
     """First non-empty of explicit arg, env vars (``.env`` aware), stored state, then *default*."""
     from hermes_cli.config import get_env_value
-
     candidates = (
         explicit, *(get_env_value(var) for var in env_vars),
         state.get(state_key) if isinstance(state, dict) else None, default,
@@ -246,27 +242,6 @@ def _spotify_token_post(
     return payload
 
 
-def _spotify_exchange_code_for_tokens(
-    *, client_id: str, code: str, redirect_uri: str, code_verifier: str, accounts_base_url: str,
-    timeout_seconds: float = 20.0,
-) -> Dict[str, Any]:
-    return _spotify_token_post(
-        accounts_base_url,
-        {
-            "client_id": client_id,
-            "grant_type": "authorization_code",
-            "code": code,
-            "redirect_uri": redirect_uri,
-            "code_verifier": code_verifier,
-        },
-        timeout_seconds=timeout_seconds,
-        what="token exchange",
-        failed_code="spotify_token_exchange_failed",
-        invalid_code="spotify_token_exchange_invalid",
-        invalid_message="Spotify token response did not include an access_token.",
-    )
-
-
 def _refresh_spotify_oauth_state(state: Dict[str, Any], *, timeout_seconds: float = 20.0) -> Dict[str, Any]:
     refresh_token = _clean(state.get("refresh_token"))
     if not refresh_token:
@@ -370,7 +345,6 @@ def _spotify_interactive_setup(redirect_uri_hint: str) -> str:
     """Walk the user through creating a Spotify developer app; persist the client_id to ~/.hermes/.env."""
     from hermes_cli.auth import _is_remote_session
     from hermes_cli.config import save_env_value
-
     print(
         f"\n{'=' * 70}\nSpotify first-time setup\n{'=' * 70}\n\n"
         "Spotify requires every user to register their own lightweight\n"
@@ -396,7 +370,6 @@ def _spotify_interactive_setup(redirect_uri_hint: str) -> str:
             pass
 
     from hermes_cli.cli_output import line_input
-
     try:
         raw = line_input("Spotify Client ID: ").strip()
     except (EOFError, KeyboardInterrupt):
@@ -469,10 +442,19 @@ def login_spotify_command(args) -> None:
     if callback.get("state") != state_nonce:
         raise SystemExit("Spotify authorization failed: state mismatch.")
 
-    token_payload = _spotify_exchange_code_for_tokens(
-        client_id=client_id, code=str(callback.get("code") or ""), redirect_uri=redirect_uri,
-        code_verifier=code_verifier, accounts_base_url=accounts_base_url,
+    token_payload = _spotify_token_post(
+        accounts_base_url,
+        {
+            "client_id": client_id,
+            "grant_type": "authorization_code",
+            "code": str(callback.get("code") or ""),
+            "redirect_uri": redirect_uri,
+            "code_verifier": code_verifier,
+        },
         timeout_seconds=float(getattr(args, "timeout", None) or 20.0),
+        what="token exchange", failed_code="spotify_token_exchange_failed",
+        invalid_code="spotify_token_exchange_invalid",
+        invalid_message="Spotify token response did not include an access_token.",
     )
     spotify_state = _spotify_token_payload_to_state(
         token_payload, client_id=client_id, redirect_uri=redirect_uri, requested_scope=scope,

hermes_cli/auth_xai.py

@@ -26,7 +26,6 @@ from utils import env_float
 
 if TYPE_CHECKING:  # annotation-only; the runtime import would be a cycle
     from hermes_cli.auth import ProviderConfig
-
 # Log-record parity with the origin module (caplog tests pin "hermes_cli.auth").
 logger = logging.getLogger("hermes_cli.auth")
 
@@ -436,7 +435,6 @@ def resolve_xai_oauth_runtime_credentials(
     refresh_skew_seconds: Optional[int] = None,
 ) -> Dict[str, Any]:
     from hermes_cli.auth import _auth_store_lock, _is_terminal_xai_oauth_refresh_error, _refresh_xai_oauth_tokens, _xai_oauth_discovery
-
     def _view(data: Dict[str, Any]) -> tuple[Dict[str, Any], str, str, str, bool]:
         tokens = dict(data["tokens"])
         access_token = _clean(tokens.get("access_token"))
@@ -546,7 +544,6 @@ def _xai_oauth_poll_device_token(
     poll_interval: int,
 ) -> Dict[str, Any]:
     from hermes_cli.auth import _poll_device_token_generic
-
     def _validate(payload: Dict[str, Any]) -> None:
         for field_name, article in (("access_token", "an"), ("refresh_token", "a")):
             if not payload.get(field_name):

hermes_cli/auth_zai_kimi.py

@@ -69,7 +69,6 @@ def _probe_single_zai_endpoint(api_key: str, endpoint: tuple, timeout: float) ->
 def detect_zai_endpoint(api_key: str, timeout: float = 8.0) -> Optional[Dict[str, str]]:
     """Probe z.ai endpoints in parallel; first working one in ZAI_ENDPOINTS priority order, or None."""
     from concurrent.futures import ThreadPoolExecutor, as_completed
-
     # No `with` block: a context manager would join ALL probe threads on exit, defeating the early
     # return below. shutdown(wait=False) lets surviving probes drain in the background.
     pool = ThreadPoolExecutor(max_workers=len(ZAI_ENDPOINTS))

hermes_cli/blueprint_cmd.py

@@ -29,7 +29,6 @@ def _resolve_origin(explicit: Optional[Dict[str, Any]]) -> Optional[Dict[str, An
         return explicit
     try:
         from gateway.session_context import get_session_env
-
         platform = get_session_env("HERMES_SESSION_PLATFORM")
         chat_id = get_session_env("HERMES_SESSION_CHAT_ID")
         if platform and chat_id:
@@ -74,7 +73,6 @@ def match_blueprint(query: str) -> Tuple[Optional[Any], List[Any]]:
     anywhere in key/title/description, then a difflib fuzzy pass on keys.
     """
     from cron.blueprint_catalog import CATALOG, get_blueprint
-
     q = (query or "").strip().lower()
     if not q:
         return None, []
@@ -97,7 +95,6 @@ def match_blueprint(query: str) -> Tuple[Optional[Any], List[Any]]:
 
 def _humanize_schedule(blueprint) -> str:
     from cron.blueprint_catalog import _humanize_schedule as _h
-
     try:
         return _h(blueprint)
     except Exception:
@@ -112,7 +109,6 @@ def build_blueprint_seed(blueprint) -> str:
     rendered prompt. Defaults are stated so the agent can offer them.
     """
     from cron.blueprint_catalog import WEEKDAY_PRESETS
-
     lines: List[str] = [
         f"Set up the '{blueprint.title}' automation for me (automation blueprint "
         f"'{blueprint.key}'). {blueprint.description}",
@@ -148,7 +144,6 @@ def build_blueprint_seed(blueprint) -> str:
 
 def _fmt_catalog() -> str:
     from cron.blueprint_catalog import CATALOG
-
     lines = ["Automation Blueprints — `/blueprint <name>` and I'll ask you what I need:\n"]
     for r in CATALOG:
         lines.append(f"  • {r.key} — {r.title}")
@@ -169,7 +164,6 @@ def _fmt_candidates(query: str, candidates: List[Any]) -> str:
 
 def _fmt_no_match(query: str) -> str:
     from cron.blueprint_catalog import CATALOG
-
     close = difflib.get_close_matches((query or "").lower(), [r.key for r in CATALOG], n=3, cutoff=0.4)
     msg = f"No automation blueprint matches '{query}'."
     if close:
@@ -229,7 +223,6 @@ def handle_blueprint_command(
 
     try:
         from cron.scheduler import CronSchedulerRegistrationError, create_job_with_scheduler_registration
-
         job = create_job_with_scheduler_registration(**spec)
     except CronSchedulerRegistrationError as e:
         return BlueprintCommandResult(e.user_message())