Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 43 additions & 3 deletions hermes_cli/banner.py
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,24 @@ def get_available_skills() -> Dict[str, List[str]]:
_UPDATE_CHECK_CACHE_SECONDS = 6 * 3600


def _git_rev(repo_dir: Path, rev: str) -> Optional[str]:
"""Resolve a git revision to its full hash, or None on failure."""
try:
result = subprocess.run(
["git", "rev-parse", rev],
capture_output=True,
text=True,
timeout=5,
cwd=str(repo_dir),
)
except Exception:
return None
if result.returncode != 0:
return None
value = (result.stdout or "").strip()
return value or None


def check_for_updates() -> Optional[int]:
"""Check how many commits behind origin/main the local repo is.

Expand All @@ -140,13 +158,23 @@ def check_for_updates() -> Optional[int]:
if not (repo_dir / ".git").exists():
return None

# Read cache
current_head = _git_rev(repo_dir, "HEAD")
current_upstream = _git_rev(repo_dir, "origin/main")

# Read cache. Only trust a fresh cache if the local and upstream refs
# still match the refs that were current when the cache was written.
now = time.time()
try:
if cache_file.exists():
cached = json.loads(cache_file.read_text())
if now - cached.get("ts", 0) < _UPDATE_CHECK_CACHE_SECONDS:
return cached.get("behind")
cached_head = cached.get("head")
cached_upstream = cached.get("upstream")
if (
cached_head == current_head
and cached_upstream == current_upstream
):
return cached.get("behind")
except Exception:
pass

Expand All @@ -160,6 +188,9 @@ def check_for_updates() -> Optional[int]:
except Exception:
pass # Offline or timeout — use stale refs, that's fine

refreshed_head = _git_rev(repo_dir, "HEAD")
refreshed_upstream = _git_rev(repo_dir, "origin/main")

# Count commits behind
try:
result = subprocess.run(
Expand All @@ -176,7 +207,16 @@ def check_for_updates() -> Optional[int]:

# Write cache
try:
cache_file.write_text(json.dumps({"ts": now, "behind": behind}))
cache_file.write_text(
json.dumps(
{
"ts": now,
"behind": behind,
"head": refreshed_head,
"upstream": refreshed_upstream,
}
)
)
except Exception:
pass

Expand Down
100 changes: 91 additions & 9 deletions tests/hermes_cli/test_update_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ def test_version_string_no_v_prefix():


def test_check_for_updates_uses_cache(tmp_path, monkeypatch):
"""When cache is fresh, check_for_updates should return cached value without calling git."""
"""When cache is fresh and refs match, check_for_updates should return cached value without calling git fetch."""
from hermes_cli.banner import check_for_updates

# Create a fake git repo and fresh cache
Expand All @@ -26,14 +26,76 @@ def test_check_for_updates_uses_cache(tmp_path, monkeypatch):
(repo_dir / ".git").mkdir()

cache_file = tmp_path / ".update_check"
cache_file.write_text(json.dumps({"ts": time.time(), "behind": 3}))
cache_file.write_text(
json.dumps(
{
"ts": time.time(),
"behind": 3,
"head": "head-sha",
"upstream": "upstream-sha",
}
)
)

def fake_run(cmd, *args, **kwargs):
if cmd[:2] == ["git", "rev-parse"]:
rev = cmd[2]
if rev == "HEAD":
return MagicMock(returncode=0, stdout="head-sha\n")
if rev == "origin/main":
return MagicMock(returncode=0, stdout="upstream-sha\n")
raise AssertionError(f"Unexpected subprocess call: {cmd}")

monkeypatch.setenv("HERMES_HOME", str(tmp_path))
with patch("hermes_cli.banner.subprocess.run") as mock_run:
with patch("hermes_cli.banner.subprocess.run", side_effect=fake_run) as mock_run:
result = check_for_updates()

assert result == 3
mock_run.assert_not_called()
assert mock_run.call_count == 2 # rev-parse HEAD + rev-parse origin/main


def test_check_for_updates_refreshes_when_cached_refs_do_not_match(tmp_path, monkeypatch):
"""Fresh cache should be ignored when HEAD/origin refs changed since it was written."""
from hermes_cli.banner import check_for_updates

repo_dir = tmp_path / "hermes-agent"
repo_dir.mkdir()
(repo_dir / ".git").mkdir()

cache_file = tmp_path / ".update_check"
cache_file.write_text(
json.dumps(
{
"ts": time.time(),
"behind": 497,
"head": "old-head",
"upstream": "old-upstream",
}
)
)

def fake_run(cmd, *args, **kwargs):
if cmd[:2] == ["git", "rev-parse"]:
rev = cmd[2]
if rev == "HEAD":
return MagicMock(returncode=0, stdout="new-head\n")
if rev == "origin/main":
return MagicMock(returncode=0, stdout="new-upstream\n")
if cmd[:3] == ["git", "fetch", "origin"]:
return MagicMock(returncode=0, stdout="")
if cmd[:3] == ["git", "rev-list", "--count"]:
return MagicMock(returncode=0, stdout="1\n")
raise AssertionError(f"Unexpected subprocess call: {cmd}")

monkeypatch.setenv("HERMES_HOME", str(tmp_path))
with patch("hermes_cli.banner.subprocess.run", side_effect=fake_run):
result = check_for_updates()

assert result == 1
rewritten = json.loads(cache_file.read_text())
assert rewritten["behind"] == 1
assert rewritten["head"] == "new-head"
assert rewritten["upstream"] == "new-upstream"


def test_check_for_updates_expired_cache(tmp_path, monkeypatch):
Expand All @@ -46,16 +108,36 @@ def test_check_for_updates_expired_cache(tmp_path, monkeypatch):

# Write an expired cache (timestamp far in the past)
cache_file = tmp_path / ".update_check"
cache_file.write_text(json.dumps({"ts": 0, "behind": 1}))

mock_result = MagicMock(returncode=0, stdout="5\n")
cache_file.write_text(
json.dumps(
{
"ts": 0,
"behind": 1,
"head": "old-head",
"upstream": "old-upstream",
}
)
)

def fake_run(cmd, *args, **kwargs):
if cmd[:2] == ["git", "rev-parse"]:
rev = cmd[2]
if rev == "HEAD":
return MagicMock(returncode=0, stdout="current-head\n")
if rev == "origin/main":
return MagicMock(returncode=0, stdout="current-upstream\n")
if cmd[:3] == ["git", "fetch", "origin"]:
return MagicMock(returncode=0, stdout="")
if cmd[:3] == ["git", "rev-list", "--count"]:
return MagicMock(returncode=0, stdout="5\n")
raise AssertionError(f"Unexpected subprocess call: {cmd}")

monkeypatch.setenv("HERMES_HOME", str(tmp_path))
with patch("hermes_cli.banner.subprocess.run", return_value=mock_result) as mock_run:
with patch("hermes_cli.banner.subprocess.run", side_effect=fake_run) as mock_run:
result = check_for_updates()

assert result == 5
assert mock_run.call_count == 2 # git fetch + git rev-list
assert mock_run.call_count == 6 # 2 pre-cache rev-parse + fetch + 2 refreshed rev-parse + rev-list


def test_check_for_updates_no_git_dir(tmp_path, monkeypatch):
Expand Down