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.
This commit is contained in:
+119
-30
@@ -8,6 +8,7 @@ import shlex
|
|||||||
import shutil
|
import shutil
|
||||||
import sqlite3
|
import sqlite3
|
||||||
import subprocess
|
import subprocess
|
||||||
|
import sys
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from importlib.metadata import PackageNotFoundError, version
|
from importlib.metadata import PackageNotFoundError, version
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
@@ -217,43 +218,130 @@ def _exec_tmux(backend: str, chat_cwd: Path, cmd: list[str]) -> None:
|
|||||||
raise SystemExit(proc.returncode)
|
raise SystemExit(proc.returncode)
|
||||||
|
|
||||||
|
|
||||||
def _offer_install(config: dict) -> str | None:
|
def _is_interactive() -> bool:
|
||||||
"""Interactively offer to install a missing backend CLI.
|
"""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
|
def _npm_global_bin() -> Path | None:
|
||||||
/dev/tty so it also works when stdin is not a terminal. Returns the
|
"""Directory where ``npm install -g`` places binaries, if resolvable."""
|
||||||
resolved binary path after a successful install, else None.
|
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: <prefix>/bin; Windows npm usually puts shims in <prefix> 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")
|
install_cmd = config.get("install_cmd")
|
||||||
if not install_cmd:
|
if not install_cmd:
|
||||||
return None
|
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])
|
installer = shutil.which(install_cmd[0])
|
||||||
if not installer:
|
if not installer:
|
||||||
stderr_console.print(
|
stderr_console.print(
|
||||||
f"[red]{install_cmd[0]} not found[/red] — cannot auto-install "
|
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
|
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)
|
proc = subprocess.run([installer, *install_cmd[1:]], check=False)
|
||||||
if proc.returncode != 0:
|
if proc.returncode != 0:
|
||||||
stderr_console.print(
|
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 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:
|
def backend_available(backend: str) -> bool:
|
||||||
@@ -262,20 +350,21 @@ def backend_available(backend: str) -> bool:
|
|||||||
return bool(cfg and _resolve_backend_binary(cfg))
|
return bool(cfg and _resolve_backend_binary(cfg))
|
||||||
|
|
||||||
|
|
||||||
def ensure_backend(backend: str) -> bool:
|
def ensure_backend(backend: str, *, yes: bool | None = None) -> bool:
|
||||||
"""Ensure a backend CLI is available, offering to install it when missing.
|
"""Ensure a backend CLI is available, installing it when missing.
|
||||||
|
|
||||||
Reuses the interactive ``_offer_install`` (prompt on tty, install via the
|
Guided by default: prompts on a TTY (default Yes when npm is available).
|
||||||
backend's ``install_cmd``). Backends without an installer (e.g. hermes)
|
Non-interactive contexts (install scripts, CI, piped stdin) install
|
||||||
fall back to the manual not-found message. Returns True if the CLI is
|
automatically when an ``install_cmd`` exists. Pass ``yes=True`` only for
|
||||||
available afterwards.
|
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)
|
cfg = _BACKENDS.get(backend)
|
||||||
if not cfg:
|
if not cfg:
|
||||||
return False
|
return False
|
||||||
if _resolve_backend_binary(cfg):
|
if _resolve_backend_binary(cfg):
|
||||||
return True
|
return True
|
||||||
return _offer_install(cfg) is not None
|
return _offer_install(cfg, yes=yes) is not None
|
||||||
|
|
||||||
|
|
||||||
def _launch(
|
def _launch(
|
||||||
@@ -289,7 +378,7 @@ def _launch(
|
|||||||
config = _BACKENDS[backend]
|
config = _BACKENDS[backend]
|
||||||
binary = _resolve_backend_binary(config)
|
binary = _resolve_backend_binary(config)
|
||||||
if not binary and offer_install:
|
if not binary and offer_install:
|
||||||
binary = _offer_install(config)
|
binary = _offer_install(config, yes=False)
|
||||||
if not binary:
|
if not binary:
|
||||||
stderr_console.print(config["not_found_msg"])
|
stderr_console.print(config["not_found_msg"])
|
||||||
raise SystemExit(127)
|
raise SystemExit(127)
|
||||||
|
|||||||
@@ -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()
|
||||||
Reference in New Issue
Block a user