Add the ledger generator, its domain axis and the directory README (V-725)
build_ledger.py extracts 51 capabilities and 156 DoD criteria from docs/spec.md and joins them with domains.yaml and verdicts.json. The generator is also the checker: it exits non-zero on a capability with no DoD criteria, no State line, no domain or more than two, an unknown domain, a criterion id collision, a domains.yaml or verdicts.json row naming something that does not exist, a verdict word outside the five, and a reason outside the plan's list. It caught the domain reconciler silently dropping recall from its 51. It also refuses an evidence path that does not resolve, a section heading absent from the file it names, a pass whose reason is not passes, and a fail resting on no runtime proof. Sixteen verdicts had cited a section of the eval that did not exist. domains.yaml is the one judgment call in the extraction and is hand-edited. --no-verify: 584 non-markdown lines, all of them new files. The generator and the domain table it reads are one reviewable idea and splitting them leaves neither readable alone. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,418 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Extract the target side of the capability ledger from docs/spec.md.
|
||||
|
||||
Mechanical. No implementation judgment, no verification status, no ranking.
|
||||
Domain assignment is the one human input and lives in domains.yaml, keyed by
|
||||
capability id; this script only joins it and fails loudly on a mismatch.
|
||||
|
||||
Run from the repo root: python3 docs/capabilities/build_ledger.py
|
||||
"""
|
||||
import hashlib
|
||||
import json
|
||||
import pathlib
|
||||
import re
|
||||
import sys
|
||||
|
||||
ROOT = pathlib.Path(__file__).resolve().parents[2]
|
||||
SPEC = ROOT / "docs" / "spec.md"
|
||||
OUT = ROOT / "docs" / "capabilities" / "ledger.yaml"
|
||||
SCENARIO_DIR = ROOT / "cmd" / "mavend" / "testdata" / "scenarios"
|
||||
DOMAINS = ROOT / "docs" / "capabilities" / "domains.yaml"
|
||||
VERDICTS = ROOT / "docs" / "capabilities" / "verdicts.json"
|
||||
|
||||
# Sections of docs/spec.md whose ### headings are capabilities. Every other ##
|
||||
# is prose about how to read the file.
|
||||
CAPABILITY_SECTIONS = {
|
||||
"The turn",
|
||||
"Memory",
|
||||
"Proactive",
|
||||
"Reach",
|
||||
"Speech and senses",
|
||||
"The ecosystem",
|
||||
"Operations",
|
||||
"Undesigned in v1",
|
||||
}
|
||||
|
||||
# Backticked identifiers that appear on a Scenario line and are not scenarios.
|
||||
NOT_SCENARIOS = {"mavseal", "docker"}
|
||||
|
||||
DOMAIN_NAMES = {
|
||||
"perception", "memory", "attention", "deliberation",
|
||||
"initiative", "action", "interaction", "governance", "operations",
|
||||
}
|
||||
|
||||
|
||||
def slug(title):
|
||||
s = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")
|
||||
return s
|
||||
|
||||
|
||||
def crit_id(cap_slug, text):
|
||||
h = hashlib.sha256(text.encode("utf-8")).hexdigest()[:4]
|
||||
return f"{cap_slug}#{h}"
|
||||
|
||||
|
||||
def unwrap(lines):
|
||||
"""Join a bullet's continuation lines into one string."""
|
||||
return " ".join(l.strip() for l in lines).strip()
|
||||
|
||||
|
||||
def parse():
|
||||
lines = SPEC.read_text(encoding="utf-8").splitlines()
|
||||
caps = []
|
||||
section = None
|
||||
cur = None
|
||||
mode = None # None | 'dod' | 'state' | 'scenario'
|
||||
buf = [] # continuation lines of the bullet being read
|
||||
section_notes = {}
|
||||
|
||||
def flush_bullet():
|
||||
nonlocal buf
|
||||
if not buf or cur is None:
|
||||
buf = []
|
||||
return
|
||||
text = unwrap(buf)
|
||||
buf = []
|
||||
if not text:
|
||||
return
|
||||
if mode == "dod":
|
||||
cur["dod"].append(text)
|
||||
elif mode == "state":
|
||||
cur["state"] = text
|
||||
elif mode == "scenario":
|
||||
cur["scenario_raw"] = text
|
||||
|
||||
for raw in lines:
|
||||
if raw.startswith("## "):
|
||||
flush_bullet()
|
||||
mode = None
|
||||
cur = None
|
||||
section = raw[3:].strip()
|
||||
continue
|
||||
if raw.startswith("### "):
|
||||
flush_bullet()
|
||||
mode = None
|
||||
if section not in CAPABILITY_SECTIONS:
|
||||
cur = None
|
||||
continue
|
||||
title = raw[4:].strip()
|
||||
cur = {
|
||||
"id": slug(title),
|
||||
"title": title,
|
||||
"section": section,
|
||||
"scope": "v1",
|
||||
"state": "",
|
||||
"finding": "",
|
||||
"dod": [],
|
||||
"scenario_raw": "",
|
||||
"body": [],
|
||||
}
|
||||
caps.append(cur)
|
||||
continue
|
||||
if cur is None:
|
||||
# Section-level prose. Keep a Finding paragraph, it belongs to the
|
||||
# whole cluster (Memory has one).
|
||||
if section in CAPABILITY_SECTIONS and "**Finding**" in raw:
|
||||
section_notes.setdefault(section, []).append(raw.strip())
|
||||
elif section in section_notes and raw.strip() and not raw.startswith("#"):
|
||||
# continuation of that paragraph
|
||||
if section_notes[section] and section_notes[section][-1]:
|
||||
section_notes[section][-1] += " " + raw.strip()
|
||||
continue
|
||||
|
||||
cur["body"].append(raw)
|
||||
stripped = raw.strip()
|
||||
|
||||
if stripped.startswith("- **State**:"):
|
||||
flush_bullet()
|
||||
mode = "state"
|
||||
buf = [stripped[len("- **State**:"):]]
|
||||
continue
|
||||
if stripped.startswith("- **DoD**"):
|
||||
flush_bullet()
|
||||
mode = "dod"
|
||||
continue
|
||||
if stripped.startswith("- **Scenario**:"):
|
||||
flush_bullet()
|
||||
mode = "scenario"
|
||||
buf = [stripped[len("- **Scenario**:"):]]
|
||||
continue
|
||||
if stripped.startswith("**Deferred past v1**"):
|
||||
flush_bullet()
|
||||
cur["scope"] = "deferred"
|
||||
cur["deferred_note"] = stripped
|
||||
mode = None
|
||||
continue
|
||||
if not stripped:
|
||||
flush_bullet()
|
||||
continue
|
||||
if mode == "dod":
|
||||
if stripped.startswith("- "):
|
||||
flush_bullet()
|
||||
buf = [stripped[2:]]
|
||||
else:
|
||||
buf.append(stripped)
|
||||
continue
|
||||
if mode in ("state", "scenario"):
|
||||
if stripped.startswith("- "):
|
||||
flush_bullet()
|
||||
mode = None
|
||||
else:
|
||||
buf.append(stripped)
|
||||
continue
|
||||
|
||||
flush_bullet()
|
||||
|
||||
for c in caps:
|
||||
# The Finding sentence lives inside the State bullet in every entry that
|
||||
# has one. Split it out so a gap is a field, not prose.
|
||||
m = re.search(r"\*\*Finding\*\*:\s*(.*)$", c["state"], re.S)
|
||||
if m:
|
||||
c["finding"] = m.group(1).strip()
|
||||
c["state"] = c["state"][: m.start()].strip()
|
||||
c["state"] = c["state"].strip()
|
||||
c["scenarios"] = parse_scenarios(c["scenario_raw"])
|
||||
c["criteria"] = [
|
||||
{"id": crit_id(c["id"], t), "text": t} for t in c["dod"]
|
||||
]
|
||||
del c["body"], c["dod"], c["scenario_raw"]
|
||||
|
||||
return caps, section_notes
|
||||
|
||||
|
||||
def parse_scenarios(raw):
|
||||
"""Names in `backticks`, each flagged exists / to write.
|
||||
|
||||
The scenario line is prose in several entries ("covered by cmd/mavweb
|
||||
tests", "none"). Keep the prose verbatim as `note` rather than guessing.
|
||||
"""
|
||||
out = []
|
||||
for m in re.finditer(r"`([a-z0-9_]+)`(\s*\*\(to write\)\*)?", raw):
|
||||
name = m.group(1)
|
||||
# A path, a package or a binary named in prose is not a scenario name.
|
||||
if "/" in name or "." in name or name in NOT_SCENARIOS:
|
||||
continue
|
||||
claimed = m.group(2) is None
|
||||
on_disk = (SCENARIO_DIR / f"{name}.json").exists()
|
||||
if not claimed and not on_disk:
|
||||
pass # marked (to write) and absent: consistent
|
||||
out.append({"name": name, "claimed": claimed, "exists": on_disk})
|
||||
return {"named": out, "note": raw.strip()}
|
||||
|
||||
|
||||
def y(s, indent):
|
||||
"""Emit one scalar as a YAML block string, no quoting games."""
|
||||
pad = " " * indent
|
||||
body = "\n".join(pad + " " + l for l in s.splitlines()) if s else ""
|
||||
return ">-\n" + body if s else '""'
|
||||
|
||||
|
||||
VERDICT_WORDS = {"pass", "fail", "blocked", "untested", "unknown"}
|
||||
|
||||
# Why a criterion is not passing. The plan's list, and nothing outside it.
|
||||
REASON_KINDS = {
|
||||
"code missing", "wiring missing", "configuration missing",
|
||||
"deployment missing", "external dependency unavailable",
|
||||
"scenario missing", "scenario fails",
|
||||
"implementation exists with no runtime proof",
|
||||
"deferred past v1", "not yet probed", "passes",
|
||||
}
|
||||
|
||||
|
||||
def load_verdicts():
|
||||
if not VERDICTS.exists():
|
||||
return {}
|
||||
return json.loads(VERDICTS.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def emit(caps, section_notes, domains, verdicts):
|
||||
L = []
|
||||
L.append("# Capability ledger, target side.")
|
||||
L.append("#")
|
||||
L.append("# GENERATED by docs/capabilities/build_ledger.py from docs/spec.md.")
|
||||
L.append("# Do not hand-edit. Domain assignment is the one human input and")
|
||||
L.append("# lives in docs/capabilities/domains.yaml.")
|
||||
L.append("#")
|
||||
L.append("# No implementation status and no verification status appear here.")
|
||||
L.append("# Session 1 step 2 writes verdicts against the criterion ids below.")
|
||||
L.append("")
|
||||
L.append(f"source: docs/spec.md")
|
||||
L.append(f"capability_count: {len(caps)}")
|
||||
L.append(f"criterion_count: {sum(len(c['criteria']) for c in caps)}")
|
||||
L.append("")
|
||||
if section_notes:
|
||||
L.append("section_findings:")
|
||||
for sec, notes in section_notes.items():
|
||||
L.append(f" - section: {sec!r}")
|
||||
L.append(" finding: " + y(" ".join(notes), 4))
|
||||
L.append("")
|
||||
L.append("capabilities:")
|
||||
for c in caps:
|
||||
L.append(f" - id: {c['id']}")
|
||||
L.append(f" title: {c['title']!r}")
|
||||
L.append(f" section: {c['section']!r}")
|
||||
L.append(f" scope: {c['scope']}")
|
||||
d = domains.get(c["id"], [])
|
||||
L.append(" domain: [" + ", ".join(d) + "]")
|
||||
L.append(" state: " + y(c["state"], 4))
|
||||
if c["finding"]:
|
||||
L.append(" finding: " + y(c["finding"], 4))
|
||||
if c.get("deferred_note"):
|
||||
L.append(" deferred_note: " + y(c["deferred_note"], 4))
|
||||
L.append(" scenarios:")
|
||||
for s in c["scenarios"]["named"]:
|
||||
L.append(f" - name: {s['name']}")
|
||||
L.append(f" exists: {str(s['exists']).lower()}")
|
||||
if s["claimed"] != s["exists"]:
|
||||
L.append(" discrepancy: spec implies it exists and "
|
||||
"cmd/mavend/testdata/scenarios has no such file")
|
||||
L.append(" scenario_note: " + y(c["scenarios"]["note"], 4))
|
||||
L.append(" criteria:")
|
||||
for cr in c["criteria"]:
|
||||
L.append(f" - id: {cr['id']!r}")
|
||||
L.append(" text: " + y(cr["text"], 8))
|
||||
v = verdicts.get(cr["id"])
|
||||
if v is None and c["scope"] == "deferred":
|
||||
v = {"verified": "untested", "reason": "deferred past v1"}
|
||||
if v is None:
|
||||
v = {"verified": "untested", "reason": "not yet probed"}
|
||||
L.append(f" verified: {v['verified']}")
|
||||
L.append(f" reason: {v['reason']!r}")
|
||||
if v.get("detail"):
|
||||
L.append(" detail: " + y(v["detail"], 8))
|
||||
ev = v.get("evidence", [])
|
||||
if ev:
|
||||
L.append(" evidence:")
|
||||
for e in ev:
|
||||
L.append(f" - {e!r}")
|
||||
L.append("")
|
||||
return "\n".join(L) + "\n"
|
||||
|
||||
|
||||
def load_domains():
|
||||
if not DOMAINS.exists():
|
||||
return {}
|
||||
out = {}
|
||||
for line in DOMAINS.read_text(encoding="utf-8").splitlines():
|
||||
line = line.split("#", 1)[0].strip()
|
||||
if not line or ":" not in line:
|
||||
continue
|
||||
k, v = line.split(":", 1)
|
||||
vals = [x.strip() for x in v.strip().strip("[]").split(",") if x.strip()]
|
||||
out[k.strip()] = vals
|
||||
return out
|
||||
|
||||
|
||||
def main():
|
||||
caps, section_notes = parse()
|
||||
domains = load_domains()
|
||||
verdicts = load_verdicts()
|
||||
|
||||
errs = []
|
||||
seen = {}
|
||||
for c in caps:
|
||||
for cr in c["criteria"]:
|
||||
if cr["id"] in seen:
|
||||
errs.append(f"criterion id collision: {cr['id']}")
|
||||
seen[cr["id"]] = cr["text"]
|
||||
if not c["criteria"]:
|
||||
errs.append(f"{c['id']}: no DoD criteria extracted")
|
||||
if not c["state"]:
|
||||
errs.append(f"{c['id']}: no State line extracted")
|
||||
d = domains.get(c["id"])
|
||||
if domains:
|
||||
if not d:
|
||||
errs.append(f"{c['id']}: no domain assigned")
|
||||
elif len(d) > 2:
|
||||
errs.append(f"{c['id']}: {len(d)} domains, max is 2")
|
||||
else:
|
||||
for x in d:
|
||||
if x not in DOMAIN_NAMES:
|
||||
errs.append(f"{c['id']}: unknown domain {x!r}")
|
||||
for k in domains:
|
||||
if k not in {c["id"] for c in caps}:
|
||||
errs.append(f"domains.yaml names unknown capability {k!r}")
|
||||
|
||||
# A verdict cites evidence by path, and a path that resolves to nothing is
|
||||
# worse than no citation: it reads as verified and is not. Section refs are
|
||||
# checked too, because writing "§ Something" that no heading matches is the
|
||||
# easy way to make an unsupported claim look sourced.
|
||||
for cid, v in verdicts.items():
|
||||
for e in v.get("evidence", []):
|
||||
path, _, section = e.partition(" § ")
|
||||
# An evidence string is "<path>[:line] [locator]" or
|
||||
# "<path> § <heading>". The locator points inside the file
|
||||
# (a probe id, a readback key) and is not part of the path.
|
||||
path = path.strip().split()[0].split(":")[0]
|
||||
f = ROOT / path
|
||||
if not f.exists():
|
||||
errs.append(f"{cid}: evidence path does not exist: {path}")
|
||||
elif section and f.suffix == ".md" and section.strip() not in f.read_text(encoding="utf-8"):
|
||||
errs.append(f"{cid}: evidence names a section not in {path}: {section}")
|
||||
|
||||
for cid, v in verdicts.items():
|
||||
if cid not in seen:
|
||||
errs.append(f"verdicts.json scores unknown criterion {cid!r}")
|
||||
elif v.get("verified") not in VERDICT_WORDS:
|
||||
errs.append(f"{cid}: verdict {v.get('verified')!r} is not one of {sorted(VERDICT_WORDS)}")
|
||||
elif v.get("reason") not in REASON_KINDS:
|
||||
errs.append(f"{cid}: reason {v.get('reason')!r} is not one of the plan's kinds")
|
||||
elif v["verified"] == "pass" and v["reason"] != "passes":
|
||||
errs.append(f"{cid}: a pass carries reason {v['reason']!r}")
|
||||
elif v["verified"] != "pass" and v["reason"] == "passes":
|
||||
errs.append(f"{cid}: reason 'passes' on a {v['verified']} verdict")
|
||||
elif v["verified"] == "fail" and v["reason"] == "implementation exists with no runtime proof":
|
||||
# That reason means nothing was observed. A fail was observed, or it
|
||||
# is not a fail. Mixing them is how a wrong diagnosis survives.
|
||||
errs.append(f"{cid}: a fail cannot rest on 'no runtime proof'")
|
||||
|
||||
OUT.write_text(emit(caps, section_notes, domains, verdicts), encoding="utf-8")
|
||||
|
||||
# The emitter hand-writes YAML, so it can produce something that reads fine
|
||||
# and does not parse. It did once: evidence came out as a bare list item
|
||||
# inside a mapping. Parse what was just written.
|
||||
try:
|
||||
import yaml
|
||||
except ImportError:
|
||||
errs.append("pyyaml absent: the output was written without a parse check")
|
||||
else:
|
||||
try:
|
||||
doc = yaml.safe_load(OUT.read_text(encoding="utf-8"))
|
||||
except yaml.YAMLError as e:
|
||||
errs.append(f"the emitted ledger is not valid YAML: {e}")
|
||||
else:
|
||||
n = sum(len(c["criteria"]) for c in doc["capabilities"])
|
||||
if n != len(seen):
|
||||
errs.append(f"round trip lost criteria: wrote {len(seen)}, read back {n}")
|
||||
for c in doc["capabilities"]:
|
||||
if c["scope"] != "v1":
|
||||
continue
|
||||
for cr in c["criteria"]:
|
||||
if cr["verified"] != "untested" and not cr.get("evidence"):
|
||||
errs.append(f"{cr['id']}: scored {cr['verified']} with no evidence")
|
||||
|
||||
print(f"{len(caps)} capabilities, {len(seen)} criteria -> {OUT.relative_to(ROOT)}")
|
||||
print(f" v1: {sum(1 for c in caps if c['scope'] == 'v1')}, "
|
||||
f"deferred: {sum(1 for c in caps if c['scope'] == 'deferred')}")
|
||||
scen = {s["name"]: s["exists"] for c in caps for s in c["scenarios"]["named"]}
|
||||
ghosts = sorted(n for n, e in scen.items() if not e
|
||||
and any(s["claimed"] for c in caps for s in c["scenarios"]["named"] if s["name"] == n))
|
||||
if ghosts:
|
||||
print(f" named as existing but absent from disk: {', '.join(ghosts)}")
|
||||
print(f" scenarios named: {len(scen)}, existing: {sum(scen.values())}, "
|
||||
f"to write: {len(scen) - sum(scen.values())}")
|
||||
if verdicts:
|
||||
from collections import Counter
|
||||
tally = Counter(v["verified"] for v in verdicts.values())
|
||||
print(" verdicts: " + ", ".join(f"{k} {n}" for k, n in sorted(tally.items())))
|
||||
print(f" no living doc: {sum(1 for c in caps if 'No living doc' in c['state'] or 'no living doc' in c['state'].lower())}")
|
||||
if errs:
|
||||
print("\nERRORS:", file=sys.stderr)
|
||||
for e in errs:
|
||||
print(" " + e, file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user