From f8e71375a791d42ac2a23a635f256a89d3c5150a Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:18:56 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20spotify=20=E2=80=94?= =?UTF-8?q?=20inline=20single-use=20exchange/scope=20helpers;=20minimax=20?= =?UTF-8?q?=E2=80=94=20simpler=20bounded=20body=20reader;=20drop=20post-im?= =?UTF-8?q?port=20blanks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/approval_mode.py | 2 -- hermes_cli/approvals_suggest.py | 6 ----- hermes_cli/approvals_test.py | 1 - hermes_cli/auth_minimax.py | 28 ++++++------------- hermes_cli/auth_model_picker.py | 7 ----- hermes_cli/auth_spotify.py | 48 +++++++++++---------------------- hermes_cli/auth_xai.py | 3 --- hermes_cli/auth_zai_kimi.py | 1 - hermes_cli/blueprint_cmd.py | 7 ----- 9 files changed, 23 insertions(+), 80 deletions(-) diff --git a/hermes_cli/approval_mode.py b/hermes_cli/approval_mode.py index 8e835e0b3d..6f019bbd01 100644 --- a/hermes_cli/approval_mode.py +++ b/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): diff --git a/hermes_cli/approvals_suggest.py b/hermes_cli/approvals_suggest.py index 0dcd67f764..245cfc6c98 100644 --- a/hermes_cli/approvals_suggest.py +++ b/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, diff --git a/hermes_cli/approvals_test.py b/hermes_cli/approvals_test.py index 6400c61cab..cf822b5c6a 100644 --- a/hermes_cli/approvals_test.py +++ b/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: diff --git a/hermes_cli/auth_minimax.py b/hermes_cli/auth_minimax.py index 95b51e3fb5..dcccaaa914 100644 --- a/hermes_cli/auth_minimax.py +++ b/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() diff --git a/hermes_cli/auth_model_picker.py b/hermes_cli/auth_model_picker.py index 36616ffea9..e8e4ad49e2 100644 --- a/hermes_cli/auth_model_picker.py +++ b/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): diff --git a/hermes_cli/auth_spotify.py b/hermes_cli/auth_spotify.py index bbf56e4672..370aa96545 100644 --- a/hermes_cli/auth_spotify.py +++ b/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, diff --git a/hermes_cli/auth_xai.py b/hermes_cli/auth_xai.py index 1b43082b3e..ae5db55109 100644 --- a/hermes_cli/auth_xai.py +++ b/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): diff --git a/hermes_cli/auth_zai_kimi.py b/hermes_cli/auth_zai_kimi.py index 78b206a9cf..ff03a7b55d 100644 --- a/hermes_cli/auth_zai_kimi.py +++ b/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)) diff --git a/hermes_cli/blueprint_cmd.py b/hermes_cli/blueprint_cmd.py index 68a2413045..c177d1c75e 100644 --- a/hermes_cli/blueprint_cmd.py +++ b/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 ` 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())