Compare commits

...
3 changed files with 588 additions and 0 deletions
+27
View File
@@ -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,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()
+389
View File
@@ -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
+172
View File
@@ -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