From 0d32adbbb5f7dca0ee24edcbd9d90c1da7f4c155 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 16:11:38 -0700 Subject: [PATCH] refactor(hermes_cli/main): extract provider-setup helpers to main_provider_setup.py --- hermes_cli/main.py | 1109 +---------------------------- hermes_cli/main_provider_setup.py | 1100 ++++++++++++++++++++++++++++ 2 files changed, 1132 insertions(+), 1077 deletions(-) create mode 100644 hermes_cli/main_provider_setup.py diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 8b9f3e2f2f..eb6a4a8f3d 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -873,6 +873,38 @@ from hermes_cli.model_setup_flows import ( _model_flow_ai_gateway, ) logger = logging.getLogger(__name__) +from hermes_cli.main_provider_setup import ( # noqa: E402,F401 (re-exported; tests patch hermes_cli.main.) + _AUX_TASKS, + _DEFAULT_QWEN_PORTAL_MODELS, + _DELEGATION_TASK_DESC, + _DELEGATION_TASK_KEY, + _DELEGATION_TASK_NAME, + _GENERIC_API_KEY_PROVIDERS, + _all_aux_tasks, + _auto_provider_name, + _aux_config_menu, + _aux_flow_custom_endpoint, + _aux_flow_provider_model, + _aux_select_for_task, + _aux_task_display_name, + _clear_stale_openai_base_url, + _custom_provider_api_key_config_value, + _custom_provider_base_url_config_value, + _delegation_cfg_as_task, + _format_aux_current, + _infer_stepfun_region, + _is_profile_api_key_provider, + _prompt_api_key, + _prompt_custom_api_mode_selection, + _prompt_provider_choice, + _prompt_reasoning_effort_selection, + _remove_custom_provider, + _reset_aux_to_auto, + _run_anthropic_oauth_flow, + _save_aux_choice, + _save_custom_provider, + _stepfun_base_url_for_region, +) from hermes_cli.main_install_repair import ( # noqa: E402,F401 (re-exported; tests patch hermes_cli.main.) ShimQuarantineError, _LAZY_REFRESH_IMPORT_PROBES, @@ -2387,21 +2419,6 @@ def cmd_model(args): ) -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. - """ - try: - from providers import get_provider_profile - _p = get_provider_profile(provider_id) - return _p is not None and _p.auth_type == "api_key" - except Exception: - return False - - # Provider id -> flow(config, current_model, args). Lambdas resolve the # _model_flow_* names at call time so test monkeypatches on hermes_cli.main # keep intercepting. Custom-slug providers (always ``custom:*``), @@ -2427,13 +2444,6 @@ _PROVIDER_MODEL_FLOWS = { "azure-foundry": lambda c, m, a: _model_flow_azure_foundry(c, m), } -_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", -}) - def select_provider_and_model(args=None): """Core provider selection + model picking logic. @@ -2841,36 +2851,6 @@ def select_provider_and_model(args=None): _clear_stale_openai_base_url() -def _clear_stale_openai_base_url(): - """Remove OPENAI_BASE_URL from ~/.hermes/.env if the active provider is not '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. - """ - 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 = "" - - 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})" - ) - - # ───────────────────────────────────────────────────────────────────────────── # Auxiliary model configuration # @@ -2884,743 +2864,6 @@ def _clear_stale_openai_base_url(): # configure new providers through the normal `hermes model` flow first. # ───────────────────────────────────────────────────────────────────────────── -# (task_key, display_name, short_description) -_AUX_TASKS: list[tuple[str, str, str]] = [ - ("vision", "Vision", "image/screenshot analysis"), - ("compression", "Compression", "context summarization"), - ("approval", "Approval", "smart command approval"), - ("mcp", "MCP", "MCP tool reasoning"), - ("title_generation", "Title generation", "session titles"), - ("review", "Review", "/review reviewer subagent"), - ("memory_query_rewrite", "Memory query rewrite", "memory retrieval queries"), - ("tts_audio_tags", "TTS audio tags", "Gemini TTS tag insertion"), - ("skills_hub", "Skills hub", "skills search/install"), - ("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"), -] - -# 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. -_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. - """ - tasks = list(_AUX_TASKS) - try: - from hermes_cli.plugins import get_plugin_auxiliary_tasks - 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 - return tasks - - -def _format_aux_current(task_cfg: dict) -> str: - """Render the current aux config for display in the task menu.""" - if not isinstance(task_cfg, dict): - return "auto" - base_url = str(task_cfg.get("base_url") or "").strip() - 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 "") - if provider == "auto": - return "auto" + (f" · {model}" if model else "") - if model: - return f"{provider} · {model}" - return provider - - -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". - """ - 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(), - } - - -def _aux_task_display_name(task: str) -> str: - """Display name for a task key, covering the special delegation entry.""" - if task == _DELEGATION_TASK_KEY: - return _DELEGATION_TASK_NAME - 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. - """ - 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["provider"] = provider - entry["model"] = model or "" - entry["base_url"] = base_url or "" - entry["api_key"] = api_key or "" - save_config(cfg) - - -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. - """ - 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 - count = 0 - for task, _name, _desc in _all_aux_tasks(): - entry = aux.setdefault(task, {}) - if not isinstance(entry, dict): - entry = {} - aux[task] = entry - changed = False - if entry.get("provider") not in {None, "", "auto"}: - entry["provider"] = "auto" - changed = True - for field in ("model", "base_url", "api_key"): - if entry.get(field): - entry[field] = "" - changed = True - # 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. - dele = cfg.get("delegation") - if isinstance(dele, dict): - changed = False - for field in ("provider", "model", "base_url", "api_key"): - if dele.get(field): - dele[field] = "" - changed = True - if changed: - count += 1 - save_config(cfg) - return count - - -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. - """ - from hermes_cli.config import load_config - - while True: - cfg = load_config() - aux = cfg.get("auxiliary", {}) if isinstance(cfg.get("auxiliary"), dict) else {} - - 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) - ] - 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.append(("__reset__", "Reset all to auto")) - entries.append(("__back__", "Back")) - - idx = _prompt_provider_choice( - [label for _, label in entries], - default=0, - ) - if idx is None: - return - key = entries[idx][0] - if key == "__back__": - 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() - 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. - - 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. - """ - 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 {} - 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, - ) - 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) - 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() - - 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__": - _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) - - -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. - 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() - return - selected = val or "" - else: - 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="" - ) - if selected: - print(f"{display_name}: {provider_slug} · {selected}") - else: - print(f"{display_name}: {provider_slug} (provider default 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() - 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() - 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() - 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 "")) - - -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. - """ - try: - from hermes_cli.setup import _curses_prompt_choice - - idx = _curses_prompt_choice(title, choices, default) - if idx >= 0: - print() - return idx - 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() - 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 - - -_DEFAULT_QWEN_PORTAL_MODELS = [ - "qwen3-coder-plus", - "qwen3-coder", -] - - -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. - """ - 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 "" - - 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") - suffix = f" [{' / '.join(markers)}]" if markers else "" - print(f" {idx}. {label}{suffix}") - print(f" {description}") - - try: - 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" - - 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. - """ - import re - - clean = base_url.replace("https://", "").replace("http://", "").rstrip("/") - clean = re.sub(r"/v1/?$", "", clean) - name = clean.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 - - -def _custom_provider_api_key_config_value(provider_info, resolved_api_key=""): - """Return the value that should be persisted for a custom provider 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() - - -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). - """ - from hermes_cli.config import load_config, save_config - - cfg = load_config() - providers = cfg.get("custom_providers") or [] - 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("/"): - changed = False - if model and entry.get("model") != model: - entry["model"] = model - changed = True - if model and context_length: - models_cfg = entry.get("models", {}) - if not isinstance(models_cfg, dict): - models_cfg = {} - models_cfg[model] = {"context_length": context_length} - entry["models"] = models_cfg - changed = True - if api_mode: - if entry.get("api_mode") != api_mode: - entry["api_mode"] = api_mode - changed = True - elif "api_mode" in entry: - entry.pop("api_mode", None) - changed = True - if key_env and (entry.get("key_env") != key_env or entry.get("api_key")): - entry["key_env"] = key_env - entry.pop("api_key", None) - changed = True - if changed: - cfg["custom_providers"] = providers - 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 - elif api_key: - entry["api_key"] = api_key - if model: - entry["model"] = model - if api_mode: - entry["api_mode"] = api_mode - if model and context_length: - entry["models"] = {model: {"context_length": context_length}} - - providers.append(entry) - cfg["custom_providers"] = providers - save_config(cfg) - print(f' 💾 Saved to custom providers as "{name}" (edit in config.yaml)') - - -def _remove_custom_provider(config): - """Let the user remove a saved custom provider from config.yaml.""" - from hermes_cli.config import load_config, save_config - - cfg = load_config() - providers = cfg.get("custom_providers") or [] - if not isinstance(providers, list) or not providers: - print("No custom providers configured.") - 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.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, - ) - print() - if idx < 0: - idx = None - except (ImportError, NotImplementedError, OSError, subprocess.SubprocessError): - for i, c in enumerate(choices, 1): - print(f" {i}. {c}") - print() - try: - val = input(f"Choice [1-{len(choices)}]: ").strip() - idx = int(val) - 1 if val else None - except (ValueError, KeyboardInterrupt, EOFError): - idx = None - - if idx is None or idx >= len(providers): - print("No change.") - return - - removed = providers.pop(idx) - cfg["custom_providers"] = providers - save_config(cfg) - removed_name = ( - removed.get("name", "unnamed") if isinstance(removed, dict) else str(removed) - ) - print(f'✅ Removed "{removed_name}" from custom providers.') - # Lazy-export the model catalog at module level. Tests and a handful of # downstream call sites read `hermes_cli.main._PROVIDER_MODELS` directly, @@ -3821,294 +3064,6 @@ def __getattr__(name): raise AttributeError(f"module {__name__!r} has no attribute {name!r}") -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() - ) - ) - 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) - if not ordered: - return None - - def _label(effort): - if effort == current_effort: - return f"{effort} ← currently in use" - return effort - - disable_label = "Disable reasoning" - skip_label = "Skip (keep current)" - - if current_effort == "none": - default_idx = len(ordered) - elif current_effort in ordered: - default_idx = ordered.index(current_effort) - elif "medium" in ordered: - default_idx = ordered.index("medium") - else: - default_idx = 0 - - 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, - ) - 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): - return None - - -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. - """ - 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: - 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() - return "" - if not entered and provider_id == "lmstudio" and allow_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: - return "", True - new_key = _prompt_new_key(allow_lmstudio_default=True) - if not new_key: - print("Cancelled.") - return "", True - save_env_value(key_env, new_key) - print("API key saved.") - print() - return new_key, False - - # 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 "" - print(f" {pconfig.name} API key: {existing_key[:8]}... ✓{source_suffix}") - if not key_env: - # Nothing we can rewrite; just acknowledge and move on. - 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" - - if choice.startswith("r"): - new_key = _prompt_new_key(allow_lmstudio_default=False) - if not new_key: - print(" No change.") - print() - return existing_key, False - save_env_value(key_env, new_key) - print(" API key updated.") - print() - 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." - ) - return "", True - - # Keep (default, or any other input) - print() - return existing_key, False - - -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" - - -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, - ) - - 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, - ) - - 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")) - ): - 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." - ) - return True - return False - - try: - print() - print(" Running 'claude setup-token' — follow the prompts below.") - print(" A browser window will open for you to authorize access.") - print() - token = run_oauth_setup_token() - if token: - if _activate_claude_code_credentials_if_available(): - return True - save_anthropic_oauth_token(token, save_fn=save_env_value) - print(" ✓ OAuth credentials saved.") - 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() - return False - if manual_token: - save_anthropic_oauth_token(manual_token, save_fn=save_env_value) - print(" ✓ Setup-token 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() - return False - if token: - save_anthropic_oauth_token(token, save_fn=save_env_value) - print(" ✓ Setup-token saved.") - return True - print(" Cancelled — install Claude Code and try again.") - return False - - def cmd_login(args): """Authenticate Hermes CLI with a provider.""" from hermes_cli.auth import login_command diff --git a/hermes_cli/main_provider_setup.py b/hermes_cli/main_provider_setup.py new file mode 100644 index 0000000000..8be7e431a7 --- /dev/null +++ b/hermes_cli/main_provider_setup.py @@ -0,0 +1,1100 @@ +"""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). +""" + +import subprocess + +from typing import Optional +from hermes_cli.cli_output import line_input + + +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. + """ + try: + from providers import get_provider_profile + _p = get_provider_profile(provider_id) + return _p is not None and _p.auth_type == "api_key" + except Exception: + return False + + +_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", +}) + + +def _clear_stale_openai_base_url(): + """Remove OPENAI_BASE_URL from ~/.hermes/.env if the active provider is not '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. + """ + 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 = "" + + 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})" + ) + + +# (task_key, display_name, short_description) +_AUX_TASKS: list[tuple[str, str, str]] = [ + ("vision", "Vision", "image/screenshot analysis"), + ("compression", "Compression", "context summarization"), + ("approval", "Approval", "smart command approval"), + ("mcp", "MCP", "MCP tool reasoning"), + ("title_generation", "Title generation", "session titles"), + ("review", "Review", "/review reviewer subagent"), + ("memory_query_rewrite", "Memory query rewrite", "memory retrieval queries"), + ("tts_audio_tags", "TTS audio tags", "Gemini TTS tag insertion"), + ("skills_hub", "Skills hub", "skills search/install"), + ("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"), +] + + +# 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. +_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. + """ + from hermes_cli.main import _AUX_TASKS + tasks = list(_AUX_TASKS) + try: + from hermes_cli.plugins import get_plugin_auxiliary_tasks + 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 + return tasks + + +def _format_aux_current(task_cfg: dict) -> str: + """Render the current aux config for display in the task menu.""" + if not isinstance(task_cfg, dict): + return "auto" + base_url = str(task_cfg.get("base_url") or "").strip() + 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 "") + if provider == "auto": + return "auto" + (f" · {model}" if model else "") + if model: + return f"{provider} · {model}" + return provider + + +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". + """ + 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(), + } + + +def _aux_task_display_name(task: str) -> str: + """Display name for a task key, covering the special delegation entry.""" + if task == _DELEGATION_TASK_KEY: + return _DELEGATION_TASK_NAME + 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. + """ + 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["provider"] = provider + entry["model"] = model or "" + entry["base_url"] = base_url or "" + entry["api_key"] = api_key or "" + save_config(cfg) + + +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. + """ + 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 + count = 0 + for task, _name, _desc in _all_aux_tasks(): + entry = aux.setdefault(task, {}) + if not isinstance(entry, dict): + entry = {} + aux[task] = entry + changed = False + if entry.get("provider") not in {None, "", "auto"}: + entry["provider"] = "auto" + changed = True + for field in ("model", "base_url", "api_key"): + if entry.get(field): + entry[field] = "" + changed = True + # 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. + dele = cfg.get("delegation") + if isinstance(dele, dict): + changed = False + for field in ("provider", "model", "base_url", "api_key"): + if dele.get(field): + dele[field] = "" + changed = True + if changed: + count += 1 + save_config(cfg) + return count + + +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. + """ + 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 {} + + 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) + ] + 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.append(("__reset__", "Reset all to auto")) + entries.append(("__back__", "Back")) + + idx = _prompt_provider_choice( + [label for _, label in entries], + default=0, + ) + if idx is None: + return + key = entries[idx][0] + if key == "__back__": + 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() + 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. + + 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. + """ + 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 {} + 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, + ) + 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) + 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() + + 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__": + _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) + + +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. + 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() + return + selected = val or "" + else: + 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="" + ) + if selected: + print(f"{display_name}: {provider_slug} · {selected}") + else: + print(f"{display_name}: {provider_slug} (provider default 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() + 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() + 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() + 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 "")) + + +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. + """ + try: + from hermes_cli.setup import _curses_prompt_choice + + idx = _curses_prompt_choice(title, choices, default) + if idx >= 0: + print() + return idx + 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() + 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 + + +_DEFAULT_QWEN_PORTAL_MODELS = [ + "qwen3-coder-plus", + "qwen3-coder", +] + + +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. + """ + 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 "" + + 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") + suffix = f" [{' / '.join(markers)}]" if markers else "" + print(f" {idx}. {label}{suffix}") + print(f" {description}") + + try: + 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" + + 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. + """ + import re + + clean = base_url.replace("https://", "").replace("http://", "").rstrip("/") + clean = re.sub(r"/v1/?$", "", clean) + name = clean.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 + + +def _custom_provider_api_key_config_value(provider_info, resolved_api_key=""): + """Return the value that should be persisted for a custom provider 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() + + +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). + """ + from hermes_cli.config import load_config, save_config + + cfg = load_config() + providers = cfg.get("custom_providers") or [] + 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("/"): + changed = False + if model and entry.get("model") != model: + entry["model"] = model + changed = True + if model and context_length: + models_cfg = entry.get("models", {}) + if not isinstance(models_cfg, dict): + models_cfg = {} + models_cfg[model] = {"context_length": context_length} + entry["models"] = models_cfg + changed = True + if api_mode: + if entry.get("api_mode") != api_mode: + entry["api_mode"] = api_mode + changed = True + elif "api_mode" in entry: + entry.pop("api_mode", None) + changed = True + if key_env and (entry.get("key_env") != key_env or entry.get("api_key")): + entry["key_env"] = key_env + entry.pop("api_key", None) + changed = True + if changed: + cfg["custom_providers"] = providers + 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 + elif api_key: + entry["api_key"] = api_key + if model: + entry["model"] = model + if api_mode: + entry["api_mode"] = api_mode + if model and context_length: + entry["models"] = {model: {"context_length": context_length}} + + providers.append(entry) + cfg["custom_providers"] = providers + save_config(cfg) + print(f' 💾 Saved to custom providers as "{name}" (edit in config.yaml)') + + +def _remove_custom_provider(config): + """Let the user remove a saved custom provider from config.yaml.""" + from hermes_cli.config import load_config, save_config + + cfg = load_config() + providers = cfg.get("custom_providers") or [] + if not isinstance(providers, list) or not providers: + print("No custom providers configured.") + 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.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, + ) + print() + if idx < 0: + idx = None + except (ImportError, NotImplementedError, OSError, subprocess.SubprocessError): + for i, c in enumerate(choices, 1): + print(f" {i}. {c}") + print() + try: + val = input(f"Choice [1-{len(choices)}]: ").strip() + idx = int(val) - 1 if val else None + except (ValueError, KeyboardInterrupt, EOFError): + idx = None + + if idx is None or idx >= len(providers): + print("No change.") + return + + removed = providers.pop(idx) + cfg["custom_providers"] = providers + save_config(cfg) + 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() + ) + ) + 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) + if not ordered: + return None + + def _label(effort): + if effort == current_effort: + return f"{effort} ← currently in use" + return effort + + disable_label = "Disable reasoning" + skip_label = "Skip (keep current)" + + if current_effort == "none": + default_idx = len(ordered) + elif current_effort in ordered: + default_idx = ordered.index(current_effort) + elif "medium" in ordered: + default_idx = ordered.index("medium") + else: + default_idx = 0 + + 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, + ) + 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): + return None + + +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. + """ + 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: + 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() + return "" + if not entered and provider_id == "lmstudio" and allow_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: + return "", True + new_key = _prompt_new_key(allow_lmstudio_default=True) + if not new_key: + print("Cancelled.") + return "", True + save_env_value(key_env, new_key) + print("API key saved.") + print() + return new_key, False + + # 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 "" + print(f" {pconfig.name} API key: {existing_key[:8]}... ✓{source_suffix}") + if not key_env: + # Nothing we can rewrite; just acknowledge and move on. + 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" + + if choice.startswith("r"): + new_key = _prompt_new_key(allow_lmstudio_default=False) + if not new_key: + print(" No change.") + print() + return existing_key, False + save_env_value(key_env, new_key) + print(" API key updated.") + print() + 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." + ) + return "", True + + # Keep (default, or any other input) + print() + return existing_key, False + + +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" + + +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, + ) + + 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, + ) + + 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")) + ): + 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." + ) + return True + return False + + try: + print() + print(" Running 'claude setup-token' — follow the prompts below.") + print(" A browser window will open for you to authorize access.") + print() + token = run_oauth_setup_token() + if token: + if _activate_claude_code_credentials_if_available(): + return True + save_anthropic_oauth_token(token, save_fn=save_env_value) + print(" ✓ OAuth credentials saved.") + 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() + return False + if manual_token: + save_anthropic_oauth_token(manual_token, save_fn=save_env_value) + print(" ✓ Setup-token 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() + return False + if token: + save_anthropic_oauth_token(token, save_fn=save_env_value) + print(" ✓ Setup-token saved.") + return True + print(" Cancelled — install Claude Code and try again.") + return False