Harden setup provider flows

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Carlos
2026-04-09 13:48:36 -05:00
committed by Teknium
parent 45034b746f
commit 38ccd9eb95
8 changed files with 354 additions and 36 deletions

View File

@@ -2581,7 +2581,7 @@ def _prompt_model_selection(
custom = input("Enter model name: ").strip()
return custom if custom else None
return None
except (ImportError, NotImplementedError):
except (ImportError, NotImplementedError, OSError, subprocess.SubprocessError):
pass
# Fallback: numbered list

View File

@@ -858,7 +858,6 @@ def cmd_whatsapp(args):
def cmd_setup(args):
"""Interactive setup wizard."""
_require_tty("setup")
from hermes_cli.setup import run_setup_wizard
run_setup_wizard(args)
@@ -968,10 +967,11 @@ def select_provider_and_model(args=None):
("alibaba", "Alibaba Cloud / DashScope Coding (Qwen + multi-provider)"),
]
# Add user-defined custom providers from config.yaml
custom_providers_cfg = config.get("custom_providers") or []
_custom_provider_map = {} # key → {name, base_url, api_key}
if isinstance(custom_providers_cfg, list):
def _named_custom_provider_map(cfg) -> dict[str, dict[str, str]]:
custom_providers_cfg = cfg.get("custom_providers") or []
custom_provider_map = {}
if not isinstance(custom_providers_cfg, list):
return custom_provider_map
for entry in custom_providers_cfg:
if not isinstance(entry, dict):
continue
@@ -980,16 +980,23 @@ def select_provider_and_model(args=None):
if not name or not base_url:
continue
key = "custom:" + name.lower().replace(" ", "-")
short_url = base_url.replace("https://", "").replace("http://", "").rstrip("/")
saved_model = entry.get("model", "")
model_hint = f"{saved_model}" if saved_model else ""
top_providers.append((key, f"{name} ({short_url}){model_hint}"))
_custom_provider_map[key] = {
custom_provider_map[key] = {
"name": name,
"base_url": base_url,
"api_key": entry.get("api_key", ""),
"model": saved_model,
"model": entry.get("model", ""),
}
return custom_provider_map
# Add user-defined custom providers from config.yaml
_custom_provider_map = _named_custom_provider_map(config) # key → {name, base_url, api_key}
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("/")
saved_model = provider_info.get("model", "")
model_hint = f"{saved_model}" if saved_model else ""
top_providers.append((key, f"{name} ({short_url}){model_hint}"))
top_keys = {k for k, _ in top_providers}
extended_keys = {k for k, _ in extended_providers}
@@ -1054,8 +1061,15 @@ def select_provider_and_model(args=None):
_model_flow_copilot(config, current_model)
elif selected_provider == "custom":
_model_flow_custom(config)
elif selected_provider.startswith("custom:") and selected_provider in _custom_provider_map:
_model_flow_named_custom(config, _custom_provider_map[selected_provider])
elif selected_provider.startswith("custom:"):
provider_info = _named_custom_provider_map(load_config()).get(selected_provider)
if provider_info is None:
print(
"Warning: the selected saved custom provider is no longer available. "
"It may have been removed from config.yaml. No change."
)
return
_model_flow_named_custom(config, provider_info)
elif selected_provider == "remove-custom":
_remove_custom_provider(config)
elif selected_provider == "anthropic":
@@ -1659,7 +1673,7 @@ def _remove_custom_provider(config):
)
idx = menu.show()
print()
except (ImportError, NotImplementedError):
except (ImportError, NotImplementedError, OSError, subprocess.SubprocessError):
for i, c in enumerate(choices, 1):
print(f" {i}. {c}")
print()
@@ -1740,7 +1754,7 @@ def _model_flow_named_custom(config, provider_info):
print("Cancelled.")
return
model_name = models[idx]
except (ImportError, NotImplementedError):
except (ImportError, NotImplementedError, OSError, subprocess.SubprocessError):
for i, m in enumerate(models, 1):
print(f" {i}. {m}")
print(f" {len(models) + 1}. Cancel")
@@ -1861,7 +1875,7 @@ def _prompt_reasoning_effort_selection(efforts, current_effort=""):
if idx == len(ordered):
return "none"
return None
except (ImportError, NotImplementedError):
except (ImportError, NotImplementedError, OSError, subprocess.SubprocessError):
pass
print("Select reasoning effort:")
@@ -4472,12 +4486,12 @@ For more help on a command:
"setup",
help="Interactive setup wizard",
description="Configure Hermes Agent with an interactive wizard. "
"Run a specific section: hermes setup model|terminal|gateway|tools|agent"
"Run a specific section: hermes setup model|tts|terminal|gateway|tools|agent"
)
setup_parser.add_argument(
"section",
nargs="?",
choices=["model", "terminal", "gateway", "tools", "agent"],
choices=["model", "tts", "terminal", "gateway", "tools", "agent"],
default=None,
help="Run a specific setup section instead of the full wizard"
)

View File

@@ -16,6 +16,7 @@ import logging
import os
import shutil
import sys
import copy
from pathlib import Path
from typing import Optional, Dict, Any
@@ -316,6 +317,7 @@ def _setup_provider_model_selection(config, provider_id, current_model, prompt_c
# Import config helpers
from hermes_cli.config import (
DEFAULT_CONFIG,
get_hermes_home,
get_config_path,
get_env_path,
@@ -921,8 +923,10 @@ def setup_model_provider(config: dict, *, quick: bool = False):
# changes with stale values (#4172).
_refreshed = load_config()
config["model"] = _refreshed.get("model", config.get("model"))
if _refreshed.get("custom_providers"):
if "custom_providers" in _refreshed:
config["custom_providers"] = _refreshed["custom_providers"]
else:
config.pop("custom_providers", None)
# Derive the selected provider for downstream steps (vision setup).
selected_provider = None
@@ -1006,8 +1010,6 @@ def setup_model_provider(config: dict, *, quick: bool = False):
strategy_value = ["fill_first", "round_robin", "random"][strategy_idx]
_set_credential_pool_strategy(config, selected_provider, strategy_value)
print_success(f"Saved {selected_provider} rotation strategy: {strategy_value}")
else:
_set_credential_pool_strategy(config, selected_provider, "fill_first")
except Exception as exc:
logger.debug("Could not configure same-provider fallback in setup: %s", exc)
@@ -2844,6 +2846,7 @@ def run_setup_wizard(args):
Supports full, quick, and section-specific setup:
hermes setup — full or quick (auto-detected)
hermes setup model — just model/provider
hermes setup tts — just text-to-speech
hermes setup terminal — just terminal backend
hermes setup gateway — just messaging platforms
hermes setup tools — just tool configuration
@@ -2855,6 +2858,11 @@ def run_setup_wizard(args):
return
ensure_hermes_home()
reset_requested = bool(getattr(args, "reset", False))
if reset_requested:
save_config(copy.deepcopy(DEFAULT_CONFIG))
print_success("Configuration reset to defaults.")
config = load_config()
hermes_home = get_hermes_home()
@@ -2955,18 +2963,13 @@ def run_setup_wizard(args):
menu_choices = [
"Quick Setup - configure missing items only",
"Full Setup - reconfigure everything",
"---",
"Model & Provider",
"Terminal Backend",
"Messaging Platforms (Gateway)",
"Tools",
"Agent Settings",
"---",
"Exit",
]
# Separator indices (not selectable, but prompt_choice doesn't filter them,
# so we handle them below)
choice = prompt_choice("What would you like to do?", menu_choices, 0)
if choice == 0:
@@ -2976,18 +2979,14 @@ def run_setup_wizard(args):
elif choice == 1:
# Full setup — fall through to run all sections
pass
elif choice in (2, 8):
# Separator — treat as exit
elif choice == 7:
print_info("Exiting. Run 'hermes setup' again when ready.")
return
elif choice == 9:
print_info("Exiting. Run 'hermes setup' again when ready.")
return
elif 3 <= choice <= 7:
elif 2 <= choice <= 6:
# Individual section — map by key, not by position.
# SETUP_SECTIONS includes TTS but the returning-user menu skips it,
# so positional indexing (choice - 3) would dispatch the wrong section.
section_key = RETURNING_USER_MENU_SECTION_KEYS[choice - 3]
# so positional indexing (choice - 2) would dispatch the wrong section.
section_key = RETURNING_USER_MENU_SECTION_KEYS[choice - 2]
section = next((s for s in SETUP_SECTIONS if s[0] == section_key), None)
if section:
_, label, func = section

View File

@@ -633,6 +633,7 @@ class TestHasAnyProviderConfigured:
hermes_home.mkdir()
monkeypatch.setattr(config_module, "get_env_path", lambda: hermes_home / ".env")
monkeypatch.setattr(config_module, "get_hermes_home", lambda: hermes_home)
monkeypatch.setattr("hermes_cli.copilot_auth.resolve_copilot_token", lambda: ("", ""))
# Clear all provider env vars so earlier checks don't short-circuit
_all_vars = {"OPENROUTER_API_KEY", "OPENAI_API_KEY", "ANTHROPIC_API_KEY",
"ANTHROPIC_TOKEN", "OPENAI_BASE_URL"}
@@ -727,6 +728,7 @@ class TestHasAnyProviderConfigured:
monkeypatch.setattr(config_module, "get_env_path", lambda: hermes_home / ".env")
monkeypatch.setattr(config_module, "get_hermes_home", lambda: hermes_home)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr("hermes_cli.copilot_auth.resolve_copilot_token", lambda: ("", ""))
_all_vars = {"OPENROUTER_API_KEY", "OPENAI_API_KEY", "ANTHROPIC_API_KEY",
"ANTHROPIC_TOKEN", "OPENAI_BASE_URL"}
for pconfig in PROVIDER_REGISTRY.values():

View File

@@ -142,6 +142,31 @@ def test_setup_custom_providers_synced(tmp_path, monkeypatch):
assert reloaded.get("custom_providers") == [{"name": "Local", "base_url": "http://localhost:8080/v1"}]
def test_setup_syncs_custom_provider_removal_from_disk(tmp_path, monkeypatch):
"""Removing the last custom provider in model setup should persist."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
config = load_config()
config["custom_providers"] = [{"name": "Local", "base_url": "http://localhost:8080/v1"}]
save_config(config)
def fake_select():
cfg = load_config()
cfg["model"] = {"provider": "openrouter", "default": "anthropic/claude-opus-4.6"}
cfg["custom_providers"] = []
save_config(cfg)
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert reloaded.get("custom_providers") == []
def test_setup_cancel_preserves_existing_config(tmp_path, monkeypatch):
"""When the user cancels provider selection, existing config is preserved."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
@@ -201,6 +226,38 @@ def test_setup_keyboard_interrupt_gracefully_handled(tmp_path, monkeypatch):
setup_model_provider(config)
def test_select_provider_and_model_warns_if_named_custom_provider_disappears(
tmp_path, monkeypatch, capsys
):
"""If a saved custom provider is deleted mid-selection, show a warning instead of silently doing nothing."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
cfg = load_config()
cfg["custom_providers"] = [{"name": "Local", "base_url": "http://localhost:8080/v1"}]
save_config(cfg)
def fake_prompt_provider_choice(choices, default=0):
current = load_config()
current["custom_providers"] = []
save_config(current)
return next(i for i, label in enumerate(choices) if label.startswith("Local (localhost:8080/v1)"))
monkeypatch.setattr("hermes_cli.auth.resolve_provider", lambda provider: None)
monkeypatch.setattr("hermes_cli.main._prompt_provider_choice", fake_prompt_provider_choice)
monkeypatch.setattr(
"hermes_cli.main._model_flow_named_custom",
lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("named custom flow should not run")),
)
from hermes_cli.main import select_provider_and_model
select_provider_and_model()
out = capsys.readouterr().out
assert "selected saved custom provider is no longer available" in out
def test_codex_setup_uses_runtime_access_token_for_live_model_list(tmp_path, monkeypatch):
"""Codex model list fetching uses the runtime access token."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))

View File

@@ -230,6 +230,39 @@ def test_setup_same_provider_fallback_can_add_another_credential(tmp_path, monke
assert config.get("credential_pool_strategies", {}).get("openrouter") == "fill_first"
def test_setup_same_provider_single_credential_keeps_existing_rotation_strategy(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
save_env_value("OPENROUTER_API_KEY", "or-key")
_write_model_config("openrouter", "", "anthropic/claude-opus-4.6")
config = load_config()
config["credential_pool_strategies"] = {"openrouter": "round_robin"}
save_config(config)
class _Entry:
def __init__(self, label):
self.label = label
class _Pool:
def entries(self):
return [_Entry("primary")]
def fake_select():
pass
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
_stub_tts(monkeypatch)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("agent.credential_pool.load_pool", lambda provider: _Pool())
monkeypatch.setattr("agent.auxiliary_client.get_available_vision_backends", lambda: [])
setup_model_provider(config)
assert config.get("credential_pool_strategies", {}).get("openrouter") == "round_robin"
def test_setup_pool_step_shows_manual_vs_auto_detected_counts(tmp_path, monkeypatch, capsys):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)

View File

@@ -4,6 +4,7 @@ from argparse import Namespace
from unittest.mock import MagicMock, patch
import pytest
from hermes_cli.config import DEFAULT_CONFIG, load_config, save_config
def _make_setup_args(**overrides):
@@ -34,6 +35,36 @@ def _make_chat_args(**overrides):
class TestNonInteractiveSetup:
"""Verify setup paths exit cleanly in headless/non-interactive environments."""
def test_cmd_setup_allows_noninteractive_flag_without_tty(self):
"""The CLI entrypoint should not block --non-interactive before setup.py handles it."""
from hermes_cli.main import cmd_setup
args = _make_setup_args(non_interactive=True)
with (
patch("hermes_cli.setup.run_setup_wizard") as mock_run_setup,
patch("sys.stdin") as mock_stdin,
):
mock_stdin.isatty.return_value = False
cmd_setup(args)
mock_run_setup.assert_called_once_with(args)
def test_cmd_setup_defers_no_tty_handling_to_setup_wizard(self):
"""Bare `hermes setup` should reach setup.py, which prints headless guidance."""
from hermes_cli.main import cmd_setup
args = _make_setup_args(non_interactive=False)
with (
patch("hermes_cli.setup.run_setup_wizard") as mock_run_setup,
patch("sys.stdin") as mock_stdin,
):
mock_stdin.isatty.return_value = False
cmd_setup(args)
mock_run_setup.assert_called_once_with(args)
def test_non_interactive_flag_skips_wizard(self, capsys):
"""--non-interactive should print guidance and not enter the wizard."""
from hermes_cli.setup import run_setup_wizard
@@ -72,6 +103,26 @@ class TestNonInteractiveSetup:
out = capsys.readouterr().out
assert "hermes config set model.provider custom" in out
def test_reset_flag_rewrites_config_before_noninteractive_exit(self, tmp_path, monkeypatch, capsys):
"""--reset should rewrite config.yaml even when the wizard cannot run interactively."""
from hermes_cli.setup import run_setup_wizard
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
cfg = load_config()
cfg["model"] = {"provider": "custom", "base_url": "http://localhost:8080/v1", "default": "llama3"}
cfg["agent"]["max_turns"] = 12
save_config(cfg)
args = _make_setup_args(non_interactive=True, reset=True)
run_setup_wizard(args)
reloaded = load_config()
assert reloaded["model"] == DEFAULT_CONFIG["model"]
assert reloaded["agent"]["max_turns"] == DEFAULT_CONFIG["agent"]["max_turns"]
out = capsys.readouterr().out
assert "Configuration reset to defaults." in out
def test_chat_first_run_headless_skips_setup_prompt(self, capsys):
"""Bare `hermes` should not prompt for input when no provider exists and stdin is headless."""
from hermes_cli.main import cmd_chat
@@ -117,7 +168,7 @@ class TestNonInteractiveSetup:
side_effect=lambda key: "sk-test" if key == "OPENROUTER_API_KEY" else "",
),
patch("hermes_cli.auth.get_active_provider", return_value=None),
patch.object(setup_mod, "prompt_choice", return_value=4),
patch.object(setup_mod, "prompt_choice", return_value=3),
patch.object(
setup_mod,
"SETUP_SECTIONS",
@@ -137,3 +188,59 @@ class TestNonInteractiveSetup:
terminal_section.assert_called_once_with(config)
tts_section.assert_not_called()
def test_returning_user_menu_does_not_show_separator_rows(self, tmp_path):
"""Returning-user menu should only show selectable actions."""
from hermes_cli import setup as setup_mod
args = _make_setup_args()
captured = {}
def fake_prompt_choice(question, choices, default=0):
captured["question"] = question
captured["choices"] = list(choices)
return len(choices) - 1
with (
patch.object(setup_mod, "ensure_hermes_home"),
patch.object(setup_mod, "load_config", return_value={}),
patch.object(setup_mod, "get_hermes_home", return_value=tmp_path),
patch.object(setup_mod, "is_interactive_stdin", return_value=True),
patch.object(
setup_mod,
"get_env_value",
side_effect=lambda key: "sk-test" if key == "OPENROUTER_API_KEY" else "",
),
patch("hermes_cli.auth.get_active_provider", return_value=None),
patch.object(setup_mod, "prompt_choice", side_effect=fake_prompt_choice),
):
setup_mod.run_setup_wizard(args)
assert captured["question"] == "What would you like to do?"
assert "---" not in captured["choices"]
assert captured["choices"] == [
"Quick Setup - configure missing items only",
"Full Setup - reconfigure everything",
"Model & Provider",
"Terminal Backend",
"Messaging Platforms (Gateway)",
"Tools",
"Agent Settings",
"Exit",
]
def test_main_accepts_tts_setup_section(self, monkeypatch):
"""`hermes setup tts` should parse and dispatch like other setup sections."""
from hermes_cli import main as main_mod
received = {}
def fake_cmd_setup(args):
received["section"] = args.section
monkeypatch.setattr(main_mod, "cmd_setup", fake_cmd_setup)
monkeypatch.setattr("sys.argv", ["hermes", "setup", "tts"])
main_mod.main()
assert received["section"] == "tts"

View File

@@ -0,0 +1,106 @@
"""Regression tests for numbered fallbacks when TerminalMenu cannot initialize."""
import subprocess
import sys
import types
from hermes_cli.config import load_config, save_config
class _BrokenTerminalMenu:
def __init__(self, *args, **kwargs):
raise subprocess.CalledProcessError(2, ["tput", "clear"])
def test_prompt_model_selection_falls_back_on_terminalmenu_runtime_error(monkeypatch):
from hermes_cli.auth import _prompt_model_selection
monkeypatch.setitem(
sys.modules,
"simple_term_menu",
types.SimpleNamespace(TerminalMenu=_BrokenTerminalMenu),
)
responses = iter(["2"])
monkeypatch.setattr("builtins.input", lambda _prompt="": next(responses))
selected = _prompt_model_selection(["model-a", "model-b"])
assert selected == "model-b"
def test_prompt_reasoning_effort_falls_back_on_terminalmenu_runtime_error(monkeypatch):
from hermes_cli.main import _prompt_reasoning_effort_selection
monkeypatch.setitem(
sys.modules,
"simple_term_menu",
types.SimpleNamespace(TerminalMenu=_BrokenTerminalMenu),
)
responses = iter(["3"])
monkeypatch.setattr("builtins.input", lambda _prompt="": next(responses))
selected = _prompt_reasoning_effort_selection(["low", "medium", "high"], current_effort="")
assert selected == "high"
def test_remove_custom_provider_falls_back_on_terminalmenu_runtime_error(tmp_path, monkeypatch):
from hermes_cli.main import _remove_custom_provider
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setitem(
sys.modules,
"simple_term_menu",
types.SimpleNamespace(TerminalMenu=_BrokenTerminalMenu),
)
cfg = load_config()
cfg["custom_providers"] = [
{"name": "Local A", "base_url": "http://localhost:8001/v1"},
{"name": "Local B", "base_url": "http://localhost:8002/v1"},
]
save_config(cfg)
responses = iter(["1"])
monkeypatch.setattr("builtins.input", lambda _prompt="": next(responses))
_remove_custom_provider(cfg)
reloaded = load_config()
assert reloaded["custom_providers"] == [
{"name": "Local B", "base_url": "http://localhost:8002/v1"},
]
def test_named_custom_provider_model_picker_falls_back_on_terminalmenu_runtime_error(tmp_path, monkeypatch):
from hermes_cli.main import _model_flow_named_custom
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setitem(
sys.modules,
"simple_term_menu",
types.SimpleNamespace(TerminalMenu=_BrokenTerminalMenu),
)
monkeypatch.setattr("hermes_cli.models.fetch_api_models", lambda *args, **kwargs: ["model-a", "model-b"])
monkeypatch.setattr("hermes_cli.auth.deactivate_provider", lambda: None)
cfg = load_config()
save_config(cfg)
responses = iter(["2"])
monkeypatch.setattr("builtins.input", lambda _prompt="": next(responses))
_model_flow_named_custom(
cfg,
{
"name": "Local",
"base_url": "http://localhost:8000/v1",
"api_key": "",
"model": "",
},
)
reloaded = load_config()
assert reloaded["model"]["provider"] == "custom"
assert reloaded["model"]["base_url"] == "http://localhost:8000/v1"
assert reloaded["model"]["default"] == "model-b"