diff --git a/strix/interface/main.py b/strix/interface/main.py index 2403200df..4b2946266 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -5,6 +5,7 @@ import argparse import asyncio +import os import shutil import sys from datetime import UTC, datetime @@ -32,6 +33,13 @@ from strix.core.paths import run_dir_for, runtime_state_dir from strix.interface.cli import run_cli from strix.interface.tui import run_tui +from strix.interface.update_check import ( + is_binary_install, + notify_update, + prompt_update_if_available, + self_update, + start_background_check, +) from strix.interface.utils import ( assign_workspace_subdirs, build_final_stats_text, @@ -447,6 +455,14 @@ def parse_arguments() -> argparse.Namespace: version=f"strix {get_version()}", ) + parser.add_argument( + "--update", + action="store_true", + help="Update strix to the latest version and exit. Self-updates the " + "standalone binary install; for pip/pipx/uv installs, prints the " + "matching upgrade command instead.", + ) + parser.add_argument( "-t", "--target", @@ -565,6 +581,9 @@ def parse_arguments() -> argparse.Namespace: args = parser.parse_args() + if args.update: + sys.exit(0 if self_update() else 1) + if args.instruction and args.instruction_file: parser.error( "Cannot specify both --instruction and --instruction-file. Use one or the other." @@ -788,6 +807,8 @@ def display_completion_message(args: argparse.Namespace, results_path: Path) -> "[#60a5fa]discord.gg/strix-ai[/]" ) console.print() + if not args.non_interactive: + notify_update(console) def pull_docker_image() -> None: @@ -851,6 +872,12 @@ def main() -> None: if args.config: apply_config_override(validate_config_file(args.config)) + start_background_check() + if not args.non_interactive and prompt_update_if_available(Console()): + if is_binary_install() and sys.platform != "win32": + os.execv(sys.executable, sys.argv) # noqa: S606 # nosec B606 + sys.exit(0) + check_docker_installed() pull_docker_image() diff --git a/strix/interface/update_check.py b/strix/interface/update_check.py new file mode 100644 index 000000000..56ed4cd53 --- /dev/null +++ b/strix/interface/update_check.py @@ -0,0 +1,389 @@ +"""Update notifications and self-update for the strix CLI. + +Follows the pattern used by tools like gh, uv, and pip: a background, +rate-limited (once per 24h) check against the release source, a cached +result in ``~/.strix``, a non-intrusive notice with the upgrade command +for the detected install method, and a ``strix --update`` self-update +path for the standalone binary install. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import platform +import shutil +import stat +import subprocess +import sys +import tarfile +import tempfile +import threading +import time +import zipfile +from pathlib import Path +from typing import cast + +import requests +from rich.console import Console +from rich.prompt import Prompt + +from strix.telemetry._common import get_version + + +logger = logging.getLogger(__name__) + +GITHUB_REPO = "usestrix/strix" +PYPI_PACKAGE = "strix-agent" +CHECK_INTERVAL_SECONDS = 24 * 60 * 60 +REQUEST_TIMEOUT_SECONDS = 5 + +_CACHE_PATH = Path.home() / ".strix" / "update-check.json" + +_background_thread: threading.Thread | None = None + + +def _is_disabled() -> bool: + return bool(os.environ.get("STRIX_NO_UPDATE_CHECK")) or any( + os.environ.get(key) + for key in ("CI", "GITHUB_ACTIONS", "GITLAB_CI", "JENKINS_URL", "BUILDKITE", "CIRCLECI") + ) + + +def is_binary_install() -> bool: + return bool(getattr(sys, "frozen", False)) + + +def get_install_method() -> str: + if is_binary_install(): + return "binary" + prefix = str(Path(sys.prefix)).replace("\\", "/") + if "/pipx/" in prefix or prefix.endswith("/pipx"): + return "pipx" + if "/uv/tools/" in prefix: + return "uv" + return "pip" + + +def get_upgrade_command(method: str | None = None) -> str: + method = method or get_install_method() + commands = { + "binary": "strix --update", + "pipx": "pipx upgrade strix-agent", + "uv": "uv tool upgrade strix-agent", + "pip": "pip install --upgrade strix-agent", + } + return commands[method] + + +def _parse_version(value: str) -> tuple[int, ...] | None: + parts = value.strip().lstrip("v").split(".") + try: + return tuple(int(part) for part in parts) + except ValueError: + return None + + +def _is_newer(latest: str, current: str) -> bool: + latest_parts = _parse_version(latest) + current_parts = _parse_version(current) + if latest_parts is None or current_parts is None: + return False + return latest_parts > current_parts + + +def _fetch_latest_version() -> str | None: + try: + if is_binary_install(): + response = requests.get( + f"https://api.github.com/repos/{GITHUB_REPO}/releases/latest", + timeout=REQUEST_TIMEOUT_SECONDS, + ) + response.raise_for_status() + tag = response.json().get("tag_name", "") + return tag.lstrip("v") or None + response = requests.get( + f"https://pypi.org/pypi/{PYPI_PACKAGE}/json", + timeout=REQUEST_TIMEOUT_SECONDS, + ) + response.raise_for_status() + version = response.json().get("info", {}).get("version") + return str(version) if version else None + except Exception: # noqa: BLE001 + logger.debug("update check failed", exc_info=True) + return None + + +def _fetch_asset_digest(version: str, filename: str) -> str | None: + """Return the expected sha256 (hex) for a release asset, if the API provides one.""" + try: + response = requests.get( + f"https://api.github.com/repos/{GITHUB_REPO}/releases/tags/v{version}", + timeout=REQUEST_TIMEOUT_SECONDS, + ) + response.raise_for_status() + for asset in response.json().get("assets", []): + if asset.get("name") == filename: + digest = asset.get("digest") or "" + if digest.startswith("sha256:"): + return digest.removeprefix("sha256:") + except Exception: # noqa: BLE001 + logger.debug("release asset digest lookup failed", exc_info=True) + return None + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as f: + for chunk in iter(lambda: f.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _read_cache() -> dict[str, object]: + try: + with _CACHE_PATH.open(encoding="utf-8") as f: + data = json.load(f) + if isinstance(data, dict): + return cast("dict[str, object]", data) + except Exception: # noqa: BLE001, S110 + pass # nosec B110 + return {} + + +def _write_cache(**fields: object) -> None: + try: + cache = _read_cache() + cache.update(fields) + _CACHE_PATH.parent.mkdir(parents=True, exist_ok=True) + _CACHE_PATH.write_text(json.dumps(cache), encoding="utf-8") + except Exception: # noqa: BLE001, S110 + pass # nosec B110 + + +def skip_version(version: str) -> None: + """Remember not to prompt again for this version (newer releases still notify).""" + _write_cache(skipped_version=version) + + +def _refresh_cache() -> None: + latest = _fetch_latest_version() + if latest: + _write_cache(latest_version=latest, checked_at=time.time()) + + +def start_background_check() -> None: + """Refresh the cached latest-version info in a daemon thread (at most once per 24h).""" + global _background_thread # noqa: PLW0603 + if _is_disabled(): + return + cache = _read_cache() + checked_at = cache.get("checked_at") + if isinstance(checked_at, int | float) and time.time() - checked_at < CHECK_INTERVAL_SECONDS: + return + _background_thread = threading.Thread(target=_refresh_cache, daemon=True) + _background_thread.start() + + +def get_available_update(*, respect_skip: bool = True) -> str | None: + """Return the newer version from the cache, or None if up to date / unknown.""" + if _is_disabled(): + return None + if _background_thread is not None: + _background_thread.join(timeout=0.2) + cache = _read_cache() + latest = cache.get("latest_version") + current = get_version() + if not isinstance(latest, str) or current == "unknown" or not _is_newer(latest, current): + return None + if respect_skip and cache.get("skipped_version") == latest: + return None + return latest + + +def notify_update(console: Console) -> None: + latest = get_available_update() + if not latest: + return + console.print( + f"[#eab308]A new version of strix is available:[/] " + f"[dim]{get_version()}[/] [dim]→[/] [bold #22c55e]{latest}[/]" + f" [dim]·[/] [#60a5fa]{get_upgrade_command()}[/]" + ) + console.print() + + +def run_package_upgrade(console: Console, method: str) -> bool: + """Upgrade a package-manager install by running its upgrade command.""" + command = get_upgrade_command(method).split() + console.print(f"[dim]Running[/] [#60a5fa]{' '.join(command)}[/]") + try: + result = subprocess.run(command, check=False) # noqa: S603 + except OSError as e: + console.print(f"[bold red]Update failed:[/] {e}") + return False + if result.returncode != 0: + console.print( + f"[bold red]Update failed[/] [dim](exit code {result.returncode}).[/] " + f"Run it manually: [#60a5fa]{get_upgrade_command(method)}[/]" + ) + return False + console.print("[#22c55e]✓ strix updated — restart the scan to use the new version[/]") + return True + + +def prompt_update_if_available(console: Console) -> bool: + """Offer an interactive update before a scan starts. + + Returns True if strix was updated (caller should re-exec / exit). + """ + latest = get_available_update() + if not latest or not sys.stdin.isatty() or not sys.stdout.isatty(): + return False + console.print() + console.print( + f"[#eab308]A new version of strix is available:[/] " + f"[dim]{get_version()}[/] [dim]→[/] [bold #22c55e]{latest}[/]" + ) + console.print( + "[dim] y — update now n — not now (ask again next run) s — skip this version[/]" + ) + choice = Prompt.ask("Update strix?", choices=["y", "n", "s"], default="n") + console.print() + if choice == "s": + skip_version(latest) + return False + if choice != "y": + return False + method = get_install_method() + if method == "binary": + return self_update(console, version=latest) + return run_package_upgrade(console, method) + + +def _release_target() -> str | None: + raw_os = platform.system().lower() + os_name = {"darwin": "macos", "linux": "linux", "windows": "windows"}.get(raw_os) + arch = platform.machine().lower() + arch = {"aarch64": "arm64", "amd64": "x86_64"}.get(arch, arch) + if os_name is None: + return None + target = f"{os_name}-{arch}" + supported = {"linux-x86_64", "macos-x86_64", "macos-arm64", "windows-x86_64"} + return target if target in supported else None + + +def _download_and_replace(version: str, target: str, console: Console) -> bool: + is_windows = target.startswith("windows") + archive_ext = ".zip" if is_windows else ".tar.gz" + filename = f"strix-{version}-{target}{archive_ext}" + url = f"https://github.com/{GITHUB_REPO}/releases/download/v{version}/{filename}" + binary_name = f"strix-{version}-{target}" + (".exe" if is_windows else "") + current_exe = Path(sys.executable).resolve() + + with tempfile.TemporaryDirectory() as tmp: + tmp_dir = Path(tmp) + archive_path = tmp_dir / filename + console.print(f"[dim]Downloading[/] {url}") + with requests.get( # nosec B113 + url, + stream=True, + timeout=REQUEST_TIMEOUT_SECONDS * 12, + ) as response: + response.raise_for_status() + with archive_path.open("wb") as f: + for chunk in response.iter_content(chunk_size=1 << 20): + f.write(chunk) + + expected_digest = _fetch_asset_digest(version, filename) + if expected_digest: + actual_digest = _sha256_file(archive_path) + if actual_digest != expected_digest: + raise RuntimeError( + f"checksum mismatch for {filename}: " + f"expected sha256 {expected_digest}, got {actual_digest}" + ) + else: + console.print("[dim yellow]No published checksum available; skipping verification[/]") + + if is_windows: + with zipfile.ZipFile(archive_path) as zf: + zf.extract(binary_name, tmp_dir) + else: + with tarfile.open(archive_path, "r:gz") as tf: + tf.extract(binary_name, tmp_dir, filter="data") + + new_binary = tmp_dir / binary_name + new_binary.chmod(new_binary.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) + + staged = current_exe.with_name(current_exe.name + ".new") + try: + shutil.copy2(new_binary, staged) + if is_windows: + # Windows can't replace a running executable in place; move it aside first. + old = current_exe.with_name(current_exe.name + ".old") + old.unlink(missing_ok=True) + current_exe.rename(old) + try: + staged.replace(current_exe) + except Exception: + old.rename(current_exe) + raise + else: + staged.replace(current_exe) + except Exception: + staged.unlink(missing_ok=True) + raise + return True + + +def self_update(console: Console | None = None, version: str | None = None) -> bool: + """Replace the running standalone binary with the latest release. + + Returns True on success. For package-manager installs this only + prints the right upgrade command and returns False. + """ + console = console or Console() + + if not is_binary_install(): + method = get_install_method() + console.print( + f"[#eab308]This strix was installed via {method};[/] " + f"upgrade it with: [#60a5fa]{get_upgrade_command(method)}[/]" + ) + return False + + latest = version or _fetch_latest_version() + if not latest: + console.print("[bold red]Could not determine the latest strix version.[/]") + return False + + current = get_version() + if current != "unknown" and not _is_newer(latest, current): + console.print(f"[#22c55e]strix {current} is already the latest version.[/]") + return True + + target = _release_target() + if not target: + console.print( + f"[bold red]No prebuilt binary for this platform " + f"({platform.system()}/{platform.machine()}).[/]" + ) + return False + + try: + _download_and_replace(latest, target, console) + except Exception as e: # noqa: BLE001 + logger.debug("self-update failed", exc_info=True) + console.print(f"[bold red]Update failed:[/] {e}") + console.print( + "[dim]You can reinstall manually with:[/] " + "[#60a5fa]curl -sSL https://strix.ai/install | bash[/]" + ) + return False + + _write_cache(latest_version=latest, checked_at=time.time()) + console.print(f"[#22c55e]✓ Updated strix to {latest}[/]") + return True diff --git a/tests/test_update_check.py b/tests/test_update_check.py new file mode 100644 index 000000000..b602f2855 --- /dev/null +++ b/tests/test_update_check.py @@ -0,0 +1,172 @@ +import hashlib +import io +import json +import platform +import time +from pathlib import Path + +import pytest +from rich.console import Console + +from strix.interface import update_check + + +@pytest.fixture(autouse=True) +def _isolated_cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(update_check, "_CACHE_PATH", tmp_path / "update-check.json") + monkeypatch.setattr(update_check, "_background_thread", None) + monkeypatch.delenv("STRIX_NO_UPDATE_CHECK", raising=False) + for key in ("CI", "GITHUB_ACTIONS", "GITLAB_CI", "JENKINS_URL", "BUILDKITE", "CIRCLECI"): + monkeypatch.delenv(key, raising=False) + + +@pytest.mark.parametrize( + ("latest", "current", "expected"), + [ + ("1.2.0", "1.1.0", True), + ("1.1.0", "1.1.0", False), + ("1.0.9", "1.1.0", False), + ("2.0.0", "1.99.99", True), + ("1.10.0", "1.9.0", True), + ("v1.2.0", "1.1.0", True), + ("not-a-version", "1.1.0", False), + ("1.2.0", "unknown", False), + ], +) +def test_is_newer(latest: str, current: str, expected: bool) -> None: + assert update_check._is_newer(latest, current) is expected + + +def test_get_available_update_from_cache(monkeypatch: pytest.MonkeyPatch) -> None: + update_check._CACHE_PATH.write_text( + json.dumps({"latest_version": "9.9.9", "checked_at": time.time()}) + ) + monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0") + assert update_check.get_available_update() == "9.9.9" + + +def test_get_available_update_up_to_date(monkeypatch: pytest.MonkeyPatch) -> None: + update_check._CACHE_PATH.write_text( + json.dumps({"latest_version": "1.0.0", "checked_at": time.time()}) + ) + monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0") + assert update_check.get_available_update() is None + + +def test_get_available_update_disabled_by_env(monkeypatch: pytest.MonkeyPatch) -> None: + update_check._CACHE_PATH.write_text( + json.dumps({"latest_version": "9.9.9", "checked_at": time.time()}) + ) + monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0") + monkeypatch.setenv("STRIX_NO_UPDATE_CHECK", "1") + assert update_check.get_available_update() is None + + +def test_get_available_update_disabled_in_ci(monkeypatch: pytest.MonkeyPatch) -> None: + update_check._CACHE_PATH.write_text( + json.dumps({"latest_version": "9.9.9", "checked_at": time.time()}) + ) + monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0") + monkeypatch.setenv("CI", "true") + assert update_check.get_available_update() is None + + +def test_get_available_update_corrupt_cache(monkeypatch: pytest.MonkeyPatch) -> None: + update_check._CACHE_PATH.write_text("{not json") + monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0") + assert update_check.get_available_update() is None + + +def test_background_check_skipped_when_fresh(monkeypatch: pytest.MonkeyPatch) -> None: + update_check._CACHE_PATH.write_text( + json.dumps({"latest_version": "1.0.0", "checked_at": time.time()}) + ) + called = False + + def fake_refresh() -> None: + nonlocal called + called = True + + monkeypatch.setattr(update_check, "_refresh_cache", fake_refresh) + update_check.start_background_check() + assert update_check._background_thread is None + assert called is False + + +def test_background_check_runs_when_stale(monkeypatch: pytest.MonkeyPatch) -> None: + update_check._CACHE_PATH.write_text( + json.dumps({"latest_version": "1.0.0", "checked_at": time.time() - 2 * 24 * 60 * 60}) + ) + monkeypatch.setattr(update_check, "_fetch_latest_version", lambda: "1.2.3") + update_check.start_background_check() + assert update_check._background_thread is not None + update_check._background_thread.join(timeout=5) + cache = json.loads(update_check._CACHE_PATH.read_text()) + assert cache["latest_version"] == "1.2.3" + + +def test_skipped_version_suppresses_update(monkeypatch: pytest.MonkeyPatch) -> None: + update_check._CACHE_PATH.write_text( + json.dumps({"latest_version": "9.9.9", "checked_at": time.time()}) + ) + monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0") + update_check.skip_version("9.9.9") + assert update_check.get_available_update() is None + assert update_check.get_available_update(respect_skip=False) == "9.9.9" + + +def test_newer_release_overrides_skipped_version(monkeypatch: pytest.MonkeyPatch) -> None: + update_check._CACHE_PATH.write_text( + json.dumps( + {"latest_version": "9.9.10", "checked_at": time.time(), "skipped_version": "9.9.9"} + ) + ) + monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0") + assert update_check.get_available_update() == "9.9.10" + + +def test_write_cache_preserves_existing_fields() -> None: + update_check.skip_version("9.9.9") + update_check._write_cache(latest_version="1.2.3", checked_at=123.0) + cache = json.loads(update_check._CACHE_PATH.read_text()) + assert cache == {"latest_version": "1.2.3", "checked_at": 123.0, "skipped_version": "9.9.9"} + + +def test_get_upgrade_command_all_methods() -> None: + assert update_check.get_upgrade_command("binary") == "strix --update" + assert update_check.get_upgrade_command("pipx") == "pipx upgrade strix-agent" + assert update_check.get_upgrade_command("uv") == "uv tool upgrade strix-agent" + assert update_check.get_upgrade_command("pip") == "pip install --upgrade strix-agent" + + +def test_self_update_non_binary_prints_command(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(update_check, "is_binary_install", lambda: False) + buffer = io.StringIO() + assert update_check.self_update(Console(file=buffer)) is False + assert "upgrade" in buffer.getvalue() + + +def test_self_update_already_latest(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(update_check, "is_binary_install", lambda: True) + monkeypatch.setattr(update_check, "_fetch_latest_version", lambda: "1.0.0") + monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0") + assert update_check.self_update() is True + + +def test_sha256_file(tmp_path: Path) -> None: + path = tmp_path / "blob" + path.write_bytes(b"strix") + assert update_check._sha256_file(path) == hashlib.sha256(b"strix").hexdigest() + + +def test_release_target(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(platform, "system", lambda: "Linux") + monkeypatch.setattr(platform, "machine", lambda: "x86_64") + assert update_check._release_target() == "linux-x86_64" + + monkeypatch.setattr(platform, "system", lambda: "Darwin") + monkeypatch.setattr(platform, "machine", lambda: "arm64") + assert update_check._release_target() == "macos-arm64" + + monkeypatch.setattr(platform, "machine", lambda: "riscv64") + assert update_check._release_target() is None