feat(memory): A47 memory freshness checker, tests, and CI/CD contract
This commit is contained in:
ochenstarik-ui
parent
80aab00ee2
commit
5abcb7d525
2 changed files with 307 additions and 0 deletions
205
scripts/check_memory_freshness.py
Normal file
205
scripts/check_memory_freshness.py
Normal file
|
|
@ -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 <path>] [--repo-path <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())
|
||||
102
tests/test_memory_freshness_a47.py
Normal file
102
tests/test_memory_freshness_a47.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue