clay-borg/tools/rule-coverage.py

#!/usr/bin/env python3
"""AM-1 / M-D1-COV and AM-1b / M-D1-LNK, over **every** numbered spec.

Two denominators, deliberately separate because they are measured
differently:

  ground  GR-rules in specs/GroundRules.md, covered by `covers:` lists in
          scenarios/ground/*.yaml, and linked to the aggregate source.
  kernel  K-rules in specs/GameKernel.md. **Link only.** K-rules are
          kernel invariants, not game rules: there is no kernel aggregate,
          no setup preset and no command vocabulary, so a
          `scenarios/kernel/*.yaml` with `covers: [K11]` would be a tag in
          a directory the runner cannot dispatch. Claiming scenario
          coverage for them would be the inflation this gate exists to
          prevent.

Until CB-WP-0005 T01, `AGGREGATE` was one file and the rule pattern
matched `GR-` only, so the kernel spec was outside the instrument
entirely. K10, K14 and K18 were unimplemented for four workplans while
`make coverage` printed `58/58 (100%)`.

Stated limit (InnerLoop implementation rule 4), now doubly important:
**this gate counts names.** It proves no rule is unclaimed, no claimed
rule is invented, and no rule is absent from the source. It does NOT
prove anything fails when a rule is violated — that is M-D1-MUT
(`make mutation-check`, CB-WP-0005 T02), and four of the seven defects
found in CB-RES-0004 were invisible to a name-based check.

Positive control (InnerLoop v1.1 §Step 5): every denominator asserts it
actually found rules. Before this was added, a broken spec regex yielded
rules=[] and missing=[] and the tool exited 0 reporting "0/0" — the
harness-does-nothing class, in the tool that reports our headline
coverage number. A new denominator inherits the control, or it
reintroduces the defect the old one was fixed for.

Usage:
  python3 tools/rule-coverage.py
  python3 tools/rule-coverage.py --self-test
"""
import datetime
import glob
import os
import re
import sys

from repo import enter_root

RULE_RE = r"\*\*(GR-[A-Z]+\d+)"
COVERS_RE = r"covers: \[(.*?)\]"
AGGREGATE = "games/ground/src/lib.rs"
# CB-WP-0003 T08: a provisional default with no expiry can shape the kernel
# indefinitely while looking handled. CI warns; it does not break the build,
# because the ruling is a ground-game decision we cannot make for them.
PROVISIONAL_WARN_DAYS = 30
ID_RE = r"GR-[A-Z]+\d+"

# ADR-0005 §5: a newly widened denominator is not a regression, so the
# kernel arm reports without feeding the exit code — but only until a
# date that lives in the tool rather than in prose. An open-ended "we
# will gate it later" is how AM-4's targets went unratified for four
# workplans. The remaining days are printed on every run.
KERNEL_GATES_FROM = datetime.date(2026, 8, 31)

# Source roots searched for rule IDs. A list, not one file: K-rules live
# in cb-kernel, cb-events and cb-game-runtime, so a single-file AGGREGATE
# would report every one of them unlinked forever.
SOURCE_ROOTS = ("crates", "games", "tools")


def source_files(roots=SOURCE_ROOTS):
    """Every .rs file under the given roots, excluding build output."""
    out = []
    for root in roots:
        for dirpath, dirnames, files in os.walk(root):
            dirnames[:] = [d for d in dirnames if d != "target"]
            out += [os.path.join(dirpath, f) for f in sorted(files)
                    if f.endswith(".rs")]
    return sorted(out)


def parse_rules(spec_text, pattern=RULE_RE):
    return sorted(set(re.findall(pattern, spec_text)))


def parse_code_ids(text, pattern=ID_RE):
    """Rule IDs named anywhere in the source (T09)."""
    return set(re.findall(pattern, text))


def code_ids_over(paths, pattern):
    """Union of rule IDs named across many source files."""
    found = set()
    for p in paths:
        found |= parse_code_ids(open(p).read(), pattern)
    return found


def provisional_items(paths):
    """(path, owner, raised) for every scenario encoding a U-item default."""
    out = []
    for path in paths:
        text = open(path).read()
        if not re.search(r"^provisional:\s*true", text, re.M):
            continue
        owner = re.search(r"^provisional_owner:\s*(\S+)", text, re.M)
        raised = re.search(r"^provisional_raised:\s*(\S+)", text, re.M)
        out.append((path,
                    owner.group(1) if owner else None,
                    raised.group(1) if raised else None))
    return out


def parse_covers(text):
    match = re.search(COVERS_RE, text, re.S)
    if not match:
        return set()
    return {c.strip() for c in match.group(1).split(",") if c.strip()}


def _silent_output():
    """Whatever the kernel arm prints with quiet=True — must be nothing."""
    import io
    from contextlib import redirect_stdout

    buf = io.StringIO()
    with redirect_stdout(buf):
        kernel_arm(today=datetime.date(2026, 1, 1), quiet=True)
    return buf.getvalue()


def self_test():
    """Each assertion pins a failure this tool must detect."""
    results = []

    def check(name, ok, detail=""):
        results.append((name, ok, detail))

    # The defect that motivated this control: a spec that parses to zero
    # rules must not be reportable as coverage.
    check("zero rules detected as a failure", parse_rules("no rules here") == [],
          "empty spec yields no rules; main() now aborts on this")
    # The matcher must actually match the real format.
    check("rule matcher works on real spec format",
          parse_rules("**GR-R06** something\n**GR-A12** other")
          == ["GR-A12", "GR-R06"])
    # covers: parsing, including the empty case.
    check("covers matcher works", parse_covers("covers: [GR-R06, GR-A12]")
          == {"GR-R06", "GR-A12"})
    check("missing covers yields empty set", parse_covers("no covers key") == set())
    # T09: the spec -> code link must be detectable.
    check("code-id matcher finds ids in source",
          parse_code_ids("// GR-R06: lead first\nfn f(){} // GR-A12")
          == {"GR-R06", "GR-A12"})
    check("code-id matcher finds none in unmarked source",
          parse_code_ids("fn f() { let x = 1; }") == set())
    # T08: every provisional scenario must carry an owner and a date.
    import glob as _g
    prov = provisional_items(sorted(_g.glob("scenarios/ground/*.yaml")))
    check("every provisional item has an owner and a raised date",
          bool(prov) and all(o and r for _, o, r in prov),
          f"{len(prov)} provisional item(s)")

    # --- CB-WP-0005 T01: the kernel denominator ---
    # The control the old arm was fixed for, replicated. A pattern that
    # stops matching must abort, not report 0/0 as though it were 100%.
    check("kernel: zero rules detected as a failure",
          parse_rules("no kernel rules here", r"\*\*(K\d+)\*\*") == [])
    check("kernel: matcher works on the real spec format",
          parse_rules("- **K10** A replay bundle\n- **K9** A snapshot",
                      r"\*\*(K\d+)\*\*") == ["K10", "K9"])
    check("kernel: matcher does not match GR-rules",
          parse_rules("**GR-R06** lead first", r"\*\*(K\d+)\*\*") == [])
    # A single-file AGGREGATE reported every K-rule unlinked forever; the
    # union across roots is the fix, so assert it actually unions.
    # Compute once and report the same value that was asserted. Building
    # the detail string with a second, re-escaped copy of the pattern
    # printed "0 K-ids" beside a passing ">5" assertion — a label that
    # contradicts its own check is worse than no label.
    srcs = source_files()
    k_in_src = code_ids_over(srcs, r"\bK\d+\b")
    check("kernel: ids union across many files, not just one",
          code_ids_over([os.devnull], r"\bK\d+\b") == set() and len(k_in_src) > 5,
          f"{len(k_in_src)} K-ids across {len(srcs)} files")
    check("kernel: source roots resolve to real files",
          len(srcs) >= 5, f"{len(srcs)} .rs files")

    # The gate date must actually change behaviour, in both directions.
    # A "binds later" that never binds is the AM-4 failure this replaces.
    # What the gate must do depends on whether anything is unlinked, so
    # compute that rather than assuming it. An earlier version hardcoded
    # "unlinked rules exist today" and failed the moment T07 linked the
    # last one — correctly, but for the wrong reason.
    _k = parse_rules(open(os.path.join("specs", "GameKernel.md")).read(),
                     r"\*\*(K\d+)\*\*")
    _named = code_ids_over(source_files(), r"\bK\d+\b")
    _unlinked = [r for r in _k if r not in _named]
    before = kernel_arm(today=datetime.date(2026, 1, 1), quiet=True)
    after = kernel_arm(today=datetime.date(2027, 1, 1), quiet=True)

    # The reporting path must be exercised, not only the quiet one. When
    # this control ran `quiet=True` exclusively, a broken `say()` made
    # every real `make coverage` die with RecursionError while the
    # self-test printed all-ok — a positive control that named the
    # behaviour without asserting it, which is exactly the defect
    # CB-RES-0004 is about.
    import io
    from contextlib import redirect_stdout

    buf = io.StringIO()
    with redirect_stdout(buf):
        loud = kernel_arm(today=datetime.date(2026, 1, 1))
    out = buf.getvalue()
    check("kernel: the reporting path actually prints",
          loud == before and "AM-1b kernel spec->code link:" in out
          and "gate:" in out,
          f"{len(out.splitlines())} lines")
    check("kernel: quiet suppresses output, loud does not",
          out.strip() != "" and _silent_output() == "")
    check("kernel: gate never fails before the binding date",
          before == 0, f"before={before}")
    check("kernel: after the binding date the gate fails iff rules are unlinked",
          after == (2 if _unlinked else 0),
          f"after={after}, {len(_unlinked)} unlinked"
          + (f" ({' '.join(_unlinked)})" if _unlinked else " — all linked"))

    print("rule-coverage self-test (positive control)")
    ok = True
    for name, passed, detail in results:
        print(f"  [{'ok ' if passed else 'FAIL'}] {name}"
              + (f"  — {detail}" if detail else ""))
        ok &= passed
    return 0 if ok else 1


def main():
    # T01: inputs are repo-relative, so anchor to the repo rather than
    # requiring the caller to `cd` first.
    enter_root()
    if "--self-test" in sys.argv:
        return self_test()

    rules = parse_rules(open("specs/GroundRules.md").read())
    paths = sorted(glob.glob("scenarios/ground/*.yaml"))

    # Positive control: refuse to report a percentage over nothing.
    if not rules:
        print("ERROR — no GR-rules parsed from specs/GroundRules.md; "
              "refusing to report coverage", file=sys.stderr)
        return 1
    if not paths:
        print("ERROR — no scenarios found in scenarios/ground/; "
              "refusing to report coverage", file=sys.stderr)
        return 1

    covered = set()
    for path in paths:
        covered |= parse_covers(open(path).read())

    known = set(rules)
    hit = sorted(known & covered)
    missing = [r for r in rules if r not in covered]
    invented = sorted(covered - known)

    # T09: the spec -> code -> scenario chain, made mechanical. A rule a
    # scenario claims should also be named in the aggregate, or the claim
    # rests on nothing but a tag.
    code_ids = parse_code_ids(open(AGGREGATE).read())
    unlinked = sorted((known & covered) - code_ids)
    phantom = sorted(code_ids - known)

    pct = 100 * len(hit) // len(rules)
    linked = len((known & covered) & code_ids)
    print(f"AM-1 rule coverage: {len(hit)}/{len(rules)} ({pct}%) "
          f"over {len(paths)} scenarios")
    print(f"AM-1b spec->code link: {linked}/{len(hit)} claimed rules also "
          f"named in {AGGREGATE}")
    print("  NOTE: counts tags; does not prove a scenario exercises what it names")
    if unlinked:
        print("  unlinked (claimed by a scenario, absent from the aggregate):")
        print("   ", " ".join(unlinked))
    if phantom:
        print("  ERROR — rule id in code that the spec does not define:",
              " ".join(phantom), file=sys.stderr)
        return 1
    # T08: provisional items are reported with an owner and an age.
    prov = provisional_items(paths)
    if prov:
        import datetime
        today = datetime.date.today()
        print(f"\nprovisional U-item defaults: {len(prov)}")
        unowned, stale = [], []
        for path, owner, raised in prov:
            age = "?"
            if raised:
                try:
                    age = (today - datetime.date.fromisoformat(raised)).days
                except ValueError:
                    age = "?"
            name = path.split("/")[-1]
            print(f"  {name:<34} owner={owner or 'NONE':<12} age={age}d")
            if not owner:
                unowned.append(name)
            if isinstance(age, int) and age > PROVISIONAL_WARN_DAYS:
                stale.append(f"{name} ({age}d)")
        if unowned:
            print("  WARN — provisional with no owner:", " ".join(unowned))
        if stale:
            print(f"  WARN — provisional for over {PROVISIONAL_WARN_DAYS} days:",
                  " ".join(stale))
        print("  NOTE: evidence files must list these; a ruling flips the "
              "scenario, not the kernel")

    if missing:
        print("  uncovered:", " ".join(missing))
    if invented:
        print("  ERROR — claimed but not defined in the spec:", " ".join(invented),
              file=sys.stderr)
        return 1

    kernel_rc = kernel_arm()
    if kernel_rc:
        return kernel_rc
    return 0 if not missing else 2


def kernel_arm(today=None, quiet=False):
    """AM-1b over specs/GameKernel.md × every crate (CB-WP-0005 T01).

    Link only — see the module docstring for why K-rules cannot use the
    scenario `covers:` mechanism. Returns a non-zero code only once
    KERNEL_GATES_FROM has passed.
    """
    today = today or datetime.date.today()

    def say(*a, **kw):
        if not quiet:
            print(*a, **kw)

    spec = os.path.join("specs", "GameKernel.md")
    rules = parse_rules(open(spec).read(), r"\*\*(K\d+)\*\*")

    # The inherited positive control. A pattern that stops matching must
    # abort, not report 0/0 as though it were an answer.
    if not rules:
        print(f"\nERROR — no K-rules parsed from {spec}; refusing to report "
              f"kernel coverage", file=sys.stderr)
        return 1
    paths = source_files()
    if not paths:
        print("\nERROR — no source files found under "
              f"{'/, '.join(SOURCE_ROOTS)}/; refusing to report kernel coverage",
              file=sys.stderr)
        return 1

    named = code_ids_over(paths, r"\bK\d+\b")
    known = set(rules)
    linked = sorted(known & named)
    unlinked = [r for r in rules if r not in named]
    phantom = sorted(named - known)

    days = (KERNEL_GATES_FROM - today).days
    binding = days <= 0
    pct = 100 * len(linked) // len(rules)
    say(f"\nAM-1b kernel spec->code link: {len(linked)}/{len(rules)} ({pct}%) "
          f"K-rules named across {len(paths)} source files")
    say("  NOTE: link only — K-rules are kernel invariants with no scenario "
          "mechanism; and this counts names, not assertions (see "
          "`make mutation-check`)")
    if binding:
        say(f"  gate: BINDING since {KERNEL_GATES_FROM}")
    else:
        say(f"  gate: reporting only for {days} more day(s), binds "
              f"{KERNEL_GATES_FROM} (ADR-0005 §5)")
    if unlinked:
        say("  unlinked (declared in the spec, named nowhere in source):")
        say("   ", " ".join(unlinked))
    if phantom:
        say("  ERROR — K-id in code that the spec does not define:",
              " ".join(phantom), file=sys.stderr)
        return 1
    if unlinked and binding:
        return 2
    return 0


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