From b5e82edb835208a85fbe6135d76c165e17dafba5 Mon Sep 17 00:00:00 2001 From: Alex Schapiro Date: Sun, 19 Jul 2026 16:56:50 +0000 Subject: [PATCH] feat(cli): update notifications + self-update (strix --update) --- strix/interface/main.py | 24 +++ strix/interface/update_check.py | 308 ++++++++++++++++++++++++++++++++ tests/test_update_check.py | 138 ++++++++++++++ 3 files changed, 470 insertions(+) create mode 100644 strix/interface/update_check.py create mode 100644 tests/test_update_check.py diff --git a/strix/interface/main.py b/strix/interface/main.py index 2403200d..cd36b778 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -5,6 +5,7 @@ Strix Agent Interface import argparse import asyncio +import os import shutil import sys from datetime import UTC, datetime @@ -32,6 +33,13 @@ from strix.config.models import ( 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 @@ Examples: 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 @@ Examples: 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,7 @@ def display_completion_message(args: argparse.Namespace, results_path: Path) -> "[#60a5fa]discord.gg/strix-ai[/]" ) console.print() + notify_update(console) def pull_docker_image() -> None: @@ -851,6 +871,10 @@ def main() -> None: if args.config: apply_config_override(validate_config_file(args.config)) + start_background_check() + if prompt_update_if_available(Console()) and is_binary_install() and sys.platform != "win32": + os.execv(sys.executable, sys.argv) # noqa: S606 # nosec B606 + 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 00000000..b30ca335 --- /dev/null +++ b/strix/interface/update_check.py @@ -0,0 +1,308 @@ +"""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 json +import logging +import os +import platform +import shutil +import stat +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 Confirm + +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 _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(latest_version: str) -> None: + try: + _CACHE_PATH.parent.mkdir(parents=True, exist_ok=True) + _CACHE_PATH.write_text( + json.dumps({"latest_version": latest_version, "checked_at": time.time()}), + encoding="utf-8", + ) + except Exception: # noqa: BLE001, S110 + pass # nosec B110 + + +def _refresh_cache() -> None: + latest = _fetch_latest_version() + if latest: + _write_cache(latest) + + +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() -> 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) + latest = _read_cache().get("latest_version") + current = get_version() + if isinstance(latest, str) and current != "unknown" and _is_newer(latest, current): + return latest + return None + + +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 prompt_update_if_available(console: Console) -> bool: + """Offer an interactive self-update before a scan starts. + + Returns True if the binary was updated (caller should re-exec). + """ + 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}[/]" + ) + if not is_binary_install(): + console.print(f"[dim]Upgrade with:[/] [#60a5fa]{get_upgrade_command()}[/]") + console.print() + return False + if not Confirm.ask("Update now?", default=False): + console.print() + return False + return self_update(console, version=latest) + + +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) + + 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") + 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) + staged.replace(current_exe) + 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) + 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 00000000..a62a6822 --- /dev/null +++ b/tests/test_update_check.py @@ -0,0 +1,138 @@ +import json +import platform +import time +from pathlib import Path + +import pytest + +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_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, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setattr(update_check, "is_binary_install", lambda: False) + assert update_check.self_update() is False + out = capsys.readouterr().out + assert "upgrade" in out + + +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_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