tai/research/audit.py

#!/usr/bin/env python3
"""research/audit.py — verify the research archive is internally consistent.

Run after any research session, and any time someone asks "is research/ up
to date": `python3 research/audit.py`. Stdlib only, no dependencies.

Checks, per subject folder:
  - every source .md has a Tier metadata row (1 or 2) and the required
    Source URL / Document type fields;
  - every Tier 1 dossier has non-empty Pros/Cons sections (an empty one is
    the "wasn't actually read critically" smell research/README.md calls out);
  - the counts in this script's own scan match what research/INDEX.md claims
    (catches INDEX.md drift, the thing that bit this archive once already).

This does NOT check whether a research task that happened got archived at
all — that failure mode (skipping archival silently) is caught by the
PostToolUse/WebSearch hook in .claude/settings.json, not by auditing files
that were never written. This script only audits what's already on disk.
"""
import re
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent
TIER_RE = re.compile(r"^\|\s*Tier\s*\|\s*(1|2|N/A)\s*\|", re.MULTILINE)
DOCTYPE_RE = re.compile(r"^\|\s*Document type\s*\|\s*(.+?)\s*\|", re.MULTILINE)
SOURCE_URL_RE = re.compile(r"^\|\s*Source URL\s*\|", re.MULTILINE)
INDEX_SUBJECT_ROW_RE = re.compile(
    r"\[`([a-z0-9-]+)`\]\([a-z0-9-]+/\)\s*\|\s*(\d+)\s*\(\s*(\d+)\s*/\s*(\d+)\s*\)\s*\|\s*(\d+)\s*\|"
)


def subject_dirs():
    return sorted(p for p in ROOT.iterdir() if p.is_dir() and p.name != "conclusions")


def scan_subject(path):
    sources = sorted(p for p in path.glob("*.md"))
    conclusions = sorted((path / "conclusions").glob("*.md")) if (path / "conclusions").is_dir() else []
    tier1 = tier2 = 0
    problems = []
    for src in sources:
        text = src.read_text(encoding="utf-8", errors="replace")
        tier_match = TIER_RE.search(text)
        if not tier_match:
            problems.append("%s: no 'Tier' metadata row" % src.name)
            continue
        tier = tier_match.group(1)
        if tier == "1":
            tier1 += 1
            if "## Pros / strengths" in text and "## Cons / limitations / caveats" in text:
                pros = text.split("## Pros / strengths", 1)[1].split("## Cons", 1)[0].strip()
                cons = text.split("## Cons / limitations / caveats", 1)[1].split("## Relevance", 1)[0].strip()
                if not pros:
                    problems.append("%s: Tier 1 but empty Pros section" % src.name)
                if not cons:
                    problems.append("%s: Tier 1 but empty Cons section" % src.name)
            else:
                problems.append("%s: Tier 1 missing Pros/Cons sections entirely" % src.name)
            if not SOURCE_URL_RE.search(text):
                problems.append("%s: Tier 1 missing Source URL field" % src.name)
        elif tier == "2":
            tier2 += 1
        else:
            problems.append("%s: unexpected Tier value" % src.name)
    for conc in conclusions:
        text = conc.read_text(encoding="utf-8", errors="replace")
        for required in ("## Question / problem statement", "## Conclusion", "## Action taken"):
            if required not in text:
                problems.append("%s: missing required section %r" % (conc.name, required))
    return {
        "name": path.name,
        "sources": len(sources),
        "tier1": tier1,
        "tier2": tier2,
        "conclusions": len(conclusions),
        "problems": problems,
    }


def parse_index_claims():
    index_path = ROOT / "INDEX.md"
    if not index_path.exists():
        return {}
    text = index_path.read_text(encoding="utf-8", errors="replace")
    claims = {}
    for match in INDEX_SUBJECT_ROW_RE.finditer(text):
        subject, total, t1, t2, conclusions = match.groups()
        claims[subject] = {"sources": int(total), "tier1": int(t1), "tier2": int(t2), "conclusions": int(conclusions)}
    return claims


def main():
    subjects = subject_dirs()
    if not subjects:
        print("no subject folders found under research/ — nothing to audit")
        return 0

    claims = parse_index_claims()
    exit_code = 0
    total_sources = total_tier1 = total_tier2 = total_conclusions = 0

    for path in subjects:
        report = scan_subject(path)
        total_sources += report["sources"]
        total_tier1 += report["tier1"]
        total_tier2 += report["tier2"]
        total_conclusions += report["conclusions"]

        print("%s: %d sources (T1=%d T2=%d), %d conclusions" % (
            report["name"], report["sources"], report["tier1"], report["tier2"], report["conclusions"]
        ))

        for problem in report["problems"]:
            print("  PROBLEM: %s" % problem)
            exit_code = 1

        claim = claims.get(report["name"])
        if claim is None:
            print("  PROBLEM: INDEX.md has no row for this subject")
            exit_code = 1
        elif claim != {"sources": report["sources"], "tier1": report["tier1"], "tier2": report["tier2"], "conclusions": report["conclusions"]}:
            print("  PROBLEM: INDEX.md claims %s, disk has sources=%d (T1=%d T2=%d) conclusions=%d"
                  % (claim, report["sources"], report["tier1"], report["tier2"], report["conclusions"]))
            exit_code = 1

    print("\ntotal: %d subjects, %d sources (T1=%d T2=%d), %d conclusions" % (
        len(subjects), total_sources, total_tier1, total_tier2, total_conclusions
    ))
    print("PASS" if exit_code == 0 else "FAIL -- see PROBLEM lines above")
    return exit_code


if __name__ == "__main__":
    sys.exit(main())