refactor(hermes_cli): AST-neutral layout compaction across group D

This commit is contained in:
Teknium
2026-09-02 20:47:53 -07:00
parent bfb3b5807b
commit 076ede8d97
9 changed files with 33 additions and 67 deletions
+3 -6
View File
@@ -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.
+1 -2
View File
@@ -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:
+5 -10
View File
@@ -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)
+10 -20
View File
@@ -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
+3 -6
View File
@@ -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
+2 -4
View File
@@ -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
+3 -6
View File
@@ -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):
+5 -11
View File
@@ -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:
+1 -2
View File
@@ -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]