diff --git a/roar/cli/commands/init.py b/roar/cli/commands/init.py index 2cf21c99..dc259129 100644 --- a/roar/cli/commands/init.py +++ b/roar/cli/commands/init.py @@ -364,6 +364,7 @@ def _print_version_header() -> None: def _maybe_print_init_hints(*, in_git_repo: bool, gitignore_action: str | None) -> None: """Print git-style `hint:` lines for next steps. Amber-colored to match git's hint convention. Suppressed in quiet/non-TTY contexts.""" + from ...version_check import upgrade_hint_text from .._format import hints_should_print, make_hint_printer if not hints_should_print(): @@ -395,6 +396,12 @@ def _maybe_print_init_hints(*, in_git_repo: bool, gitignore_action: str | None) hint() hint("Commit the .gitignore change before your first `roar run`:") hint(" git add .gitignore && git commit -m 'ignore .roar/'") + + upgrade = upgrade_hint_text() + if upgrade: + hint() + hint(upgrade) + hint() hint("Docs: https://glaas.ai/docs") hint("Disable these hints with `roar config set hints.enabled false`.") diff --git a/roar/telemetry/uploader.py b/roar/telemetry/uploader.py index 1cfe9b42..faac222e 100644 --- a/roar/telemetry/uploader.py +++ b/roar/telemetry/uploader.py @@ -69,6 +69,14 @@ def drain_queue( def main() -> int: drain_queue() + # Piggyback the "newer roar available?" pypi check on this already-detached + # background process so the foreground command never makes a network call. + try: + from ..version_check import refresh_pypi_version_cache + + refresh_pypi_version_cache() + except Exception: # pragma: no cover - opportunistic, must never break + pass return 0 diff --git a/roar/version_check.py b/roar/version_check.py new file mode 100644 index 00000000..89aeb29d --- /dev/null +++ b/roar/version_check.py @@ -0,0 +1,105 @@ +"""Best-effort "a newer roar is available" nudge. + +Design constraints (all deliberate): + +* **Zero foreground latency.** The pypi lookup runs *only* in the background + telemetry uploader subprocess (``roar.telemetry.uploader``), which is already + detached and fire-and-forget. The foreground (the ``hint:`` line) only ever + reads a small on-disk cache — the command path never touches the network. +* **Cached, throttled.** The lookup is skipped when the cache is younger than + ``_TTL_SECONDS`` so we don't hammer pypi on every telemetry flush. +* **Fail-open, always.** Every path swallows its exceptions. A version check + must never block, slow, or break the CLI — if pypi is slow/unreachable or the + cache is corrupt, the nudge simply doesn't appear. +""" + +from __future__ import annotations + +import json +import time +import urllib.error +import urllib.request +from collections.abc import Mapping +from pathlib import Path + +from . import __version__ +from .telemetry.paths import resolve_paths + +_PYPI_URL = "https://pypi.org/pypi/roar-cli/json" +_TTL_SECONDS = 24 * 60 * 60 +_TIMEOUT_SECONDS = 3.0 + + +def _cache_path(environ: Mapping[str, str] | None = None) -> Path: + return resolve_paths(environ).cache_dir / "version_check.json" + + +def _read_cache(environ: Mapping[str, str] | None = None) -> dict | None: + try: + return json.loads(_cache_path(environ).read_text(encoding="utf-8")) + except Exception: + return None + + +def refresh_pypi_version_cache( + environ: Mapping[str, str] | None = None, + *, + force: bool = False, +) -> None: + """Fetch the latest ``roar-cli`` version from pypi and cache it. + + Intended to be called only from the background telemetry uploader. No-op + (no network) when the cache is younger than the TTL unless ``force``. + Swallows every error. + """ + try: + cached = _read_cache(environ) + if ( + not force + and cached + and (time.time() - float(cached.get("checked_at", 0))) < _TTL_SECONDS + ): + return + request = urllib.request.Request(_PYPI_URL, headers={"User-Agent": "roar-version-check/1"}) + with urllib.request.urlopen(request, timeout=_TIMEOUT_SECONDS) as response: + payload = json.loads(response.read().decode("utf-8")) + latest = str(payload["info"]["version"]) + path = _cache_path(environ) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps({"latest": latest, "checked_at": time.time()}), + encoding="utf-8", + ) + except Exception: + return # fail-open: a version check must never raise + + +def pending_upgrade_version(environ: Mapping[str, str] | None = None) -> str | None: + """Return the cached latest version when it's newer than the running + version, else ``None``. Reads the cache only (never the network).""" + try: + cached = _read_cache(environ) + if not cached: + return None + latest = str(cached.get("latest") or "") + if not latest: + return None + from packaging.version import InvalidVersion, Version + + try: + return latest if Version(latest) > Version(__version__) else None + except InvalidVersion: + return None + except Exception: + return None + + +def upgrade_hint_text(environ: Mapping[str, str] | None = None) -> str | None: + """Ready-to-print ``hint:`` body when a newer roar is available, else None.""" + latest = pending_upgrade_version(environ) + if not latest: + return None + return ( + f"roar {latest} is available (you have {__version__}). " + f"Upgrade with `uv tool upgrade roar-cli` or `pip install -U roar-cli`." + ) diff --git a/tests/unit/test_version_check.py b/tests/unit/test_version_check.py new file mode 100644 index 00000000..f1e7cdb7 --- /dev/null +++ b/tests/unit/test_version_check.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import json +import time +from pathlib import Path +from unittest.mock import patch + +from roar import version_check + + +def _env(tmp_path: Path) -> dict[str, str]: + return { + "HOME": str(tmp_path), + "XDG_CACHE_HOME": str(tmp_path / "cache"), + "XDG_CONFIG_HOME": str(tmp_path / "config"), + } + + +def _write_cache(env: dict[str, str], latest: str, checked_at: float | None = None) -> None: + path = version_check._cache_path(env) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps({"latest": latest, "checked_at": checked_at or time.time()}), + encoding="utf-8", + ) + + +class _Resp: + def __init__(self, body: bytes) -> None: + self._body = body + + def __enter__(self) -> _Resp: + return self + + def __exit__(self, *_: object) -> bool: + return False + + def read(self) -> bytes: + return self._body + + +def test_upgrade_hint_when_newer_available(tmp_path: Path) -> None: + env = _env(tmp_path) + _write_cache(env, "9.9.9") + with patch("roar.version_check.__version__", "0.3.3"): + text = version_check.upgrade_hint_text(env) + assert text is not None + assert "9.9.9" in text and "0.3.3" in text + + +def test_no_hint_when_current_is_latest(tmp_path: Path) -> None: + env = _env(tmp_path) + _write_cache(env, "0.3.3") + with patch("roar.version_check.__version__", "0.3.3"): + assert version_check.upgrade_hint_text(env) is None + + +def test_no_hint_when_current_is_newer(tmp_path: Path) -> None: + env = _env(tmp_path) + _write_cache(env, "0.3.3") + with patch("roar.version_check.__version__", "0.4.0"): + assert version_check.upgrade_hint_text(env) is None + + +def test_no_hint_without_cache(tmp_path: Path) -> None: + assert version_check.upgrade_hint_text(_env(tmp_path)) is None + + +def test_no_hint_on_corrupt_cache(tmp_path: Path) -> None: + env = _env(tmp_path) + path = version_check._cache_path(env) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("{not json", encoding="utf-8") + assert version_check.upgrade_hint_text(env) is None + + +def test_refresh_writes_cache(tmp_path: Path) -> None: + env = _env(tmp_path) + body = json.dumps({"info": {"version": "1.2.3"}}).encode("utf-8") + with patch("roar.version_check.urllib.request.urlopen", return_value=_Resp(body)): + version_check.refresh_pypi_version_cache(env, force=True) + cached = json.loads(version_check._cache_path(env).read_text(encoding="utf-8")) + assert cached["latest"] == "1.2.3" + assert "checked_at" in cached + + +def test_refresh_skips_network_when_cache_fresh(tmp_path: Path) -> None: + env = _env(tmp_path) + _write_cache(env, "0.3.3", checked_at=time.time()) # fresh + with patch("roar.version_check.urllib.request.urlopen") as urlopen: + version_check.refresh_pypi_version_cache(env) # not forced + urlopen.assert_not_called() + + +def test_refresh_failopen_on_network_error(tmp_path: Path) -> None: + env = _env(tmp_path) + with patch("roar.version_check.urllib.request.urlopen", side_effect=OSError("no net")): + version_check.refresh_pypi_version_cache(env, force=True) # must not raise + assert not version_check._cache_path(env).exists()