From 076ede8d97e88c50f1f11c5b9e2a6c253a5e4a2d Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:47:53 -0700 Subject: [PATCH] refactor(hermes_cli): AST-neutral layout compaction across group D --- hermes_cli/setup_summary.py | 9 +++------ hermes_cli/setup_terminal.py | 3 +-- hermes_cli/setup_tts.py | 15 +++++---------- hermes_cli/setup_whatsapp_cloud.py | 30 ++++++++++-------------------- hermes_cli/sqlite_runtime.py | 9 +++------ hermes_cli/sqlite_safe_read.py | 6 ++---- hermes_cli/status.py | 9 +++------ hermes_cli/status_auth.py | 16 +++++----------- hermes_cli/stdio.py | 3 +-- 9 files changed, 33 insertions(+), 67 deletions(-) diff --git a/hermes_cli/setup_summary.py b/hermes_cli/setup_summary.py index 8e78231e27..858bfc4861 100644 --- a/hermes_cli/setup_summary.py +++ b/hermes_cli/setup_summary.py @@ -43,8 +43,7 @@ _WEB_MISSING = ("EXA_API_KEY, PARALLEL_API_KEY, FIRECRAWL_API_KEY/FIRECRAWL_API_ _DONE_BANNER = ( "┌─────────────────────────────────────────────────────────┐", "│ ✓ Setup Complete! │", - "└─────────────────────────────────────────────────────────┘", -) + "└─────────────────────────────────────────────────────────┘") # (command, description) rows; the description carries its own alignment padding. _EDIT_WIZARD_ROWS = ( ("hermes setup", " Re-run the full wizard"), @@ -219,8 +218,7 @@ def _always_on_rows(config, feats): _TOOL_ROW_BUILDERS = ( _vision_row, _web_row, _browser_row, _image_gen_row, _video_gen_row, _tts_row, _stt_row, - _modal_row, _home_assistant_row, _spotify_row, _skills_hub_row, _always_on_rows, -) + _modal_row, _home_assistant_row, _spotify_row, _skills_hub_row, _always_on_rows) def _print_cmd_rows(rows): @@ -242,8 +240,7 @@ def _print_setup_summary(config: dict, hermes_home): """Print the setup completion summary.""" from hermes_cli.setup import ( color, Colors, get_config_path, get_env_path, get_nous_subscription_features, _info, print_header, - print_warning, - ) + print_warning) from hermes_constants import display_hermes_home as _dhh # Provider readiness — the one thing setup must produce. A user who cancelled the API-key # prompt mid-wizard used to exit "successfully" with NO working model; say so loudly. diff --git a/hermes_cli/setup_terminal.py b/hermes_cli/setup_terminal.py index d4a5a0a951..d811e81536 100644 --- a/hermes_cli/setup_terminal.py +++ b/hermes_cli/setup_terminal.py @@ -43,8 +43,7 @@ def _prompt_vercel_sandbox_settings(config: dict): # (key, prompt label, default, parser) — unparseable input leaves the value untouched. for key, label, default, parse in ( ("container_cpu", " CPU cores", 1, float), - ("container_memory", " Memory in MB (5120 = 5GB)", 5120, int), - ): + ("container_memory", " Memory in MB (5120 = 5GB)", 5120, int)): try: terminal[key] = parse(prompt(label, str(terminal.get(key, default)))) except ValueError: diff --git a/hermes_cli/setup_tts.py b/hermes_cli/setup_tts.py index 87c6bf86cc..363bd7c57f 100644 --- a/hermes_cli/setup_tts.py +++ b/hermes_cli/setup_tts.py @@ -96,8 +96,7 @@ def _run_xai_oauth_login_from_setup() -> bool: try: from hermes_cli.auth import ( _is_remote_session, _save_xai_oauth_tokens, _xai_oauth_device_code_login, - unsuppress_credential_source, - ) + unsuppress_credential_source) except Exception as exc: print_warning(f"xAI Grok OAuth helpers unavailable: {exc}") return False @@ -108,8 +107,7 @@ def _run_xai_oauth_login_from_setup() -> bool: creds = _xai_oauth_device_code_login(open_browser=open_browser) _save_xai_oauth_tokens( creds["tokens"], discovery=creds.get("discovery"), redirect_uri=creds.get("redirect_uri", ""), - last_refresh=creds.get("last_refresh"), auth_mode="oauth_device_code", set_active=False, - ) + last_refresh=creds.get("last_refresh"), auth_mode="oauth_device_code", set_active=False) # Mirror model/dashboard re-login: clear device_code suppression so the pool can seed # from the singleton after a prior `auth remove`. unsuppress_credential_source("xai-oauth", "device_code") @@ -198,8 +196,7 @@ def _tts_xai_step(config: dict) -> str: XAI_API_KEY > offer both paths — xAI TTS works with OAuth bearer tokens too.""" from hermes_cli.setup import ( get_env_value, print_success, print_warning, prompt, prompt_choice, _run_xai_oauth_login_from_setup, - save_env_value, _xai_oauth_logged_in_for_setup, - ) + save_env_value, _xai_oauth_logged_in_for_setup) selected = "xai" if _xai_oauth_logged_in_for_setup(): print_success("xAI TTS will use your xAI Grok OAuth (SuperGrok / Premium+) credentials") @@ -211,8 +208,7 @@ def _tts_xai_step(config: dict) -> str: "How do you want xAI TTS to authenticate?", choices=["Sign in with xAI Grok OAuth (SuperGrok / Premium+) — browser login", "Paste an xAI API key (console.x.ai)", "Skip → fallback to Edge TTS"], - default=0, - ) + default=0) if choice_idx == 0: if _run_xai_oauth_login_from_setup(): print_success("Logged in — xAI TTS will use these OAuth credentials") @@ -246,8 +242,7 @@ def _setup_tts_provider(config: dict): """Interactive TTS provider selection with install flow for local engines.""" from hermes_cli.setup import ( get_env_value, get_nous_subscription_features, _info, managed_nous_tools_enabled, print_header, - print_info, print_success, print_warning, prompt_choice, save_config, - ) + print_info, print_success, print_warning, prompt_choice, save_config) current_provider = config.get("tts", {}).get("provider", "edge") current_label = _TTS_PROVIDER_LABELS.get(current_provider, current_provider) diff --git a/hermes_cli/setup_whatsapp_cloud.py b/hermes_cli/setup_whatsapp_cloud.py index ca972ace76..56e7ec0c37 100644 --- a/hermes_cli/setup_whatsapp_cloud.py +++ b/hermes_cli/setup_whatsapp_cloud.py @@ -36,8 +36,7 @@ def _validate_phone_number_id(value: str) -> tuple[bool, Optional[str]]: "That looks like a phone number — but this field needs the " "Phone Number ID (Meta's internal ID, 15-17 digits, e.g. " "'7794189252778687'). Look just BELOW the 'From' dropdown in " - "API Setup → it's labelled 'Phone number ID'." - ) + "API Setup → it's labelled 'Phone number ID'.") if len(s) < 13: return False, "Phone Number ID looks too short (expected 13-18 digits)" if len(s) > 20: @@ -83,8 +82,7 @@ def _validate_app_secret(value: str) -> tuple[bool, Optional[str]]: return False, ( "App Secret should be a hex string (only digits 0-9 and " "letters a-f). Make sure you copied the 'App secret' from " - "Settings → Basic, not some other token." - ) + "Settings → Basic, not some other token.") if len(s) != 32: return False, f"App Secret should be exactly 32 hex characters (got {len(s)})" return True, None @@ -103,8 +101,7 @@ def _validate_access_token(value: str) -> tuple[bool, Optional[str]]: "Meta WhatsApp access tokens start with 'EAA'. Check that " "you're copying from the right place (API Setup → 'Generate " "access token', or Business Settings → System Users → " - "'Generate token' for a permanent one)." - ) + "'Generate token' for a permanent one).") if len(s) < 100: return False, f"Access token looks too short ({len(s)} chars, expected 100+)" return True, None @@ -133,8 +130,7 @@ def _prompt(message: str, default: Optional[str] = None, secret: bool = False) - def _prompt_validated( message: str, validator, *, current: Optional[str] = None, help_text: Optional[str] = None, - secret: bool = False, -) -> Optional[str]: + secret: bool = False) -> Optional[str]: """Repeat the prompt until the user enters a valid value or aborts. Returns the validated value, or None if the user gave up (empty response after an error, or @@ -291,8 +287,7 @@ def run_whatsapp_cloud_setup() -> int: " → 'Connect with customers through WhatsApp'", " 2. App Dashboard → WhatsApp → API Setup", " 3. Click 'Generate access token' (temp 24h token is fine to", - " start; switch to a System User permanent token later)", "", - ) + " start; switch to a System User permanent token later)", "") try: input("Press Enter to continue, or Ctrl+C to abort... ") except (EOFError, KeyboardInterrupt): @@ -311,8 +306,7 @@ def run_whatsapp_cloud_setup() -> int: current = get_env_value(env_var) or None value = _prompt_validated( label, lambda v, _val=validator: (True, None) if not v else _val(v), - current=current, help_text=help_text, - ) + current=current, help_text=help_text) _save_optional(env_var, value, current) ids[env_var] = value or current print() @@ -344,8 +338,7 @@ def run_whatsapp_cloud_setup() -> int: "", " Who is allowed to message the bot? (Comma-separated phone", " numbers with country code, no '+' / spaces / dashes. Use '*'", " to allow anyone — only safe if you've also configured Meta's", - " recipient whitelist for app-development mode.)", "", - ) + " recipient whitelist for app-development mode.)", "") allow_default = get_env_value("WHATSAPP_CLOUD_ALLOWED_USERS") or None try: allowed = line_input( @@ -396,8 +389,7 @@ def run_whatsapp_cloud_setup() -> int: " 6. Add your phone to Meta's recipient list:", " App Dashboard → WhatsApp → API Setup → 'To' →", " 'Manage phone number list'", "", - " 7. DM the bot's test number from your phone.", "", - ) + " 7. DM the bot's test number from your phone.", "") _header("Optional: polish your bot's WhatsApp profile") effective_waba = ids["WHATSAPP_CLOUD_WABA_ID"] _lines( @@ -407,8 +399,7 @@ def run_whatsapp_cloud_setup() -> int: " it once you're up and running:", "", " • Display name + profile picture:", " https://business.facebook.com/wa/manage/phone-numbers/" - + (f"?waba_id={effective_waba}" if effective_waba else ""), - ) + + (f"?waba_id={effective_waba}" if effective_waba else "")) if not effective_waba: print(" (select your WhatsApp Business Account on that page)") _lines( @@ -419,6 +410,5 @@ def run_whatsapp_cloud_setup() -> int: " Requires Meta's business verification process —", " Business Manager → Security Center → Start Verification.", "", " Docs: https://hermes-agent.nousresearch.com/docs/user-guide/", - " messaging/whatsapp-cloud", "", - ) + " messaging/whatsapp-cloud", "") return 0 diff --git a/hermes_cli/sqlite_runtime.py b/hermes_cli/sqlite_runtime.py index 5e019c3172..edaf986c20 100644 --- a/hermes_cli/sqlite_runtime.py +++ b/hermes_cli/sqlite_runtime.py @@ -27,8 +27,7 @@ def is_sqlite_wal_reset_vulnerable(version_info: tuple[int, ...]) -> bool: info < (3, 7, 0) or info >= (3, 51, 3) or (3, 50, 7) <= info < (3, 51, 0) - or (3, 44, 6) <= info < (3, 45, 0) - ) + or (3, 44, 6) <= info < (3, 45, 0)) @dataclass(frozen=True) @@ -84,8 +83,7 @@ def probe_sqlite_runtime(python: str | Path, *, timeout: float = 30.0) -> SQLite try: result = subprocess.run( [str(python), "-I", "-c", _PROBE_SCRIPT], capture_output=True, text=True, timeout=timeout, - check=False, env=isolated_interpreter_env(), - ) + check=False, env=isolated_interpreter_env()) except (OSError, subprocess.TimeoutExpired): return None if result.returncode != 0: @@ -98,7 +96,6 @@ def probe_sqlite_runtime(python: str | Path, *, timeout: float = 30.0) -> SQLite python_version=_version_tuple(payload["python_version"]), sqlite_version=_version_tuple(payload["sqlite_version"]), sqlite_version_string=str(payload["sqlite_version_string"]), - sqlite_source_id=str(payload.get("sqlite_source_id", "")), - ) + sqlite_source_id=str(payload.get("sqlite_source_id", ""))) except (KeyError, TypeError, ValueError, json.JSONDecodeError): return None diff --git a/hermes_cli/sqlite_safe_read.py b/hermes_cli/sqlite_safe_read.py index 5171fb7008..8629ce2b58 100644 --- a/hermes_cli/sqlite_safe_read.py +++ b/hermes_cli/sqlite_safe_read.py @@ -234,8 +234,7 @@ def read_header_bytes_preopen(path: Path | str, *, length: int = 100, force: boo logger.debug( "refusing byte-level read of %s: a live connection exists in " "this process and close() would cancel its POSIX locks", - path, - ) + path) return None try: with open(path, "rb") as handle: @@ -258,6 +257,5 @@ def offline_file_access(path: Path | str, *, what: str = "read"): f"Refusing to {what} {path}: a connection to it is still open " "in this process, and raw file access would cancel that " "connection's POSIX advisory locks. Close all database " - "handles (stop the gateway/dashboard) and retry." - ) + "handles (stop the gateway/dashboard) and retry.") yield diff --git a/hermes_cli/status.py b/hermes_cli/status.py index e309d2e52d..78d3915539 100644 --- a/hermes_cli/status.py +++ b/hermes_cli/status.py @@ -17,8 +17,7 @@ from hermes_cli.models import provider_label from hermes_cli.runtime_provider import resolve_requested_provider from hermes_cli.vercel_auth import describe_vercel_auth from hermes_cli.status_auth import ( # renderers wired into _SECTIONS below - _render_api_keys, _render_apikey_providers, _render_auth_providers, _render_nous_gateway, -) + _render_api_keys, _render_apikey_providers, _render_auth_providers, _render_nous_gateway) from hermes_constants import OPENROUTER_MODELS_URL from hermes_constants import is_termux as _is_termux @@ -313,8 +312,7 @@ def _render_sessions(ctx): # runtime/active_sessions.json by hand. try: from hermes_cli.active_sessions import ( - active_session_registry_snapshot, format_age, resolve_max_concurrent_sessions, - ) + active_session_registry_snapshot, format_age, resolve_max_concurrent_sessions) cap = resolve_max_concurrent_sessions(ctx.config) except Exception: cap = None @@ -365,8 +363,7 @@ def _render_footer(ctx): _SECTIONS = ( _render_header, _render_environment, _render_api_keys, _render_auth_providers, _render_nous_gateway, _render_apikey_providers, _render_terminal, _render_platforms, _render_gateway, _render_cron, - _render_sessions, _render_deep, _render_footer, -) + _render_sessions, _render_deep, _render_footer) def show_status(args): diff --git a/hermes_cli/status_auth.py b/hermes_cli/status_auth.py index effba64a61..57ba3acf6d 100644 --- a/hermes_cli/status_auth.py +++ b/hermes_cli/status_auth.py @@ -8,9 +8,7 @@ Origin helpers (``_row``, ``_first_env_value``, ...) are imported lazily from from hermes_cli.auth import AuthError from hermes_cli.nous_account import ( - format_nous_portal_entitlement_message, - get_nous_portal_account_info, -) + format_nous_portal_entitlement_message, get_nous_portal_account_info) from hermes_cli.nous_subscription import get_nous_subscription_features from tools.tool_backend_helpers import managed_nous_tools_enabled @@ -87,13 +85,11 @@ _OAUTH_BLOCKS = ( ("Qwen OAuth", "get_qwen_auth_status", "qwen auth qwen-oauth", ( ("Auth file:", "auth_file", None, None), ("Access exp:", "expires_at_ms", _qwen_expiry, None), - ("Error:", "error", None, False), - )), + ("Error:", "error", None, False))), ("MiniMax OAuth", "get_minimax_oauth_auth_status", "hermes auth add minimax-oauth", ( ("Region:", "region", None, True), ("Access exp:", "expires_at", None, None), - ("Error:", "error", None, False), - )), + ("Error:", "error", None, False))), ("xAI OAuth", "get_xai_oauth_auth_status", "hermes auth add xai-oauth", _FILE_REFRESH_ROWS), ) @@ -155,8 +151,7 @@ def _render_auth_providers(ctx): "Nous Portal", logged_in, "logged in" if logged_in else "not logged in (Nous inference key configured)" if inference - else "not logged in (run: hermes portal)", - ) + else "not logged in (run: hermes portal)") portal_url = nous_status.get("portal_base_url") or "(unknown)" inference_url = nous_status.get("inference_base_url") or (info.inference_base_url if info else None) for label, value, show in ( @@ -168,8 +163,7 @@ def _render_auth_providers(ctx): logged_in or inference or nous_status.get("agent_key_expires_at")), ("Refresh:", "yes" if nous_status.get("has_refresh_token") else "no", logged_in or nous_status.get("has_refresh_token")), - ("Error:", nous_error, nous_error), - ): + ("Error:", nous_error, nous_error)): if show: _detail(label, value) for name, getter, hint, rows in _OAUTH_BLOCKS: diff --git a/hermes_cli/stdio.py b/hermes_cli/stdio.py index 75161f14d9..8fb6a8e492 100644 --- a/hermes_cli/stdio.py +++ b/hermes_cli/stdio.py @@ -114,8 +114,7 @@ def _augment_path_with_known_tools() -> None: os.path.join(local_appdata, "hermes", "git", "bin"), os.path.join(local_appdata, "hermes", "git", "usr", "bin"), os.path.join(local_appdata, "hermes", "hermes-agent", "venv", "Scripts"), - os.path.join(local_appdata, "Microsoft", "WinGet", "Links"), - ] + os.path.join(local_appdata, "Microsoft", "WinGet", "Links")] existing = os.environ.get("PATH", "") existing_lower = {p.lower() for p in existing.split(os.pathsep) if p} prepend = [d for d in candidate_dirs if os.path.isdir(d) and d.lower() not in existing_lower]