tegwick
53c460c992
Duplicated-fact drift is the fourth error class and the only one with no executable rule. No positive control catches it (both copies are internally consistent) and re-derivation does not either (the copy reproduces whatever it was copied from). It is caught only by reading a copy against its source, which nothing in the loop required. facts.toml holds 15 facts and is GENERATED by `make facts-gen` from cb-cost, dep-weight and rule-coverage. The trap this task named — a hand-maintained registry that becomes another drifting copy — is closed by facts-check re-running the instruments and failing when the committed registry disagrees with them. A stale registry cannot certify stale artifacts. An artifact quoting a fact tags it: **$93.15** <!-- fact --> with the key. 17 occurrences across 5 artifacts are now checked. Falsified before being believed: changing CostAccounting.md line 158 from $93.15 to $92.87 — the exact historical drift — produced exit 1 naming the file, the line and the expected value. Tested against the class it exists to catch, on a real artifact, not only in its self-test. It then caught a live tag inside its own documentation example in InnerLoop.md within the hour. Third time a gate has failed on its own pass's work. What it does not close is stated rather than implied: 22 untagged literal copies remain and are reported, not failed. Tagging is opt-in, a number can legitimately recur, and a gate that fires on coincidence gets routed around. Naming the uncovered surface beats claiming the class is closed. InnerLoop single-source-of-fact moves from prose to executable — v1.3. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
369 lines
14 KiB
Python
369 lines
14 KiB
Python
#!/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")
|
|
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)
|