diff --git a/hermes_cli/setup.py b/hermes_cli/setup.py index 6d0ec0f4594..d99ed06340f 100644 --- a/hermes_cli/setup.py +++ b/hermes_cli/setup.py @@ -20,11 +20,9 @@ from pathlib import Path from typing import Optional, Dict, Any -from hermes_cli.nous_subscription import ( - apply_nous_provider_defaults, - get_nous_subscription_features, -) +from hermes_cli.nous_subscription import get_nous_subscription_features from tools.tool_backend_helpers import managed_nous_tools_enabled +from utils import base_url_hostname from hermes_constants import get_optional_skills_dir logger = logging.getLogger(__name__) @@ -92,20 +90,19 @@ def _supports_same_provider_pool_setup(provider: str) -> bool: "grok-code-fast-1", ], "gemini": [ - "gemini-3.1-pro-preview", "gemini-3-flash-preview", "gemini-3.1-flash-lite-preview", - "gemini-2.5-pro", "gemini-2.5-flash", "gemini-2.5-flash-lite", - "gemma-4-31b-it", "gemma-4-26b-it", + "gemini-3.1-pro-preview", "gemini-3-pro-preview", + "gemini-3-flash-preview", "gemini-3.1-flash-lite-preview", ], "zai": ["glm-5.1", "glm-5", "glm-4.7", "glm-4.5", "glm-4.5-flash"], - "kimi-coding": ["kimi-k2.5", "kimi-k2-thinking", "kimi-k2-turbo-preview"], - "kimi-coding-cn": ["kimi-k2.5", "kimi-k2-thinking", "kimi-k2-turbo-preview"], + "kimi-coding": ["kimi-k2.6", "kimi-k2.5", "kimi-k2-thinking", "kimi-k2-turbo-preview"], + "kimi-coding-cn": ["kimi-k2.6", "kimi-k2.5", "kimi-k2-thinking", "kimi-k2-turbo-preview"], "arcee": ["trinity-large-thinking", "trinity-large-preview", "trinity-mini"], "minimax": ["MiniMax-M2.7", "MiniMax-M2.5", "MiniMax-M2.1", "MiniMax-M2"], "minimax-cn": ["MiniMax-M2.7", "MiniMax-M2.5", "MiniMax-M2.1", "MiniMax-M2"], "ai-gateway": ["anthropic/claude-opus-4.6", "anthropic/claude-sonnet-4.6", "openai/gpt-5", "google/gemini-3-flash"], "kilocode": ["anthropic/claude-opus-4.6", "anthropic/claude-sonnet-4.6", "openai/gpt-5.4", "google/gemini-3-pro-preview", "google/gemini-3-flash-preview"], "opencode-zen": ["gpt-5.4", "gpt-5.3-codex", "claude-sonnet-4-6", "gemini-3-flash", "glm-5", "kimi-k2.5", "minimax-m2.7"], - "opencode-go": ["glm-5", "kimi-k2.5", "mimo-v2-pro", "mimo-v2-omni", "minimax-m2.5", "minimax-m2.7"], + "opencode-go": ["glm-5.1", "glm-5", "kimi-k2.5", "mimo-v2-pro", "mimo-v2-omni", "minimax-m2.5", "minimax-m2.7"], "huggingface": [ "Qwen/Qwen3.5-397B-A17B", "Qwen/Qwen3-235B-A22B-Thinking-2507", "Qwen/Qwen3-Coder-480B-A35B-Instruct", "deepseek-ai/DeepSeek-R1-0528", @@ -213,20 +210,20 @@ def prompt(question: str, default: str = None, password: bool = False) -> str: sys.exit(1) -def _curses_prompt_choice(question: str, choices: list, default: int = 0) -> int: +def _curses_prompt_choice(question: str, choices: list, default: int = 0, description: str | None = None) -> int: """Single-select menu using curses. Delegates to curses_radiolist.""" from hermes_cli.curses_ui import curses_radiolist - return curses_radiolist(question, choices, selected=default, cancel_returns=-1) + return curses_radiolist(question, choices, selected=default, cancel_returns=-1, description=description) -def prompt_choice(question: str, choices: list, default: int = 0) -> int: +def prompt_choice(question: str, choices: list, default: int = 0, description: str | None = None) -> int: """Prompt for a choice from a list with arrow key navigation. Escape keeps the current default (skips the question). Ctrl+C exits the wizard. """ - idx = _curses_prompt_choice(question, choices, default) + idx = _curses_prompt_choice(question, choices, default, description=description) if idx >= 0: if idx == default: print_info(" Skipped (keeping current)") @@ -433,9 +430,10 @@ def _print_setup_summary(config: dict, hermes_home): tool_status.append(("Text-to-Speech (MiniMax)", True, None)) elif tts_provider == "mistral" and get_env_value("MISTRAL_API_KEY"): tool_status.append(("Text-to-Speech (Mistral Voxtral)", True, None)) + elif tts_provider == "gemini" and (get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY")): + tool_status.append(("Text-to-Speech (Google Gemini)", True, None)) elif tts_provider == "neutts": try: - import importlib.util neutts_ok = importlib.util.find_spec("neutts") is not None except Exception: neutts_ok = False @@ -443,6 +441,16 @@ def _print_setup_summary(config: dict, hermes_home): tool_status.append(("Text-to-Speech (NeuTTS local)", True, None)) else: tool_status.append(("Text-to-Speech (NeuTTS — not installed)", False, "run 'hermes setup tts'")) + elif tts_provider == "kittentts": + try: + import importlib.util + kittentts_ok = importlib.util.find_spec("kittentts") is not None + except Exception: + kittentts_ok = False + if kittentts_ok: + tool_status.append(("Text-to-Speech (KittenTTS local)", True, None)) + else: + tool_status.append(("Text-to-Speech (KittenTTS — not installed)", False, "run 'hermes setup tts'")) else: tool_status.append(("Text-to-Speech (Edge TTS)", True, None)) @@ -805,7 +813,8 @@ def setup_model_provider(config: dict, *, quick: bool = False): elif _vision_idx == 1: # OpenAI-compatible endpoint _base_url = prompt(" Base URL (blank for OpenAI)").strip() or "https://api.openai.com/v1" _api_key_label = " API key" - if "api.openai.com" in _base_url.lower(): + _is_native_openai = base_url_hostname(_base_url) == "api.openai.com" + if _is_native_openai: _api_key_label = " OpenAI API key" _oai_key = prompt(_api_key_label, password=True).strip() if _oai_key: @@ -813,7 +822,7 @@ def setup_model_provider(config: dict, *, quick: bool = False): # Save vision base URL to config (not .env — only secrets go there) _vaux = config.setdefault("auxiliary", {}).setdefault("vision", {}) _vaux["base_url"] = _base_url - if "api.openai.com" in _base_url.lower(): + if _is_native_openai: _oai_vision_models = ["gpt-4o", "gpt-4o-mini", "gpt-4.1", "gpt-4.1-mini", "gpt-4.1-nano"] _vm_choices = _oai_vision_models + ["Use default (gpt-4o-mini)"] _vm_idx = prompt_choice("Select vision model:", _vm_choices, 0) @@ -835,14 +844,7 @@ def setup_model_provider(config: dict, *, quick: bool = False): print_info("Skipped — add later with 'hermes setup' or configure AUXILIARY_VISION_* settings") - if selected_provider == "nous" and nous_subscription_selected: - changed_defaults = apply_nous_provider_defaults(config) - current_tts = str(config.get("tts", {}).get("provider") or "edge") - if "tts" in changed_defaults: - print_success("TTS provider set to: OpenAI TTS via your Nous subscription") - else: - print_info(f"Keeping your existing TTS provider: {current_tts}") - + # Tool Gateway prompt is already shown by _model_flow_nous() above. save_config(config) if not quick and selected_provider != "nous": @@ -856,7 +858,6 @@ def setup_model_provider(config: dict, *, quick: bool = False): def _check_espeak_ng() -> bool: """Check if espeak-ng is installed.""" - import shutil return shutil.which("espeak-ng") is not None or shutil.which("espeak") is not None @@ -910,6 +911,31 @@ def _install_neutts_deps() -> bool: return False +def _install_kittentts_deps() -> bool: + """Install KittenTTS dependencies with user approval. Returns True on success.""" + import subprocess + import sys + + wheel_url = ( + "https://github.com/KittenML/KittenTTS/releases/download/" + "0.8.1/kittentts-0.8.1-py3-none-any.whl" + ) + print() + print_info("Installing kittentts Python package (~25-80MB model downloaded on first use)...") + print() + try: + subprocess.run( + [sys.executable, "-m", "pip", "install", "-U", wheel_url, "soundfile", "--quiet"], + check=True, timeout=300, + ) + print_success("kittentts installed successfully") + return True + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: + print_error(f"Failed to install kittentts: {e}") + print_info(f"Try manually: python -m pip install -U '{wheel_url}' soundfile") + return False + + def _setup_tts_provider(config: dict): """Interactive TTS provider selection with install flow for NeuTTS.""" tts_config = config.get("tts", {}) @@ -920,9 +946,12 @@ def _setup_tts_provider(config: dict): "edge": "Edge TTS", "elevenlabs": "ElevenLabs", "openai": "OpenAI TTS", + "xai": "xAI TTS", "minimax": "MiniMax TTS", "mistral": "Mistral Voxtral TTS", + "gemini": "Google Gemini TTS", "neutts": "NeuTTS", + "kittentts": "KittenTTS", } current_label = provider_labels.get(current_provider, current_provider) @@ -941,12 +970,15 @@ def _setup_tts_provider(config: dict): "Edge TTS (free, cloud-based, no setup needed)", "ElevenLabs (premium quality, needs API key)", "OpenAI TTS (good quality, needs API key)", + "xAI TTS (Grok voices, needs API key)", "MiniMax TTS (high quality with voice cloning, needs API key)", "Mistral Voxtral TTS (multilingual, native Opus, needs API key)", + "Google Gemini TTS (30 prebuilt voices, prompt-controllable, needs API key)", "NeuTTS (local on-device, free, ~300MB model download)", + "KittenTTS (local on-device, free, lightweight ~25-80MB ONNX)", ] ) - providers.extend(["edge", "elevenlabs", "openai", "minimax", "mistral", "neutts"]) + providers.extend(["edge", "elevenlabs", "openai", "xai", "minimax", "mistral", "gemini", "neutts", "kittentts"]) choices.append(f"Keep current ({current_label})") keep_current_idx = len(choices) - 1 idx = prompt_choice("Select TTS provider:", choices, keep_current_idx) @@ -967,7 +999,6 @@ def _setup_tts_provider(config: dict): if selected == "neutts": # Check if already installed try: - import importlib.util already_installed = importlib.util.find_spec("neutts") is not None except Exception: already_installed = False @@ -1012,6 +1043,23 @@ def _setup_tts_provider(config: dict): print_warning("No API key provided. Falling back to Edge TTS.") selected = "edge" + elif selected == "xai": + existing = get_env_value("XAI_API_KEY") + if not existing: + print() + api_key = prompt("xAI API key for TTS", password=True) + if api_key: + save_env_value("XAI_API_KEY", api_key) + print_success("xAI TTS API key saved") + else: + from hermes_constants import display_hermes_home as _dhh + print_warning( + "No xAI API key provided for TTS. Configure XAI_API_KEY via " + f"hermes setup model or {_dhh()}/.env to use xAI TTS. " + "Falling back to Edge TTS." + ) + selected = "edge" + elif selected == "minimax": existing = get_env_value("MINIMAX_API_KEY") if not existing: @@ -1036,6 +1084,42 @@ def _setup_tts_provider(config: dict): print_warning("No API key provided. Falling back to Edge TTS.") selected = "edge" + elif selected == "gemini": + existing = get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY") + if not existing: + print() + print_info("Get a free API key at https://aistudio.google.com/app/apikey") + api_key = prompt("Gemini API key for TTS", password=True) + if api_key: + save_env_value("GEMINI_API_KEY", api_key) + print_success("Gemini TTS API key saved") + else: + print_warning("No API key provided. Falling back to Edge TTS.") + selected = "edge" + + elif selected == "kittentts": + # Check if already installed + try: + import importlib.util + already_installed = importlib.util.find_spec("kittentts") is not None + except Exception: + already_installed = False + + if already_installed: + print_success("KittenTTS is already installed") + else: + print() + print_info("KittenTTS is lightweight (~25-80MB, CPU-only, no API key required).") + print_info("Voices: Jasper, Bella, Luna, Bruno, Rosie, Hugo, Kiki, Leo") + print() + if prompt_yes_no("Install KittenTTS now?", True): + if not _install_kittentts_deps(): + print_warning("KittenTTS installation incomplete. Falling back to Edge TTS.") + selected = "edge" + else: + print_info("Skipping install. Set tts.provider to 'kittentts' after installing manually.") + selected = "edge" + # Save the selection if "tts" not in config: config["tts"] = {} @@ -1057,8 +1141,6 @@ def setup_tts(config: dict): def setup_terminal_backend(config: dict): """Configure the terminal execution backend.""" import platform as _platform - import shutil - print_header("Terminal Backend") print_info("Choose where Hermes runs shell commands and code.") print_info("This affects tool execution, file access, and isolation.") @@ -1435,7 +1517,9 @@ def setup_agent_settings(config: dict): ) print_info("Maximum tool-calling iterations per conversation.") print_info("Higher = more complex tasks, but costs more tokens.") - print_info("Default is 90, which works for most tasks. Use 150+ for open exploration.") + print_info( + f"Press Enter to keep {current_max}. Use 90 for most tasks or 150+ for open exploration." + ) max_iter_str = prompt("Max iterations", current_max) try: @@ -1611,9 +1695,19 @@ def _setup_telegram(): return print_info("Create a bot via @BotFather on Telegram") - token = prompt("Telegram bot token", password=True) - if not token: - return + import re + + while True: + token = prompt("Telegram bot token", password=True) + if not token: + return + if not re.match(r"^\d+:[A-Za-z0-9_-]{30,}$", token): + print_error( + "Invalid token format. Expected: : " + "(e.g., 123456789:ABCdefGHI-jklMNOpqrSTUvwxYZ)" + ) + continue + break save_env_value("TELEGRAM_BOT_TOKEN", token) print_success("Telegram token saved") @@ -1969,6 +2063,8 @@ def _setup_wecom_callback(): _gw_setup() + + def _setup_bluebubbles(): """Configure BlueBubbles iMessage gateway.""" print_header("BlueBubbles (iMessage)") @@ -2034,6 +2130,12 @@ def _setup_bluebubbles(): print_info(" Install: https://docs.bluebubbles.app/helper-bundle/installation") +def _setup_qqbot(): + """Configure QQ Bot (Official API v2) via gateway setup.""" + from hermes_cli.gateway import _setup_qqbot as _gateway_setup_qqbot + _gateway_setup_qqbot() + + def _setup_webhooks(): """Configure webhook integration.""" print_header("Webhooks") @@ -2097,6 +2199,7 @@ def _setup_webhooks(): ("WeCom Callback (Self-Built App)", "WECOM_CALLBACK_CORP_ID", _setup_wecom_callback), ("Weixin (WeChat)", "WEIXIN_ACCOUNT_ID", _setup_weixin), ("BlueBubbles (iMessage)", "BLUEBUBBLES_SERVER_URL", _setup_bluebubbles), + ("QQ Bot", "QQ_APP_ID", _setup_qqbot), ("Webhooks (GitHub, GitLab, etc.)", "WEBHOOK_ENABLED", _setup_webhooks), ] @@ -2148,6 +2251,7 @@ def setup_gateway(config: dict): or get_env_value("WECOM_BOT_ID") or get_env_value("WEIXIN_ACCOUNT_ID") or get_env_value("BLUEBUBBLES_SERVER_URL") + or get_env_value("QQ_APP_ID") or get_env_value("WEBHOOK_ENABLED") ) if any_messaging: @@ -2169,6 +2273,10 @@ def setup_gateway(config: dict): missing_home.append("Slack") if get_env_value("BLUEBUBBLES_SERVER_URL") and not get_env_value("BLUEBUBBLES_HOME_CHANNEL"): missing_home.append("BlueBubbles") + if get_env_value("QQ_APP_ID") and not ( + get_env_value("QQBOT_HOME_CHANNEL") or get_env_value("QQ_HOME_CHANNEL") + ): + missing_home.append("QQBot") if missing_home: print() @@ -2192,8 +2300,10 @@ def setup_gateway(config: dict): _is_service_running, supports_systemd_services, has_conflicting_systemd_units, + has_legacy_hermes_units, install_linux_gateway_from_setup, print_systemd_scope_conflict_warning, + print_legacy_unit_warning, systemd_start, systemd_restart, launchd_install, @@ -2211,6 +2321,10 @@ def setup_gateway(config: dict): print_systemd_scope_conflict_warning() print() + if supports_systemd and has_legacy_hermes_units(): + print_legacy_unit_warning() + print() + if service_running: if prompt_yes_no(" Restart the gateway to pick up changes?", True): try: @@ -2301,6 +2415,74 @@ def setup_tools(config: dict, first_install: bool = False): # ============================================================================= +def _model_section_has_credentials(config: dict) -> bool: + """Return True when any known inference provider has usable credentials. + + Sources of truth: + * ``PROVIDER_REGISTRY`` in ``hermes_cli.auth`` — lists every supported + provider along with its ``api_key_env_vars``. + * ``active_provider`` in the auth store — covers OAuth device-code / + external-OAuth providers (Nous, Codex, Qwen, Gemini CLI, ...). + * The legacy OpenRouter aggregator env vars, which route generic + ``OPENAI_API_KEY`` / ``OPENROUTER_API_KEY`` values through OpenRouter. + """ + try: + from hermes_cli.auth import get_active_provider + if get_active_provider(): + return True + except Exception: + pass + + try: + from hermes_cli.auth import PROVIDER_REGISTRY + except Exception: + PROVIDER_REGISTRY = {} # type: ignore[assignment] + + def _has_key(pconfig) -> bool: + for env_var in pconfig.api_key_env_vars: + # CLAUDE_CODE_OAUTH_TOKEN is set by Claude Code itself, not by + # the user — mirrors is_provider_explicitly_configured in auth.py. + if env_var == "CLAUDE_CODE_OAUTH_TOKEN": + continue + if get_env_value(env_var): + return True + return False + + # Prefer the provider declared in config.yaml, avoids false positives + # from stray env vars (GH_TOKEN, etc.) when the user has already picked + # a different provider. + model_cfg = config.get("model") if isinstance(config, dict) else None + if isinstance(model_cfg, dict): + provider_id = (model_cfg.get("provider") or "").strip().lower() + if provider_id in PROVIDER_REGISTRY: + if _has_key(PROVIDER_REGISTRY[provider_id]): + return True + if provider_id == "openrouter": + for env_var in ("OPENROUTER_API_KEY", "OPENAI_API_KEY"): + if get_env_value(env_var): + return True + + # OpenRouter aggregator fallback (no provider declared in config). + for env_var in ("OPENROUTER_API_KEY", "OPENAI_API_KEY"): + if get_env_value(env_var): + return True + + for pid, pconfig in PROVIDER_REGISTRY.items(): + # Skip copilot in auto-detect: GH_TOKEN / GITHUB_TOKEN are + # commonly set for git tooling. Mirrors resolve_provider in auth.py. + if pid == "copilot": + continue + if _has_key(pconfig): + return True + return False + + +def _gateway_platform_short_label(label: str) -> str: + """Strip trailing parenthetical qualifiers from a gateway platform label.""" + base = label.split("(", 1)[0].strip() + return base or label + + def _get_section_config_summary(config: dict, section_key: str) -> Optional[str]: """Return a short summary if a setup section is already configured, else None. @@ -2309,20 +2491,7 @@ def _get_section_config_summary(config: dict, section_key: str) -> Optional[str] so that test patches on ``setup_mod.get_env_value`` take effect. """ if section_key == "model": - has_key = bool( - get_env_value("OPENROUTER_API_KEY") - or get_env_value("OPENAI_API_KEY") - or get_env_value("ANTHROPIC_API_KEY") - ) - if not has_key: - # Check for OAuth providers - try: - from hermes_cli.auth import get_active_provider - if get_active_provider(): - has_key = True - except Exception: - pass - if not has_key: + if not _model_section_has_credentials(config): return None model = config.get("model") if isinstance(model, str) and model.strip(): @@ -2340,37 +2509,11 @@ def _get_section_config_summary(config: dict, section_key: str) -> Optional[str] return f"max turns: {max_turns}" elif section_key == "gateway": - platforms = [] - if get_env_value("TELEGRAM_BOT_TOKEN"): - platforms.append("Telegram") - if get_env_value("DISCORD_BOT_TOKEN"): - platforms.append("Discord") - if get_env_value("SLACK_BOT_TOKEN"): - platforms.append("Slack") - if get_env_value("SIGNAL_ACCOUNT"): - platforms.append("Signal") - if get_env_value("EMAIL_ADDRESS"): - platforms.append("Email") - if get_env_value("TWILIO_ACCOUNT_SID"): - platforms.append("SMS") - if get_env_value("MATRIX_ACCESS_TOKEN") or get_env_value("MATRIX_PASSWORD"): - platforms.append("Matrix") - if get_env_value("MATTERMOST_TOKEN"): - platforms.append("Mattermost") - if get_env_value("WHATSAPP_PHONE_NUMBER_ID"): - platforms.append("WhatsApp") - if get_env_value("DINGTALK_CLIENT_ID"): - platforms.append("DingTalk") - if get_env_value("FEISHU_APP_ID"): - platforms.append("Feishu") - if get_env_value("WECOM_BOT_ID"): - platforms.append("WeCom") - if get_env_value("WEIXIN_ACCOUNT_ID"): - platforms.append("Weixin") - if get_env_value("BLUEBUBBLES_SERVER_URL"): - platforms.append("BlueBubbles") - if get_env_value("WEBHOOK_ENABLED"): - platforms.append("Webhooks") + platforms = [ + _gateway_platform_short_label(label) + for label, env_var, _ in _GATEWAY_PLATFORMS + if get_env_value(env_var) + ] if platforms: return ", ".join(platforms) return None # No platforms configured — section must run @@ -2914,6 +3057,27 @@ def _resolve_hermes_chat_argv() -> Optional[list[str]]: return None +def _reattach_stdin_to_tty() -> bool: + """Point stdin back at a real terminal before relaunching chat.""" + try: + if sys.stdin.isatty(): + return True + except Exception: + pass + + tty_path = "CONIN$" if os.name == "nt" else "/dev/tty" + try: + tty_fd = os.open(tty_path, os.O_RDONLY) + except OSError: + return False + + try: + os.dup2(tty_fd, 0) + finally: + os.close(tty_fd) + return True + + def _offer_launch_chat(): """Prompt the user to jump straight into chat after setup.""" print() @@ -2924,6 +3088,9 @@ def _offer_launch_chat(): if not chat_argv: print_info("Could not relaunch Hermes automatically. Run 'hermes chat' manually.") return + if not _reattach_stdin_to_tty(): + print_info("Could not relaunch Hermes automatically. Run 'hermes chat' manually.") + return os.execvp(chat_argv[0], chat_argv) diff --git a/tests/hermes_cli/test_setup.py b/tests/hermes_cli/test_setup.py index 2c07d3d6671..2718f339ef8 100644 --- a/tests/hermes_cli/test_setup.py +++ b/tests/hermes_cli/test_setup.py @@ -363,7 +363,7 @@ def fake_select(): def test_modal_setup_can_use_nous_subscription_without_modal_creds(tmp_path, monkeypatch, capsys): - monkeypatch.setenv("HERMES_ENABLE_NOUS_MANAGED_TOOLS", "1") + monkeypatch.setattr("hermes_cli.setup.managed_nous_tools_enabled", lambda: True) monkeypatch.setenv("HERMES_HOME", str(tmp_path)) config = load_config() @@ -405,7 +405,7 @@ def fake_prompt(message, *args, **kwargs): def test_modal_setup_persists_direct_mode_when_user_chooses_their_own_account(tmp_path, monkeypatch): - monkeypatch.setenv("HERMES_ENABLE_NOUS_MANAGED_TOOLS", "1") + monkeypatch.setattr("hermes_cli.setup.managed_nous_tools_enabled", lambda: True) monkeypatch.setenv("HERMES_HOME", str(tmp_path)) monkeypatch.delenv("MODAL_TOKEN_ID", raising=False) monkeypatch.delenv("MODAL_TOKEN_SECRET", raising=False) @@ -467,6 +467,7 @@ def test_offer_launch_chat_execs_fresh_process(monkeypatch): monkeypatch.setattr(setup_mod, "prompt_yes_no", lambda *_args, **_kwargs: True) monkeypatch.setattr(setup_mod, "_resolve_hermes_chat_argv", lambda: ["/usr/local/bin/hermes", "chat"]) + monkeypatch.setattr(setup_mod, "_reattach_stdin_to_tty", lambda: True) exec_calls = [] @@ -492,3 +493,43 @@ def test_offer_launch_chat_manual_fallback_when_unresolvable(monkeypatch, capsys captured = capsys.readouterr() assert "Run 'hermes chat' manually" in captured.out + + +def test_reattach_stdin_to_tty_noop_when_stdin_is_already_a_tty(monkeypatch): + from hermes_cli import setup as setup_mod + + monkeypatch.setattr(setup_mod.sys.stdin, "isatty", lambda: True) + + assert setup_mod._reattach_stdin_to_tty() is True + + +def test_reattach_stdin_to_tty_rebinds_dev_tty(monkeypatch): + from hermes_cli import setup as setup_mod + + calls = [] + + monkeypatch.setattr(setup_mod.sys.stdin, "isatty", lambda: False) + monkeypatch.setattr(setup_mod.os, "open", lambda path, flags: calls.append(("open", path, flags)) or 42) + monkeypatch.setattr(setup_mod.os, "dup2", lambda src, dst: calls.append(("dup2", src, dst))) + monkeypatch.setattr(setup_mod.os, "close", lambda fd: calls.append(("close", fd))) + monkeypatch.setattr(setup_mod.os, "name", "posix") + + assert setup_mod._reattach_stdin_to_tty() is True + assert calls == [ + ("open", "/dev/tty", setup_mod.os.O_RDONLY), + ("dup2", 42, 0), + ("close", 42), + ] + + +def test_offer_launch_chat_manual_fallback_when_tty_unavailable(monkeypatch, capsys): + from hermes_cli import setup as setup_mod + + monkeypatch.setattr(setup_mod, "prompt_yes_no", lambda *_args, **_kwargs: True) + monkeypatch.setattr(setup_mod, "_resolve_hermes_chat_argv", lambda: ["/usr/local/bin/hermes", "chat"]) + monkeypatch.setattr(setup_mod, "_reattach_stdin_to_tty", lambda: False) + + setup_mod._offer_launch_chat() + + captured = capsys.readouterr() + assert "Run 'hermes chat' manually" in captured.out