Current section
Files
Jump to
Current section
Files
priv/toolchain/check_join.py
#!/usr/bin/env python3
"""check_join.py — the mechanical core of /dds-check (protocol 2).
Everything rule-shaped in the check runs here, deterministically:
comparison 1 (structure ↔ canon), the register-world canonical parse
(subjects, datums, activities, governance — facet completeness and
vocabulary validated, holes and draft flags explicit), the cube's
mechanical legs (stale pins, pin→datum resolution, uncovered person
stores, dormancy), the five-row coverage table with its hard-tier
read-off (class membership from the risk files' ratified `Class:`
declarations), delta/background tagging, nominee selection for the
judged tier, verdict-completeness enforcement, and the derived
findings. The LLM's remaining jobs are (a) per-file fact extraction for
code files (see extraction.md — every fact carries a quote span this
script verifies against the file bytes) and (b) the narrow per-ADR
conformance verdicts this script asks for by name (verified by the
shared verdicts.py — rule_quote against the ADR's ## Rule section,
evidence replayed against the tree).
The walk doc carries `protocol: 2` and a `register` section — the
canonical parse of the claim world. Downstream consumers (the portal)
ingest ONLY that artifact; no second markdown parser exists (the seam
principle, spec/protocol-2.md P1).
Usage:
check_join.py --base SHA [--repo DIR] [--facts DIR]
[--verdicts FILE] [--json OUT] [--write-status]
check_join.py --selftest spec/fixtures (fixture conformance run)
Exit codes: 0 = PASS, 1 = FAIL, 2 = INCOMPLETE (missing witness or
extractions, unverifiable quotes, or nominees without verdicts — the
agent must supply facts and re-run; a verdictless run never passes).
Stdlib only; python3.9+. verdicts.py (same dir) is the shared verdict
verifier; tsq.py (same dir) is the optional replay engine.
"""
import argparse
import json
import os
import re
import subprocess
import sys
from collections import defaultdict
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from verdicts import VERDICTS as VERDICT_ENUM_LIST, Verifier, norm_ws, rule_scope
PROTOCOL = 2
VERDICT_ENUM = set(VERDICT_ENUM_LIST)
CLASS_ORDER = ["gating", "erasure", "audit", "surface", "credential"]
CLASS_SET = set(CLASS_ORDER)
# The claim-file law is DATA, not code: regimes/_core.json carries the
# regime-free core (facet law PER PARTY KIND, activity law, governance
# stems); obligation profiles (regimes/<name>.json) overlay it. This
# embedded copy is the fallback when the carried file is unreadable.
DEFAULT_CORE = {
"subject_kinds": ["person", "organization", "internal"],
"facet_law": {
"person": {
"required": ["Identifiability", "Category", "Provenance"],
"vocab": {
"Identifiability": ["direct", "indirect", "association", "linked_only"],
"Category": [
"identity", "contact", "affiliation", "content",
"health", "behavioral", "credential", "commercial",
],
"Provenance": ["provided", "observed", "derived", "inferred"],
},
"source_party_when_provided": True,
},
"organization": {
"required": ["Classification"],
"vocab": {
"Classification": ["public", "internal", "confidential", "restricted"],
"Concern": ["confidentiality", "integrity", "availability"],
},
"multi": ["Concern"],
},
},
"activity_law": {
"required": ["Purpose", "Function", "Automation"],
"vocab": {
"Function": ["service_delivery", "safeguarding", "administration", "security_accountability"],
"Automation": ["manual", "assisted", "automated", "automated_review"],
},
"authority_field": "Lawful basis",
"authority_grammar": "^basis\\.[\\w.-]+$",
},
"governance_stems": ["basis", "retention", "terms"],
"retention_direction": ["ceiling", "floor"],
}
# engine defaults — platform profiles (assets/platforms/*.json) override
PERSON_SIGNAL = re.compile(
r"first_name|last_name|full_name|\bemail\b|date_of_birth|\bdob\b|phone", re.I
)
AUDIT_STORE = re.compile(r"audit|_history|_versions$")
AUTH_PLUG = re.compile(r"auth|require|access", re.I)
# the open-state markers (conventions.md §11): a hole is honest structure
HOLE_MARKERS = re.compile(r"basis\.TODO|needs-ratification|needs-legal-review|\bTODO\b|⚠")
HOLE_VALUE = re.compile(r"TODO|needs-ratification|needs-legal-review|⚠")
# a drafted (machine-written, unratified) claim section's marker
DRAFT_MARK = re.compile(r"°|unratified|needs-ratification")
STATUS_OK = re.compile(r"^(Proposed|Accepted \((interim, )?\d{4}-\d{2}-\d{2}\))$")
TRAILER = re.compile(r"^(Risk|Concern):\s*(\S+)\s*$", re.M)
CLASS_LINE = re.compile(r"^Class:\s*(\S+)\s*$", re.M)
ADDRESSED_LINE = re.compile(r"^Addressed by:\s*(.+?)\s*$", re.M)
# register-doc section shape: `### <Name> · `<id>``
SECTION_HEAD = re.compile(
r"^###\s+(.+?)\s+·\s+`([^`]+)`\s*$(.*?)(?=^###\s|\Z)", re.M | re.S
)
FACET_ROW = re.compile(
r"^\|\s*([A-Z][\w ]*?)\s*\|\s*(.*?)\s*\|(?:\s*([^|]*?)\s*\|)?\s*$", re.M
)
ADR_SECTIONS = ["Rule", "Premises"]
RISK_SECTIONS = ["Stake", "Currency", "No strategy looks like", "Current state"]
OVERLAY_KINDS = {"store", "router", "secrets", "empty", "degraded"}
def sh(repo, *args):
return subprocess.run(
["git", "-C", repo, *args], capture_output=True, text=True
).stdout
def read(repo, rel):
p = os.path.join(repo, rel)
try:
with open(p, encoding="utf-8") as f:
return f.read()
except OSError:
return None
def md_section(text, name):
"""A `## <name>`-headed section's body (any heading level), or None."""
m = re.search(rf"^#+\s*{re.escape(name)}\s*$(.*?)(?=^#+\s|\Z)", text, re.M | re.S)
return m.group(1).strip() if m else None
def clean_facet(value):
"""A facet cell's value, stripped of rendering chips (`→ `, backticks,
the ° draft mark)."""
return re.sub(r"^→\s*", "", value).replace("°", "").strip().strip("`").strip()
class Check:
def __init__(self, repo, base, facts_dir, verdicts_path, prior_path=None):
self.repo = repo
self.base = base
self.facts_dir = facts_dir
self.verdicts_path = verdicts_path
self.prior_path = prior_path
self.findings = [] # {tier, artifact, text, delta}
self.degraded = []
self.needs_extraction = []
self.delta_files = set()
self.witness = {"files": {}, "router": None, "secrets": []}
self.baseline_files = {}
self.platforms = [] # active profiles, with detection evidence
self.core = DEFAULT_CORE
self.regimes = [] # active obligation profiles (declared, never detected)
# ── platforms: evidence-based detection, config-driven engine ────────
def detect_platforms(self):
pdir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "platforms")
if not os.path.isdir(pdir):
self.degraded.append("no platforms/ dir next to the instrument")
return
for name in sorted(os.listdir(pdir)):
if not name.endswith(".json"):
continue
try:
prof = json.loads(open(os.path.join(pdir, name), encoding="utf-8").read())
except (OSError, json.JSONDecodeError) as e:
self.degraded.append(f"platform profile {name} unparseable: {e}")
continue
for d in prof.get("detect", []):
text = read(self.repo, d["file"])
if not text:
continue
m = re.search(d["pattern"], text, re.M)
if m:
line = next(
(ln for ln in text.splitlines() if m.group(0) in ln),
m.group(0),
)
prof["_evidence"] = f"{d['file']}: {line.strip()}"
self.platforms.append(prof)
break
if not self.platforms:
self.degraded.append(
"no platform profile detected — code witnessing skipped, "
"docs-side checks only (see assets/platforms/README.md)"
)
def pconf(self, key):
"""union of a config value across active profiles."""
return [p[key] for p in self.platforms if key in p]
def pmatch(self, key, value, default=None):
pats = self.pconf(key)
if not pats and default is not None:
return bool(default.search(value))
return any(re.search(pat, value, re.I) for pat in pats)
def finding(self, tier, artifact, text):
self.findings.append(
{
"tier": tier,
"artifact": artifact,
"text": text,
"delta": artifact in self.delta_files,
}
)
# ── delta ────────────────────────────────────────────────────────────
def load_delta(self):
out = sh(self.repo, "diff", "--name-only", f"{self.base}..HEAD")
self.delta_files = {ln.strip() for ln in out.splitlines() if ln.strip()}
out = sh(self.repo, "status", "--porcelain")
for ln in out.splitlines():
if len(ln) > 3:
path = ln[3:].strip()
# rename entries read `R old -> new`; the new path is the delta
if " -> " in path:
path = path.split(" -> ", 1)[1].strip()
self.delta_files.add(path)
# ── obligation profiles: declared in governance.md, never detected ──
def load_regimes(self):
rdir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "regimes")
registry = {}
if os.path.isdir(rdir):
for name in sorted(os.listdir(rdir)):
if not name.endswith(".json"):
continue
try:
prof = json.loads(open(os.path.join(rdir, name), encoding="utf-8").read())
except (OSError, json.JSONDecodeError) as e:
self.degraded.append(f"obligation profile {name} unparseable: {e}")
continue
if name == "_core.json":
self.core = prof
else:
registry[prof.get("name") or name[:-5]] = prof
else:
self.degraded.append("no regimes/ dir next to the instrument — embedded core law only")
self._gov_text = read(self.repo, "docs/governance.md") or ""
declared, todo = [], False
for label in ("Regime", "Scope"):
m = re.search(rf"{label}:\s*(.+)$", self._gov_text, re.M)
if not m:
continue
seg = re.split(r"[—(]", m.group(1))[0]
for tok in re.split(r"[,·]", seg):
tok = re.sub(r"[^\w-]", "", tok).lower()
if tok == "todo":
todo = True
elif tok:
declared.append(tok)
for tok in declared:
prof = registry.get(tok)
if prof is None:
self.finding(
"STRUCTURAL",
"docs/governance.md",
f"declared regime/scope `{tok}` matches no carried profile "
f"(carried: {sorted(registry)})",
)
else:
self.regimes.append(prof)
if todo or not declared:
self.degraded.append(
"regime/scope undeclared (`Regime: TODO`) — obligation profiles "
"inactive; core facet law only (declare in docs/governance.md)"
)
# ── canon conformance: divergences are DECLARED, never diffed ───────
# A consuming repo's canon files are verbatim copies of the carried
# canon EXCEPT inside declared divergence blocks. The block's anchor
# quote is its currency pin against the canon (same primitive as
# residuals/signals/mappings): upstream rewrites the passage → the
# anchor dies → stale divergence, re-base or drop. Upgrade becomes a
# computation: overwrite with the new canon, re-apply the declared
# blocks, surface the stale ones as precisely-localized conflicts.
CANON_MAP = {
"docs/principles.md": "principles.md",
"docs/conventions.md": "conventions.md",
"docs/registers.md": "registers.md",
"docs/claim-files.md": "claim-files.md",
"docs/README.md": "docs/README.md",
"docs/adr/README.md": "docs/adr-README.md",
"docs/goals/README.md": "docs/goals-README.md",
"docs/domain/README.md": "docs/domain-README.md",
}
DIV_OPEN = re.compile(r"<!--\s*dds-divergence\b(.*?)-->", re.S)
DIV_CLOSE = re.compile(r"<!--\s*/dds-divergence\s*-->")
_FENCE = re.compile(r"^[ \t]*(`{3,}|~{3,})[^\n]*\n.*?^[ \t]*\1[ \t]*$", re.M | re.S)
_HTML_COMMENT = re.compile(r"<!--.*?-->", re.S)
_INLINE_CODE = re.compile(r"`+[^`\n]+`+")
@staticmethod
def _mask_code(text):
"""A copy of `text` with Markdown code spans blanked (same
length — offsets stay valid), so a divergence marker inside a
fenced example or inline code is DOCUMENTATION, never a
declaration (issue #28: the canon's own §12 example tripped
the parser). Inline spans inside a real HTML comment are kept
— a genuine anchor quote may legitimately contain backticks."""
buf = list(text)
def blank(a, b):
for i in range(a, b):
if buf[i] != "\n":
buf[i] = " "
for m in Check._FENCE.finditer(text):
blank(m.start(), m.end())
fence_masked = "".join(buf)
protected = [
(m.start(), m.end()) for m in Check._HTML_COMMENT.finditer(fence_masked)
]
for m in Check._INLINE_CODE.finditer(fence_masked):
if not any(s <= m.start() < e for s, e in protected):
blank(m.start(), m.end())
return "".join(buf)
def parse_divergences(self, rel, text):
"""([{slug, kind, why, anchor, start, end}], whole_file) — emits
findings for malformed blocks. Kinds: replace / delete (an
`instead-of` anchor), insert (an `after` anchor), file (the
whole-file escape hatch — the honest .dds-overrides successor).
Markers are scanned on a code-masked copy (identical offsets):
a marker inside a fence or inline code is documentation, never
a declaration (issue #28)."""
scan = self._mask_code(text)
first = self.DIV_OPEN.search(scan)
if first and re.match(r"\s*file\b", first.group(1)):
h = first.group(1)
why = re.search(r"why:\s*([^\n]+)", h)
return (
[
{
"slug": "file",
"kind": "file",
"why": why.group(1).strip() if why else None,
"anchor": None,
}
],
True,
)
blocks, pos = [], 0
while True:
m = self.DIV_OPEN.search(scan, pos)
if not m:
break
header = m.group(1)
mslug = re.match(r"\s*([\w-]+)", header)
slug = mslug.group(1) if mslug else "unnamed"
mc = self.DIV_CLOSE.search(scan, m.end())
if not mc:
self.finding(
"STRUCTURAL", rel, f"malformed divergence `{slug}`: unclosed block"
)
break
instead = re.search(r'instead-of:\s*"(.*?)"', header, re.S)
after = re.search(r'after:\s*"(.*?)"', header, re.S)
why = re.search(r"why:\s*([^\n]+)", header)
if bool(instead) == bool(after):
self.finding(
"STRUCTURAL",
rel,
f"malformed divergence `{slug}`: exactly one of "
'`instead-of: "…"` / `after: "…"` is required',
)
pos = mc.end()
continue
body = text[m.end() : mc.start()]
kind = "insert" if after else ("delete" if not body.strip() else "replace")
blocks.append(
{
"slug": slug,
"kind": kind,
"why": why.group(1).strip() if why else None,
"anchor": (instead or after).group(1),
"start": m.start(),
"end": mc.end(),
}
)
pos = mc.end()
return blocks, False
def check_canon_file(self, rel, local, carried):
"""One canon file: parse declarations, verify anchors against the
carried canon, reconstruct, compare. Returns the ledger entries."""
blocks, whole = self.parse_divergences(rel, local)
ledger = []
if whole:
b = blocks[0]
ledger.append(
{"file": rel, "slug": b["slug"], "kind": "file", "why": b["why"], "stale": False}
)
return ledger
flat_carried = norm_ws(carried)
stale = 0
for b in blocks:
b_stale = norm_ws(b["anchor"]) not in flat_carried
if b_stale:
stale += 1
self.finding(
"CUBE",
rel,
f"stale divergence `{b['slug']}`: its anchor quote no longer "
"appears in the carried canon — re-base the divergence onto "
"the new text or drop it",
)
ledger.append(
{
"file": rel,
"slug": b["slug"],
"kind": b["kind"],
"why": b["why"],
"stale": b_stale,
}
)
parts, last = [], 0
for b in sorted(blocks, key=lambda x: x["start"]):
parts.append(local[last : b["start"]])
if b["kind"] in ("replace", "delete"):
parts.append(b["anchor"])
last = b["end"]
parts.append(local[last:])
if norm_ws("".join(parts)) != norm_ws(carried):
self.finding(
"STRUCTURAL",
rel,
"undeclared canon drift: the file does not reconstruct to the "
"carried canon outside its declared divergences"
+ (f" ({stale} stale divergence(s) may explain part of it)" if stale else "")
+ " — wrap local edits in dds-divergence blocks, revert them, "
"or run /dds-upgrade if the canon version moved",
)
return ledger
def canon_conformance(self):
# the carried canon: beside the skills in a checkout/apm_modules,
# or shipped as package data (dds/canon) in an installed wheel
here = os.path.dirname(os.path.abspath(__file__))
carried_root = next(
(
p
for p in (
os.path.join(here, "..", "..", "dds-sync", "assets"),
os.path.join(here, "canon"),
)
if os.path.isdir(p)
),
None,
)
ledger = []
if carried_root is None:
self.degraded.append(
"carried canon not found (../dds-sync/assets or package data) — "
"canon conformance skipped"
)
return ledger
for rel, carried_rel in self.CANON_MAP.items():
local = read(self.repo, rel)
if local is None:
continue
try:
carried = open(
os.path.join(carried_root, carried_rel), encoding="utf-8"
).read()
except OSError:
self.degraded.append(f"carried canon {carried_rel} unreadable — {rel} not checked")
continue
ledger += self.check_canon_file(rel, local, carried)
return ledger
# ── the framework pin (protocol declaration is tree-grain) ──────────
def check_dds_version(self):
text = read(self.repo, ".dds-version") or ""
if not re.search(rf"^protocol:\s*{PROTOCOL}\s*$", text, re.M):
self.finding(
"STRUCTURAL",
".dds-version",
f"missing `protocol: {PROTOCOL}` declaration — the protocol "
"generation is a property of the tree; run /dds-upgrade "
"(the protocol-2 migration pins it)",
)
# ── witness: baseline + overlay, quote-verified ──────────────────────
def load_witness(self):
# class signals are JUDGMENT: they live in the risk files' own
# `## Signals` ledgers (parsed by parse_risks, bound to classes
# through the risk's `Class:` declarations) — no JSON side-file,
# and no in-witness fallback (protocol 2).
witness_adopted = False
committed = read(self.repo, "docs/registers/witness.json")
if committed:
try:
loaded = json.loads(committed)
if loaded.get("protocol") != PROTOCOL:
self.degraded.append(
f"witness.json declares protocol {loaded.get('protocol')!r} — "
f"protocol {PROTOCOL} required; regenerate it (witness.py, or "
"re-extract on unprofiled platforms); the stale witness is "
"not adopted"
)
else:
loaded.setdefault("files", {})
loaded.setdefault("secrets", [])
self.witness = loaded
witness_adopted = True
except json.JSONDecodeError as e:
self.degraded.append(f"docs/registers/witness.json unparseable: {e}")
overlay = {}
if self.facts_dir and os.path.isdir(self.facts_dir):
for name in sorted(os.listdir(self.facts_dir)):
if not name.endswith(".json"):
continue
path = os.path.join(self.facts_dir, name)
try:
doc = json.loads(open(path, encoding="utf-8").read())
except (OSError, json.JSONDecodeError) as e:
self.degraded.append(f"facts overlay {name} unparseable: {e}")
continue
rel = doc.get("file")
if not rel:
self.degraded.append(f"facts overlay {name}: no 'file' key")
continue
kind = doc.get("kind")
if kind not in OVERLAY_KINDS:
self.needs_extraction.append(
{
"file": rel,
"reason": f"record missing/unknown 'kind' ({kind!r}) — "
f"protocol 2 records declare one of {sorted(OVERLAY_KINDS)}",
}
)
continue
bad = self.verify_quotes(rel, doc)
if bad:
self.needs_extraction.append(
{"file": rel, "reason": f"unverified quotes: {bad}"}
)
continue
overlay[rel] = doc
# overlay wins over committed baseline
self.baseline_files = dict(self.witness.get("files", {}))
for rel, doc in overlay.items():
kind = doc["kind"]
if kind == "router":
self.witness["router"] = doc.get("router")
elif kind == "secrets":
self.witness["secrets"] = doc.get("secrets") or []
elif kind == "degraded":
self.degraded.append(f"{rel}: extraction degraded: {doc.get('degraded')}")
self.witness["files"][rel] = doc
else: # store | empty
self.witness["files"][rel] = doc
# a checked tree is never witnessed by silence: a profiled platform
# with no witness DOC at all is INCOMPLETE, whatever the delta looks
# like. (An adopted protocol-2 witness that is honestly EMPTY — a
# profiled repo with nothing to witness yet — is a statement, not
# silence, and passes this guard.)
if self.platforms and not witness_adopted and not overlay:
self.needs_extraction.append(
{
"file": "docs/registers/witness.json",
"reason": "profiled platform with no witness — regenerate it "
"(witness.py) or supply extraction records; silence never passes",
}
)
# delta code files must have verified facts (overlay or committed-current)
code_res = self.pconf("code_file_regex")
# a mechanically generated witness parsed every file at check time —
# its facts are current by construction; the overlay obligation
# exists for LLM-witnessed platforms, whose committed baseline can
# be stale for exactly the delta.
mechanical = self.witness.get("mode") == "mechanical"
for rel in sorted(self.delta_files):
if not any(re.match(cr, rel) for cr in code_res):
continue
if rel in overlay:
continue
if not os.path.exists(os.path.join(self.repo, rel)):
# deleted file: its committed witness entry must go
if rel in self.witness["files"]:
del self.witness["files"][rel]
continue
if mechanical:
continue
self.needs_extraction.append(
{"file": rel, "reason": "delta code file with no facts overlay"}
)
def verify_quotes(self, rel, doc, _seen=None):
"""every 'quote' value anywhere in the doc must appear in the file."""
content = read(self.repo, rel)
if content is None:
return ["file missing"]
flat = norm_ws(content)
bad = []
def walk(node):
if isinstance(node, dict):
for k, v in node.items():
if k == "quote" and isinstance(v, str):
if norm_ws(v) not in flat:
bad.append(v[:60])
else:
walk(v)
elif isinstance(node, list):
for v in node:
walk(v)
walk(doc)
return bad
# ── docs parsing (deterministic, no LLM) ─────────────────────────────
def md_files(self, subdir):
root = os.path.join(self.repo, subdir)
if not os.path.isdir(root):
return []
return sorted(
os.path.join(subdir, f)
for f in os.listdir(root)
if f.endswith(".md") and f != "README.md"
)
def parse_adr(self, text, rel):
status = None
m = re.search(r"^Status:\s*(.+?)\s*$", text, re.M)
if m:
status = m.group(1)
conformance = None
m = re.search(r"^Conformance:\s*(.+?)\s*$", text, re.M)
if m:
conformance = m.group(1)
trailers = [t[1] for t in TRAILER.findall(text)]
rule = ""
m = re.search(r"^#+\s*Rule\s*$(.*?)(?=^#+\s|\Z)", text, re.M | re.S)
if m:
rule = m.group(1).strip()
return {
"path": rel,
"status": status,
"conformance": conformance,
"trailers": trailers,
"rule": rule,
"text": text,
}
def adr_corpus(self, at_base=False):
corpus = []
for rel in self.md_files("docs/adr"):
text = (
sh(self.repo, "show", f"{self.base}:{rel}")
if at_base
else read(self.repo, rel)
)
if text:
corpus.append(self.parse_adr(text, rel))
if at_base:
# base may have ADRs since deleted/renamed; list from git
names = sh(self.repo, "ls-tree", "--name-only", self.base, "docs/adr/")
have = {a["path"] for a in corpus}
for rel in names.splitlines():
rel = rel.strip()
if (
rel.endswith(".md")
and not rel.endswith("README.md")
and rel not in have
):
text = sh(self.repo, "show", f"{self.base}:{rel}")
if text:
corpus.append(self.parse_adr(text, rel))
return corpus
# ── risk files: one parse, every consumer reads it ──────────────────
SIGNAL = re.compile(r"^- (.*?)\s*\[([^|\]]+)\|\s*\"(.+?)\"\s*\]$", re.S)
def parse_signals(self, rel, text):
"""The risk file's `## Signals` ledger: witnessed finding-shaped
facts the schema-level derivations can't see, one line each —
`- <text> [<file> | "<quote>"]` — bound to classes through the
risk's own `Class:` declarations. Quote-verified like residuals:
a signal whose quote died is stale, never silently kept."""
m = re.search(
r"^## Signals\s*\n(.*?)(?=^## |^Addressed by:|^Class:|^Codified-by:|\Z)",
text,
re.M | re.S,
)
if not m:
return []
out = []
for raw in re.split(r"^- ", m.group(1), flags=re.M)[1:]:
entry = "- " + re.sub(r"\s+", " ", raw.strip())
mm = self.SIGNAL.match(entry)
if not mm:
self.finding(
"STRUCTURAL", rel, f"malformed signal line: {entry[:90]}"
)
continue
stext, sfile, quote = mm.group(1).strip(), mm.group(2).strip(), mm.group(3)
if self.verify_quotes(sfile, {"quote": quote}):
self.finding(
"CUBE",
rel,
f"stale signal: its witness quote no longer verifies in "
f"{sfile} — re-witness it or remove the line ({stext[:60]})",
)
continue
out.append({"text": stext, "file": sfile, "quote": quote})
return out
def parse_risks(self):
out = []
for rel in self.md_files("docs/risks"):
text = read(self.repo, rel) or ""
stem = os.path.basename(rel)[:-3]
m = re.search(r"^#\s+(.+?)\s*$", text, re.M)
addressed_lines = ADDRESSED_LINE.findall(text)
addressed = []
if addressed_lines:
addressed = [
re.sub(r"[^\w-]", "", tok)
for tok in re.split(r"[,\s]+", addressed_lines[0].strip())
]
addressed = [t for t in addressed if t]
out.append(
{
"id": stem,
"path": rel,
"title": m.group(1).strip() if m else stem,
"stake": md_section(text, "Stake"),
"currency": md_section(text, "Currency"),
"classes": CLASS_LINE.findall(text),
"addressed_by": addressed,
"addressed_lines": addressed_lines,
"signals": self.parse_signals(rel, text),
"stub": "dds-stub" in text,
"holes": sorted(set(HOLE_MARKERS.findall(text))),
"text": text,
}
)
return out
def base_risk_class_map(self):
"""{risk stem → declared classes} at the base sha."""
out = {}
names = sh(self.repo, "ls-tree", "--name-only", self.base, "docs/risks/")
for rel in names.splitlines():
rel = rel.strip()
if not rel.endswith(".md") or rel.endswith("README.md"):
continue
text = sh(self.repo, "show", f"{self.base}:{rel}") or ""
out[os.path.basename(rel)[:-3]] = set(CLASS_LINE.findall(text))
return out
# ── residuals: the ledger of known gaps (authored, never by the check) ──
RESIDUAL = re.compile(
r"^- `(res\.[\w-]+)`\s*·\s*(open|accepted [0-9]{4}-[0-9]{2}-[0-9]{2}[^·\[]*|closed [0-9a-f]{7,})\s*·\s*(.*?)(?:\[([^|\]]+)\|\s*\"(.+?)\"\s*\])?\s*$",
re.S,
)
def residuals(self):
"""[{id, state, risk, text, file, quote}] from every risk file's
## Residuals section; malformed lines and duplicate ids are
structural findings; open/accepted quotes are verified — a
residual whose quote died is reported stale (close or
re-witness?), so an acknowledgment cannot outlive its code."""
out, seen = [], {}
for rel in self.md_files("docs/risks"):
text = read(self.repo, rel) or ""
m = re.search(
r"^## Residuals\s*\n(.*?)(?=^## |^Addressed by:|^Class:|^Codified-by:|\Z)",
text,
re.M | re.S,
)
if not m:
continue
block = m.group(1)
entries = re.split(r"^- ", block, flags=re.M)[1:]
for raw in entries:
entry = "- " + re.sub(r"\s+", " ", raw.strip())
mm = self.RESIDUAL.match(entry)
if not mm:
self.finding("STRUCTURAL", rel, f"malformed residual line: {entry[:90]}")
continue
rid, state, rtext, rfile, quote = mm.groups()
state_word = state.split()[0]
if rid in seen:
self.finding(
"STRUCTURAL", rel, f"duplicate residual id `{rid}` (also in {seen[rid]})"
)
continue
seen[rid] = rel
r = {
"id": rid,
"state": state.strip(),
"state_word": state_word,
"risk": os.path.basename(rel)[:-3],
"text": rtext.strip(),
"file": rfile.strip() if rfile else None,
"quote": quote,
"verified": False,
}
if state_word in ("open", "accepted"):
if not rfile:
self.finding(
"STRUCTURAL", rel, f"residual `{rid}` ({state_word}) has no witness"
)
elif self.verify_quotes(r["file"], {"quote": quote or ""}):
self.finding(
"CUBE",
rel,
f"stale residual `{rid}`: its witness quote no longer verifies "
f"in {r['file']} — close it (fixed?) or re-witness it",
)
else:
r["verified"] = True
out.append(r)
return out
# a finding is KNOWN when a ratified open/accepted residual with a
# LIVE witness quote names the same artifact. Currency is BYTES, not
# PR-coincidence: the quote is the acknowledgment's pin — if the
# code moves enough to change the declaration, the quote dies, the
# residual reports stale, and the finding gates again. A stale
# residual acknowledges nothing; HARD findings are never
# acknowledgeable (coverage is the floor); docs-side findings are
# structurally uncoverable (a residual's witness is a code file, so
# the artifact never matches — structure is fixed, never carried).
def reconcile_residuals(self, res):
by_file = defaultdict(list)
for r in res:
if r["state_word"] in ("open", "accepted") and r.get("verified"):
by_file[r["file"]].append(r)
for f in self.findings:
if f["tier"] == "HARD":
continue
for r in by_file.get(f["artifact"], []):
f["known"] = r["id"]
break
# decision debt made visible, never gated: a Proposed ADR untouched
# for `days` is a NOTE — the check cannot force a decision, but it
# can refuse to let indecision be invisible.
def aging_proposed(self, adrs, days=60):
notes = []
for a in adrs:
if not (a["status"] or "").startswith("Proposed"):
continue
ts = sh(self.repo, "log", "-1", "--format=%ct", "--", a["path"]).strip()
if not ts.isdigit():
continue
now = sh(self.repo, "log", "-1", "--format=%ct").strip()
if not now.isdigit():
continue
age = (int(now) - int(ts)) // 86_400
if age > days:
notes.append(
f"decision debt: {a['path']} has been Proposed and untouched for {age} days"
)
return notes
# ── the risk-grain view: each risk → its covering ADRs → coverage ────
# The spine IS the decision of what must be covered: every risk file
# deserves an Accepted strategy, and the check walks them one by one.
# An uncovered risk is a STRUCTURAL finding, and it GATES until
# decided (a docs-side finding — no residual can carry it); the
# honest way to HOLD a risk open is an `Accepted (interim, …)` ADR,
# never silence.
def risk_table(self, adrs, res):
by_risk_res = _residuals_by_risk(res)
rows = []
for rel in self.md_files("docs/risks"):
stem = os.path.basename(rel)[:-3]
covering = [
{"adr": a["path"], "stem": os.path.basename(a["path"])[:-3], "status": a["status"]}
for a in adrs
if stem in a["trailers"]
]
covered = any((c["status"] or "").startswith("Accepted") for c in covering)
if not covered:
self.finding(
"STRUCTURAL",
rel,
f"risk `{stem}` has no Accepted covering ADR "
f"({[(c['stem'], c['status']) for c in covering] or 'no trailers name it'}) "
f"— a risk with no strategy is a finding, not a silence; hold it open "
f"honestly with an Accepted (interim, …) ADR if it cannot be decided yet",
)
rows.append(
{
"risk": stem,
"covered": covered,
"adrs": covering,
"residuals": by_risk_res.get(stem, {"open": 0, "accepted": 0, "closed": 0}),
}
)
return rows
# ── comparison 1: structure ↔ canon ──────────────────────────────────
def check_adr_structure(self, adrs, stems):
stem_list = ", ".join(sorted(stems))
for adr in adrs:
rel, text = adr["path"], adr["text"]
missing = [s for s in ADR_SECTIONS if not re.search(rf"^#+\s*{s}\b", text, re.M)]
if missing:
self.finding("STRUCTURAL", rel, f"missing required sections: {missing}")
if adr["status"] and not STATUS_OK.match(adr["status"]):
self.finding("STRUCTURAL", rel, f"Status not in vocabulary: {adr['status']!r}")
if adr["conformance"] and adr["conformance"] not in ("partial", "pending"):
self.finding(
"STRUCTURAL", rel, f"Conformance not in vocabulary: {adr['conformance']!r}"
)
seen = set()
for t in adr["trailers"]:
if t in seen:
self.finding(
"STRUCTURAL", rel, f"duplicate trailer `{t}` (well-formedness fault)"
)
seen.add(t)
if t not in stems:
self.finding(
"STRUCTURAL",
rel,
f"dangling edge: trailer `{t}` resolves to no docs/risks/ stem "
f"(stems checked: {stem_list}) — DAMAGE, not a correction",
)
def check_risk_structure(self, risks, adr_stems):
for r in risks:
rel, text = r["path"], r["text"]
missing = [s for s in RISK_SECTIONS if md_section(text, s) is None]
if missing:
self.finding("STRUCTURAL", rel, f"missing required sections: {missing}")
if re.search(r"^#+\s*Validation question", text, re.M):
self.finding(
"STRUCTURAL",
rel,
"carries a LEGACY `Validation question` section — report for removal",
)
has_marker = "dds-stub" in text
has_holes = bool(HOLE_MARKERS.search(text))
if has_marker != has_holes:
self.finding(
"STRUCTURAL",
rel,
f"stub-marker mismatch: marker={'present' if has_marker else 'absent'}, "
f"holes={'present' if has_holes else 'absent'}",
)
for c in r["classes"]:
if c not in CLASS_SET:
self.finding(
"STRUCTURAL",
rel,
f"Class value `{c}` not in the class vocabulary {CLASS_ORDER}",
)
if len(r["addressed_lines"]) > 1:
self.finding(
"STRUCTURAL",
rel,
"multiple `Addressed by:` lines — protocol 2 allows exactly one "
"(comma-separate the stems)",
)
if r["addressed_lines"] and "`" in r["addressed_lines"][0]:
self.finding(
"STRUCTURAL",
rel,
"`Addressed by:` values are backticked — protocol 2 requires bare stems",
)
for stem in r["addressed_by"]:
if stem not in adr_stems:
self.finding(
"STRUCTURAL",
rel,
f"`Addressed by:` names `{stem}` — resolves to no docs/adr file",
)
if r["signals"] and not r["classes"]:
self.finding(
"STRUCTURAL",
rel,
"signals with no `Class:` declaration — a signal needs a class "
"home; declare the class this risk covers (or move the signal "
"to the risk that does)",
)
def coverage_closure(self, adrs, risks):
"""ADR Risk trailers ↔ risk-file Addressed by, both directions."""
addressed = {r["id"]: set(r["addressed_by"]) for r in risks}
for adr in adrs:
adr_stem = os.path.basename(adr["path"])[:-3]
for t in adr["trailers"]:
if t in addressed and adr_stem not in addressed[t] and addressed[t]:
self.finding(
"STRUCTURAL",
adr["path"],
f"coverage closure: names `{t}` but "
f"docs/risks/{t}.md's `Addressed by:` does not name {adr_stem}",
)
# ── the register world: canonical parse + facet law ─────────────────
def facet_values(self, seg):
"""{Field → cleaned value} from a section's facet table rows."""
out = {}
for m in FACET_ROW.finditer(seg):
field, value = m.group(1), m.group(2)
if field == "Field":
continue
out[field] = clean_facet(value)
return out
# the mapping ledger: WHICH COLUMNS EXIST is witness (recomputed);
# WHICH DATUM a column carries is judgment — a `Held by:` line in the
# datum's own section, evidence = the column's declaring quote. The
# quote IS the currency pin: dead quote = stale mapping, re-confirm.
HELD_LINE = re.compile(
r"^- (°\s*)?([\w-]+)\.([\w-]+)\s*\[([^|\]]+)\|\s*\"(.+?)\"\s*\]$", re.S
)
def parse_held_by(self, rel, dname, body):
m = re.search(r"^Held by:\s*$\n((?:(?:- |\s+).*\n?)*)", body, re.M)
if not m:
return []
out = []
for raw in re.split(r"^- ", m.group(1), flags=re.M)[1:]:
entry = "- " + re.sub(r"\s+", " ", raw.strip())
mm = self.HELD_LINE.match(entry)
if not mm:
self.finding(
"STRUCTURAL", rel, f"datum `{dname}`: malformed `Held by` line: {entry[:90]}"
)
continue
draft_mark, store, column, sfile, quote = mm.groups()
rec = {
"store": store,
"column": column,
"file": sfile.strip(),
"quote": quote,
"draft": bool(draft_mark),
"stale": False,
}
if self.verify_quotes(rec["file"], {"quote": quote}):
rec["stale"] = True
self.finding(
"CUBE",
rel,
f"stale mapping: `{store}.{column}` → `{dname}` — its declaring "
f"quote no longer verifies in {rec['file']}; re-confirm the "
"mapping or remove the line (stale — re-confirm, not unwitnessed)",
)
out.append(rec)
return out
def kind_law(self, kind):
"""The facet law for a party kind (internal shares organization's)."""
law = self.core.get("facet_law") or DEFAULT_CORE["facet_law"]
return law.get(kind) or law.get("organization") if kind == "internal" else law.get(
kind, law["person"]
)
def validate_datum_facets(self, rel, dname, facets, holes, kind="person"):
law = self.kind_law(kind)
vocab = law.get("vocab") or {}
multi = set(law.get("multi") or [])
for f in law.get("required") or []:
v = facets.get(f)
if v is None:
self.finding("STRUCTURAL", rel, f"datum `{dname}` missing required facet `{f}`")
holes.add(f"{f}: missing")
elif HOLE_VALUE.search(v):
holes.add(f"{f}: {v}")
elif f in vocab and v not in vocab[f]:
self.finding(
"STRUCTURAL",
rel,
f"datum `{dname}` facet `{f}` value `{v}` not in vocabulary "
f"{sorted(vocab[f])}",
)
# optional facets still keep their vocabulary (multi = ·-separated)
for f, allowed in vocab.items():
v = facets.get(f)
if v is None or f in (law.get("required") or []) or HOLE_VALUE.search(v):
continue
values = [t.strip() for t in re.split(r"[·,]", v)] if f in multi else [v]
for val in values:
if val and val not in allowed:
self.finding(
"STRUCTURAL",
rel,
f"datum `{dname}` facet `{f}` value `{val}` not in vocabulary "
f"{sorted(allowed)}",
)
if (
law.get("source_party_when_provided")
and facets.get("Provenance") == "provided"
and "Source party" not in facets
):
self.finding(
"STRUCTURAL",
rel,
f"datum `{dname}` is `provided` but carries no `Source party` facet",
)
holes.add("Source party: missing")
def parse_subjects(self):
subjects, datums = [], []
for rel in self.md_files("docs/registers/subjects"):
text = read(self.repo, rel) or ""
head = re.search(r"^#\s+(.+?)\s+·\s+`([\w-]+)`", text, re.M)
if head:
sname, sid = head.group(1).strip(), head.group(2)
else:
self.finding(
"STRUCTURAL", rel, "subject file missing its `# <Name> · `<id>`` header"
)
sid = sname = os.path.basename(rel)[:-3]
header_seg = re.split(r"^###\s", text, maxsplit=1, flags=re.M)[0]
header_facets = self.facet_values(header_seg)
kind = header_facets.get("Kind", "person")
if kind not in (self.core.get("subject_kinds") or DEFAULT_CORE["subject_kinds"]):
self.finding(
"STRUCTURAL",
rel,
f"subject `Kind` value `{kind}` not in vocabulary "
f"{self.core.get('subject_kinds') or DEFAULT_CORE['subject_kinds']}",
)
kind = "person"
subjects.append(
{
"id": sid,
"name": sname,
"kind": kind,
"file": rel,
"facets": header_facets,
"holes": sorted(set(HOLE_MARKERS.findall(header_seg))),
"draft": bool(DRAFT_MARK.search(header_seg)),
}
)
facet_sets = []
for m in SECTION_HEAD.finditer(text):
dname, key, body = m.group(1).strip(), m.group(2), m.group(3)
facets = self.facet_values(body)
holes = set(HOLE_MARKERS.findall(body))
if key.split(".", 1)[0] != sid:
self.finding(
"STRUCTURAL",
rel,
f"datum key `{key}` does not carry its subject's stem `{sid}.`",
)
self.validate_datum_facets(rel, dname, facets, holes, kind=kind)
datums.append(
{
"key": key,
"subject": sid,
"name": dname,
"file": rel,
"facets": facets,
"held_by": self.parse_held_by(rel, dname, body),
"holes": sorted(holes),
"draft": bool(DRAFT_MARK.search(body)),
}
)
if facets:
facet_sets.append((dname, set(facets)))
# a ratified G-cell does not vanish without a PR-visible decision:
# sibling datums carry the modal facet-row set
if len(facet_sets) >= 2:
modal = max(
(fs for _, fs in facet_sets),
key=lambda s: sum(1 for _, o in facet_sets if o == s),
)
for title, fs in facet_sets:
lost = modal - fs
if lost:
self.finding(
"STRUCTURAL",
rel,
f"datum `{title}` is missing facet row(s) {sorted(lost)} "
f"that sibling datums carry — a ratified G-cell does not "
f"vanish without a PR-visible decision",
)
return subjects, datums
def parse_activities_file(self):
rel = "docs/registers/processing-activities.md"
text = read(self.repo, rel)
if text is None:
return []
acts = []
for m in SECTION_HEAD.finditer(text):
name, sid, body = m.group(1).strip(), m.group(2), m.group(3)
facets = self.facet_values(body)
holes = set(HOLE_MARKERS.findall(body))
summary = next(
(s.strip() for s in re.findall(r"^>\s*(.+)$", body, re.M) if "°" not in s),
None,
)
alaw = self.core.get("activity_law") or DEFAULT_CORE["activity_law"]
authority = alaw.get("authority_field", "Lawful basis")
authority_required = any(p.get("require_activity_authority") for p in self.regimes)
lb = facets.get(authority)
if lb is None:
if authority_required:
names = [p["name"] for p in self.regimes if p.get("require_activity_authority")]
self.finding(
"STRUCTURAL",
rel,
f"activity `{name}` missing `{authority}` — required by "
f"regime(s) {names} (a `basis.TODO.<slug>` hole is the honest start)",
)
holes.add(f"{authority}: missing")
elif lb.startswith("basis.TODO") or HOLE_VALUE.search(lb):
holes.add(f"{authority}: {lb}")
elif not re.match(alaw.get("authority_grammar", r"^basis\.[\w.-]+$"), lb):
self.finding(
"STRUCTURAL",
rel,
f"activity `{name}` `{authority}` value `{lb}` must reference a "
"`basis.*` id (or `basis.TODO.<slug>` as the honest hole)",
)
if not facets.get("Purpose"):
self.finding("STRUCTURAL", rel, f"activity `{name}` missing `Purpose`")
holes.add("Purpose: missing")
for f in alaw.get("required") or []:
if f == "Purpose":
continue
v = facets.get(f)
vocab = (alaw.get("vocab") or {}).get(f)
if v is None:
self.finding("STRUCTURAL", rel, f"activity `{name}` missing `{f}`")
holes.add(f"{f}: missing")
elif HOLE_VALUE.search(v):
holes.add(f"{f}: {v}")
elif vocab and v not in vocab:
self.finding(
"STRUCTURAL",
rel,
f"activity `{name}` `{f}` value `{v}` not in vocabulary {sorted(vocab)}",
)
acts.append(
{
"stable_id": sid,
"name": name,
"file": rel,
"summary": summary,
"purpose": facets.get("Purpose"),
"lawful_basis": lb,
"function": facets.get("Function"),
"automation": facets.get("Automation"),
"holes": sorted(holes),
"draft": bool(DRAFT_MARK.search(body)),
}
)
return acts
def parse_governance_file(self):
rel = "docs/governance.md"
text = read(self.repo, rel)
if text is None:
return []
out = []
stems = self.core.get("governance_stems") or DEFAULT_CORE["governance_stems"]
for m in SECTION_HEAD.finditer(text):
name, sid, body = m.group(1).strip(), m.group(2), m.group(3)
kind = next((s for s in stems if sid.startswith(s + ".")), None)
if kind is None:
self.finding(
"STRUCTURAL",
rel,
f"governance id `{sid}` must use one of the stems "
f"{[s + '.*' for s in stems]}",
)
kind = "other"
fields = {}
for fm in FACET_ROW.finditer(body):
field, value, mark = fm.group(1), fm.group(2), fm.group(3)
if field == "Field":
continue
fields[field] = {
"value": clean_facet(value),
"mark": mark.strip() if mark and mark.strip() else None,
}
direction = (fields.get("Direction") or {}).get("value")
if kind == "retention" and direction and not HOLE_VALUE.search(direction):
allowed = self.core.get("retention_direction") or DEFAULT_CORE["retention_direction"]
if direction not in allowed:
self.finding(
"STRUCTURAL",
rel,
f"retention `{sid}` `Direction` value `{direction}` not in "
f"vocabulary {allowed} (ceiling = delete by; floor = preserve until)",
)
summary = next((s.strip() for s in re.findall(r"^>\s*(.+)$", body, re.M)), None)
out.append(
{
"stable_id": sid,
"kind": kind,
"name": name,
"file": rel,
"summary": summary,
"fields": fields,
"holes": sorted(set(HOLE_MARKERS.findall(body))),
}
)
return out
def register_world(self):
# holdings.json is RETIRED: the file's presence is format debt
if read(self.repo, "docs/registers/holdings.json") is not None:
self.finding(
"STRUCTURAL",
"docs/registers/holdings.json",
"holdings.json is RETIRED (protocol 2): column→datum mappings are "
"`Held by:` ledger lines in each datum's claim-file section — run "
"the 0.11 migration to convert the pins, then delete this file",
)
subjects, datums = self.parse_subjects()
# regime rules over the parsed claim world (a FIXED rule set —
# profiles parameterize; they never carry logic)
for prof in self.regimes:
for f in prof.get("required_person_subject_facets") or []:
for s in subjects:
if s["kind"] == "person" and f not in s["facets"]:
self.finding(
"STRUCTURAL",
s["file"],
f"regime `{prof['name']}`: person subject `{s['id']}` "
f"missing `{f}` facet",
)
for stem in prof.get("required_governance_stems") or []:
if f"{stem}." not in self._gov_text:
self.finding(
"STRUCTURAL",
"docs/governance.md",
f"regime `{prof['name']}` requires a `{stem}.*` entry — "
f"a `{stem}.TODO` hole is the honest start",
)
return {
"subjects": subjects,
"datums": datums,
"activities": self.parse_activities_file(),
"governance": self.parse_governance_file(),
}
def scope_rules(self, stores, register):
"""Class-grain protection rules for non-person data (the
confidentiality scope): a store carrying a classified
org/internal datum via a live mapping must have a policy gate."""
tiers, active = set(), []
for p in self.regimes:
if p.get("classified_store_gating"):
active.append(p["name"])
tiers |= set(p.get("classified_tiers") or ["confidential", "restricted"])
if not active:
return
kind_by_subject = {s["id"]: s["kind"] for s in register["subjects"]}
for d in register["datums"]:
if kind_by_subject.get(d["subject"]) not in ("organization", "internal"):
continue
cls = (d["facets"].get("Classification") or "").lower()
if cls not in tiers:
continue
for h in d.get("held_by") or []:
if h["stale"] or h["draft"]:
continue
doc = stores.get(h["store"])
if doc and not (doc.get("policies") or {}).get("present"):
self.finding(
"CUBE",
doc["file"],
f"classified store `{h['store']}`: carries `{d['key']}` "
f"(Classification {cls}) with no policy gate — scope "
f"{active}",
)
def derived_holdings(self, register):
"""The holdings VIEW — computed every run from the ratified,
non-stale `Held by` lines; committed nowhere. The consumer
contract shape ({store, column, datum_key}) is unchanged."""
return [
{"store": h["store"], "column": h["column"], "datum_key": d["key"]}
for d in register["datums"]
if not d["draft"]
for h in d.get("held_by", [])
if not h["stale"] and not h["draft"]
]
# ── cube: coverage, dormancy ─────────────────────────────────────────
def cube(self, holdings):
stores = {}
for rel, doc in self.witness["files"].items():
if doc.get("store"):
stores[doc["store"]] = dict(doc, file=rel)
mapped_stores = {h["store"] for h in holdings}
for store, doc in sorted(stores.items()):
person = any(
self.pmatch("person_signal", c["name"], PERSON_SIGNAL)
for c in doc.get("columns", [])
) or doc.get("person_store")
if not person:
continue
# coverage = the ratified mapping ledger: a person store is
# covered when at least one of its columns lands on a claimed
# datum via a live `Held by` line
covered = store in mapped_stores
refs = self.reference_count(doc)
if not covered:
self.finding(
"CUBE",
doc["file"],
f"uncovered surface: person store `{store}` "
f"({[c['name'] for c in doc.get('columns', []) if PERSON_SIGNAL.search(c['name'])]}) "
f"has no `Held by` mapping landing on a claimed datum",
)
if refs == 0:
self.finding(
"CUBE",
doc["file"],
f"dormant store `{store}`: module referenced nowhere outside its "
f"own file"
+ (
"; not registered in its domain"
if doc.get("domain_registered") is False
else ""
),
)
return stores
def reference_count(self, doc):
mod = doc.get("module", "")
base = mod.rsplit(".", 1)[-1] if mod else None
if not base:
return -1
count = 0
roots = [r for rs in self.pconf("reference_roots") for r in rs] or ["lib", "test"]
suffixes = tuple(
s for ss in self.pconf("reference_file_suffixes") for s in ss
) or (".ex", ".exs")
for root in roots:
top = os.path.join(self.repo, root)
for dirpath, _, names in os.walk(top):
for n in names:
if not n.endswith(suffixes):
continue
rel = os.path.relpath(os.path.join(dirpath, n), self.repo)
if rel == doc["file"]:
continue
text = read(self.repo, rel) or ""
if re.search(rf"\b{re.escape(base)}\b", text):
count += 1
return count
# ── presence, table, read-off ────────────────────────────────────────
def presence(self, stores, risks):
signals = defaultdict(list)
for store, doc in sorted(stores.items()):
person = any(
self.pmatch("person_signal", c["name"], PERSON_SIGNAL)
for c in doc.get("columns", [])
) or doc.get("person_store")
pol = doc.get("policies") or {}
if person and not pol.get("present"):
signals["gating"].append(f"ungated person store `{store}`")
acts = doc.get("actions") or {}
has_erasure = acts.get("destroy") or acts.get("archive") or acts.get("redaction")
if person and not has_erasure:
signals["erasure"].append(f"person store `{store}` with no erasure path")
if self.pmatch("audit_store", store, AUDIT_STORE):
refs = self.reference_count(doc)
if refs == 0 or not (doc.get("policies") or {}).get("present"):
signals["audit"].append(
f"audit-shaped store `{store}` "
+ ("dormant" if refs == 0 else "with no policy block")
)
router = self.witness.get("router") or {}
for p in router.get("pipelines", []):
if not any(
self.pmatch("auth_plug", pl, AUTH_PLUG) for pl in p.get("plugs", [])
):
signals["surface"].append(f"pipeline `{p['name']}` carries no auth-shaped plug")
if self.witness.get("secrets"):
signals["credential"].append(
f"{len(self.witness['secrets'])} census-witnessed secrets"
)
# ratified signals ride the risk files' `## Signals` ledgers,
# bound to classes through each risk's `Class:` declarations
for r in risks:
for cls in r["classes"]:
if cls not in CLASS_SET:
continue
for sig in r["signals"]:
signals[cls].append(sig["text"])
return signals
def table(self, adrs, signals, class_map):
"""The five-row coverage table. Protocol 2: an ADR belongs to a
class through its Risk trailers — the named risks' ratified
`Class:` declarations — never through keyword matching."""
rows = []
for cls in CLASS_ORDER:
members = [
{"adr": a["path"], "status": a["status"], "trailers": a["trailers"]}
for a in adrs
if any(cls in class_map.get(t, ()) for t in a["trailers"])
]
accepted = any(
(m["status"] or "").startswith("Accepted") for m in members
)
rows.append(
{
"class": cls,
"present": bool(signals.get(cls)),
"signals": signals.get(cls, []),
"members": members,
"any_accepted": accepted,
}
)
return rows
def hard_tier(self, rows, base_rows):
base_ok = {r["class"]: r["any_accepted"] for r in base_rows}
for r in rows:
if r["present"] and not r["any_accepted"]:
was_covered = base_ok.get(r["class"], False)
tag = "delta" if was_covered else "background"
self.findings.append(
{
"tier": "HARD",
"artifact": f"coverage-table:{r['class']}",
"text": (
f"present class `{r['class']}` has NO Accepted ADR in its row "
f"(members: {[(m['adr'], m['status']) for m in r['members']]}; "
f"signals: {r['signals']})"
+ (
" — the delta emptied this row (it had Accepted cover at base)"
if was_covered
else " — uncovered at base too"
)
),
"delta": tag == "delta",
}
)
# ── changed classes + nominees ───────────────────────────────────────
def changed_classes(self, rows, base_rows):
changed = set()
for rel in self.delta_files:
doc = self.witness["files"].get(rel)
baseline = self.baseline_files.get(rel)
if doc and doc.get("store"):
if baseline is None:
changed.add("gating")
else:
if (doc.get("policies") or {}) != (baseline.get("policies") or {}):
changed.add("gating")
if {c["name"] for c in doc.get("columns", [])} != {
c["name"] for c in baseline.get("columns", [])
}:
changed.add("gating")
a, b = doc.get("actions") or {}, (baseline or {}).get("actions") or {}
for key in ("destroy", "archive", "redaction"):
if a.get(key) != b.get(key):
changed.add("erasure")
if AUDIT_STORE.search(doc.get("store", "")):
changed.add("audit")
if self.pmatch("surface_file_regex", rel):
changed.add("surface")
if self.pmatch("credential_file_regex", rel):
changed.add("credential")
# a doc edit that changes a row's membership re-examines that class
base_members = {r["class"]: {m["adr"] for m in r["members"]} for r in base_rows}
for r in rows:
if {m["adr"] for m in r["members"]} != base_members.get(r["class"], set()):
changed.add(r["class"])
return sorted(changed)
def nominees(self, rows, changed, adrs, stems):
"""Every ADR carrying a resolving Risk trailer is judged every
run — the per-ADR implementation verdict is the report's spine
(supersedes changed-class-only nomination; the untouched-ADR
lesson holds trivially when everything is judged). Class
membership is attached where it exists, for the judge's facts."""
by_class = {}
for r in rows:
for m in r["members"]:
by_class.setdefault(m["adr"], []).append(r["class"])
out = []
for a in adrs:
if any(t in stems for t in a["trailers"]):
out.append(
{
"adr": a["path"],
"status": a["status"],
"conformance": a["conformance"],
"classes": sorted(set(by_class.get(a["path"], []))),
"changed": bool(set(by_class.get(a["path"], [])) & set(changed)),
}
)
return sorted(out, key=lambda n: n["adr"])
# ── verdicts: completeness + shared verification + derived findings ──
def _tsq(self):
"""The structural-search engine, or None (no tree-sitter here)."""
if not hasattr(self, "_tsq_mod"):
try:
import tsq as _tsq_mod
self._tsq_mod = _tsq_mod
except Exception:
self._tsq_mod = None
return self._tsq_mod
def merge_verdicts(self, nominees):
if not self.verdicts_path:
return None, [n["adr"] for n in nominees]
try:
verdicts = json.loads(open(self.verdicts_path, encoding="utf-8").read())
except (OSError, json.JSONDecodeError) as e:
self.degraded.append(f"verdicts file unparseable: {e}")
return None, [n["adr"] for n in nominees]
if not isinstance(verdicts, list):
self.degraded.append(
"verdicts file must be a BARE LIST of records (protocol 2 — "
"no envelope); rejected"
)
return None, [n["adr"] for n in nominees]
verdicts = [v for v in verdicts if isinstance(v, dict) and v.get("adr")]
by_adr = {v["adr"]: v for v in verdicts}
missing = [n["adr"] for n in nominees if n["adr"] not in by_adr]
rejected = set()
profile_queries = {}
for p in self.platforms:
profile_queries.update(p.get("queries") or {})
verifier = Verifier(self.repo, profile_queries, self._tsq())
replayable = bool(self.platforms and self.platforms[0].get("language"))
for n in nominees:
v = by_adr.get(n["adr"])
if not v:
continue
if v.get("verdict") not in VERDICT_ENUM:
self.degraded.append(
f"verdict for {n['adr']} not in enum: {v.get('verdict')} — "
"rejected (the record shape is the contract)"
)
rejected.add(n["adr"])
missing.append(n["adr"])
continue
adr_text = read(self.repo, n["adr"]) or ""
scope = rule_scope(adr_text)
if scope is None:
self.degraded.append(
f"{n['adr']}: no ## Rule section — rule_quote checked against the whole file"
)
scope = adr_text
# the shared verifier: rule_quote against the Rule section,
# evidence REPLAYED against the tree — a claim that does not
# reproduce rejects the verdict outright; it must be redone.
problems, notes = verifier.validate(scope, v, replayable=replayable)
for note in notes:
self.degraded.append(f"{n['adr']}: {note}")
if problems:
for prob in problems:
self.degraded.append(f"verdict for {n['adr']}: {prob}")
rejected.add(n["adr"])
missing.append(n["adr"])
continue
accepted = (n["status"] or "").startswith("Accepted")
conformance_trailer = re.search(r"^Conformance:", adr_text, re.M)
if accepted and v["verdict"] == "partial" and not conformance_trailer:
self.finding(
"JUDGED",
n["adr"],
"derived silent_conformance_mismatch: Accepted + judged partial + "
"no `Conformance:` trailer",
)
if v["verdict"] in ("unwitnessed", "partial") and v.get("dormant_mechanism"):
self.finding(
"JUDGED",
n["adr"],
f"derived dormant_mechanism: {v.get('dormant_mechanism')}",
)
# a rejected verdict never enters the doc — its claims did not survive
verdicts = [v for v in verdicts if v["adr"] not in rejected]
return verdicts, missing
# ── run ──────────────────────────────────────────────────────────────
def run(self):
self.detect_platforms()
self.load_delta()
self.load_witness()
self.load_regimes()
self.check_dds_version()
divergences = self.canon_conformance()
adrs = self.adr_corpus()
base_adrs = self.adr_corpus(at_base=True)
risks = self.parse_risks()
stems = {r["id"] for r in risks}
adr_stems = {os.path.basename(a["path"])[:-3] for a in adrs}
class_map = {r["id"]: set(r["classes"]) for r in risks}
base_class_map = self.base_risk_class_map()
self.check_adr_structure(adrs, stems)
self.check_risk_structure(risks, adr_stems)
self.coverage_closure(adrs, risks)
register = self.register_world()
register["holdings"] = self.derived_holdings(register)
stores = self.cube(register["holdings"])
self.scope_rules(stores, register)
signals = self.presence(stores, risks)
rows = self.table(adrs, signals, class_map)
base_rows = self.table(base_adrs, signals, base_class_map)
self.hard_tier(rows, base_rows)
changed = self.changed_classes(rows, base_rows)
noms = self.nominees(rows, changed, adrs, stems)
verdicts, missing_verdicts = self.merge_verdicts(noms)
# judge-reliability instrumentation (the evidence that decides
# whether any rule ever needs a frozen evaluation): compare
# against a prior run's verdicts (--prior-verdicts, e.g. the
# reuse step's .dds-seed.json). A flip with no relevant delta is
# instability or a doc-only change; consecutive not_verifiable
# is evaluability debt — pressure on the RULE, not the judge.
stability_notes = []
if self.prior_path and verdicts:
try:
prior_list = json.loads(open(self.prior_path, encoding="utf-8").read())
prior = {
v["adr"]: v.get("verdict")
for v in prior_list
if isinstance(v, dict) and v.get("adr")
}
except (OSError, json.JSONDecodeError) as e:
prior = {}
self.degraded.append(f"prior verdicts unreadable: {e}")
changed_map = {n["adr"]: n["changed"] for n in noms}
for v in verdicts:
pv = prior.get(v["adr"])
if pv is None:
continue
if v["verdict"] != pv and not changed_map.get(v["adr"]):
stability_notes.append(
f"verdict flipped without relevant delta: {v['adr']} "
f"{pv} → {v['verdict']} — judge instability or a doc-only "
"change; an evaluability signal"
)
if v["verdict"] == pv == "not_verifiable":
stability_notes.append(
f"evaluability debt: {v['adr']} not_verifiable across "
"consecutive runs — sharpen the Rule, or reclassify it as "
"a process rule outside the judged corpus"
)
res = self.residuals()
risk_rows = self.risk_table(adrs, res)
self.reconcile_residuals(res)
res_by_risk = defaultdict(list)
for r in res:
res_by_risk[r["risk"]].append(
{"id": r["id"], "state": r["state"], "text": r["text"], "file": r["file"]}
)
register["risks"] = [
{
"id": r["id"],
"title": r["title"],
"stake": r["stake"],
"currency": r["currency"],
"classes": r["classes"],
"addressed_by": r["addressed_by"],
"signals": r["signals"],
"stub": r["stub"],
"holes": r["holes"],
"residuals": res_by_risk.get(r["id"], []),
}
for r in risks
]
for f in self.findings:
f["fix"] = remedy(f)
hard_fail = any(f["tier"] == "HARD" for f in self.findings)
# THE RESIDUAL LEDGER IS THE ONLY EXCEPTION CHANNEL: a mechanical
# finding (structural or cube) gates unless a ratified residual
# carries it. When it was introduced confers nothing — the delta
# tag is provenance, never policy (time-of-introduction was a
# silent second exception mechanism, and there is no fourth
# exit). Judged verdicts remain advisory: their declared-
# exception forms are the Conformance trailer and interim status.
unacknowledged = any(
f["tier"] in ("STRUCTURAL", "CUBE") and not f.get("known")
for f in self.findings
)
# missing verdicts gate whether or not a verdicts file was
# supplied: a run with nominees and no judgments is INCOMPLETE,
# never a PASS (run 29704740180: an empty-delta attestation
# passed with zero verdicts because this guard keyed on the
# flag's presence — silence passed).
incomplete = bool(self.needs_extraction) or bool(missing_verdicts)
status = (
"INCOMPLETE"
if incomplete
else ("FAIL" if (hard_fail or unacknowledged) else "PASS")
)
return {
"status": status,
"protocol": PROTOCOL,
"base": self.base,
"regimes": [
{
"name": p["name"],
"kind": p.get("kind", "regime"),
"sketch": p.get("sketch", False),
"terminology": p.get("terminology") or {},
}
for p in self.regimes
],
"platforms": [
{
"name": p["name"],
"evidence": p.get("_evidence"),
"extraction_guide": p.get("extraction_guide"),
"sketch": p.get("sketch", False),
}
for p in self.platforms
],
"delta_files": sorted(self.delta_files),
"changed_classes": changed,
"coverage_table": rows,
"findings": self.findings,
"nominees": noms,
"missing_verdicts": missing_verdicts,
"verdicts": verdicts,
"needs_extraction": self.needs_extraction,
"degraded": self.degraded,
"risks": risk_rows,
"register": register,
"residuals": {
"open": [r["id"] for r in res if r["state_word"] == "open"],
"accepted": [r["id"] for r in res if r["state_word"] == "accepted"],
"closed": [r["id"] for r in res if r["state_word"] == "closed"],
"by_risk": _residuals_by_risk(res),
},
"divergences": divergences,
"stability_notes": stability_notes,
"aging_proposed": self.aging_proposed(adrs),
"not_checked": [
"governance provenance-tag semantics (D/C/G cell truthfulness) — "
"the marks are parsed and carried; their semantics are not judged",
"AGENTS.md/CLAUDE.md dds-section canon conformance + vendored-rule "
"inlining — mixed-pen files, not yet mechanized",
],
}
# ── remedies: every finding knows its fix ───────────────────────────
# The three-exits doctrine, attached per finding: kind ∈ referential
# (a link/id/vocabulary to correct) · mechanical (/dds-sync, a
# migration, or a regeneration repairs it) · code (the tree needs the
# change) · judgmental (a human decision, via PR). First match wins;
# specific entries precede generic ones.
REMEDY_TABLE = [
("has no Accepted covering ADR", "judgmental",
"decide a strategy: /dds-suggest-adrs, then ratify — or hold it open honestly with an Accepted (interim, …) ADR"),
("present class `", "judgmental",
"the class is live in code with no Accepted cover — /dds-suggest-adrs; coverage is the floor and cannot be residualized"),
("missing required sections", "mechanical",
"/dds-sync repairs structure (or add the section yourself)"),
("stub-marker mismatch", "mechanical",
"/dds-sync fixes the marker (present iff holes remain)"),
("carries a LEGACY", "mechanical",
"remove the section (0.11 migration step 3); quote its text in the PR"),
("holdings.json is RETIRED", "mechanical",
"0.11 migration step 4: convert pins to `Held by` lines, delete the file"),
("missing `protocol:", "mechanical",
"/dds-upgrade pins it (0.11 migration step 1)"),
("undeclared canon drift", "mechanical",
"wrap the edit in a dds-divergence block, revert it, or /dds-upgrade if the canon version moved"),
("malformed", "referential",
"fix the line to its grammar (risks-README / conventions §12 / claim-files.md)"),
("duplicate residual id", "referential",
"residual ids are corpus-unique — rename one; write `see res.<id>` from the other"),
("dangling edge", "referential",
"point the trailer at an existing docs/risks stem (or create the risk)"),
("duplicate trailer", "referential", "delete the repeated trailer line"),
("coverage closure", "referential",
"add the ADR's stem to the risk's `Addressed by:` line (or drop the trailer)"),
("resolves to no docs/adr", "referential",
"fix the `Addressed by:` stem (or remove it until the ADR exists)"),
("backticked", "referential", "bare stems on `Addressed by:` — drop the backticks"),
("multiple `Addressed by", "referential", "collapse to one comma-separated line"),
("matches no carried profile", "referential",
"fix the Regime:/Scope: token — the carried profiles are listed in the finding"),
("must use one of the stems", "referential",
"re-mint the id on a basis.* / retention.* / terms.* stem"),
("does not carry its subject's stem", "referential",
"datum keys are <subject>.<slug> — rekey it"),
("Status not in vocabulary", "referential",
"Proposed · Accepted (YYYY-MM-DD) · Accepted (interim, YYYY-MM-DD)"),
("Conformance not in vocabulary", "referential",
"partial · pending (absence asserts realized)"),
("Class value", "referential",
"declare one of gating · erasure · audit · surface · credential"),
("`Direction` value", "referential",
"ceiling (delete by) · floor (preserve until)"),
("`Kind` value", "referential", "person · organization · internal"),
("missing required facet", "judgmental",
"fill the facet — needs-ratification is the honest hole; /dds-sync drafts from evidence"),
("not in vocabulary", "judgmental",
"pick a value from the vocabulary in the finding — or leave an honest hole (needs-ratification)"),
("missing `Minor`", "judgmental",
"declare the person subject's Minor facet (a regime requirement)"),
("requires a `", "judgmental",
"add the governance entry — its TODO hole is the honest start; counsel sources the value"),
("missing `Lawful basis`", "judgmental",
"bind the activity to a basis.* (basis.TODO.<slug> is the honest start)"),
("missing `Purpose`", "judgmental", "one plain sentence an auditor understands"),
("signals with no `Class:", "judgmental",
"declare the class this risk covers, or move the signal to the risk that does"),
("stale residual", "judgmental",
"the code moved: close it (fixed?) or re-witness with a fresh quote"),
("stale mapping", "judgmental",
"re-confirm the `Held by` line against the new declaration, or remove it"),
("stale signal", "judgmental", "re-witness the signal or remove the line"),
("stale divergence", "judgmental",
"re-base the divergence onto the new canon text, or drop it"),
("uncovered surface", "judgmental",
"/dds-sync drafts the claim + mapping from evidence; merging ratifies"),
("dormant store", "judgmental",
"a decision, not a defect: wire it, remove it, or accept it via /dds-suggest-adrs"),
("classified store", "code",
"add the policy gate — or consciously carry it (ratify a residual)"),
("silent_conformance_mismatch", "referential",
"declare it (`Conformance: partial`) — or fix the code and re-judge"),
("dormant_mechanism", "judgmental",
"built but unwired: wire it, or ratify the interim honestly"),
]
def remedy(f):
for needle, kind, hint in REMEDY_TABLE:
if needle in f["text"]:
return {"kind": kind, "hint": hint}
return {
"kind": "judgmental" if f["tier"] in ("CUBE", "JUDGED", "HARD") else "mechanical",
"hint": "see the finding text — /dds-sync for mechanical repair; a PR for judgment",
}
def _residuals_by_risk(res):
# counts per state PLUS the ids behind the open/accepted counts — the
# consumers (the report, the portal's ingested walk) name the ledger
# entries instead of an anonymous tally.
by = {}
for r in res:
by.setdefault(
r["risk"],
{"open": 0, "accepted": 0, "closed": 0, "open_ids": [], "accepted_ids": []},
)
by[r["risk"]][r["state_word"]] += 1
if r["state_word"] in ("open", "accepted"):
by[r["risk"]][r["state_word"] + "_ids"].append(r["id"])
return by
def render(doc):
lines = [
f"status: {doc['status']} (base {doc['base']})",
f"protocol: {doc.get('protocol')}",
"",
]
for p in doc.get("platforms", []):
sk = " [SKETCH — unproven profile]" if p.get("sketch") else ""
lines.append(
f"platform: {p['name']}{sk} — {p['evidence']} — extraction guide: {p['extraction_guide']}"
)
if doc.get("platforms"):
lines.append("")
regs = doc.get("regimes") or []
if regs:
lines.append(
"obligation profiles: "
+ " · ".join(
f"{r['name']} ({r['kind']}{', SKETCH' if r.get('sketch') else ''})" for r in regs
)
)
lines.append("")
divs = doc.get("divergences") or []
if divs:
lines.append("canon divergences (declared):")
for d in divs:
st = " [STALE]" if d.get("stale") else ""
lines.append(
f" {d['file']} · `{d['slug']}` ({d['kind']}){st}"
+ (f" — {d['why']}" if d.get("why") else "")
)
lines.append("")
reg = doc.get("register") or {}
if reg:
d = reg.get("datums") or []
lines.append(
f"register: {len(reg.get('subjects') or [])} subjects · {len(d)} datums "
f"({sum(1 for x in d if x.get('draft'))} draft · "
f"{sum(1 for x in d if x.get('holes'))} with holes) · "
f"{len(reg.get('activities') or [])} activities · "
f"{len(reg.get('governance') or [])} governance · "
f"{len(reg.get('holdings') or [])} holdings"
)
lines.append("")
lines.append(f"delta files ({len(doc['delta_files'])}):")
for f in doc["delta_files"]:
lines.append(f" {f}")
lines.append("")
lines.append("coverage table:")
for r in doc["coverage_table"]:
mark = "PRESENT" if r["present"] else "absent"
acc = "yes" if r["any_accepted"] else "NO"
members = ", ".join(f"{os.path.basename(m['adr'])}({m['status']})" for m in r["members"]) or "—"
lines.append(f" {r['class']:<10} {mark:<8} accepted:{acc:<4} {members}")
for s in r["signals"]:
lines.append(f" · {s}")
lines.append("")
if doc["changed_classes"]:
lines.append(f"changed classes: {', '.join(doc['changed_classes'])}")
if doc["nominees"]:
lines.append("nominees to judge (one verdict EACH — see SKILL.md):")
for n in doc["nominees"]:
lines.append(f" {n['adr']} [{','.join(n['classes'])}] ({n['status']})")
lines.append("")
verdicts_by_adr = {v["adr"]: v for v in (doc.get("verdicts") or [])}
if doc.get("risks"):
lines.append("risks:")
for r in doc["risks"]:
resc = r["residuals"]
cover = "covered" if r["covered"] else "GAP"
lines.append(
f" {r['risk']} — {cover} — residuals: {resc['open']} open · {resc['accepted']} accepted"
)
for c in r["adrs"]:
v = verdicts_by_adr.get(c["adr"])
verdict = f" — {v['verdict']}" if v else ""
lines.append(f" · {c['stem']} ({c['status']}){verdict}")
if not r["adrs"]:
lines.append(" · (no ADR names this risk)")
lines.append("")
resid = doc.get("residuals") or {}
if any(resid.get(k) for k in ("open", "accepted", "closed")):
lines.append(
f"residuals: {len(resid.get('open', []))} open · "
f"{len(resid.get('accepted', []))} accepted · "
f"{len(resid.get('closed', []))} closed"
)
for risk, counts in sorted((resid.get("by_risk") or {}).items()):
lines.append(f" {risk}: {counts['open']} open · {counts['accepted']} accepted")
lines.append("")
for f in doc["findings"]:
tag = "delta" if f["delta"] else "background"
known = f" [known: {f['known']}]" if f.get("known") else ""
lines.append(f"[{f['tier']}][{tag}]{known} {f['artifact']}: {f['text']}")
for n in doc["needs_extraction"]:
lines.append(f"[NEEDS-EXTRACTION] {n['file']}: {n['reason']}")
for d in doc["degraded"]:
lines.append(f"[DEGRADED] {d}")
if doc.get("missing_verdicts"):
lines.append(f"[MISSING-VERDICTS] {doc['missing_verdicts']}")
for a in doc.get("stability_notes") or []:
lines.append(f"[NOTE] {a}")
for a in doc.get("aging_proposed") or []:
lines.append(f"[NOTE] {a}")
return "\n".join(lines)
# ── selftest: the fixture conformance run (spec/fixtures) ───────────────
def _subset(actual, expected, path, failures):
"""expected is a subset-shape of actual: dict keys must exist and
match; each expected list element must subset-match SOME actual
element; scalars must be equal."""
if isinstance(expected, dict):
if not isinstance(actual, dict):
failures.append(f"{path}: expected an object, got {type(actual).__name__}")
return
for k, v in expected.items():
if k not in actual:
failures.append(f"{path}.{k}: missing")
continue
_subset(actual[k], v, f"{path}.{k}", failures)
elif isinstance(expected, list):
if not isinstance(actual, list):
failures.append(f"{path}: expected a list, got {type(actual).__name__}")
return
for i, ev in enumerate(expected):
matched = False
for av in actual:
trial = []
_subset(av, ev, path, trial)
if not trial:
matched = True
break
if not matched:
failures.append(f"{path}[{i}]: no element matches {json.dumps(ev)[:80]}")
else:
if actual != expected:
failures.append(f"{path}: {actual!r} != {expected!r}")
def selftest(fixtures_dir):
"""Run the docs-side grammar over each fixture tree and assert its
fixture.json expectations. The fixtures ARE the format spec's
examples — a parser change that shifts behavior fails here first."""
failures, ran = [], 0
for name in sorted(os.listdir(fixtures_dir)):
fdir = os.path.join(fixtures_dir, name)
spec_path = os.path.join(fdir, "fixture.json")
if not os.path.isfile(spec_path):
continue
ran += 1
spec = json.loads(open(spec_path, encoding="utf-8").read())
chk = Check(os.path.abspath(fdir), None, None, None)
chk.delta_files = set()
chk.load_regimes()
adrs = chk.adr_corpus()
risks = chk.parse_risks()
adr_stems = {os.path.basename(a["path"])[:-3] for a in adrs}
chk.check_adr_structure(adrs, {r["id"] for r in risks})
chk.check_risk_structure(risks, adr_stems)
chk.coverage_closure(adrs, risks)
register = chk.register_world()
register["holdings"] = chk.derived_holdings(register)
res = chk.residuals()
chk.risk_table(adrs, res)
div_ledger = []
for pair in spec.get("canon") or []:
local = open(os.path.join(fdir, pair["local"]), encoding="utf-8").read()
carried = open(os.path.join(fdir, pair["carried"]), encoding="utf-8").read()
div_ledger += chk.check_canon_file(pair["file"], local, carried)
dump = {
"register": register,
"risks": [{k: v for k, v in r.items() if k != "text"} for r in risks],
"regimes": [p["name"] for p in chk.regimes],
"divergences": div_ledger,
"findings": chk.findings,
"degraded": chk.degraded,
}
texts = [f["text"] for f in chk.findings]
for s in spec.get("expect_findings", []):
if not any(s in t for t in texts):
failures.append(f"{name}: no finding contains {s!r}")
for s in spec.get("absent_findings", []):
hits = [t for t in texts if s in t]
if hits:
failures.append(f"{name}: unexpected finding ~ {s!r}: {hits[0][:90]}")
if "expect" in spec:
_subset(dump, spec["expect"], name, failures)
if failures:
for f in failures:
print(f"FIXTURE FAIL {f}")
print(f"-- selftest: {ran} fixture(s), {len(failures)} failure(s)")
return 1
print(f"-- selftest: {ran} fixture(s), all conform")
return 0
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--repo", default=".")
ap.add_argument("--base")
ap.add_argument("--facts", default=".dds-check/facts")
ap.add_argument("--verdicts")
ap.add_argument("--prior-verdicts", help="a prior run's verdicts (e.g. .dds-seed.json) for stability/evaluability notes")
ap.add_argument("--json")
ap.add_argument("--write-status", action="store_true")
ap.add_argument("--selftest", metavar="FIXTURES_DIR")
args = ap.parse_args()
if args.selftest:
sys.exit(selftest(args.selftest))
if not args.base:
ap.error("--base is required (or --selftest FIXTURES_DIR)")
chk = Check(
os.path.abspath(args.repo), args.base, args.facts, args.verdicts,
prior_path=args.prior_verdicts,
)
doc = chk.run()
text = render(doc)
print(text)
if args.json:
with open(args.json, "w", encoding="utf-8") as f:
json.dump(doc, f, indent=2)
if args.write_status:
first = "FAIL" if doc["status"] in ("FAIL", "INCOMPLETE") else "PASS"
with open(os.path.join(chk.repo, ".dds-check-status"), "w", encoding="utf-8") as f:
f.write(first + "\n\n" + text + "\n")
sys.exit({"PASS": 0, "FAIL": 1, "INCOMPLETE": 2}[doc["status"]])
if __name__ == "__main__":
main()