From 630e224578e690dd869ff63b29fb57605eb31fd3 Mon Sep 17 00:00:00 2001 From: ZdenekSrotyr Date: Wed, 6 May 2026 15:58:32 +0200 Subject: [PATCH] feat(cli): add agnes self-upgrade with smoke test + rollback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reuses cli.update_check.check() for the version probe — extended with bypass_disabled=True so explicit user-typed self-upgrade is not silenced by AGNES_NO_UPDATE_CHECK (which is for the implicit warning loop). Install path: uv tool install --force when uv is on PATH; otherwise curl + pip via sys.executable (NOT system python3, NOT --user — both would land outside the agnes venv and silently no-op the upgrade). Smoke test execs the binary at the install-resolved path (uv tool dir joined with agnes-the-ai-analyst/bin/agnes, or sys.executable's sibling agnes for pip) — never via shutil.which, which can resolve a stale shadow on PATH and produce a false-positive smoke pass on the OLD version. Smoke also asserts --version output contains info.latest via PEP 440 Version() equality (so 0.40.0 does not falsely match 0.40.10). On smoke fail: rollback to last_known_good.json (written only after a previous run's smoke passed). Rollback rc is captured and surfaced on stderr if it also fails. First-ever upgrade or unrecoverable rollback prints the canonical bootstrap recovery: curl -fsSL /cli/install.sh | bash. AGNES_SELF_UPGRADE_IN_PROGRESS=1 is set for the duration of the run and propagated to the smoke-test subprocess. Layer B's _check_version_headers honors the sentinel and skips the < min hard-stop, so an in-flight upgrade can never sys.exit(2) itself. --force invalidates the update_check cache BEFORE probing. --force + offline = exit 1 with explicit stderr (without --force, offline is silent). --quiet suppresses progress output but never gags failure stderr. --- cli/commands/self_upgrade.py | 288 +++++++++++++++++++++++++++++++ cli/main.py | 2 + cli/update_check.py | 14 +- tests/test_cli_update_check.py | 21 +++ tests/test_self_upgrade.py | 304 +++++++++++++++++++++++++++++++++ 5 files changed, 627 insertions(+), 2 deletions(-) create mode 100644 cli/commands/self_upgrade.py create mode 100644 tests/test_self_upgrade.py diff --git a/cli/commands/self_upgrade.py b/cli/commands/self_upgrade.py new file mode 100644 index 0000000..4c06a2a --- /dev/null +++ b/cli/commands/self_upgrade.py @@ -0,0 +1,288 @@ +"""`agnes self-upgrade` — pull the wheel from the server, reinstall, smoke-test, +roll back on failure.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path +from typing import Optional, Union + +import typer + +from cli.config import _config_dir, get_server_url +from cli.update_check import UpdateInfo, check, format_outdated_notice + +self_upgrade_app = typer.Typer( + name="self-upgrade", + help="Reinstall the CLI from the server's currently-shipped wheel.", + invoke_without_command=True, +) + +_SENTINEL_ENV = "AGNES_SELF_UPGRADE_IN_PROGRESS" + + +class _Unreachable: + """Sentinel returned by _resolve_info when --force was specified but the + server probe failed. Distinguishes 'explicitly requested an upgrade and + we couldn't reach the server' (exit 1, stderr) from 'no upgrade needed' + (exit 0, silent).""" + + +_UNREACHABLE = _Unreachable() + + +def _invalidate_update_cache() -> None: + """Drop update_check.json so the next CLI invocation re-probes /cli/latest.""" + (_config_dir() / "update_check.json").unlink(missing_ok=True) + + +def _last_known_good_path() -> Path: + return _config_dir() / "last_known_good.json" + + +def _read_last_known_good() -> Optional[str]: + p = _last_known_good_path() + if not p.exists(): + return None + try: + return json.loads(p.read_text(encoding="utf-8")).get("download_url") + except (OSError, json.JSONDecodeError): + return None + + +def _record_last_known_good(download_url: str) -> None: + p = _last_known_good_path() + try: + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(json.dumps({"download_url": download_url}), encoding="utf-8") + except OSError: + pass # best-effort — failure to record must not break the flow + + +def _uv_tool_bin_path() -> Optional[Path]: + """Locate the agnes shim uv installed. + + Tries `uv tool dir --bin` (uv >= 0.5). Falls back to uv's documented + default install location on older uv where `--bin` is rejected. + """ + bin_dir: Optional[Path] = None + try: + out = subprocess.run( + ["uv", "tool", "dir", "--bin"], capture_output=True, text=True, timeout=5, + ) + if out.returncode == 0: + bin_dir = Path(out.stdout.strip()) + except (OSError, subprocess.TimeoutExpired): + bin_dir = None + + if bin_dir is None: + if sys.platform == "win32": + appdata = os.environ.get("APPDATA") + if appdata: + bin_dir = Path(appdata) / "uv" / "tools" / "bin" + else: + bin_dir = Path.home() / ".local" / "bin" + + if bin_dir is None or not bin_dir.exists(): + return None + + for name in ("agnes.exe", "agnes"): + candidate = bin_dir / name + if candidate.exists(): + return candidate + return None + + +def _pip_bin_path() -> Optional[Path]: + """`/bin/agnes` (POSIX) or `\\Scripts\\agnes.exe` (Windows).""" + parent = Path(sys.executable).parent + name = "agnes.exe" if sys.platform == "win32" else "agnes" + candidate = parent / name + return candidate if candidate.exists() else None + + +def _install_with_uv(download_url: str, *, quiet: bool) -> int: + out = subprocess.DEVNULL if quiet else None + return subprocess.run( + ["uv", "tool", "install", "--force", download_url], stdout=out + ).returncode + + +def _install_with_pip(download_url: str, *, quiet: bool) -> int: + """Install into the SAME interpreter that's running this command. + + sys.executable resolves to the venv that owns the live `agnes` binary. + `python3` would PATH-resolve to system python on macOS, landing the + wheel outside the agnes venv. `--user` is wrong inside a uv-tool venv + (targets ~/.local outside the venv). + """ + out = subprocess.DEVNULL if quiet else None + with tempfile.TemporaryDirectory(prefix="agnes_cli.") as td: + wheel_path = Path(td) / "agnes.whl" + rc = subprocess.run( + ["curl", "-fsSL", "-o", str(wheel_path), download_url], stdout=out + ).returncode + if rc != 0: + return rc + return subprocess.run( + [sys.executable, "-m", "pip", "install", + "--force-reinstall", "--no-deps", str(wheel_path)], + stdout=out, + ).returncode + + +def _smoke_test_new_binary(install_method: str, expected_version: str) -> tuple[bool, str]: + """Exec `/agnes --version` and confirm it boots AND reports + the expected version. Resolves the binary at the install-method-specific + path rather than via PATH — defends against a stale shadow ahead of the + freshly-installed binary in $PATH.""" + binary = _uv_tool_bin_path() if install_method == "uv" else _pip_bin_path() + if binary is None: + return False, f"agnes binary not found at expected {install_method} install path" + try: + env = {**os.environ, "AGNES_NO_UPDATE_CHECK": "1", _SENTINEL_ENV: "1"} + out = subprocess.run( + [str(binary), "--version"], + capture_output=True, text=True, timeout=10, env=env, + ) + if out.returncode != 0: + return False, f"exit {out.returncode}: {out.stderr.strip()[:200]}" + # Use Version() equality (PEP 440-aware) so "0.40.0" doesn't match "0.40.10". + from packaging.version import InvalidVersion, Version + tokens = out.stdout.strip().split() + actual_str = tokens[-1] if tokens else "" + try: + if Version(actual_str) != Version(expected_version): + return False, ( + f"version mismatch: expected {expected_version}, " + f"got {actual_str}" + ) + except InvalidVersion: + return False, f"unparseable version output: {out.stdout.strip()[:80]}" + return True, out.stdout.strip() + except (subprocess.TimeoutExpired, OSError) as e: + return False, f"{type(e).__name__}: {e}" + + +def _resolve_info(force: bool) -> Union[UpdateInfo, _Unreachable, None]: + """Returns: + UpdateInfo — install this wheel + _UNREACHABLE — --force specified, server probe failed + None — nothing to do (current, or offline without --force) + """ + if force: + _invalidate_update_cache() + info = check(get_server_url(), bypass_disabled=True) + if info is None: + return _UNREACHABLE if force else None + if not info.download_url: + return None + if not force and not info.is_outdated(): + return None + return info + + +def _do_install_with_smoke_and_rollback( + info: UpdateInfo, *, quiet: bool +) -> int: + """Returns the exit code typer should use (0 success, 1 failure).""" + prior_url = _read_last_known_good() # may be None on first upgrade + + if shutil.which("uv"): + rc = _install_with_uv(info.download_url, quiet=quiet) + method = "uv" + else: + rc = _install_with_pip(info.download_url, quiet=quiet) + method = "pip" + + if rc != 0: + sys.stderr.write(f"agnes self-upgrade: install failed with exit {rc}\n") + return 1 + + ok, detail = _smoke_test_new_binary(method, expected_version=info.latest) + if not ok: + sys.stderr.write( + f"agnes self-upgrade: new binary failed smoke test ({detail}).\n" + ) + server = get_server_url().rstrip("/") + bootstrap_recovery = f" Manual recovery: curl -fsSL {server}/cli/install.sh | bash\n" + if prior_url and prior_url != info.download_url: + sys.stderr.write(f" rolling back to {prior_url}\n") + rb_rc = ( + _install_with_uv(prior_url, quiet=True) + if method == "uv" + else _install_with_pip(prior_url, quiet=True) + ) + if rb_rc != 0: + sys.stderr.write( + f" rollback ALSO failed (rc={rb_rc}); CLI is in a broken state.\n" + ) + sys.stderr.write(bootstrap_recovery) + else: + sys.stderr.write( + " no prior wheel URL on record; rollback skipped.\n" + ) + sys.stderr.write(bootstrap_recovery) + return 1 + + # Convention: record then invalidate. No correctness consequence either way. + _record_last_known_good(info.download_url) + _invalidate_update_cache() + if not quiet: + typer.echo(f"agnes self-upgrade: installed {info.latest}", err=True) + return 0 + + +@self_upgrade_app.callback() +def self_upgrade( + quiet: bool = typer.Option( + False, "--quiet", + help="Suppress progress output. Failures still surface on stderr.", + ), + check_only: bool = typer.Option( + False, "--check-only", + help="Print status, don't install. Exit 1 if outdated.", + ), + force: bool = typer.Option( + False, "--force", + help="Reinstall the server's current wheel even when already on the latest version.", + ), +) -> None: + # Snapshot any prior sentinel so we restore (rather than destroy) it + # in finally — we own the namespace but a wrapper could legitimately + # set it. + prior_sentinel = os.environ.get(_SENTINEL_ENV) + os.environ[_SENTINEL_ENV] = "1" + try: + info = _resolve_info(force) + + # --check-only is read-only intent — never exit non-zero on + # transport errors. If unreachable, treat as "can't tell, current" + # and exit 0 silently. + if check_only: + if isinstance(info, _Unreachable) or info is None or not info.is_outdated(): + raise typer.Exit(0) + typer.echo(format_outdated_notice(info), err=True) + raise typer.Exit(1) + + if isinstance(info, _Unreachable): + sys.stderr.write( + f"agnes self-upgrade: cannot reach {get_server_url()}/cli/latest\n" + ) + raise typer.Exit(1) + + if info is None: + raise typer.Exit(0) # nothing to do, silent + + rc = _do_install_with_smoke_and_rollback(info, quiet=quiet) + raise typer.Exit(rc) + finally: + if prior_sentinel is None: + os.environ.pop(_SENTINEL_ENV, None) + else: + os.environ[_SENTINEL_ENV] = prior_sentinel diff --git a/cli/main.py b/cli/main.py index e4a8e90..7b7dc6b 100644 --- a/cli/main.py +++ b/cli/main.py @@ -33,6 +33,7 @@ from cli.commands.status import status_app from cli.commands.admin import admin_app from cli.commands.diagnose import diagnose_app from cli.commands.skills import skills_app +from cli.commands.self_upgrade import self_upgrade_app from cli.commands.setup import setup_app from cli.commands.server import server_app from cli.commands.explore import explore_app @@ -115,6 +116,7 @@ app.add_typer(status_app, name="status") app.add_typer(admin_app, name="admin") app.add_typer(diagnose_app, name="diagnose") app.add_typer(skills_app, name="skills") +app.add_typer(self_upgrade_app, name="self-upgrade") app.add_typer(setup_app, name="setup") app.add_typer(server_app, name="server") app.add_typer(explore_app, name="explore") diff --git a/cli/update_check.py b/cli/update_check.py index 90cb2f2..30ace40 100644 --- a/cli/update_check.py +++ b/cli/update_check.py @@ -114,13 +114,23 @@ def _fetch_latest(server_url: str) -> Optional[dict]: return None -def check(server_url: Optional[str]) -> Optional[UpdateInfo]: +def check( + server_url: Optional[str], *, bypass_disabled: bool = False +) -> Optional[UpdateInfo]: """Return UpdateInfo if a check ran (cached or fresh), else None. Silent on every failure path: no server configured, CLI package not installed, network down, malformed response, cache unreadable. + + `bypass_disabled=True` ignores `AGNES_NO_UPDATE_CHECK`. The env var + silences the implicit warning loop in the root callback; an explicit + user-typed `agnes self-upgrade` is not the implicit loop and must + still probe. Default keeps existing call sites (root callback) silent + when the env var is set. """ - if is_disabled() or not server_url: + if not bypass_disabled and is_disabled(): + return None + if not server_url: return None installed = _installed_version() diff --git a/tests/test_cli_update_check.py b/tests/test_cli_update_check.py index fe5b980..022f936 100644 --- a/tests/test_cli_update_check.py +++ b/tests/test_cli_update_check.py @@ -35,6 +35,27 @@ def test_check_returns_none_when_server_url_missing(tmp_config): assert update_check.check(None) is None # type: ignore[arg-type] +def test_check_bypass_disabled_overrides_env(monkeypatch, tmp_config): + """`AGNES_NO_UPDATE_CHECK=1` silences the implicit warning loop, but + explicit callers (e.g. `agnes self-upgrade`) pass `bypass_disabled=True` + and must NOT become a silent no-op.""" + from cli import update_check + + monkeypatch.setenv("AGNES_NO_UPDATE_CHECK", "1") + payload = { + "version": "9.9.9", + "wheel_filename": "x.whl", + "download_url_path": "/cli/wheel/x.whl", + } + with patch("cli.update_check._installed_version", return_value="2.0.0"): + with patch("cli.update_check._fetch_latest", return_value=payload): + # Default: env var wins, returns None. + assert update_check.check("http://server.test") is None + # Bypass: env var ignored. + info = update_check.check("http://server.test", bypass_disabled=True) + assert info is not None and info.latest == "9.9.9" + + def test_check_returns_none_when_installed_version_unknown(tmp_config): from cli import update_check with patch("cli.update_check._installed_version", return_value="unknown"): diff --git a/tests/test_self_upgrade.py b/tests/test_self_upgrade.py new file mode 100644 index 0000000..929aa5c --- /dev/null +++ b/tests/test_self_upgrade.py @@ -0,0 +1,304 @@ +"""Tests for `agnes self-upgrade` — install path, smoke test, rollback +(with rc capture), recursion barrier, --force offline failure, AGNES_NO_UPDATE_CHECK +bypass for explicit upgrades, --quiet stderr behavior, version-mismatch +smoke detection.""" + +import os +import sys +from unittest.mock import patch, MagicMock + +import pytest +from typer.testing import CliRunner + +from cli.main import app +from cli.update_check import UpdateInfo + +runner = CliRunner() + + +@pytest.fixture(autouse=True) +def _ensure_no_sentinel_leak(monkeypatch): + """Pytest test order is not guaranteed; explicitly clear the recursion + sentinel before every test so a leaked value from a prior test doesn't + produce a false-positive 'cleared on exit' assertion.""" + monkeypatch.delenv("AGNES_SELF_UPGRADE_IN_PROGRESS", raising=False) + yield + + +_OUTDATED_URL = "http://server.test/cli/wheel/agnes-0.40.0-py3-none-any.whl" +_PRIOR_URL = "http://server.test/cli/wheel/agnes-0.35.0-py3-none-any.whl" + + +def _outdated_info(): + return UpdateInfo(installed="0.30.0", latest="0.40.0", download_url=_OUTDATED_URL) + + +def _current_info(): + return UpdateInfo(installed="0.40.0", latest="0.40.0", download_url=None) + + +def _smoke_pass(): + return (True, "agnes 0.40.0") + + +def _smoke_fail(): + return (False, "exit 1: ImportError: cannot import name 'foo'") + + +def test_check_only_when_outdated_exits_1(): + with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()): + result = runner.invoke(app, ["self-upgrade", "--check-only"]) + assert result.exit_code == 1 + assert "out of date" in result.output + + +def test_check_only_when_current_exits_0(): + with patch("cli.commands.self_upgrade.check", return_value=_current_info()): + result = runner.invoke(app, ["self-upgrade", "--check-only"]) + assert result.exit_code == 0 + + +def test_when_current_short_circuits_no_install(): + with patch("cli.commands.self_upgrade.check", return_value=_current_info()), \ + patch("cli.commands.self_upgrade.subprocess.run") as mock_run: + result = runner.invoke(app, ["self-upgrade"]) + assert result.exit_code == 0 + mock_run.assert_not_called() + + +def test_uv_path_when_uv_available(): + with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \ + patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \ + patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \ + patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_pass()), \ + patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \ + patch("cli.commands.self_upgrade._record_last_known_good"), \ + patch("cli.commands.self_upgrade._invalidate_update_cache"): + mock_run.return_value = MagicMock(returncode=0) + result = runner.invoke(app, ["self-upgrade"]) + assert result.exit_code == 0 + args = mock_run.call_args_list[0].args[0] + assert args[:3] == ["uv", "tool", "install"] + assert "--force" in args + assert _OUTDATED_URL in args + + +def test_pip_fallback_uses_sys_executable_not_user(): + """pip path must target the running interpreter's venv, never --user.""" + with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \ + patch("cli.commands.self_upgrade.shutil.which", return_value=None), \ + patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \ + patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_pass()), \ + patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \ + patch("cli.commands.self_upgrade._record_last_known_good"), \ + patch("cli.commands.self_upgrade._invalidate_update_cache"): + mock_run.return_value = MagicMock(returncode=0) + result = runner.invoke(app, ["self-upgrade"]) + assert result.exit_code == 0 + cmds = [c.args[0] for c in mock_run.call_args_list] + assert any(cmd[0] == "curl" for cmd in cmds), cmds + pip_cmd = next(cmd for cmd in cmds if "pip" in cmd) + assert pip_cmd[0] == sys.executable, pip_cmd + assert "--force-reinstall" in pip_cmd + assert "--user" not in pip_cmd + + +def test_force_invalidates_cache_before_check(): + """--force must drop the cached download_url before probing /cli/latest.""" + fresh_current_with_url = UpdateInfo(installed="0.40.0", latest="0.40.0", + download_url=_OUTDATED_URL) + with patch("cli.commands.self_upgrade._invalidate_update_cache") as mock_invalidate, \ + patch("cli.commands.self_upgrade.check", return_value=fresh_current_with_url) as mock_check, \ + patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \ + patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \ + patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_pass()), \ + patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \ + patch("cli.commands.self_upgrade._record_last_known_good"): + mock_run.return_value = MagicMock(returncode=0) + result = runner.invoke(app, ["self-upgrade", "--force"]) + assert result.exit_code == 0 + assert mock_invalidate.call_count == 2 + mock_check.assert_called_once() + + +def test_force_offline_exits_1_with_stderr(): + """--force + server unreachable: exit 1 with explicit stderr.""" + with patch("cli.commands.self_upgrade.check", return_value=None), \ + patch("cli.commands.self_upgrade.get_server_url", + return_value="http://server.test"), \ + patch("cli.commands.self_upgrade._invalidate_update_cache"): + result = runner.invoke(app, ["self-upgrade", "--force"]) + assert result.exit_code == 1 + assert "cannot reach" in result.stderr + assert "server.test" in result.stderr + + +def test_offline_without_force_is_silent(): + """No --force, server unreachable: exit 0 silently from self-upgrade + itself. (The root callback's warning loop in cli/main.py may still emit + `[update] …` to stderr — that's a separate code path; this test only + pins that self-upgrade does not add a `cannot reach …` error.)""" + with patch("cli.commands.self_upgrade.check", return_value=None), \ + patch("cli.commands.self_upgrade._invalidate_update_cache"): + result = runner.invoke(app, ["self-upgrade"]) + assert result.exit_code == 0 + assert "cannot reach" not in result.stderr + assert "self-upgrade:" not in result.stderr + + +def test_self_upgrade_passes_bypass_disabled_to_check(): + """AGNES_NO_UPDATE_CHECK silences the implicit warning loop, but + explicit `agnes self-upgrade` must NOT be a silent no-op when set.""" + with patch("cli.commands.self_upgrade.check", return_value=_current_info()) as mock_check: + result = runner.invoke(app, ["self-upgrade", "--check-only"]) + assert result.exit_code == 0 + kwargs = mock_check.call_args.kwargs + assert kwargs.get("bypass_disabled") is True + + +def test_quiet_does_not_suppress_install_failure_stderr(): + """--quiet suppresses progress but install/smoke failures always surface.""" + with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \ + patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \ + patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \ + patch("cli.commands.self_upgrade._read_last_known_good", return_value=None): + mock_run.return_value = MagicMock(returncode=42) + result = runner.invoke(app, ["self-upgrade", "--quiet"]) + assert result.exit_code == 1 + assert "install failed" in result.stderr + + +def test_smoke_fail_triggers_rollback_when_prior_url_known(): + """Broken new wheel: smoke fails, rollback to last-known-good URL, exit 1.""" + with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \ + patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \ + patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \ + patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_fail()), \ + patch("cli.commands.self_upgrade._read_last_known_good", return_value=_PRIOR_URL), \ + patch("cli.commands.self_upgrade._record_last_known_good") as mock_record: + mock_run.return_value = MagicMock(returncode=0) + result = runner.invoke(app, ["self-upgrade"]) + assert result.exit_code == 1 + urls_installed = [ + arg for c in mock_run.call_args_list + for arg in c.args[0] if isinstance(arg, str) and arg.startswith("http") + ] + assert _OUTDATED_URL in urls_installed + assert _PRIOR_URL in urls_installed + mock_record.assert_not_called() + assert "smoke test" in result.stderr + + +def test_smoke_fail_with_rollback_failure_surfaces_rc(): + """Forward install ok, smoke fail, rollback ALSO fails: stderr surfaces rc + recovery.""" + install_results = [MagicMock(returncode=0), MagicMock(returncode=99)] + with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \ + patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \ + patch("cli.commands.self_upgrade.subprocess.run", side_effect=install_results), \ + patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_fail()), \ + patch("cli.commands.self_upgrade._read_last_known_good", return_value=_PRIOR_URL), \ + patch("cli.commands.self_upgrade.get_server_url", + return_value="http://server.test"): + result = runner.invoke(app, ["self-upgrade"]) + assert result.exit_code == 1 + assert "rollback ALSO failed" in result.stderr + assert "rc=99" in result.stderr + assert "/cli/install.sh" in result.stderr + + +def test_smoke_fail_no_prior_url_prints_install_sh_recovery(): + """First-ever upgrade with no rollback target: stderr points at bootstrap path.""" + with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \ + patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \ + patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \ + patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_fail()), \ + patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \ + patch("cli.commands.self_upgrade.get_server_url", + return_value="http://server.test"): + mock_run.return_value = MagicMock(returncode=0) + result = runner.invoke(app, ["self-upgrade"]) + assert result.exit_code == 1 + assert "/cli/install.sh" in result.stderr + assert "server.test" in result.stderr + + +def test_smoke_pass_records_last_known_good_then_invalidates_cache(): + """Convention: record before invalidate.""" + call_order = [] + with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \ + patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \ + patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \ + patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_pass()), \ + patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \ + patch("cli.commands.self_upgrade._record_last_known_good", + side_effect=lambda url: call_order.append(("record", url))), \ + patch("cli.commands.self_upgrade._invalidate_update_cache", + side_effect=lambda: call_order.append(("invalidate", None))): + mock_run.return_value = MagicMock(returncode=0) + result = runner.invoke(app, ["self-upgrade"]) + assert result.exit_code == 0 + record_idx = next(i for i, c in enumerate(call_order) if c[0] == "record") + invalidate_idx = next(i for i, c in enumerate(call_order) if c[0] == "invalidate") + assert record_idx < invalidate_idx, call_order + assert call_order[record_idx] == ("record", _OUTDATED_URL) + + +def test_self_upgrade_propagates_sentinel_to_smoke_subprocess(): + """The sentinel is set in os.environ during the run and cleared in finally.""" + captured_envs = [] + + def _fake_smoke(method, expected_version): + env = {**os.environ, "AGNES_NO_UPDATE_CHECK": "1", + "AGNES_SELF_UPGRADE_IN_PROGRESS": "1"} + captured_envs.append(env) + return _smoke_pass() + + with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \ + patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \ + patch("cli.commands.self_upgrade.subprocess.run", + return_value=MagicMock(returncode=0)), \ + patch("cli.commands.self_upgrade._smoke_test_new_binary", side_effect=_fake_smoke), \ + patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \ + patch("cli.commands.self_upgrade._record_last_known_good"), \ + patch("cli.commands.self_upgrade._invalidate_update_cache"): + result = runner.invoke(app, ["self-upgrade"]) + assert result.exit_code == 0 + assert captured_envs and captured_envs[0]["AGNES_SELF_UPGRADE_IN_PROGRESS"] == "1" + assert os.environ.get("AGNES_SELF_UPGRADE_IN_PROGRESS") is None + + +@pytest.mark.parametrize("install_method,patch_target", [ + ("uv", "_uv_tool_bin_path"), + ("pip", "_pip_bin_path"), +]) +def test_smoke_test_detects_version_mismatch(install_method, patch_target): + """Smoke test execs binary at install path (NOT shutil.which) and checks + Version equality (NOT substring). Parametrized over uv + pip.""" + from pathlib import Path + from cli.commands import self_upgrade as su + + fake_bin = f"/fake/{install_method}/bin/agnes" + with patch.object(su, patch_target, return_value=Path(fake_bin)), \ + patch.object(su.subprocess, "run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stdout="agnes 0.30.0\n", stderr="") + ok, detail = su._smoke_test_new_binary(install_method, expected_version="0.40.0") + assert ok is False + assert "version mismatch" in detail + assert "0.40.0" in detail and "0.30.0" in detail + assert mock_run.call_args.args[0][0] == fake_bin + + +def test_smoke_test_passes_with_pep440_local_version(): + """Use Version() comparison, not substring (so "0.40.0" doesn't match "0.40.10").""" + from pathlib import Path + from cli.commands import self_upgrade as su + + with patch.object(su, "_uv_tool_bin_path", return_value=Path("/fake/agnes")), \ + patch.object(su.subprocess, "run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stdout="agnes 0.40.0\n", stderr="") + ok, _ = su._smoke_test_new_binary("uv", expected_version="0.40.0") + assert ok is True + mock_run.return_value = MagicMock(returncode=0, stdout="agnes 0.40.10\n", stderr="") + ok, detail = su._smoke_test_new_binary("uv", expected_version="0.40.0") + assert ok is False + assert "version mismatch" in detail