#!/usr/bin/env python3 """Ingest-manual drift detection for the drift monitor (REQ-078, REQ-079). Flags when a repo's docs/-ingest-manual.html has drifted from the code, carries no commit stamp, or no longer conforms to ingest-manual/v1. Read-only, fail-open, and a no-op where there is no manual, so it ships safely downstream (a repo that never adopted the ingest-manual kit simply reports "nothing to drift"). Staleness is source-path-aware (REQ-079): the primary signal is whether the specific repo files/dirs the manual cites (paths inside ... ) have changed since the manual's meta.commit stamp -- silent for unrelated churn, immediate for relevant changes. This is strictly better than a commits-behind count, which is kept only as a coarse fallback for a manual that cites no extractable repo paths (nothing precise to check). Conformance is checked against the authoring kit's validator when it is vendored (ingest-manual-kit/validate_ingest_manual.py) so there is a single source of truth for the schema; where the kit is absent the check degrades to staleness only. The commits-behind helper is reused from prd_wiki_backfill so the fallback matches wiki_drift's arithmetic exactly. """ from __future__ import annotations import argparse import json import re import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) try: # reuse the exact HEAD/commits-behind logic wiki_drift uses from prd_wiki_backfill import _commits_behind except Exception: # pragma: no cover - fail-open if the sibling script moves import subprocess def _commits_behind(root, commit): if not commit or commit == "unknown": return None try: r = subprocess.run( ["git", "-C", str(root), "rev-list", "--count", f"{commit}..HEAD"], capture_output=True, text=True, timeout=15, ) if r.returncode == 0 and r.stdout.strip().isdigit(): return int(r.stdout.strip()) except Exception: pass return None _META_RE = re.compile( r']*id="ingest-manual-meta"[^>]*>(.*?)', re.DOTALL) # Repo-relative paths cited inside ... . We only trust tokens that # carry a directory separator (unambiguous) and then keep the ones that actually # resolve to a file or directory in the repo, so globs/placeholders like # skills//SKILL.md or .prd_plugin/state/*.json fall away naturally. _CODE_RE = re.compile(r"([^<>]+)") _PATH_TOKEN_RE = re.compile(r"^[A-Za-z0-9._/-]+$") def _cited_paths(text, root): """Repo-relative file/dir paths the manual cites and that exist in the repo. Excludes the docs/ tree itself so a manual never cites-and-drifts on itself.""" cited = set() for token in _CODE_RE.findall(text): token = token.strip() if "/" not in token or not _PATH_TOKEN_RE.match(token): continue rel = token.rstrip("/") if rel.startswith("docs/"): continue if (root / rel).exists(): cited.add(rel) return cited def _changed_since(root, commit): """Repo-relative paths changed between `commit` and HEAD, or None when that cannot be determined (not git / unknown commit) so the caller can fall back.""" if not commit or commit == "unknown": return None try: import subprocess r = subprocess.run( ["git", "-C", str(root), "diff", "--name-only", f"{commit}..HEAD"], capture_output=True, text=True, timeout=15, ) if r.returncode != 0: return None return {line.strip() for line in r.stdout.splitlines() if line.strip()} except Exception: return None def _cited_hits(cited, changed): """Changed paths that match a cited file exactly or fall under a cited dir.""" hits = [] for c in changed: for p in cited: if c == p or c.startswith(p + "/"): hits.append(c) break return hits def _load_validate(root): """Return the kit's validate(path, allow_name_mismatch) if the authoring kit is vendored in this repo, else None (staleness-only downstream).""" cand = Path(root) / "ingest-manual-kit" / "validate_ingest_manual.py" if not cand.is_file(): return None try: import importlib.util spec = importlib.util.spec_from_file_location("_im_validate_kit", cand) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return getattr(mod, "validate", None) except Exception: # pragma: no cover - fail-open return None def ingest_manual_drift(root, threshold=25): """Detect ingest-manual staleness for the drift monitor. Returns {status, findings, summary} -- the validator contract. When there is no manual, there is nothing to drift (status ok).""" root = Path(root) findings = [] try: docs = root / "docs" manuals = sorted(docs.glob("*-ingest-manual.html")) if docs.is_dir() else [] if not manuals: return {"status": "ok", "findings": [], "summary": "no ingest manual (nothing to drift)"} validate = _load_validate(root) drifted = stale = unstamped = nonconformant = 0 for man in manuals: rel = str(man.relative_to(root)).replace("\\", "/") text = man.read_text(encoding="utf-8-sig") if validate is not None: try: violations = validate(man, False) except Exception: # pragma: no cover - fail-open per manual violations = [] if violations: nonconformant += 1 findings.append({ "category": "ingest_manual_drift", "severity": "warning", "summary": f"ingest manual no longer conformant " f"({len(violations)} violations): {rel}"}) commit = None m = _META_RE.search(text) if m: try: commit = (json.loads(m.group(1)) or {}).get("commit") except Exception: commit = None if not commit or commit == "unknown": unstamped += 1 findings.append({ "category": "ingest_manual_drift", "severity": "info", "summary": f"ingest manual has no commit stamp in meta: {rel}"}) continue # Primary signal (REQ-079): did any file/dir the manual cites change # since it was stamped? Precise -- silent for unrelated churn. cited = _cited_paths(text, root) changed = _changed_since(root, commit) if cited and changed is not None: hits = _cited_hits(cited, changed) if hits: drifted += 1 extra = f" (+{len(hits) - 1} more)" if len(hits) > 1 else "" findings.append({ "category": "ingest_manual_drift", "severity": "warning", "summary": f"ingest manual cited source changed since stamp " f"(regenerate): {rel} -- e.g. {sorted(hits)[0]}{extra}"}) continue # Coarse fallback: no citable paths (nothing precise to check) -- fall # back to commits-behind-HEAD, matching wiki_drift. behind = _commits_behind(root, commit) if behind is not None and behind >= threshold: stale += 1 findings.append({ "category": "ingest_manual_drift", "severity": "warning", "summary": f"ingest manual {behind} commits behind HEAD " f"(no citable sources; regenerate candidate): {rel}"}) summary = (f"{len(manuals)} manual(s): {drifted} drifted (cited source changed), " f"{stale} stale (>= {threshold} behind, fallback), " f"{unstamped} unstamped, {nonconformant} non-conformant") return {"status": "attention" if findings else "ok", "findings": findings, "summary": summary} except Exception as exc: # pragma: no cover - fail-open return {"status": "ok", "findings": [], "summary": f"ingest manual drift skipped: {exc}"} def main(argv=None): parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--repo-root", default=".") parser.add_argument("--drift", action="store_true", help="Report ingest-manual staleness for the drift monitor.") parser.add_argument("--format", choices=("json", "markdown"), default=None) parser.add_argument("--output", help="Write the result here instead of stdout.") args = parser.parse_args(argv) result = ingest_manual_drift(Path(args.repo_root)) text = (result["summary"] if args.format == "markdown" else json.dumps(result, indent=2)) if args.output: out = Path(args.output) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(text, encoding="utf-8") else: print(text) return 0 if __name__ == "__main__": sys.exit(main())