clay-borg/tools/facts.py

#!/usr/bin/env python3
"""The fact registry: a number lives in one place, and copies are checked.

CB-WP-0004 T04. Duplicated-fact drift (DFD) is the fourth error class on
record and the only one with no executable gate. Two instances:

  * `specs/MetricsAndScenarios.md` §1a inlined a copy of the price sheet;
    the real sheet changed and the copy went stale within the hour.
  * the acceptance figure moved $92.21 -> $92.87 -> $93.32 -> $93.15 and
    each move had to be chased by hand across a survey, a workplan and an
    evidence file.

No positive control catches DFD — both copies are internally consistent —
and re-derivation does not either, because the copy reproduces whatever it
was copied from. It is caught only by reading a copy against its source.
That is what this does.

## The trap, and how it is avoided

A hand-maintained registry is just another copy that drifts. So the
registry is **generated from the instruments** (`make facts-gen`), never
edited: every value here is produced by cb-cost, dep-weight or
rule-coverage on the current tree. `--check` re-runs the instruments and
fails if the committed registry disagrees with them, so a stale registry
cannot silently certify stale artifacts.

## How an artifact quotes a fact

Tag the occurrence with an HTML comment naming the registry key:

    The benchmark to beat is **$93.15**. <!-- fact:pinned_total -->

`make facts-check` re-reads every tagged occurrence and fails when the
text disagrees with the registry. Tagging is opt-in, so the gate also
reports **untagged copies** — literal occurrences of a registry value in
artifacts that did not declare them — which is the drift surface that is
not yet covered.

Usage:
  python3 tools/facts.py --gen        # regenerate facts.toml from instruments
  python3 tools/facts.py --check      # gate: registry vs instruments vs artifacts
  python3 tools/facts.py --self-test
"""
import datetime
import glob
import importlib.util
import os
import re
import subprocess
import sys

from repo import ROOT, enter_root

REGISTRY = os.path.join(ROOT, "facts.toml")
PIN = "fc76445"  # CB-WP-0001 acceptance pin (specs/CostAccounting.md §7)
TAG_RE = re.compile(r"<!--\s*fact:([a-z0-9_]+)\s*-->")
# Artifact kinds that quote measured numbers. `history/` is excluded from
# the untagged sweep: it narrates corrections and legitimately states
# superseded values.
ARTIFACT_DIRS = ("specs", "evidence", "research", "decisions", "workplans")

try:
    import tomllib
except ModuleNotFoundError:  # pragma: no cover
    print("ERROR: needs Python 3.11+ for tomllib", file=sys.stderr)
    sys.exit(1)


def _load(script):
    spec = importlib.util.spec_from_file_location(
        script.replace("-", "_").replace(".py", ""),
        os.path.join(ROOT, "tools", script))
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


# --------------------------------------------------------------- generate


def measure():
    """Every registry value, produced by the instrument that owns it.

    Each entry is (value, format, instrument). The format string is how the
    value appears in prose, so the check compares like with like rather
    than re-deriving a rendering in two places.
    """
    facts = {}

    cb = _load("cb-cost.py")
    rep = cb.collect("-home-worsch-clay-borg", PIN)
    facts["pinned_total"] = (rep["total"], "${:,.2f}",
                             f"tools/cb-cost.py --pin {PIN}")
    facts["pinned_main"] = (rep["main_total"], "${:,.2f}",
                            f"tools/cb-cost.py --pin {PIN}")
    facts["pinned_subagent"] = (rep["subagent_total"], "${:,.2f}",
                                f"tools/cb-cost.py --pin {PIN}")
    facts["pinned_responses"] = (rep["responses"], "{:,}",
                                 f"tools/cb-cost.py --pin {PIN}")
    mix = rep["tool_mix"]
    facts["pinned_mechanical_cost"] = (mix["mechanical_cost"], "${:,.2f}",
                                       f"tools/cb-cost.py --pin {PIN}")
    facts["pinned_mechanical_turns"] = (mix["mechanical_turns"], "{:,}",
                                        f"tools/cb-cost.py --pin {PIN}")
    facts["pinned_mechanical_share"] = (
        round(100 * mix["mechanical_cost"] / rep["total"]), "{:d}%",
        f"tools/cb-cost.py --pin {PIN}")

    dw = _load("dep-weight.py")
    for label, key in (("shipped-runtime", "am4a_loc"), ("dev-toolchain", "am4b_loc")):
        found = dw.crates(dw.CONFIGS[label])
        total = sum(dw.source_lines(n, v) for n, v in sorted(found.items()))
        facts[key] = (total, "{:,}", "tools/dep-weight.py")
    facts["am4a_target"] = (dw.TARGETS["shipped-runtime"], "{:,}",
                            "tools/dep-weight.py TARGETS")
    facts["am4b_target"] = (dw.TARGETS["dev-toolchain"], "{:,}",
                            "tools/dep-weight.py TARGETS")

    rc = _load("rule-coverage.py")
    rules = rc.parse_rules(open(os.path.join(ROOT, "specs/GroundRules.md")).read())
    paths = sorted(glob.glob(os.path.join(ROOT, "scenarios/ground/*.yaml")))
    covered = set()
    for p in paths:
        covered |= rc.parse_covers(open(p).read())
    code_ids = rc.parse_code_ids(open(os.path.join(ROOT, rc.AGGREGATE)).read())
    hit = set(rules) & covered
    facts["gr_rules"] = (len(rules), "{:,}", "tools/rule-coverage.py")
    facts["gr_covered"] = (len(hit), "{:,}", "tools/rule-coverage.py")
    facts["gr_linked"] = (len(hit & code_ids), "{:,}", "tools/rule-coverage.py")
    facts["gr_scenarios"] = (len(paths), "{:,}", "tools/rule-coverage.py")

    # CB-WP-0005 T01: the kernel denominator, so its numbers are under the
    # DFD gate from the day they first exist rather than after they drift.
    k_rules = rc.parse_rules(
        open(os.path.join(ROOT, "specs/GameKernel.md")).read(), r"\*\*(K\d+)\*\*")
    k_named = rc.code_ids_over(rc.source_files(), r"\bK\d+\b")
    facts["k_rules"] = (len(k_rules), "{:,}", "tools/rule-coverage.py")
    facts["k_linked"] = (len(set(k_rules) & k_named), "{:,}",
                         "tools/rule-coverage.py")
    facts["k_unlinked"] = (
        " ".join(r for r in k_rules if r not in k_named), "{}",
        "tools/rule-coverage.py")
    return facts


def render(value, fmt):
    return fmt.format(value)


def generate():
    facts = measure()
    lines = [
        "# GENERATED — do not edit. `make facts-gen` rewrites this file.",
        "#",
        "# The single source of fact for numbers that appear in more than",
        "# one artifact (InnerLoop v1.2). Every value here is produced by",
        "# the instrument named in its `by` field, on the current tree.",
        "# `make facts-check` fails if this file disagrees with the",
        "# instruments, or if a tagged artifact disagrees with this file.",
        "",
        f'generated = "{datetime.date.today().isoformat()}"',
        f'pin = "{PIN}"',
        "",
    ]
    for key in sorted(facts):
        value, fmt, by = facts[key]
        lines += [
            f"[{key}]",
            f"value = {value!r}",
            f'text = "{render(value, fmt)}"',
            f'fmt = "{fmt}"',
            f'by = "{by}"',
            "",
        ]
    with open(REGISTRY, "w") as fh:
        fh.write("\n".join(lines))
    print(f"wrote {os.path.relpath(REGISTRY, ROOT)} — {len(facts)} facts")
    for key in sorted(facts):
        print(f"  {key:<26} {facts[key][1].format(facts[key][0]):>12}"
              f"   {facts[key][2]}")
    return 0


# ------------------------------------------------------------------ check


def load_registry():
    if not os.path.isfile(REGISTRY):
        return None
    with open(REGISTRY, "rb") as fh:
        return tomllib.load(fh)


def artifacts():
    for d in ARTIFACT_DIRS:
        base = os.path.join(ROOT, d)
        if not os.path.isdir(base):
            continue
        for path in sorted(glob.glob(os.path.join(base, "**", "*.md"),
                                     recursive=True)):
            yield os.path.relpath(path, ROOT)


def tagged_occurrences(text):
    """(key, line_no, the text of that line) for each fact tag."""
    out = []
    for i, line in enumerate(text.splitlines(), 1):
        for m in TAG_RE.finditer(line):
            out.append((m.group(1), i, line))
    return out


def check():
    reg = load_registry()
    if reg is None:
        print(f"ERROR — {os.path.relpath(REGISTRY, ROOT)} missing; "
              f"run `make facts-gen`", file=sys.stderr)
        return 1
    keys = {k: v for k, v in reg.items() if isinstance(v, dict)}

    findings = []

    # 1. The registry must still agree with the instruments. Without this
    #    a stale registry would happily certify stale artifacts — the
    #    trap this task was warned about.
    live = measure()
    for key, (value, fmt, _by) in sorted(live.items()):
        if key not in keys:
            findings.append(f"registry missing {key} (instrument says "
                            f"{render(value, fmt)}) — run `make facts-gen`")
            continue
        if render(value, fmt) != keys[key]["text"]:
            findings.append(
                f"registry stale: {key} = {keys[key]['text']} but "
                f"{keys[key]['by']} now measures {render(value, fmt)} "
                f"— run `make facts-gen`")
    for key in sorted(set(keys) - set(live)):
        findings.append(f"registry has {key}, no instrument produces it "
                        f"— hand-edited?")

    # 2. Every tagged occurrence must state the registry value.
    tagged = 0
    for rel in artifacts():
        text = open(os.path.join(ROOT, rel)).read()
        for key, lineno, line in tagged_occurrences(text):
            tagged += 1
            if key not in keys:
                findings.append(f"{rel}:{lineno} tags unknown fact {key!r}")
                continue
            if keys[key]["text"] not in line:
                findings.append(
                    f"{rel}:{lineno} claims fact:{key} but does not state "
                    f"{keys[key]['text']}  |  {line.strip()[:70]}")

    # 3. Positive control. A gate that checked nothing would pass silently
    #    — the harness-does-nothing class, in the tool meant to close DFD.
    if tagged == 0:
        print("ERROR — no fact tags found in any artifact; the check verified "
              "nothing. Tag at least one occurrence, or delete this gate.",
              file=sys.stderr)
        return 1

    # 4. Report untagged copies: the drift surface still uncovered. This
    #    reports and does not fail — a number can legitimately recur (a
    #    round figure, a year), and failing on that would make the gate
    #    something people route around.
    untagged = {}
    for rel in artifacts():
        text = open(os.path.join(ROOT, rel)).read()
        declared = {k for k, _, _ in tagged_occurrences(text)}
        for i, line in enumerate(text.splitlines(), 1):
            if TAG_RE.search(line):
                continue
            for key, spec in keys.items():
                if key in declared:
                    continue
                # Only distinctive values: a bare "58" is everywhere.
                if len(spec["text"]) < 5:
                    continue
                if spec["text"] in line:
                    untagged.setdefault(key, []).append(f"{rel}:{i}")

    print("facts-check — single source of fact (DFD gate)")
    print(f"  registry     {len(keys)} facts, generated {reg.get('generated')}, "
          f"pin {reg.get('pin')}")
    print(f"  tagged       {tagged} occurrence(s) checked against the registry")
    if untagged:
        n = sum(len(v) for v in untagged.values())
        print(f"  untagged     {n} literal copy/copies, not covered by the gate:")
        for key in sorted(untagged):
            print(f"    {key:<26} {' '.join(untagged[key][:6])}"
                  + (" …" if len(untagged[key]) > 6 else ""))
        print("    NOTE: reported, not failed — tag them to bring them under "
              "the gate")

    if findings:
        print(f"\n{len(findings)} finding(s):", file=sys.stderr)
        for f in findings:
            print(f"  [facts] {f}", file=sys.stderr)
        return 1
    print("  no findings")
    return 0


# --------------------------------------------------------------- self-test


def self_test():
    """Each assertion pins a failure this gate must detect.

    The class being closed is DFD, so the assertions are about *copies*:
    a copy that disagrees must fail, a copy that agrees must pass, and a
    run that checked no copies at all must not report success.
    """
    results = []

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

    check_("tag is recognised in prose",
           tagged_occurrences("cost was **$93.15** <!-- fact:pinned_total -->")
           == [("pinned_total", 1, "cost was **$93.15** <!-- fact:pinned_total -->")])
    check_("untagged prose yields no occurrence",
           tagged_occurrences("cost was $93.15") == [])
    check_("tag with spaces is recognised",
           [k for k, _, _ in tagged_occurrences("x <!--  fact:am4a_loc  -->")]
           == ["am4a_loc"])
    check_("a malformed tag is not silently accepted",
           tagged_occurrences("<!-- facts:pinned_total -->") == [])

    # The DFD detection itself: agreeing and disagreeing copies.
    spec = {"text": "$93.15"}
    agree = "the benchmark is **$93.15** <!-- fact:pinned_total -->"
    drift = "the benchmark is **$92.87** <!-- fact:pinned_total -->"
    check_("an agreeing copy passes", spec["text"] in agree)
    check_("a drifted copy is detected", spec["text"] not in drift,
           "this is the whole class: both lines are internally consistent")

    # Registry must exist, be complete, and match the instruments.
    reg = load_registry()
    check_("registry exists", reg is not None)
    if reg:
        keys = {k: v for k, v in reg.items() if isinstance(v, dict)}
        check_("registry is non-empty", len(keys) >= 8, f"{len(keys)} facts")
        check_("every fact records the instrument that produced it",
               all(v.get("by") for v in keys.values()))
        check_("every fact records both a value and its rendered text",
               all("value" in v and v.get("text") for v in keys.values()))
        check_("registry is generated, not hand-written",
               open(REGISTRY).read().startswith("# GENERATED"))

    # At least one artifact must actually be tagged, or the gate is inert.
    total = sum(len(tagged_occurrences(open(os.path.join(ROOT, r)).read()))
                for r in artifacts())
    check_("at least one artifact is tagged (gate is not inert)", total > 0,
           f"{total} tagged occurrence(s)")

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


def main():
    enter_root()
    if "--self-test" in sys.argv:
        return self_test()
    if "--gen" in sys.argv:
        return generate()
    return check()


if __name__ == "__main__":
    try:
        sys.exit(main())
    except subprocess.CalledProcessError as e:
        print(f"ERROR — {e}", file=sys.stderr)
        sys.exit(1)