From 6702f674620a4fe75d9a6d2a23eceacf226482ee Mon Sep 17 00:00:00 2001 From: Zhengshou Lai Date: Sun, 9 Aug 2026 15:54:23 +0800 Subject: [PATCH] feat: guided auto-install for missing agent CLIs Offer install with default Yes on TTY; auto-install when non-interactive (install/CI). Resolve binaries from npm global prefix after install. --- myagents/launcher.py | 149 ++++++++++++++++++++++++++++------- tests/test_ensure_backend.py | 105 ++++++++++++++++++++++++ 2 files changed, 224 insertions(+), 30 deletions(-) create mode 100644 tests/test_ensure_backend.py diff --git a/myagents/launcher.py b/myagents/launcher.py index 0e50529..74e7542 100644 --- a/myagents/launcher.py +++ b/myagents/launcher.py @@ -8,6 +8,7 @@ import shlex import shutil import sqlite3 import subprocess +import sys from datetime import datetime from importlib.metadata import PackageNotFoundError, version from pathlib import Path @@ -217,43 +218,130 @@ def _exec_tmux(backend: str, chat_cwd: Path, cmd: list[str]) -> None: raise SystemExit(proc.returncode) -def _offer_install(config: dict) -> str | None: - """Interactively offer to install a missing backend CLI. +def _is_interactive() -> bool: + """True when we can safely prompt the user (TTY, not CI).""" + if os.environ.get("CI", "").lower() in ("1", "true", "yes"): + return False + if os.environ.get("XIAOHE_ASSUME_YES", "").lower() in ("1", "true", "yes"): + return False + try: + return bool(sys.stdin.isatty() and sys.stderr.isatty()) + except Exception: # noqa: BLE001 — treat broken streams as non-interactive + return False - Only backends with a known one-shot installer (``install_cmd``) are - offered; others fall back to the manual ``not_found_msg``. Prompts on - /dev/tty so it also works when stdin is not a terminal. Returns the - resolved binary path after a successful install, else None. - """ + +def _npm_global_bin() -> Path | None: + """Directory where ``npm install -g`` places binaries, if resolvable.""" + npm = shutil.which("npm") + if not npm: + return None + try: + proc = subprocess.run( + [npm, "prefix", "-g"], + capture_output=True, + text=True, + timeout=30, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + return None + prefix = (proc.stdout or "").strip() + if proc.returncode != 0 or not prefix: + return None + root = Path(prefix) + # Unix: /bin; Windows npm usually puts shims in itself. + if sys.platform == "win32": + return root + return root / "bin" + + +def _resolve_installed_binary(config: dict) -> str | None: + """Resolve binary on PATH, then under the npm global prefix (post-install).""" + found = _resolve_backend_binary(config) + if found: + return found + names = (config["binary"], *config.get("alt_binaries", ())) + npm_bin = _npm_global_bin() + if not npm_bin or not npm_bin.is_dir(): + return None + for name in names: + candidates = [npm_bin / name] + if sys.platform == "win32": + candidates.extend( + [npm_bin / f"{name}.cmd", npm_bin / f"{name}.exe"] + ) + for cand in candidates: + if cand.is_file(): + return str(cand) + return None + + +def _run_install(config: dict) -> str | None: + """Run ``install_cmd`` and return the resolved binary path, or None.""" install_cmd = config.get("install_cmd") if not install_cmd: return None - stderr_console.print( - f"[yellow]{config['binary']} not found.[/yellow] " - f"Install with [cyan]{shlex.join(install_cmd)}[/cyan] ?" - ) - try: - answer = click.prompt( - "Install now? [y/N]", default="n", show_default=True, err=True - ) - except (click.Abort, EOFError): - return None - if answer.strip().lower() not in ("y", "yes"): - return None installer = shutil.which(install_cmd[0]) if not installer: stderr_console.print( f"[red]{install_cmd[0]} not found[/red] — cannot auto-install " - f"{config['binary']}. Install it manually." + f"{config['binary']}.\n" + " Install Node.js (e.g. [cyan]brew install node[/cyan]), then run " + "[cyan]xiaohe ensure-deps[/cyan] or [cyan]xiaohe init[/cyan]." ) return None + stderr_console.print( + f"[dim]Installing {config['binary']} via[/dim] " + f"[cyan]{shlex.join(install_cmd)}[/cyan] …" + ) proc = subprocess.run([installer, *install_cmd[1:]], check=False) if proc.returncode != 0: stderr_console.print( - f"[red]Install failed (exit {proc.returncode}).[/red]" + f"[red]Install failed (exit {proc.returncode}).[/red]\n" + " Fix the error above, then run [cyan]xiaohe ensure-deps[/cyan]." ) return None - return _resolve_backend_binary(config) + return _resolve_installed_binary(config) + + +def _offer_install(config: dict, *, yes: bool | None = None) -> str | None: + """Install a missing backend CLI (guided prompt, or auto when non-interactive). + + Only backends with a known one-shot installer (``install_cmd``) are + offered; others fall back to the manual ``not_found_msg``. + + ``yes``: True = install without asking; False = always prompt when + possible; None = auto (no prompt when non-interactive / CI). + Interactive default is Yes when the installer binary is on PATH. + """ + install_cmd = config.get("install_cmd") + if not install_cmd: + return None + + auto = bool(yes) if yes is not None else not _is_interactive() + if not auto: + stderr_console.print( + f"[yellow]{config['binary']} CLI is missing[/yellow] " + "(needed to launch this agent).\n" + f" Will run: [cyan]{shlex.join(install_cmd)}[/cyan]" + ) + default = "y" if shutil.which(install_cmd[0]) else "n" + prompt = ( + "Install now? [Y/n]" if default == "y" else "Install now? [y/N]" + ) + try: + answer = click.prompt( + prompt, default=default, show_default=False, err=True + ) + except (click.Abort, EOFError): + return None + if answer.strip().lower() not in ("y", "yes", ""): + stderr_console.print( + "[dim]Skipped.[/dim] Later: [cyan]xiaohe ensure-deps[/cyan] " + "or [cyan]xiaohe init[/cyan]." + ) + return None + return _run_install(config) def backend_available(backend: str) -> bool: @@ -262,20 +350,21 @@ def backend_available(backend: str) -> bool: return bool(cfg and _resolve_backend_binary(cfg)) -def ensure_backend(backend: str) -> bool: - """Ensure a backend CLI is available, offering to install it when missing. +def ensure_backend(backend: str, *, yes: bool | None = None) -> bool: + """Ensure a backend CLI is available, installing it when missing. - Reuses the interactive ``_offer_install`` (prompt on tty, install via the - backend's ``install_cmd``). Backends without an installer (e.g. hermes) - fall back to the manual not-found message. Returns True if the CLI is - available afterwards. + Guided by default: prompts on a TTY (default Yes when npm is available). + Non-interactive contexts (install scripts, CI, piped stdin) install + automatically when an ``install_cmd`` exists. Pass ``yes=True`` only for + explicit scripting. Backends without an installer (e.g. hermes) cannot + be auto-installed. Returns True if the CLI is available afterwards. """ cfg = _BACKENDS.get(backend) if not cfg: return False if _resolve_backend_binary(cfg): return True - return _offer_install(cfg) is not None + return _offer_install(cfg, yes=yes) is not None def _launch( @@ -289,7 +378,7 @@ def _launch( config = _BACKENDS[backend] binary = _resolve_backend_binary(config) if not binary and offer_install: - binary = _offer_install(config) + binary = _offer_install(config, yes=False) if not binary: stderr_console.print(config["not_found_msg"]) raise SystemExit(127) diff --git a/tests/test_ensure_backend.py b/tests/test_ensure_backend.py new file mode 100644 index 0000000..5d8ac43 --- /dev/null +++ b/tests/test_ensure_backend.py @@ -0,0 +1,105 @@ +"""Tests for myagents.launcher.ensure_backend / guided auto-install.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import patch + +from myagents import launcher as launcher_mod + + +class TestEnsureBackend: + def test_already_available_skips_install(self) -> None: + with ( + patch.object( + launcher_mod, + "_resolve_backend_binary", + return_value="/usr/bin/claude", + ), + patch.object(launcher_mod, "_offer_install") as offer, + ): + assert launcher_mod.ensure_backend("claude") is True + offer.assert_not_called() + + def test_noninteractive_auto_installs(self) -> None: + with ( + patch.object( + launcher_mod, "_resolve_backend_binary", return_value=None + ), + patch.object(launcher_mod, "_is_interactive", return_value=False), + patch.object( + launcher_mod, "_run_install", return_value="/opt/bin/claude" + ) as run_install, + patch.object(launcher_mod, "click") as click_mod, + ): + assert launcher_mod.ensure_backend("claude") is True + run_install.assert_called_once() + click_mod.prompt.assert_not_called() + + def test_yes_true_skips_prompt(self) -> None: + with ( + patch.object( + launcher_mod, "_resolve_backend_binary", return_value=None + ), + patch.object( + launcher_mod, "_run_install", return_value="/opt/bin/claude" + ) as run_install, + patch.object(launcher_mod, "click") as click_mod, + ): + assert launcher_mod.ensure_backend("claude", yes=True) is True + run_install.assert_called_once() + click_mod.prompt.assert_not_called() + + def test_unknown_backend_false(self) -> None: + assert launcher_mod.ensure_backend("nope") is False + + def test_hermes_cannot_auto_install(self) -> None: + with ( + patch.object( + launcher_mod, "_resolve_backend_binary", return_value=None + ), + patch.object(launcher_mod, "_is_interactive", return_value=False), + ): + assert launcher_mod.ensure_backend("hermes") is False + + +class TestResolveInstalledBinary: + def test_falls_back_to_npm_global_bin(self, tmp_path: Path) -> None: + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + claude = bin_dir / "claude" + claude.write_text("#!/bin/sh\n") + claude.chmod(0o755) + cfg = launcher_mod._BACKENDS["claude"] + with ( + patch.object( + launcher_mod, "_resolve_backend_binary", return_value=None + ), + patch.object(launcher_mod, "_npm_global_bin", return_value=bin_dir), + ): + assert launcher_mod._resolve_installed_binary(cfg) == str(claude) + + +class TestOfferInstallGuided: + def test_interactive_default_yes_when_npm_present(self) -> None: + cfg = launcher_mod._BACKENDS["claude"] + with ( + patch.object(launcher_mod, "_is_interactive", return_value=True), + patch.object( + launcher_mod.shutil, "which", return_value="/usr/bin/npm" + ), + patch.object( + launcher_mod, "_run_install", return_value="/opt/bin/claude" + ) as run_install, + patch.object( + launcher_mod.click, + "prompt", + return_value="y", + ) as prompt, + ): + assert ( + launcher_mod._offer_install(cfg, yes=None) == "/opt/bin/claude" + ) + prompt.assert_called_once() + assert prompt.call_args.kwargs.get("default") == "y" + run_install.assert_called_once()