From 3699fa9b7fee3c564435ee1e9053468ed2a4e1da Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:27:50 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20main=5Fprovider=5Fset?= =?UTF-8?q?up=20=E2=80=94=20shared=20prompt/section=20helpers,=20api-mode?= =?UTF-8?q?=20table,=20picker=20row=20builder?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/main_provider_setup.py | 929 ++++++++++-------------------- 1 file changed, 305 insertions(+), 624 deletions(-) diff --git a/hermes_cli/main_provider_setup.py b/hermes_cli/main_provider_setup.py index e2a4799a2a..2956c4b04e 100644 --- a/hermes_cli/main_provider_setup.py +++ b/hermes_cli/main_provider_setup.py @@ -1,25 +1,21 @@ -"""Provider setup wizard helpers: provider picker, custom-provider save/remove, auxiliary-model routing menu, API-key/reasoning prompts, Anthropic OAuth. +"""Provider setup wizard helpers: provider picker, custom-provider save/remove, auxiliary-model +routing menu, API-key/reasoning prompts, Anthropic OAuth. Split out of ``hermes_cli/main.py``; every moved name is re-imported there, so -``hermes_cli.main.`` keeps resolving (and monkeypatching) as before. -Names that stay in main are imported lazily inside the functions that use them -(call-time resolution keeps ``hermes_cli.main.`` patches effective and -avoids an import cycle). +``hermes_cli.main.`` keeps resolving (and monkeypatching). Names that stay in main are +imported lazily inside the functions that use them (call-time resolution keeps +``hermes_cli.main.`` patches effective and avoids an import cycle). """ import subprocess from typing import Optional -from hermes_cli.cli_output import line_input +from hermes_cli.model_setup_flows_common import _ask, _ensure_dict_section, _print_numbered, _say def _is_profile_api_key_provider(provider_id: str) -> bool: - """Return True when provider_id maps to a profile with auth_type='api_key'. - - Used as a catch-all in select_provider_and_model() so that new providers - declared in plugins/model-providers// automatically dispatch to _model_flow_api_key_provider - without requiring an explicit elif branch here. - """ + """True when *provider_id* maps to a profile with ``auth_type='api_key'`` — the catch-all + in select_provider_and_model() so plugin providers dispatch to the generic key flow.""" try: from providers import get_provider_profile _p = get_provider_profile(provider_id) @@ -32,36 +28,33 @@ _GENERIC_API_KEY_PROVIDERS = frozenset({ "openai-api", "gemini", "deepseek", "xai", "zai", "kimi-coding-cn", "minimax", "minimax-cn", "kilocode", "opencode-zen", "opencode-go", "opencode-free", "alibaba", "huggingface", "xiaomi", "arcee", "gmi", - "nvidia", "ollama-cloud", "tencent-tokenhub", "tencent-tokenplan", "lmstudio"}) + "nvidia", "ollama-cloud", "tencent-tokenhub", "tencent-tokenplan", "lmstudio", +}) + + +def _short_url(url: str) -> str: + """``https://host/path/`` -> ``host/path`` for menu labels.""" + return url.replace("https://", "").replace("http://", "").rstrip("/") def _clear_stale_openai_base_url(): - """Remove OPENAI_BASE_URL from ~/.hermes/.env if the active provider is not 'custom'. + """Remove OPENAI_BASE_URL from ~/.hermes/.env unless the active provider is 'custom'. - After a provider switch, a leftover OPENAI_BASE_URL causes auxiliary - clients (compression, vision, delegation) with provider:auto to route - requests to the old custom endpoint instead of the newly selected - provider. See issue #5161. + A leftover OPENAI_BASE_URL after a provider switch routes provider:auto auxiliary + clients (compression, vision, delegation) to the old custom endpoint. """ from hermes_cli.config import get_env_value, save_env_value, load_config - cfg = load_config() - model_cfg = cfg.get("model", {}) - if isinstance(model_cfg, dict): - provider = (model_cfg.get("provider") or "").strip().lower() - else: - provider = "" - + model_cfg = load_config().get("model", {}) + provider = (model_cfg.get("provider") or "").strip().lower() if isinstance(model_cfg, dict) else "" if provider == "custom" or not provider: return # custom provider legitimately uses OPENAI_BASE_URL stale_url = get_env_value("OPENAI_BASE_URL") if stale_url: save_env_value("OPENAI_BASE_URL", "") - print( - f"Cleared stale OPENAI_BASE_URL from .env (was: {stale_url[:40]}...)" - if len(stale_url) > 40 - else f"Cleared stale OPENAI_BASE_URL from .env (was: {stale_url})") + shown = f"{stale_url[:40]}..." if len(stale_url) > 40 else stale_url + print(f"Cleared stale OPENAI_BASE_URL from .env (was: {shown})") # (task_key, display_name, short_description) @@ -78,34 +71,22 @@ _AUX_TASKS: list[tuple[str, str, str]] = [ ("triage_specifier", "Triage specifier", "kanban spec fleshing"), ("kanban_decomposer", "Kanban decomposer", "task decomposition"), ("profile_describer", "Profile describer", "auto profile descriptions"), - ("curator", "Curator", "skill-usage review pass")] + ("curator", "Curator", "skill-usage review pass"), +] - -# Special non-auxiliary task surfaced in the same picker: subagent delegation. -# Routing lives under top-level `delegation.*` in config.yaml (NOT -# `auxiliary.delegation`) because delegate_task spawns full child agents via -# tools/delegate_tool.py::_resolve_delegation_credentials(), which reads the -# delegation section directly. "auto" here means "inherit the parent agent's -# provider/model/credentials" and is stored as empty strings — never persist -# the literal "auto", or it would be resolved as a provider name. +# Special non-auxiliary task surfaced in the same picker: subagent delegation. Routing lives +# under top-level `delegation.*` (NOT `auxiliary.delegation`) because delegate_task spawns full +# child agents via tools/delegate_tool.py::_resolve_delegation_credentials(), which reads that +# section directly. "auto" means "inherit the parent agent" and is stored as empty strings — +# never persist the literal "auto", or it would be resolved as a provider name. _DELEGATION_TASK_KEY = "delegation" - - _DELEGATION_TASK_NAME = "Delegation" - - _DELEGATION_TASK_DESC = "subagent model (delegate_task)" def _all_aux_tasks() -> list[tuple[str, str, str]]: - """Return built-in + plugin-registered auxiliary tasks for picker/menu use. - - Built-in tasks come first (preserving order), followed by plugin tasks - sorted by key. Used by ``_aux_config_menu``, ``_reset_aux_to_auto``, and - display-name lookups so plugin-registered tasks (registered via - :meth:`hermes_cli.plugins.PluginContext.register_auxiliary_task`) appear - in the same surfaces as built-in ones without core knowing about them. - """ + """Built-in aux tasks (in order) followed by plugin-registered ones + (:meth:`hermes_cli.plugins.PluginContext.register_auxiliary_task`).""" from hermes_cli.main import _AUX_TASKS tasks = list(_AUX_TASKS) try: @@ -113,9 +94,7 @@ def _all_aux_tasks() -> list[tuple[str, str, str]]: for entry in get_plugin_auxiliary_tasks(): tasks.append((entry["key"], entry["display_name"], entry["description"])) except Exception: - # Plugin discovery failure must not break the aux config UI. - # Built-in tasks remain available. - pass + pass # Plugin discovery failure must not break the aux config UI. return tasks @@ -127,8 +106,7 @@ def _format_aux_current(task_cfg: dict) -> str: provider = str(task_cfg.get("provider") or "auto").strip() or "auto" model = str(task_cfg.get("model") or "").strip() if base_url: - short = base_url.replace("https://", "").replace("http://", "").rstrip("/") - return f"custom ({short})" + (f" · {model}" if model else "") + return f"custom ({_short_url(base_url)})" + (f" · {model}" if model else "") if provider == "auto": return "auto" + (f" · {model}" if model else "") if model: @@ -137,21 +115,20 @@ def _format_aux_current(task_cfg: dict) -> str: def _delegation_cfg_as_task(cfg: dict) -> dict: - """Project the top-level ``delegation`` section into aux-task shape. - - Returns a dict with provider/model/base_url/api_key keys so the shared - rendering (``_format_aux_current``) and picker code can treat delegation - like any other task. Empty provider means "inherit parent" which renders - as "auto". - """ + """Project the top-level ``delegation`` section into aux-task shape (provider/model/ + base_url/api_key); an empty provider means "inherit parent" and renders as "auto".""" d = cfg.get("delegation") if not isinstance(d, dict): d = {} - return { - "provider": str(d.get("provider") or "").strip(), - "model": str(d.get("model") or "").strip(), - "base_url": str(d.get("base_url") or "").strip(), - "api_key": str(d.get("api_key") or "").strip()} + return {k: str(d.get(k) or "").strip() for k in ("provider", "model", "base_url", "api_key")} + + +def _aux_task_cfg(cfg: dict, task: str) -> dict: + """The stored routing dict for *task* (delegation reads its top-level section).""" + if task == _DELEGATION_TASK_KEY: + return _delegation_cfg_as_task(cfg) + aux = cfg.get("auxiliary", {}) if isinstance(cfg.get("auxiliary"), dict) else {} + return aux.get(task, {}) if isinstance(aux.get(task), dict) else {} def _aux_task_display_name(task: str) -> str: @@ -161,43 +138,19 @@ def _aux_task_display_name(task: str) -> str: return next((name for key, name, _ in _all_aux_tasks() if key == task), task) -def _save_aux_choice( - task: str, *, provider: str, model: str = "", base_url: str = "", api_key: str = "") -> None: - """Persist an auxiliary task's provider/model to config.yaml. - - Only writes the four routing fields — timeout, download_timeout, and any - other task-specific settings are preserved untouched. The main model - config (``model.default``/``model.provider``) is never modified. - - The special ``delegation`` task writes to the top-level ``delegation`` - section (consumed by ``tools/delegate_tool.py``), not ``auxiliary.*``. - There, "auto" (inherit the parent agent) is stored as an empty provider — - the literal string "auto" would be resolved as a provider name. - """ +def _save_aux_choice(task: str, *, provider: str, model: str = "", base_url: str = "", + api_key: str = "") -> None: + """Persist an aux task's four routing fields to config.yaml (timeout etc. untouched; the main + model config never modified). ``delegation`` writes the top-level section, with "auto" + stored as an empty provider.""" from hermes_cli.config import load_config, save_config cfg = load_config() - if task == _DELEGATION_TASK_KEY: - entry = cfg.setdefault("delegation", {}) - if not isinstance(entry, dict): - entry = {} - cfg["delegation"] = entry - entry["provider"] = "" if provider == "auto" else provider - entry["model"] = model or "" - entry["base_url"] = base_url or "" - entry["api_key"] = api_key or "" - save_config(cfg) - return - - aux = cfg.setdefault("auxiliary", {}) - if not isinstance(aux, dict): - aux = {} - cfg["auxiliary"] = aux - entry = aux.setdefault(task, {}) - if not isinstance(entry, dict): - entry = {} - aux[task] = entry + entry = _ensure_dict_section(cfg, "delegation") + provider = "" if provider == "auto" else provider + else: + entry = _ensure_dict_section(_ensure_dict_section(cfg, "auxiliary"), task) entry["provider"] = provider entry["model"] = model or "" entry["base_url"] = base_url or "" @@ -206,24 +159,14 @@ def _save_aux_choice( def _reset_aux_to_auto() -> int: - """Reset every known aux task back to auto/empty. Returns number reset. - - Includes plugin-registered tasks (via ``_all_aux_tasks``) so a plugin - that contributed an auxiliary task gets reset alongside built-ins. - """ + """Reset every known aux task (built-in + plugin) back to auto/empty. Returns number reset.""" from hermes_cli.config import load_config, save_config cfg = load_config() - aux = cfg.setdefault("auxiliary", {}) - if not isinstance(aux, dict): - aux = {} - cfg["auxiliary"] = aux + aux = _ensure_dict_section(cfg, "auxiliary") count = 0 for task, _name, _desc in _all_aux_tasks(): - entry = aux.setdefault(task, {}) - if not isinstance(entry, dict): - entry = {} - aux[task] = entry + entry = _ensure_dict_section(aux, task) changed = False if entry.get("provider") not in {None, "", "auto"}: entry["provider"] = "auto" @@ -235,9 +178,7 @@ def _reset_aux_to_auto() -> int: # Preserve timeout/download_timeout — those are user-tuned, not routing if changed: count += 1 - # Delegation (top-level section) — clear only the routing fields; other - # delegation settings (max_concurrent_children, max_spawn_depth, etc.) - # are not routing and must be preserved. + # Delegation: clear only the routing fields; max_concurrent_children etc. are preserved. dele = cfg.get("delegation") if isinstance(dele, dict): changed = False @@ -252,50 +193,31 @@ def _reset_aux_to_auto() -> int: def _aux_config_menu() -> None: - """Top-level auxiliary-model picker — choose a task to configure. - - Loops until the user picks "Back" so multiple tasks can be configured - without returning to the main provider menu. - """ + """Top-level auxiliary-model picker; loops until the user picks "Back".""" from hermes_cli.main import _aux_select_for_task, _prompt_provider_choice from hermes_cli.config import load_config while True: cfg = load_config() - aux = cfg.get("auxiliary", {}) if isinstance(cfg.get("auxiliary"), dict) else {} + _say("", " Auxiliary models — side-task routing", "", + " Side tasks (vision, compression, web extraction, etc.) default", + ' to your main chat model. "auto" means "use my main model" —', + " Hermes only falls back to a lightweight backend (OpenRouter,", + " Nous Portal) if the main model is unavailable. Override a", + " task below if you want it pinned to a specific provider/model.", "") - print() - print(" Auxiliary models — side-task routing") - print() - print(" Side tasks (vision, compression, web extraction, etc.) default") - print(' to your main chat model. "auto" means "use my main model" —') - print(" Hermes only falls back to a lightweight backend (OpenRouter,") - print(" Nous Portal) if the main model is unavailable. Override a") - print(" task below if you want it pinned to a specific provider/model.") - print() - - # Build the task menu with current settings inline - all_tasks = _all_aux_tasks() - menu_tasks = all_tasks + [ - (_DELEGATION_TASK_KEY, _DELEGATION_TASK_NAME, _DELEGATION_TASK_DESC)] + menu_tasks = _all_aux_tasks() + [(_DELEGATION_TASK_KEY, _DELEGATION_TASK_NAME, _DELEGATION_TASK_DESC)] name_col = max(len(name) for _, name, _ in menu_tasks) + 2 desc_col = max(len(desc) for _, _, desc in menu_tasks) + 4 - entries: list[tuple[str, str]] = [] - for task_key, name, desc in menu_tasks: - if task_key == _DELEGATION_TASK_KEY: - task_cfg = _delegation_cfg_as_task(cfg) - else: - task_cfg = ( - aux.get(task_key, {}) if isinstance(aux.get(task_key), dict) else {}) - current = _format_aux_current(task_cfg) - label = ( - f"{name.ljust(name_col)}{('(' + desc + ')').ljust(desc_col)}{current}") - entries.append((task_key, label)) + entries = [ + (task_key, f"{name.ljust(name_col)}{('(' + desc + ')').ljust(desc_col)}" + f"{_format_aux_current(_aux_task_cfg(cfg, task_key))}") + for task_key, name, desc in menu_tasks + ] entries.append(("__reset__", "Reset all to auto")) entries.append(("__back__", "Back")) - idx = _prompt_provider_choice( - [label for _, label in entries], default=0) + idx = _prompt_provider_choice([label for _, label in entries], default=0) if idx is None: return key = entries[idx][0] @@ -303,134 +225,89 @@ def _aux_config_menu() -> None: return if key == "__reset__": n = _reset_aux_to_auto() - if n: - print(f"Reset {n} auxiliary task(s) to auto.") - else: - print("All auxiliary tasks were already set to auto.") + print(f"Reset {n} auxiliary task(s) to auto." if n else "All auxiliary tasks were already set to auto.") print() continue - # Otherwise configure the specific task _aux_select_for_task(key) def _aux_select_for_task(task: str) -> None: - """Pick a provider + model for a single auxiliary task and persist it. + """Pick a provider + model for one aux task and persist it. - Provider rows come from ``build_aux_picker_rows()`` — the shared aux-picker - substrate — so this surface shows exactly what every other aux picker - shows: authenticated built-ins, the user's own ``providers:`` / - ``custom_providers:`` endpoints, and providers whose credential pool is - temporarily exhausted. Only already-configured providers appear; users set - up new ones through the normal ``hermes model`` flow, then route aux tasks - to them here. + Rows come from ``build_aux_picker_rows()`` (the shared aux-picker substrate), so only + already-configured providers appear; new ones are set up via ``hermes model`` first. """ from hermes_cli.main import _prompt_provider_choice from hermes_cli.config import load_config from hermes_cli.inventory import build_aux_picker_rows, format_aux_picker_entries - cfg = load_config() - if task == _DELEGATION_TASK_KEY: - task_cfg = _delegation_cfg_as_task(cfg) - else: - aux = cfg.get("auxiliary", {}) if isinstance(cfg.get("auxiliary"), dict) else {} - task_cfg = aux.get(task, {}) if isinstance(aux.get(task), dict) else {} + task_cfg = _aux_task_cfg(load_config(), task) current_provider = str(task_cfg.get("provider") or "auto").strip() or "auto" current_model = str(task_cfg.get("model") or "").strip() current_base_url = str(task_cfg.get("base_url") or "").strip() - display_name = _aux_task_display_name(task) - # Gather authenticated providers (has credentials + curated model list) try: - providers = build_aux_picker_rows( - current_provider=current_provider, - current_model=current_model, - current_base_url=current_base_url) + providers = build_aux_picker_rows(current_provider=current_provider, current_model=current_model, + current_base_url=current_base_url) except Exception as exc: print(f"Could not detect authenticated providers: {exc}") providers = [] - entries: list[tuple[str, str, list[str]]] = [] # (slug, label, models) - # "auto" always first - auto_marker = ( - " ← current" if current_provider == "auto" and not current_base_url else "") - auto_label = ( - "auto (inherit main agent)" if task == _DELEGATION_TASK_KEY else "auto (recommended)") - entries.append(("__auto__", f"{auto_label}{auto_marker}", [])) - - entries.extend( - format_aux_picker_entries( - providers, current_provider=current_provider, current_base_url=current_base_url)) - - # Custom endpoint (raw base_url) + # (slug, label, models); "auto" always first + auto_marker = " ← current" if current_provider == "auto" and not current_base_url else "" + auto_label = "auto (inherit main agent)" if task == _DELEGATION_TASK_KEY else "auto (recommended)" + entries: list[tuple[str, str, list[str]]] = [("__auto__", f"{auto_label}{auto_marker}", [])] + entries.extend(format_aux_picker_entries(providers, current_provider=current_provider, + current_base_url=current_base_url)) custom_marker = " ← current" if current_base_url else "" entries.append(("__custom__", f"Custom endpoint (direct URL){custom_marker}", [])) entries.append(("__back__", "Back", [])) - print() - print(f" Configure {display_name} — current: {_format_aux_current(task_cfg)}") - print() - + _say("", f" Configure {display_name} — current: {_format_aux_current(task_cfg)}", "") idx = _prompt_provider_choice([label for _, label, _ in entries], default=0) if idx is None: return slug, _label, models = entries[idx] - if slug == "__back__": return - if slug == "__auto__": _save_aux_choice(task, provider="auto", model="", base_url="", api_key="") print(f"{display_name}: reset to auto.") - return - - if slug == "__custom__": + elif slug == "__custom__": _aux_flow_custom_endpoint(task, task_cfg) - return - - # Regular provider — pick a model from its curated list - _aux_flow_provider_model(task, slug, models, current_model) + else: + _aux_flow_provider_model(task, slug, models, current_model) -def _aux_flow_provider_model( - task: str, provider_slug: str, curated_models: list, current_model: str = "") -> None: +def _aux_flow_provider_model(task: str, provider_slug: str, curated_models: list, + current_model: str = "") -> None: """Prompt for a model under an already-authenticated provider, save to aux.""" from hermes_cli.auth import _prompt_model_selection from hermes_cli.models import get_pricing_for_provider display_name = _aux_task_display_name(task) - - # Fetch live pricing for this provider (non-blocking) - pricing: dict = {} try: pricing = get_pricing_for_provider(provider_slug) or {} except Exception: pricing = {} model_list = list(curated_models) - - # Let the user pick a model. _prompt_model_selection supports "Enter custom - # model name" and cancel. When there's no curated list (rare), fall back - # to a raw input prompt. + # _prompt_model_selection supports "Enter custom model name" and cancel; with no curated + # list (rare) fall back to a raw input prompt. if not model_list: - print(f"No curated model list for {provider_slug}.") - print("Enter a model slug manually (blank = use provider default):") - try: - val = line_input("Model: ").strip() - except (KeyboardInterrupt, EOFError): - print() + _say(f"No curated model list for {provider_slug}.", "Enter a model slug manually (blank = use provider default):") + selected = _ask("Model: ", cancel_msg="") + if selected is None: return - selected = val or "" else: - selected = _prompt_model_selection( - model_list, current_model=current_model, pricing=pricing, confirm_provider=provider_slug - ) + selected = _prompt_model_selection(model_list, current_model=current_model, pricing=pricing, + confirm_provider=provider_slug) if selected is None: print("No change.") return - _save_aux_choice( - task, provider=provider_slug, model=selected or "", base_url="", api_key="") + _save_aux_choice(task, provider=provider_slug, model=selected or "", base_url="", api_key="") if selected: print(f"{display_name}: {provider_slug} · {selected}") else: @@ -439,57 +316,58 @@ def _aux_flow_provider_model( def _aux_flow_custom_endpoint(task: str, task_cfg: dict) -> None: """Prompt for a direct OpenAI-compatible base_url + optional api_key/model.""" - from hermes_cli.secret_prompt import masked_secret_prompt - display_name = _aux_task_display_name(task) current_base_url = str(task_cfg.get("base_url") or "").strip() current_model = str(task_cfg.get("model") or "").strip() - print() - print(f" Custom endpoint for {display_name}") - print(" Provide an OpenAI-compatible base URL (e.g. http://localhost:11434/v1)") - print() - try: - url_prompt = ( - f"Base URL [{current_base_url}]: " if current_base_url else "Base URL: ") - url = line_input(url_prompt).strip() - except (KeyboardInterrupt, EOFError): - print() + _say("", f" Custom endpoint for {display_name}", + " Provide an OpenAI-compatible base URL (e.g. http://localhost:11434/v1)", "") + url = _ask(f"Base URL [{current_base_url}]: " if current_base_url else "Base URL: ", cancel_msg="") + if url is None: return url = url or current_base_url if not url: print("No URL provided. No change.") return - try: - model_prompt = ( - f"Model slug (optional) [{current_model}]: " - if current_model - else "Model slug (optional): ") - model = line_input(model_prompt).strip() - except (KeyboardInterrupt, EOFError): - print() + model = _ask(f"Model slug (optional) [{current_model}]: " if current_model else "Model slug (optional): ", + cancel_msg="") + if model is None: return model = model or current_model - try: - api_key = masked_secret_prompt( - "API key (optional, blank = use OPENAI_API_KEY): ").strip() - except (KeyboardInterrupt, EOFError): - print() + api_key = _ask("API key (optional, blank = use OPENAI_API_KEY): ", secret=True, cancel_msg="") + if api_key is None: return - _save_aux_choice( - task, provider="custom", model=model, base_url=url, api_key=api_key) - short_url = url.replace("https://", "").replace("http://", "").rstrip("/") - print(f"{display_name}: custom ({short_url})" + (f" · {model}" if model else "")) + _save_aux_choice(task, provider="custom", model=model, base_url=url, api_key=api_key) + print(f"{display_name}: custom ({_short_url(url)})" + (f" · {model}" if model else "")) + + +_CANCELLED = object() + + +def _ask_index(prompt: str, count: int, *, echo_cancel: bool): + """Numbered-menu input loop: 0-based index in ``range(count)``, ``None`` on blank, + ``_CANCELLED`` on Ctrl-C/EOF (*echo_cancel* prints a blank line first).""" + while True: + try: + val = input(prompt).strip() + if not val: + return None + idx = int(val) - 1 + if 0 <= idx < count: + return idx + print(f"Please enter 1-{count}") + except ValueError: + print("Please enter a number") + except (KeyboardInterrupt, EOFError): + if echo_cancel: + print() + return _CANCELLED def _prompt_provider_choice(choices, *, default=0, title="Select provider:"): - """Show provider selection menu with curses arrow-key navigation. - - Falls back to a numbered list when curses is unavailable (e.g. piped - stdin, non-TTY environments). Returns the selected index, or None - if the user cancels. - """ + """Provider menu with curses arrow keys; numbered-list fallback when curses is unavailable + (piped stdin, non-TTY). Returns the selected index, or None if the user cancels.""" try: from hermes_cli.setup import _curses_prompt_choice @@ -500,115 +378,72 @@ def _prompt_provider_choice(choices, *, default=0, title="Select provider:"): except Exception: pass - # Fallback: numbered list - print(title) - for i, c in enumerate(choices, 1): - marker = "→" if i - 1 == default else " " - print(f" {marker} {i}. {c}") + _print_numbered(title, choices, default) print() - while True: - try: - val = input(f"Choice [1-{len(choices)}] ({default + 1}): ").strip() - if not val: - return default - idx = int(val) - 1 - if 0 <= idx < len(choices): - return idx - print(f"Please enter 1-{len(choices)}") - except ValueError: - print("Please enter a number") - except (KeyboardInterrupt, EOFError): - print() - return None + idx = _ask_index(f"Choice [1-{len(choices)}] ({default + 1}): ", len(choices), echo_cancel=True) + if idx is None: + return default + return None if idx is _CANCELLED else idx _DEFAULT_QWEN_PORTAL_MODELS = [ - "qwen3-coder-plus", "qwen3-coder"] + "qwen3-coder-plus", + "qwen3-coder", +] + + +# (mode value, label, description, accepted answers); "" = auto-detect +_CUSTOM_API_MODES = ( + ("", "Auto-detect", "Use Hermes URL heuristics; best for standard OpenAI-compatible endpoints.", + ("1", "auto", "detect", "auto-detect")), + ("chat_completions", "Chat Completions", "Use /chat/completions for standard OpenAI-compatible servers.", + ("2", "chat", "chat_completions", "completions")), + ("codex_responses", "Responses / Codex", "Use /responses for Codex-compatible tool-calling backends.", + ("3", "responses", "codex", "codex_responses")), + ("anthropic_messages", "Anthropic Messages", "Use /v1/messages for Anthropic-compatible endpoints.", + ("4", "anthropic", "anthropic_messages", "messages")), +) +_CUSTOM_API_MODE_ANSWERS = {answer: value for value, _, _, answers in _CUSTOM_API_MODES for answer in answers} def _prompt_custom_api_mode_selection(base_url: str, current_api_mode: str = "") -> Optional[str]: - """Prompt for a custom provider API mode. - - Returns an explicit mode string, or None to keep auto-detect behavior. - """ + """Prompt for a custom provider API mode: an explicit mode string, or None for auto-detect.""" from hermes_cli.runtime_provider import _detect_api_mode_for_url detected_mode = _detect_api_mode_for_url(base_url) - normalized_current = str(current_api_mode or "").strip().lower() - default_mode = normalized_current or detected_mode or "" + default_mode = str(current_api_mode or "").strip().lower() or detected_mode or "" - mode_options = [ - ( - "", - "Auto-detect", - "Use Hermes URL heuristics; best for standard OpenAI-compatible endpoints."), - ( - "chat_completions", - "Chat Completions", - "Use /chat/completions for standard OpenAI-compatible servers."), - ( - "codex_responses", - "Responses / Codex", - "Use /responses for Codex-compatible tool-calling backends."), - ( - "anthropic_messages", - "Anthropic Messages", - "Use /v1/messages for Anthropic-compatible endpoints.")] - - print() - print("Select API compatibility mode:") - for idx, (value, label, description) in enumerate(mode_options, 1): - markers = [] - if value == detected_mode: - markers.append("detected") - if value == default_mode: - markers.append("current") + _say("", "Select API compatibility mode:") + for idx, (value, label, description, _answers) in enumerate(_CUSTOM_API_MODES, 1): + markers = [m for m, hit in (("detected", value == detected_mode), ("current", value == default_mode)) if hit] suffix = f" [{' / '.join(markers)}]" if markers else "" - print(f" {idx}. {label}{suffix}") - print(f" {description}") + _say(f" {idx}. {label}{suffix}", f" {description}") try: - raw = input( - "Choice [1-4, Enter to keep current/detected]: ").strip().lower() + raw = input("Choice [1-4, Enter to keep current/detected]: ").strip().lower() except (KeyboardInterrupt, EOFError): print("\nCancelled.") raise if not raw: return default_mode or None - - if raw in {"1", "auto", "detect", "auto-detect"}: - return None - if raw in {"2", "chat", "chat_completions", "completions"}: - return "chat_completions" - if raw in {"3", "responses", "codex", "codex_responses"}: - return "codex_responses" - if raw in {"4", "anthropic", "anthropic_messages", "messages"}: - return "anthropic_messages" - + if raw in _CUSTOM_API_MODE_ANSWERS: + return _CUSTOM_API_MODE_ANSWERS[raw] or None print(f"Invalid API mode choice: {raw}. Falling back to auto-detect.") return None def _auto_provider_name(base_url: str) -> str: - """Generate a display name from a custom endpoint URL. - - Returns a human-friendly label like "Local (localhost:11434)" or - "RunPod (xyz.runpod.io)". Used as the default when prompting the - user for a display name during custom endpoint setup. - """ + """Display name from a custom endpoint URL, e.g. "Local (localhost:11434)" or + "RunPod (xyz.runpod.io)" — the default offered during custom endpoint setup.""" import re - clean = base_url.replace("https://", "").replace("http://", "").rstrip("/") - clean = re.sub(r"/v1/?$", "", clean) - name = clean.split("/")[0] + name = re.sub(r"/v1/?$", "", _short_url(base_url)).split("/")[0] if "localhost" in name or "127.0.0.1" in name: - name = f"Local ({name})" - elif "runpod" in name.lower(): - name = f"RunPod ({name})" - else: - name = name.capitalize() - return name + return f"Local ({name})" + if "runpod" in name.lower(): + return f"RunPod ({name})" + return name.capitalize() def _custom_provider_api_key_config_value(provider_info, resolved_api_key=""): @@ -616,32 +451,25 @@ def _custom_provider_api_key_config_value(provider_info, resolved_api_key=""): api_key_ref = str(provider_info.get("api_key_ref", "") or "").strip() if api_key_ref: return api_key_ref - key_env = str(provider_info.get("key_env", "") or "").strip() if key_env and not str(provider_info.get("api_key", "") or "").strip(): return f"${{{key_env}}}" - return str(resolved_api_key or "").strip() def _custom_provider_base_url_config_value(provider_info, resolved_base_url=""): """Return the value that should be persisted for a custom provider URL.""" - base_url_ref = str(provider_info.get("base_url_ref", "") or "").strip() - if base_url_ref: - return base_url_ref - return str(resolved_base_url or "").strip() + return str(provider_info.get("base_url_ref", "") or "").strip() or str(resolved_base_url or "").strip() -def _save_custom_provider( - base_url, api_key="", model="", context_length=None, name=None, api_mode=None, key_env=""): - """Save a custom endpoint to custom_providers in config.yaml. +def _save_custom_provider(base_url, api_key="", model="", context_length=None, name=None, api_mode=None, + key_env=""): + """Save a custom endpoint to ``custom_providers`` in config.yaml. - Deduplicates by base_url — if the URL already exists, updates the - model name, context_length, and api_mode but doesn't add a duplicate entry. - Uses *name* when provided, otherwise auto-generates from the URL. - - When *key_env* is set the caller has already written the key to ``.env``, - so the entry references it instead of inlining the secret (#69449). + Deduplicates by base_url: an existing entry gets its model / context_length / api_mode + updated instead of a duplicate. *name* defaults to ``_auto_provider_name``. When *key_env* + is set the caller already wrote the key to ``.env``, so the entry references it instead + of inlining the secret. """ from hermes_cli.config import load_config, save_config @@ -650,11 +478,8 @@ def _save_custom_provider( if not isinstance(providers, list): providers = [] - # Check if this URL is already saved — update model/context_length if so for entry in providers: - if isinstance(entry, dict) and entry.get("base_url", "").rstrip( - "/" - ) == base_url.rstrip("/"): + if isinstance(entry, dict) and entry.get("base_url", "").rstrip("/") == base_url.rstrip("/"): changed = False if model and entry.get("model") != model: entry["model"] = model @@ -682,10 +507,8 @@ def _save_custom_provider( save_config(cfg) return # already saved, updated if needed - # Use provided name or auto-generate from URL if not name: name = _auto_provider_name(base_url) - entry = {"name": name, "base_url": base_url} if key_env: entry["key_env"] = key_env @@ -715,23 +538,16 @@ def _remove_custom_provider(config): return print("Remove a custom provider:\n") - - choices = [] - for entry in providers: - if isinstance(entry, dict): - name = entry.get("name", "unnamed") - url = entry.get("base_url", "") - short_url = url.replace("https://", "").replace("http://", "").rstrip("/") - choices.append(f"{name} ({short_url})") - else: - choices.append(str(entry)) + choices = [ + f"{entry.get('name', 'unnamed')} ({_short_url(entry.get('base_url', ''))})" if isinstance(entry, dict) else str(entry) + for entry in providers + ] choices.append("Cancel") try: from hermes_cli.curses_ui import curses_radiolist - idx = curses_radiolist( - "Select provider to remove:", list(choices), selected=0, cancel_returns=-1) + idx = curses_radiolist("Select provider to remove:", list(choices), selected=0, cancel_returns=-1) print() if idx < 0: idx = None @@ -752,15 +568,13 @@ def _remove_custom_provider(config): removed = providers.pop(idx) cfg["custom_providers"] = providers save_config(cfg) - removed_name = ( - removed.get("name", "unnamed") if isinstance(removed, dict) else str(removed)) + removed_name = removed.get("name", "unnamed") if isinstance(removed, dict) else str(removed) print(f'✅ Removed "{removed_name}" from custom providers.') def _prompt_reasoning_effort_selection(efforts, current_effort=""): """Prompt for a reasoning effort. Returns effort, 'none', or None to keep current.""" - deduped = list( - dict.fromkeys( str(effort).strip().lower() for effort in efforts if str(effort).strip() )) + deduped = list(dict.fromkeys(str(effort).strip().lower() for effort in efforts if str(effort).strip())) canonical_order = ("minimal", "low", "medium", "high", "xhigh", "max", "ultra") ordered = [effort for effort in canonical_order if effort in deduped] ordered.extend(effort for effort in deduped if effort not in canonical_order) @@ -768,13 +582,10 @@ def _prompt_reasoning_effort_selection(efforts, current_effort=""): return None def _label(effort): - if effort == current_effort: - return f"{effort} ← currently in use" - return effort + return f"{effort} ← currently in use" if effort == current_effort else effort disable_label = "Disable reasoning" skip_label = "Skip (keep current)" - if current_effort == "none": default_idx = len(ordered) elif current_effort in ordered: @@ -784,85 +595,53 @@ def _prompt_reasoning_effort_selection(efforts, current_effort=""): else: default_idx = 0 + n = len(ordered) try: from hermes_cli.curses_ui import curses_radiolist - choices = [_label(effort) for effort in ordered] - choices.append(disable_label) - choices.append(skip_label) - idx = curses_radiolist( - "Select reasoning effort:", choices, selected=default_idx, cancel_returns=-1) + choices = [_label(effort) for effort in ordered] + [disable_label, skip_label] + idx = curses_radiolist("Select reasoning effort:", choices, selected=default_idx, cancel_returns=-1) if idx < 0: return None print() - if idx < len(ordered): - return ordered[idx] - if idx == len(ordered): - return "none" - return None except (ImportError, NotImplementedError, OSError, subprocess.SubprocessError): - pass - - print("Select reasoning effort:") - for i, effort in enumerate(ordered, 1): - print(f" {i}. {_label(effort)}") - n = len(ordered) - print(f" {n + 1}. {disable_label}") - print(f" {n + 2}. {skip_label}") - print() - - while True: - try: - choice = input(f"Choice [1-{n + 2}] (default: keep current): ").strip() - if not choice: - return None - idx = int(choice) - if 1 <= idx <= n: - return ordered[idx - 1] - if idx == n + 1: - return "none" - if idx == n + 2: - return None - print(f"Please enter 1-{n + 2}") - except ValueError: - print("Please enter a number") - except (KeyboardInterrupt, EOFError): + print("Select reasoning effort:") + for i, effort in enumerate(ordered, 1): + print(f" {i}. {_label(effort)}") + _say(f" {n + 1}. {disable_label}", f" {n + 2}. {skip_label}", "") + idx = _ask_index(f"Choice [1-{n + 2}] (default: keep current): ", n + 2, echo_cancel=False) + if idx is None or idx is _CANCELLED: return None + if idx < n: + return ordered[idx] + return "none" if idx == n else None -def _prompt_api_key( - pconfig, existing_key: str, provider_id: str = "", existing_source: str = "") -> tuple: +def _prompt_api_key(pconfig, existing_key: str, provider_id: str = "", existing_source: str = "") -> tuple: """Shared API-key entry point for ``hermes setup`` / ``hermes model``. - Handles both first-time entry and the already-configured case. When a key - is already present, offers [K]eep / [R]eplace / [C]lear so the user can - recover from a malformed paste without editing ``~/.hermes/.env`` by hand. - - Returns ``(resolved_key, abort)``. ``abort=True`` means the caller should - ``return`` immediately — the user cancelled entry, declined to replace, or - cleared the key and is now unconfigured. + First-time entry, or [K]eep / [R]eplace / [C]lear when a key is already present (so a + malformed paste is recoverable without editing ``~/.hermes/.env``). Returns + ``(resolved_key, abort)``; ``abort=True`` means the caller should ``return`` immediately. """ from hermes_cli.auth import LMSTUDIO_NOAUTH_PLACEHOLDER from hermes_cli.config import save_env_value - from hermes_cli.secret_prompt import masked_secret_prompt key_env = pconfig.api_key_env_vars[0] if pconfig.api_key_env_vars else "" def _prompt_new_key(*, allow_lmstudio_default: bool) -> str: - if provider_id == "lmstudio" and allow_lmstudio_default: + lmstudio_default = provider_id == "lmstudio" and allow_lmstudio_default + if lmstudio_default: prompt = f"{key_env} (Enter for no-auth default {LMSTUDIO_NOAUTH_PLACEHOLDER!r}): " else: prompt = f"{key_env} (or Enter to cancel): " - try: - entered = masked_secret_prompt(prompt).strip() - except (KeyboardInterrupt, EOFError): - print() + entered = _ask(prompt, secret=True, cancel_msg="") + if entered is None: return "" - if not entered and provider_id == "lmstudio" and allow_lmstudio_default: + if not entered and lmstudio_default: return LMSTUDIO_NOAUTH_PLACEHOLDER return entered - # First-time entry ──────────────────────────────────────────────────── if not existing_key: print(f"No {pconfig.name} API key configured.") if not key_env: @@ -872,11 +651,10 @@ def _prompt_api_key( print("Cancelled.") return "", True save_env_value(key_env, new_key) - print("API key saved.") - print() + _say("API key saved.", "") return new_key, False - # Already configured — offer K / R / C ──────────────────────────────── + # Already configured — offer K / R / C from hermes_cli.env_loader import format_secret_source_suffix source_suffix = format_secret_source_suffix(key_env) if key_env else "" @@ -886,33 +664,21 @@ def _prompt_api_key( print() return existing_key, False pool_backed = existing_source.startswith("credential_pool:") - menu = ( - " [K]eep / [R]eplace (default K): " - if pool_backed - else " [K]eep / [R]eplace / [C]lear (default K): ") - try: - choice = input(menu).strip().lower() - except (KeyboardInterrupt, EOFError): - print() - choice = "k" + menu = " [K]eep / [R]eplace (default K): " if pool_backed else " [K]eep / [R]eplace / [C]lear (default K): " + choice = _ask(menu, raw=True, cancel_msg="", on_cancel="k").lower() if choice.startswith("r"): new_key = _prompt_new_key(allow_lmstudio_default=False) if not new_key: - print(" No change.") - print() + _say(" No change.", "") return existing_key, False save_env_value(key_env, new_key) - print(" API key updated.") - print() + _say(" API key updated.", "") return new_key, False - if choice.startswith("c") and not pool_backed: save_env_value(key_env, "") - print( - f" API key cleared. Re-run `hermes setup` to configure {pconfig.name} again.") + print(f" API key cleared. Re-run `hermes setup` to configure {pconfig.name} again.") return "", True - # Keep (default, or any other input) print() return existing_key, False @@ -920,49 +686,46 @@ def _prompt_api_key( def _infer_stepfun_region(base_url: str) -> str: """Infer the current StepFun region from the configured endpoint.""" - normalized = (base_url or "").strip().lower() - if "api.stepfun.com" in normalized: - return "china" - return "international" + return "china" if "api.stepfun.com" in (base_url or "").strip().lower() else "international" def _stepfun_base_url_for_region(region: str) -> str: - from hermes_cli.auth import ( - STEPFUN_STEP_PLAN_CN_BASE_URL, STEPFUN_STEP_PLAN_INTL_BASE_URL) + from hermes_cli.auth import STEPFUN_STEP_PLAN_CN_BASE_URL, STEPFUN_STEP_PLAN_INTL_BASE_URL - return ( - STEPFUN_STEP_PLAN_CN_BASE_URL if region == "china" else STEPFUN_STEP_PLAN_INTL_BASE_URL) + return STEPFUN_STEP_PLAN_CN_BASE_URL if region == "china" else STEPFUN_STEP_PLAN_INTL_BASE_URL def _run_anthropic_oauth_flow(save_env_value): """Run the Claude OAuth setup-token flow. Returns True if credentials were saved.""" - from agent.anthropic_adapter import ( - run_oauth_setup_token, read_claude_code_credentials, is_claude_code_token_valid) - from hermes_cli.config import ( - save_anthropic_oauth_token, use_anthropic_claude_code_credentials) + from agent.anthropic_adapter import run_oauth_setup_token, read_claude_code_credentials, is_claude_code_token_valid + from hermes_cli.config import save_anthropic_oauth_token, use_anthropic_claude_code_credentials def _activate_claude_code_credentials_if_available() -> bool: try: creds = read_claude_code_credentials() except Exception: creds = None - if creds and ( - is_claude_code_token_valid(creds) or bool(creds.get("refreshToken"))): + if creds and (is_claude_code_token_valid(creds) or bool(creds.get("refreshToken"))): use_anthropic_claude_code_credentials(save_fn=save_env_value) print(" ✓ Claude Code credentials linked.") from hermes_constants import display_hermes_home as _dhh_fn - print( - f" Hermes will use Claude's credential store directly instead of copying a setup-token into {_dhh_fn()}/.env." - ) + print(f" Hermes will use Claude's credential store directly instead of copying a setup-token into {_dhh_fn()}/.env.") return True return False + def _paste_token(prompt: str): + """Manual setup-token entry: True saved, False empty, None cancelled.""" + token = _ask(prompt, secret=True, cancel_msg="") + if not token: + return token + save_anthropic_oauth_token(token, save_fn=save_env_value) + print(" ✓ Setup-token saved.") + return True + try: - print() - print(" Running 'claude setup-token' — follow the prompts below.") - print(" A browser window will open for you to authorize access.") - print() + _say("", " Running 'claude setup-token' — follow the prompts below.", + " A browser window will open for you to authorize access.", "") token = run_oauth_setup_token() if token: if _activate_claude_code_credentials_if_available(): @@ -972,49 +735,27 @@ def _run_anthropic_oauth_flow(save_env_value): return True # Subprocess completed but no token auto-detected — ask user to paste - print() - print(" If the setup-token was displayed above, paste it here:") - print() - from hermes_cli.secret_prompt import masked_secret_prompt - - try: - manual_token = masked_secret_prompt( - " Paste setup-token (or Enter to cancel): ").strip() - except (KeyboardInterrupt, EOFError): - print() + _say("", " If the setup-token was displayed above, paste it here:", "") + saved = _paste_token(" Paste setup-token (or Enter to cancel): ") + if saved is None: return False - if manual_token: - save_anthropic_oauth_token(manual_token, save_fn=save_env_value) - print(" ✓ Setup-token saved.") + if saved: return True - print(" ⚠ Could not detect saved credentials.") return False except FileNotFoundError: # Claude CLI not installed — guide user through manual setup - print() - print(" The 'claude' CLI is required for OAuth login.") - print() - print(" To install and authenticate:") - print() - print(" 1. Install Claude Code: npm install -g @anthropic-ai/claude-code") - print(" 2. Run: claude setup-token") - print(" 3. Follow the browser prompts to authorize") - print(" 4. Re-run: hermes model") - print() - print(" Or paste an existing setup-token now (sk-ant-oat-...):") - print() - from hermes_cli.secret_prompt import masked_secret_prompt - - try: - token = masked_secret_prompt(" Setup-token (or Enter to cancel): ").strip() - except (KeyboardInterrupt, EOFError): - print() + _say("", " The 'claude' CLI is required for OAuth login.", "", " To install and authenticate:", "", + " 1. Install Claude Code: npm install -g @anthropic-ai/claude-code", + " 2. Run: claude setup-token", + " 3. Follow the browser prompts to authorize", + " 4. Re-run: hermes model", "", + " Or paste an existing setup-token now (sk-ant-oat-...):", "") + saved = _paste_token(" Setup-token (or Enter to cancel): ") + if saved is None: return False - if token: - save_anthropic_oauth_token(token, save_fn=save_env_value) - print(" ✓ Setup-token saved.") + if saved: return True print(" Cancelled — install Claude Code and try again.") return False @@ -1025,35 +766,33 @@ def _named_custom_provider_map(cfg) -> dict[str, dict[str, str]]: from hermes_cli.config import get_compatible_custom_providers, read_raw_config from hermes_cli.providers import custom_provider_slug - # Build lookups of raw (un-expanded) templates keyed by a - # stable identity. We intentionally bypass - # ``get_compatible_custom_providers(read_raw_config())`` here because - # its ``_normalize_custom_provider_entry`` step calls ``urlparse()`` - # on ``base_url`` and drops any entry whose ``base_url`` is itself an - # env-ref template (e.g. ``${NEURALWATT_API_BASE}``). Dropping those - # entries is exactly how env-ref preservation fails for the user - # config that motivated this fix. + # Raw (un-expanded) templates keyed by identity. ``get_compatible_custom_providers( + # read_raw_config())`` is deliberately bypassed: its normalize step ``urlparse()``s + # ``base_url`` and drops entries whose base_url is itself an env-ref template. raw_api_key_refs: dict[tuple, str] = {} raw_base_url_refs: dict[tuple, str] = {} raw_cfg = read_raw_config() - def _record_raw( - name: str, provider_key: str, model: str, api_key: str, base_url: str) -> None: - template = str(api_key or "").strip() - base_template = str(base_url or "").strip() + raw_entries: list[tuple[str, str, dict]] = [] + raw_list = raw_cfg.get("custom_providers") + if isinstance(raw_list, list): + raw_entries.extend((e.get("name", ""), "", e) for e in raw_list if isinstance(e, dict)) + raw_providers = raw_cfg.get("providers") + if isinstance(raw_providers, dict): + raw_entries.extend((e.get("name", "") or k, k, e) for k, e in raw_providers.items() if isinstance(e, dict)) + for name, provider_key, raw_entry in raw_entries: + template = str(raw_entry.get("api_key", "") or "").strip() + base_template = str(raw_entry.get("base_url", "") or raw_entry.get("url", "") or raw_entry.get("api", "") or "").strip() name = str(name or "").strip() provider_key = str(provider_key or "").strip() - model = str(model or "").strip() - # Index by every plausible identity the loaded (expanded) config - # might present: (name), (name, model), (provider_key), and - # (provider_key, model). Case-insensitive on name/provider_key so - # the loaded entry matches regardless of display casing. + model = str(raw_entry.get("model", "") or raw_entry.get("default_model", "") or "").strip() + # Index by every identity the loaded (expanded) config might present: (name), + # (name, model), (provider_key), (provider_key, model); case-insensitive names. identities = [] if name: identities.extend(((name.lower(),), (name.lower(), model))) if provider_key: - identities.extend( - ((provider_key.lower(),), (provider_key.lower(), model))) + identities.extend(((provider_key.lower(),), (provider_key.lower(), model))) if "${" in template: for identity in identities: raw_api_key_refs.setdefault(identity, template) @@ -1061,40 +800,11 @@ def _named_custom_provider_map(cfg) -> dict[str, dict[str, str]]: for identity in identities: raw_base_url_refs.setdefault(identity, base_template) - raw_list = raw_cfg.get("custom_providers") - if isinstance(raw_list, list): - for raw_entry in raw_list: - if not isinstance(raw_entry, dict): - continue - _record_raw( - raw_entry.get("name", ""), - "", - raw_entry.get("model", "") or raw_entry.get("default_model", ""), - raw_entry.get("api_key", ""), - raw_entry.get("base_url", "") - or raw_entry.get("url", "") - or raw_entry.get("api", "")) - raw_providers = raw_cfg.get("providers") - if isinstance(raw_providers, dict): - for raw_key, raw_entry in raw_providers.items(): - if not isinstance(raw_entry, dict): - continue - _record_raw( - raw_entry.get("name", "") or raw_key, - raw_key, - raw_entry.get("model", "") or raw_entry.get("default_model", ""), - raw_entry.get("api_key", ""), - raw_entry.get("base_url", "") - or raw_entry.get("url", "") - or raw_entry.get("api", "")) - - def _lookup_ref( - refs: dict[tuple, str], name: str, provider_key: str, model: str) -> str: + def _lookup_ref(refs: dict[tuple, str], name: str, provider_key: str, model: str) -> str: name_lc = str(name or "").strip().lower() pkey_lc = str(provider_key or "").strip().lower() model = str(model or "").strip() - for identity in ( - (pkey_lc, model), (pkey_lc,), (name_lc, model), (name_lc,)): + for identity in ((pkey_lc, model), (pkey_lc,), (name_lc, model), (name_lc,)): if identity[0] and identity in refs: return refs[identity] return "" @@ -1108,114 +818,85 @@ def _named_custom_provider_map(cfg) -> dict[str, dict[str, str]]: if not name or not base_url: continue provider_key = (entry.get("provider_key") or "").strip() - key = custom_provider_slug(name, provider_key) - custom_provider_map[key] = { + model = entry.get("model", "") + custom_provider_map[custom_provider_slug(name, provider_key)] = { "name": name, "base_url": base_url, "api_key": entry.get("api_key", ""), "key_env": entry.get("key_env") or entry.get("api_key_env", ""), - "model": entry.get("model", ""), + "model": model, "models": entry.get("models", {}), "models_discovered": entry.get("models_discovered", False), "extra_headers": entry.get("extra_headers", {}), "discover_models": entry.get("discover_models", True), "api_mode": entry.get("api_mode", ""), "provider_key": provider_key, - "api_key_ref": _lookup_ref( - raw_api_key_refs, name, provider_key, entry.get("model", "")), - "base_url_ref": _lookup_ref( - raw_base_url_refs, name, provider_key, entry.get("model", ""))} + "api_key_ref": _lookup_ref(raw_api_key_refs, name, provider_key, model), + "base_url_ref": _lookup_ref(raw_base_url_refs, name, provider_key, model), + } return custom_provider_map -def _build_provider_picker_rows( - config: dict, - active: str, - provider_labels: dict[str, str], - custom_provider_map: dict[str, dict[str, str]]) -> tuple[list[tuple[str, str, list[str]]], int]: +def _build_provider_picker_rows(config: dict, active: str, provider_labels: dict[str, str], + custom_provider_map: dict[str, dict[str, str]]) -> tuple[list[tuple[str, str, list[str]]], int]: """Rows for the ``hermes model`` provider picker plus the pre-selected index. - Canonical providers are folded into display groups (see PROVIDER_GROUPS in - hermes_cli/models.py): a multi-member group is one row whose ``members`` - list drives a sub-picker; leaf rows have ``members == []``. Saved custom - providers and the trailing actions stay flat. Honors - ``model_catalog.excluded_providers`` (slug or alias, case-insensitive) so - the CLI hides the same providers the gateway/TUI pickers do. + Canonical providers fold into display groups (PROVIDER_GROUPS in hermes_cli/models.py): + a multi-member group is one row whose ``members`` drive a sub-picker; leaf rows have + ``members == []``. Saved custom providers and the trailing actions stay flat. Honors + ``model_catalog.excluded_providers`` (slug or alias, case-insensitive) like the + gateway/TUI pickers. """ - from hermes_cli.models import ( - CANONICAL_PROVIDERS, _PROVIDER_ALIASES, group_providers, provider_group_for_slug) + from hermes_cli.models import CANONICAL_PROVIDERS, _PROVIDER_ALIASES, group_providers, provider_group_for_slug - _custom_provider_map = custom_provider_map canonical_descs = {p.slug: p.tui_desc for p in CANONICAL_PROVIDERS} - # Honor ``model_catalog.excluded_providers`` so the CLI ``hermes model`` - # picker hides the same providers the gateway/TUI pickers do. A canonical - # provider is hidden if its slug OR any of its aliases appears in the - # exclusion list (case-insensitive), matching list_authenticated_providers' - # matching against hermes_id / alias / canonical slug. _cli_excluded = { str(p).strip().lower() for p in (config.get("model_catalog", {}) or {}).get("excluded_providers") or [] - if p} + if p + } if _cli_excluded: - _alias_to_canon = _PROVIDER_ALIASES - _names_for: dict[str, set[str]] = {} - for _p in CANONICAL_PROVIDERS: - _names_for[_p.slug] = {_p.slug.lower()} - for _alias, _canon in _alias_to_canon.items(): + # A canonical provider is hidden if its slug OR any alias is excluded. + _names_for: dict[str, set[str]] = {_p.slug: {_p.slug.lower()} for _p in CANONICAL_PROVIDERS} + for _alias, _canon in _PROVIDER_ALIASES.items(): _names_for.setdefault(_canon, {_canon.lower()}).add(_alias.lower()) - _visible_slugs = [ - p.slug for p in CANONICAL_PROVIDERS - if not _names_for.get(p.slug, {p.slug.lower()}) & _cli_excluded] + _visible_slugs = [p.slug for p in CANONICAL_PROVIDERS if not _names_for.get(p.slug, {p.slug.lower()}) & _cli_excluded] else: _visible_slugs = [p.slug for p in CANONICAL_PROVIDERS] - grouped_rows = group_providers(_visible_slugs) - # The group/slug that should be pre-selected: the active provider's group - # if it's grouped, otherwise the active slug itself. + # The active provider's group when grouped, otherwise the active slug itself. active_group = provider_group_for_slug(active) if active else "" - # ordered entries: (key, label, members) - # members == [] → leaf row, key is a provider slug / action - # members != [] → group row, key is "group:" + # (key, label, members): members == [] → leaf row (provider slug / action); + # members != [] → group row, key is "group:" ordered: list[tuple[str, str, list[str]]] = [] default_idx = 0 - for row in grouped_rows: + + def _add(key, label, members, is_active): + nonlocal default_idx + if is_active: + label = f"{label} ← currently active" + default_idx = len(ordered) + ordered.append((key, label, members)) + + for row in group_providers(_visible_slugs): if row["kind"] == "group": gid = row["group_id"] group_desc = row.get("description", "") label = f"{row['label']} ▸ ({group_desc})" if group_desc else f"{row['label']} ▸" - key = f"group:{gid}" - is_active = bool(active_group) and gid == active_group - members = row["members"] + _add(f"group:{gid}", label, row["members"], bool(active_group) and gid == active_group) else: slug = row["slug"] - label = canonical_descs.get(slug, provider_labels.get(slug, slug)) - key = slug - is_active = bool(active) and slug == active - members = [] - if is_active: - ordered.append((key, f"{label} ← currently active", members)) - default_idx = len(ordered) - 1 - else: - ordered.append((key, label, members)) + _add(slug, canonical_descs.get(slug, provider_labels.get(slug, slug)), [], bool(active) and slug == active) - for key, provider_info in _custom_provider_map.items(): - name = provider_info["name"] - base_url = provider_info["base_url"] - short_url = base_url.replace("https://", "").replace("http://", "").rstrip("/") + for key, provider_info in custom_provider_map.items(): saved_model = provider_info.get("model", "") model_hint = f" — {saved_model}" if saved_model else "" - label = f"{name} ({short_url}){model_hint}" - if active and key == active: - ordered.append((key, f"{label} ← currently active", [])) - default_idx = len(ordered) - 1 - else: - ordered.append((key, label, [])) + _add(key, f"{provider_info['name']} ({_short_url(provider_info['base_url'])}){model_hint}", [], + bool(active) and key == active) ordered.append(("custom", "Custom endpoint (enter URL manually)", [])) - _has_saved_custom_list = isinstance(config.get("custom_providers"), list) and bool( - config.get("custom_providers")) - if _has_saved_custom_list: + if isinstance(config.get("custom_providers"), list) and config.get("custom_providers"): ordered.append(("remove-custom", "Remove a saved custom provider", [])) ordered.append(("aux-config", "Configure auxiliary models...", [])) ordered.append(("cancel", "Leave unchanged", []))