diff --git a/hermes_cli/banner.py b/hermes_cli/banner.py index c8446f04d9c3c..83b146acd346f 100644 --- a/hermes_cli/banner.py +++ b/hermes_cli/banner.py @@ -130,6 +130,77 @@ def get_available_skills() -> Dict[str, List[str]]: _UPSTREAM_REPO_URL = "https://github.com/NousResearch/hermes-agent.git" +def _resolve_git_dir(repo_dir: Path) -> Optional[Path]: + """Resolve ``repo_dir/.git`` to the actual git directory. + + Supports both normal checkouts (``.git`` directory) and worktrees where + ``.git`` is a file containing ``gitdir: ...``. + """ + git_entry = repo_dir / ".git" + if git_entry.is_dir(): + return git_entry + if not git_entry.is_file(): + return None + try: + text = git_entry.read_text().strip() + except Exception: + return None + prefix = "gitdir:" + if not text.lower().startswith(prefix): + return None + raw_path = text[len(prefix):].strip() + if not raw_path: + return None + git_dir = Path(raw_path) + if not git_dir.is_absolute(): + git_dir = (repo_dir / git_dir).resolve() + return git_dir if git_dir.exists() else None + + +def _read_git_head(repo_dir: Path) -> Optional[str]: + """Return the current local HEAD commit hash without invoking git.""" + git_dir = _resolve_git_dir(repo_dir) + if git_dir is None: + return None + + try: + head_text = (git_dir / "HEAD").read_text().strip() + except Exception: + return None + + if not head_text: + return None + if not head_text.startswith("ref:"): + return head_text + + ref_name = head_text[4:].strip() + if not ref_name: + return None + + ref_path = git_dir / ref_name + try: + if ref_path.exists(): + ref_value = ref_path.read_text().strip() + return ref_value or None + except Exception: + return None + + packed_refs = git_dir / "packed-refs" + try: + if packed_refs.exists(): + for line in packed_refs.read_text().splitlines(): + line = line.strip() + if not line or line.startswith("#") or line.startswith("^"): + continue + parts = line.split(" ", 1) + if len(parts) == 2 and parts[1].strip() == ref_name: + value = parts[0].strip() + return value or None + except Exception: + return None + return None + + def _check_via_rev(local_rev: str) -> Optional[int]: """Compare an embedded git revision to upstream main via ls-remote. @@ -189,16 +260,26 @@ def check_for_updates() -> Optional[int]: hermes_home = get_hermes_home() cache_file = hermes_home / ".update_check" embedded_rev = os.environ.get("HERMES_REVISION") or None + repo_dir = None + local_head = None - # Read cache — invalidate if the embedded rev has changed since last check + if not embedded_rev: + repo_dir = hermes_home / "hermes-agent" + if not (repo_dir / ".git").exists(): + repo_dir = Path(__file__).parent.parent.resolve() + if not (repo_dir / ".git").exists(): + return None + local_head = _read_git_head(repo_dir) + + # Read cache — invalidate if the embedded rev or local HEAD changed since last check 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 - and cached.get("rev") == embedded_rev - ): + cache_is_fresh = now - cached.get("ts", 0) < _UPDATE_CHECK_CACHE_SECONDS + cache_matches_rev = cached.get("rev") == embedded_rev + cache_matches_head = bool(embedded_rev) or cached.get("local_head") == local_head + if cache_is_fresh and cache_matches_rev and cache_matches_head: return cached.get("behind") except Exception: pass @@ -206,15 +287,14 @@ def check_for_updates() -> Optional[int]: if embedded_rev: behind = _check_via_rev(embedded_rev) else: - repo_dir = hermes_home / "hermes-agent" - if not (repo_dir / ".git").exists(): - repo_dir = Path(__file__).parent.parent.resolve() - if not (repo_dir / ".git").exists(): - return None behind = _check_via_local_git(repo_dir) try: - cache_file.write_text(json.dumps({"ts": now, "behind": behind, "rev": embedded_rev})) + cache_file.write_text( + json.dumps( + {"ts": now, "behind": behind, "rev": embedded_rev, "local_head": local_head} + ) + ) except Exception: pass diff --git a/tests/hermes_cli/test_update_check.py b/tests/hermes_cli/test_update_check.py index 2bdc9b2462158..c89c1e4bdf8d4 100644 --- a/tests/hermes_cli/test_update_check.py +++ b/tests/hermes_cli/test_update_check.py @@ -17,16 +17,20 @@ 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 HEAD matches, check_for_updates should return cached value without calling git.""" from hermes_cli.banner import check_for_updates # Create a fake git repo and fresh cache repo_dir = tmp_path / "hermes-agent" repo_dir.mkdir() - (repo_dir / ".git").mkdir() + git_dir = repo_dir / ".git" + git_dir.mkdir() + (git_dir / "refs" / "heads").mkdir(parents=True) + (git_dir / "HEAD").write_text("ref: refs/heads/main\n") + (git_dir / "refs" / "heads" / "main").write_text("abc123\n") 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, "local_head": "abc123", "rev": None})) monkeypatch.setenv("HERMES_HOME", str(tmp_path)) with patch("hermes_cli.banner.subprocess.run") as mock_run: @@ -42,11 +46,15 @@ def test_check_for_updates_expired_cache(tmp_path, monkeypatch): repo_dir = tmp_path / "hermes-agent" repo_dir.mkdir() - (repo_dir / ".git").mkdir() + git_dir = repo_dir / ".git" + git_dir.mkdir() + (git_dir / "refs" / "heads").mkdir(parents=True) + (git_dir / "HEAD").write_text("ref: refs/heads/main\n") + (git_dir / "refs" / "heads" / "main").write_text("abc123\n") # 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})) + cache_file.write_text(json.dumps({"ts": 0, "behind": 1, "local_head": "abc123", "rev": None})) mock_result = MagicMock(returncode=0, stdout="5\n") @@ -58,6 +66,34 @@ def test_check_for_updates_expired_cache(tmp_path, monkeypatch): assert mock_run.call_count == 2 # git fetch + git rev-list +def test_check_for_updates_ignores_fresh_cache_when_local_head_changed(tmp_path, monkeypatch): + """Fresh cache should be invalidated when the local checkout moved to a new HEAD.""" + from hermes_cli.banner import check_for_updates + + repo_dir = tmp_path / "hermes-agent" + repo_dir.mkdir() + git_dir = repo_dir / ".git" + git_dir.mkdir() + (git_dir / "refs" / "heads").mkdir(parents=True) + (git_dir / "HEAD").write_text("ref: refs/heads/main\n") + (git_dir / "refs" / "heads" / "main").write_text("newhead\n") + + cache_file = tmp_path / ".update_check" + cache_file.write_text(json.dumps({"ts": time.time(), "behind": 1, "local_head": "oldhead", "rev": None})) + + mock_result = MagicMock(returncode=0, stdout="0\n") + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + with patch("hermes_cli.banner.subprocess.run", return_value=mock_result) as mock_run: + result = check_for_updates() + + assert result == 0 + assert mock_run.call_count == 2 # stale cache forces git fetch + rev-list + cached = json.loads(cache_file.read_text()) + assert cached["behind"] == 0 + assert cached["local_head"] == "newhead" + + def test_check_for_updates_no_git_dir(tmp_path, monkeypatch): """Returns None when .git directory doesn't exist anywhere.""" import hermes_cli.banner as banner