feat(cli): nudge to upgrade when a newer roar is available on pypi

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`.")

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
 
 

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`."
+    )

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()