diff --git a/scripts/check_memory_freshness.py b/scripts/check_memory_freshness.py new file mode 100644 index 0000000..6552845 --- /dev/null +++ b/scripts/check_memory_freshness.py @@ -0,0 +1,205 @@ +#!/usr/bin/env python3 +""" +Memory Freshness Checker (Task A47) +Validates that project memory (CURRENT_STATE.md) matches actual repository git state. + +Usage: + python scripts/check_memory_freshness.py [--strict] [--memory-path ] [--repo-path ] +""" + +from __future__ import annotations + +import argparse +import os +import re +import subprocess +import sys +from pathlib import Path + + +DEFAULT_MEMORY_ROOT = Path("/srv/projects/AI-Memory") +DEFAULT_PROJECT = "hermes-hub" + + +def get_git_commit(repo_path: Path, ref: str = "HEAD") -> str | None: + """Retrieve full or short commit hash for a given git reference.""" + try: + res = subprocess.run( + ["git", "-C", str(repo_path), "rev-parse", ref], + capture_output=True, + text=True, + check=True, + ) + return res.stdout.strip() + except (subprocess.SubprocessError, FileNotFoundError): + return None + + +def get_git_branch(repo_path: Path) -> str | None: + """Retrieve current branch name.""" + try: + res = subprocess.run( + ["git", "-C", str(repo_path), "rev-parse", "--abbrev-ref", "HEAD"], + capture_output=True, + text=True, + check=True, + ) + return res.stdout.strip() + except (subprocess.SubprocessError, FileNotFoundError): + return None + + +def is_commit_in_history(repo_path: Path, commit_hash: str) -> bool: + """Check if a commit exists in the repository object database.""" + try: + subprocess.run( + ["git", "-C", str(repo_path), "cat-file", "-e", f"{commit_hash}^{{commit}}"], + capture_output=True, + check=True, + ) + return True + except (subprocess.SubprocessError, FileNotFoundError): + return False + + +def is_ancestor(repo_path: Path, ancestor: str, descendant: str = "HEAD") -> bool: + """Check if ancestor commit is reachable from descendant commit.""" + try: + res = subprocess.run( + ["git", "-C", str(repo_path), "merge-base", "--is-ancestor", ancestor, descendant], + capture_output=True, + ) + return res.returncode == 0 + except (subprocess.SubprocessError, FileNotFoundError): + return False + + +def extract_recorded_commit(content: str) -> str | None: + """ + Extract the recorded commit hash from CURRENT_STATE.md text. + Matches patterns like: + - main = 80aab00 + - Canonical HEAD (main): `80aab00` + - HEAD (main): `c35bc48` + - commit `80aab00` + """ + patterns = [ + r"main\s*=\s*`?([0-9a-fA-F]{7,40})`?", + r"Canonical HEAD.*?:\s*`?([0-9a-fA-F]{7,40})`?", + r"HEAD\s*\(main\):\s*`?([0-9a-fA-F]{7,40})`?", + r"HEAD:\s*`?([0-9a-fA-F]{7,40})`?", + ] + for pattern in patterns: + match = re.search(pattern, content, re.IGNORECASE) + if match: + return match.group(1).lower() + return None + + +def check_memory_freshness( + repo_path: Path | None = None, + memory_file: Path | None = None, + strict: bool = False, +) -> tuple[bool, str]: + """ + Perform freshness verification. Returns (is_fresh, summary_message). + """ + if repo_path is None: + repo_path = Path(__file__).resolve().parent.parent + + if memory_file is None: + env_path = os.getenv("AI_MEMORY_PATH") + if env_path: + memory_file = Path(env_path) + if memory_file.is_dir(): + memory_file = memory_file / "01_PROJECTS" / DEFAULT_PROJECT / "CURRENT_STATE.md" + else: + memory_file = DEFAULT_MEMORY_ROOT / "01_PROJECTS" / DEFAULT_PROJECT / "CURRENT_STATE.md" + + if not memory_file.exists(): + msg = ( + f"[WARNING] Canonical memory file not found: {memory_file}\n" + f"Note: AI-Memory is canonically hosted on server 192.168.1.81. " + f"If executing in an isolated runner or off-server environment, this is expected." + ) + return (False if strict else True, msg) + + try: + content = memory_file.read_text(encoding="utf-8", errors="ignore") + except Exception as e: + return (False, f"[ERROR] Failed to read memory file {memory_file}: {e}") + + recorded_commit = extract_recorded_commit(content) + if not recorded_commit: + return (False, f"[ERROR] Could not extract recorded commit hash from {memory_file}") + + head_commit = get_git_commit(repo_path, "HEAD") + main_commit = get_git_commit(repo_path, "origin/main") or get_git_commit(repo_path, "main") + branch = get_git_branch(repo_path) or "unknown" + + if not head_commit: + return (False, f"[ERROR] Failed to inspect git repository at {repo_path}") + + head_short = head_commit[: len(recorded_commit)] + main_short = main_commit[: len(recorded_commit)] if main_commit else "N/A" + + recorded_in_history = is_commit_in_history(repo_path, recorded_commit) + is_head_match = head_short.lower() == recorded_commit.lower() + is_main_match = main_short.lower() == recorded_commit.lower() if main_commit else False + + report_lines = [ + "=== AI Memory Freshness Verification ===", + f"Memory File: {memory_file}", + f"Recorded Commit: {recorded_commit}", + f"Current Branch: {branch}", + f"Current HEAD: {head_commit[:10]}", + f"Main Commit: {main_commit[:10] if main_commit else 'N/A'}", + f"In Git History: {'YES' if recorded_in_history else 'NO'}", + ] + + if is_head_match or is_main_match: + report_lines.append("[STATUS] FRESH: Memory accurately matches canonical git repository state.") + return (True, "\n".join(report_lines)) + + # If we are on a feature branch descended from the recorded main commit + if recorded_in_history and is_ancestor(repo_path, recorded_commit, "HEAD"): + report_lines.append( + f"[STATUS] FRESH (Active Branch): Working on '{branch}' based on recorded baseline {recorded_commit}." + ) + return (True, "\n".join(report_lines)) + + if recorded_in_history: + report_lines.append( + f"[STATUS] DIVERGED: Recorded commit {recorded_commit} exists in history but differs from current main/HEAD." + ) + return (False if strict else True, "\n".join(report_lines)) + + report_lines.append( + f"[STATUS] STALE: Recorded commit {recorded_commit} does NOT exist in current repository history." + ) + return (False, "\n".join(report_lines)) + + +def main() -> int: + parser = argparse.ArgumentParser(description="Check AI-Memory freshness against git repository.") + parser.add_argument("--strict", action="store_true", help="Fail with exit code 1 on any discrepancy.") + parser.add_argument("--memory-path", type=Path, help="Path to CURRENT_STATE.md or AI-Memory root.") + parser.add_argument("--repo-path", type=Path, help="Path to git repository root.") + args = parser.parse_args() + + mem_path = args.memory_path + if mem_path and mem_path.is_dir(): + mem_path = mem_path / "01_PROJECTS" / DEFAULT_PROJECT / "CURRENT_STATE.md" + + is_fresh, summary = check_memory_freshness( + repo_path=args.repo_path, + memory_file=mem_path, + strict=args.strict, + ) + + print(summary) + return 0 if is_fresh else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_memory_freshness_a47.py b/tests/test_memory_freshness_a47.py new file mode 100644 index 0000000..9a22a7e --- /dev/null +++ b/tests/test_memory_freshness_a47.py @@ -0,0 +1,102 @@ +import subprocess +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +SCRIPTS_DIR = REPO_ROOT / "scripts" +if str(SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPTS_DIR)) + +from check_memory_freshness import ( + check_memory_freshness, + extract_recorded_commit, + get_git_commit, +) + + +def test_extract_recorded_commit_formats(): + assert extract_recorded_commit("Status: main = 80aab00\nDone") == "80aab00" + assert extract_recorded_commit("Canonical HEAD (main): `80aab00`\nOther text") == "80aab00" + assert extract_recorded_commit("- HEAD (main): c35bc48\n- Branch: main") == "c35bc48" + assert extract_recorded_commit("Some random text with no commit") is None + + +def test_check_memory_freshness_real_repo(): + canonical_memory = Path("/srv/projects/AI-Memory/01_PROJECTS/hermes-hub/CURRENT_STATE.md") + + if canonical_memory.exists(): + is_fresh, summary = check_memory_freshness( + repo_path=REPO_ROOT, + memory_file=canonical_memory, + strict=True, + ) + assert is_fresh is True + assert "FRESH" in summary + assert "80aab00" in summary + + +def test_check_memory_freshness_missing_file_strict_vs_non_strict(tmp_path): + missing_file = tmp_path / "NON_EXISTENT_CURRENT_STATE.md" + + # Non-strict mode should return True with warning for non-server runners + is_fresh_non_strict, summary_non_strict = check_memory_freshness( + repo_path=REPO_ROOT, + memory_file=missing_file, + strict=False, + ) + assert is_fresh_non_strict is True + assert "[WARNING]" in summary_non_strict + + # Strict mode should return False + is_fresh_strict, summary_strict = check_memory_freshness( + repo_path=REPO_ROOT, + memory_file=missing_file, + strict=True, + ) + assert is_fresh_strict is False + assert "[WARNING]" in summary_strict + + +def test_check_memory_freshness_stale_commit(tmp_path): + fake_memory = tmp_path / "CURRENT_STATE.md" + fake_memory.write_text("main = 0000000000000000000000000000000000000000\n", encoding="utf-8") + + is_fresh, summary = check_memory_freshness( + repo_path=REPO_ROOT, + memory_file=fake_memory, + strict=True, + ) + assert is_fresh is False + assert "STALE" in summary + + +def test_check_memory_freshness_valid_synthetic_memory(tmp_path): + head = get_git_commit(REPO_ROOT, "HEAD") + assert head is not None + + synthetic_memory = tmp_path / "CURRENT_STATE.md" + synthetic_memory.write_text(f"# Current State\n\nCanonical HEAD (main): `{head[:7]}`\n", encoding="utf-8") + + is_fresh, summary = check_memory_freshness( + repo_path=REPO_ROOT, + memory_file=synthetic_memory, + strict=True, + ) + assert is_fresh is True + assert "FRESH" in summary + + +def test_cli_execution(tmp_path): + script_path = REPO_ROOT / "scripts" / "check_memory_freshness.py" + + head = get_git_commit(REPO_ROOT, "HEAD") + synthetic_memory = tmp_path / "CURRENT_STATE.md" + synthetic_memory.write_text(f"main = {head[:7]}\n", encoding="utf-8") + + res = subprocess.run( + [sys.executable, str(script_path), "--memory-path", str(synthetic_memory), "--strict"], + capture_output=True, + text=True, + ) + assert res.returncode == 0 + assert "FRESH" in res.stdout