#!/usr/bin/env python3
"""scripts/record-audit.py -- Phase-1 deterministic record inventory (grounding audit).
Implements Phase 1 items 1, 2 and 4 of docs/audit/grounding-audit-charter.md as a
SCRIPT, not judgment: doc census, status-claim extraction (plan counts, stage
statuses, version pins, gate assertions), and reference integrity (path refs,
D-/DOCFIX-/BUNDLEFIX-/SEC-/R3-F number refs, commit hashes, file:line refs).
Scope per the operator's 2026-07-18 amendment: docs/ + runbooks/ + README.md +
CLAUDE.md + the .claude/skills/openstack-cloud-ops/ tree + (via --memory-dir) the
auto-memory directory, read as on-disk evidence.
Mutates NOTHING except writing the single report file (--out, default
docs/audit/record-inventory.md under --repo).
Deterministic counting rules (zero judgment, stated so re-runs are comparable):
- Files under docs/audit/ and the memory dir are INVENTORIED but their rows do
NOT count toward the contradiction-pair exit signal -- the audit's own record
of a defect must not keep the defect alive (charter Phase 6 criterion 1).
- A "contradiction group" is a claim key (the global plan-count set; one
version product; one stage key) holding >1 distinct value across counted
surfaces. The report lists every row; adjudication happens in the findings
register, not here.
- A second pair count EXCLUDING changelog-class files is reported alongside
(changelogs are historical by nature; GA-R2 owns their consolidation).
Usage:
python3 scripts/record-audit.py [--repo DIR] [--memory-dir DIR]
[--out FILE] [--stamp STRING] [--no-git]
Exit: 0 = zero counted contradiction groups (PROCEED), 1 = contradictions found
(HOLD), 2 = precondition missing (repo absent/unreadable). ASCII + LF.
"""
import argparse
import os
import re
import subprocess
import sys
from collections import defaultdict
from pathlib import Path
TEXT_EXT = {".md", ".sh", ".py", ".tf", ".yaml", ".yml", ".txt", ".cfg", ".conf",
".json", ".tfvars", ".php", ""}
SKIP_DIRS = {".git", ".terraform", "__pycache__", "node_modules"}
MAX_TEXT = 4 * 1024 * 1024
STATUS_TOKENS = r"(DEPLOY-READY|READY|BLOCKED|CLOSED|COMPLETED|COMPLETE|HELD|STOPPED)"
GATE_WORDS = r"(non-blocking|blocking|unblocked|unexercised|unverified|imported|verified)"
PRODUCTS = [
("tofu", r"(?:OpenTofu|opentofu|tofu)"),
("maas", r"(?:MAAS|maas)"),
("juju", r"(?:Juju|juju)"),
("opnsense", r"(?:OPNsense|opnsense)"),
("netbox", r"(?:NetBox|netbox)"),
("lxd", r"(?:LXD|lxd)"),
]
VER_RE = r"v?(\d+\.\d+(?:\.\d+)?)"
def die(msg, code):
print(f"record-audit: {msg}", file=sys.stderr)
sys.exit(code)
def run_git(repo, args):
try:
out = subprocess.run(["git", "-C", str(repo)] + args, capture_output=True,
text=True, timeout=60)
return out.returncode, out.stdout.strip()
except Exception:
return 1, ""
def is_text_candidate(p: Path):
return p.suffix.lower() in TEXT_EXT and p.is_file() and p.stat().st_size <= MAX_TEXT
def read_lines(p: Path):
try:
return p.read_text(encoding="utf-8", errors="replace").splitlines()
except Exception:
return []
def classify(rel: str):
b = os.path.basename(rel)
if rel.startswith("docs/audit/"):
return "audit"
if b.startswith("changelog-"):
return "changelog"
if b == "design-decisions.md":
return "decision"
if "ledger" in b:
return "ledger"
if "readiness" in b:
return "readiness"
if "finding" in b:
return "finding"
if "patch" in b or rel.endswith(".patch"):
return "patch"
if "workflow" in b:
return "workflow"
if rel.startswith("runbooks/"):
return "runbook"
return "reference"
def trim(line, n=110):
s = line.strip().replace("|", "\\|")
return s[:n] + ("..." if len(s) > n else "")
class Row:
__slots__ = ("kind", "key", "value", "loc", "line", "counted", "census_class")
def __init__(self, kind, key, value, loc, line, counted, census_class):
self.kind, self.key, self.value = kind, key, value
self.loc, self.line = loc, line
self.counted, self.census_class = counted, census_class
def extract_claims(rel, lines, counted, census_class):
rows = []
def add(kind, key, value, i, line):
rows.append(Row(kind, key, value, f"{rel}:{i}", line, counted, census_class))
plan_wordy = re.compile(
r"(\d+)\s*(?:add|create[sd]?)\s*/\s*(\d+(?:-or-\d+)?)\s*"
r"(?:change[sd]?|in-place|update[sd]?)\s*/\s*(\d+)\s*(?:destroy|delete[sd]?)", re.I)
plan_hyphen = re.compile(
r"(\d+)-add/\(?(\d+(?:-or-\d+)?)\)?-change/(\d+)-destroy", re.I)
plan_tofu = re.compile(r"(\d+) to add, (\d+) to change, (\d+) to destroy")
plan_compact = re.compile(r"\b(\d{1,2})/(\d{1,2})/(\d{1,2})\b")
plan_context = re.compile(r"\b(plan|apply|add|destroy|outer|inner)\b", re.I)
stage_tok = re.compile(r"\b" + STATUS_TOKENS + r"\b")
stage_key = re.compile(r"[Ss]tage[- ]?(\d+)")
gate_tok = re.compile(r"\b" + GATE_WORDS + r"\b", re.I)
# Version attribution: the version string must follow the product token with
# no digits and no OTHER product/noise token in between (kills cross-product
# bleed, Ubuntu/Django/kernel versions, URL/section numbers).
prod_words = [(k, re.compile(r"(?<![A-Za-z])" + pat + r"(?![A-Za-z])"))
for k, pat in PRODUCTS]
ver_only = re.compile(VER_RE)
ver_noise = re.compile(
r"ubuntu|django|freebsd|postgres|python|pynetbox|kernel|kea|topic|"
r"section|vault|mysql|ceph|caracal|noble|raft|charm", re.I)
for i, line in enumerate(lines, 1):
seen_span = []
for m in plan_wordy.finditer(line):
add("plan", "plan", "/".join(m.groups()), i, line)
seen_span.append(m.span())
for m in plan_hyphen.finditer(line):
add("plan", "plan", "/".join(m.groups()), i, line)
seen_span.append(m.span())
for m in plan_tofu.finditer(line):
add("plan", "plan", "/".join(m.groups()), i, line)
seen_span.append(m.span())
if plan_context.search(line):
for m in plan_compact.finditer(line):
if any(a <= m.start() < b for a, b in seen_span):
continue
# not part of a longer x/y/z/w chain, date range, or decimal
tail = line[m.end():]
head = line[:m.start()]
if tail[:1] in "/-." or head[-1:] in "/-.":
continue
add("plan", "plan", "/".join(m.groups()), i, line)
for m in stage_tok.finditer(line):
tok = m.group(1)
norm = {"DEPLOY-READY": "READY", "COMPLETED": "COMPLETE"}.get(tok, tok)
km = stage_key.search(line)
key = f"stage-{km.group(1)}" if km else "-"
add("status", key, norm, i, line)
for prod, pre in prod_words:
for pm in pre.finditer(line):
seg = line[pm.end():pm.end() + 46]
vm = ver_only.search(seg)
if not vm:
continue
between = seg[:vm.start()]
if re.search(r"\d", between) or ver_noise.search(between):
continue
if vm.start() > 0 and (seg[vm.start() - 1] == "." or
seg[vm.start() - 1].isalpha()):
continue # '.8.20' octet fragment / 'C2.3' checklist id
if any(k2 != prod and p2.search(between) for k2, p2 in prod_words):
continue
after = seg[vm.end():]
if (after[:1] == "." and after[1:2].isdigit()) or after[:1] == "%":
continue # IP fragment / percentage, not a version
add("version", prod, vm.group(1), i, line)
for m in gate_tok.finditer(line):
add("gate", "gate", m.group(1).lower(), i, line)
return rows
PATH_RE = re.compile(
r"\b((?:docs|runbooks|scripts|tests|opentofu|netbox|policies|clientdocs|"
r"creds-manifests|references)/[A-Za-z0-9._@/-]*[A-Za-z0-9])")
NUM_RE = re.compile(r"\b((?:D|DOCFIX|BUNDLEFIX|SEC)-\d{1,3}|GA-[FR]\d{1,2}|R3-F\d{2})\b")
HASH_RE = re.compile(r"\b[0-9a-f]{7,12}\b")
FILELINE_RE = re.compile(
r"\b((?:docs|runbooks|scripts|opentofu|tests)/[A-Za-z0-9._/-]+\."
r"(?:md|sh|py|tf)):(\d+)\b")
def extract_refs(rel, lines, skill_base):
paths, nums, hashes, filelines = [], [], [], []
for i, line in enumerate(lines, 1):
for m in PATH_RE.finditer(line):
p = m.group(1).rstrip(".,;:")
if any(c in p for c in "*<>{}$") or "YYYY" in p or "NN" in p:
continue
paths.append((p, f"{rel}:{i}"))
for m in NUM_RE.finditer(line):
nums.append((m.group(1), f"{rel}:{i}"))
for m in HASH_RE.finditer(line):
h = m.group(0)
if re.search(r"[a-f]", h) and re.search(r"[0-9]", h):
hashes.append((h, f"{rel}:{i}"))
for m in FILELINE_RE.finditer(line):
filelines.append((m.group(1), int(m.group(2)), f"{rel}:{i}"))
return paths, nums, hashes, filelines
def main():
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--repo", default=None)
ap.add_argument("--memory-dir", default=None)
ap.add_argument("--out", default=None)
ap.add_argument("--stamp", default="")
ap.add_argument("--no-git", action="store_true")
a = ap.parse_args()
here = Path(__file__).resolve().parent
repo = Path(a.repo).resolve() if a.repo else here.parent
if not repo.is_dir() or not (repo / "docs").is_dir():
die(f"repo dir absent or has no docs/: {repo}", 2)
out = Path(a.out) if a.out else repo / "docs" / "audit" / "record-inventory.md"
memdir = Path(a.memory_dir).resolve() if a.memory_dir else None
if memdir and not memdir.is_dir():
die(f"--memory-dir not a directory: {memdir}", 2)
use_git = not a.no_git
head = run_git(repo, ["rev-parse", "--short", "HEAD"])[1] if use_git else "(no-git)"
skill_base = repo / ".claude" / "skills" / "openstack-cloud-ops"
# ---- collect scan targets: (label, path, counted, surface) ---------------
targets = []
for base, surface in (("docs", "repo"), ("runbooks", "repo")):
d = repo / base
if d.is_dir():
for p in sorted(d.rglob("*")):
if p.is_file() and not any(s in p.parts for s in SKIP_DIRS):
rel = p.relative_to(repo).as_posix()
targets.append((rel, p, not rel.startswith("docs/audit/"), surface))
for extra in ("README.md", "CLAUDE.md"):
p = repo / extra
if p.is_file():
targets.append((extra, p, True, "repo"))
if skill_base.is_dir():
for p in sorted(skill_base.rglob("*.md")):
rel = p.relative_to(repo).as_posix()
targets.append((rel, p, True, "skill"))
if memdir:
for p in sorted(memdir.iterdir()):
if p.is_file():
targets.append((f"memory/{p.name}", p, False, "memory"))
out_rel = out.resolve()
targets = [t for t in targets if t[1].resolve() != out_rel]
# ---- census + claims + refs ---------------------------------------------
census, claims = [], []
all_paths, all_nums, all_hashes, all_filelines = [], [], [], []
contents = {}
for rel, p, counted, surface in targets:
lines = read_lines(p)
contents[rel] = lines
cls = classify(rel) if surface == "repo" else surface
if use_git and surface == "repo":
_, cdate = run_git(repo, ["log", "-1", "--format=%as", "--", rel])
cdate = cdate or "uncommitted"
else:
cdate = "n/a"
census.append((rel, cls, p.stat().st_size, len(lines), cdate))
claims.extend(extract_claims(rel, lines, counted, cls))
pa, nu, ha, fl = extract_refs(rel, lines, skill_base)
all_paths.extend(pa)
all_nums.extend(nu)
all_hashes.extend(ha)
all_filelines.extend(fl)
# ---- inbound-reference corpus (whole repo text) -------------------------
corpus = {}
for p in sorted(repo.rglob("*")):
if any(s in p.parts for s in SKIP_DIRS):
continue
if is_text_candidate(p):
rel = p.relative_to(repo).as_posix()
corpus[rel] = "\n".join(contents.get(rel) or read_lines(p))
inbound = defaultdict(list)
doc_basenames = {os.path.basename(rel): rel for rel, _, _, s in targets if s == "repo"}
for b, target_rel in sorted(doc_basenames.items()):
for rel, text in corpus.items():
if rel == target_rel:
continue
if b in text:
inbound[target_rel].append(rel)
# ---- reference integrity checks -----------------------------------------
dd_text = corpus.get("docs/design-decisions.md", "")
sec_text = corpus.get("docs/security-ledger.md", "")
broken_paths = sorted({(p, loc) for p, loc in all_paths
if not (repo / p).exists() and not (skill_base / p).exists()})
orphan_nums = []
for n, loc in sorted(set(all_nums)):
canon = n if "-" in n else n
if n.startswith("D-") and len(n) >= 4:
if canon not in dd_text:
orphan_nums.append((n, loc, "not in design-decisions.md"))
elif n.startswith("SEC-"):
if canon not in sec_text:
orphan_nums.append((n, loc, "not in security-ledger.md"))
else:
hits = sum(1 for t in corpus.values() if canon in t)
if hits <= 1:
orphan_nums.append((n, loc, "referenced nowhere else"))
unknown_hashes = []
if use_git:
for h, loc in sorted(set(all_hashes)):
rc, _ = run_git(repo, ["cat-file", "-e", f"{h}^{{commit}}"])
if rc != 0:
unknown_hashes.append((h, loc))
bad_filelines = []
for f, ln, loc in sorted(set(all_filelines)):
tgt = repo / f
if not tgt.exists():
bad_filelines.append((f, ln, loc, "target absent"))
elif ln > len(read_lines(tgt)):
bad_filelines.append((f, ln, loc, f"only {len(read_lines(tgt))} lines"))
# ---- contradiction groups -----------------------------------------------
def groups(kind):
g = defaultdict(list)
for r in claims:
if r.kind == kind:
g[r.key].append(r)
return g
contra, contra_nochg = [], []
for kind in ("plan", "version", "status"):
for key, rows in sorted(groups(kind).items()):
if kind == "status" and key == "-":
continue
counted_rows = [r for r in rows if r.counted]
vals = sorted({r.value for r in counted_rows})
if len(vals) > 1:
contra.append((kind, key, vals, counted_rows))
v2 = sorted({r.value for r in counted_rows if r.census_class != "changelog"})
if len(v2) > 1:
contra_nochg.append((kind, key, v2))
# ---- render -------------------------------------------------------------
L = []
w = L.append
w("# Record inventory -- grounding audit Phase 1 (deterministic)")
w("")
w(f"Generated by `scripts/record-audit.py` at repo HEAD `{head}`"
+ (f" ({a.stamp})" if a.stamp else "") + ".")
w("Read-only extraction; zero judgment. Counting rules are in the script header:")
w("docs/audit/* and memory-surface rows are inventoried but NOT counted toward")
w("the contradiction signal; a changelog-excluded pair count is shown alongside.")
w("")
w("## Summary")
w("")
w(f"- files scanned: {len(targets)}")
w(f"- status-claim rows: {len(claims)} "
f"(plan {sum(1 for r in claims if r.kind == 'plan')}, "
f"status {sum(1 for r in claims if r.kind == 'status')}, "
f"version {sum(1 for r in claims if r.kind == 'version')}, "
f"gate {sum(1 for r in claims if r.kind == 'gate')})")
w(f"- CONTRADICTION GROUPS (counted surfaces): {len(contra)}")
w(f"- contradiction groups excluding changelog-class files: {len(contra_nochg)}")
w(f"- broken path refs: {len(broken_paths)}; orphan number refs: {len(orphan_nums)}; "
f"unknown commit hashes: {len(unknown_hashes)}; bad file:line refs: {len(bad_filelines)}")
w("")
w("## 1. Doc census")
w("")
w("| file | class | bytes | lines | last commit | inbound refs |")
w("|---|---|---|---|---|---|")
for rel, cls, size, nlines, cdate in census:
n_in = len(inbound.get(rel, []))
w(f"| {rel} | {cls} | {size} | {nlines} | {cdate} | {n_in} |")
w("")
unref = [c for c in census if c[1] == "changelog" and not inbound.get(c[0])]
w(f"Unreferenced changelogs (consolidation candidates by default): {len(unref)}")
for rel, *_ in unref:
w(f"- {rel}")
w("")
w("## 2. Contradiction groups (mechanical)")
w("")
if not contra:
w("NONE.")
for kind, key, vals, rows in contra:
w(f"### CONTRADICTION: {kind} `{key}` -> {len(vals)} distinct values: "
+ ", ".join(f"`{v}`" for v in vals))
w("")
w("| value | where | line |")
w("|---|---|---|")
for r in sorted(rows, key=lambda r: (r.value, r.loc)):
w(f"| {r.value} | {r.loc} | {trim(r.line)} |")
w("")
w("## 3. Status-claim rows (full table)")
w("")
w("| kind | key | value | where | line |")
w("|---|---|---|---|---|")
for r in sorted((r for r in claims if r.kind != "gate"),
key=lambda r: (r.kind, r.key, r.loc)):
flag = "" if r.counted else " (not counted)"
w(f"| {r.kind}{flag} | {r.key} | {r.value} | {r.loc} | {trim(r.line, 80)} |")
w("")
w("## 4. Gate-assertion rows")
w("")
w("| word | where | line |")
w("|---|---|---|")
for r in sorted((r for r in claims if r.kind == "gate"), key=lambda r: r.loc):
w(f"| {r.value} | {r.loc} | {trim(r.line, 80)} |")
w("")
w("## 5. Reference integrity")
w("")
w("### Broken path references")
w("| ref | where |")
w("|---|---|")
for p, loc in broken_paths:
w(f"| {p} | {loc} |")
w("")
w("### Orphan / unregistered number references")
w("| ref | where | why |")
w("|---|---|---|")
for n, loc, why in orphan_nums:
w(f"| {n} | {loc} | {why} |")
w("")
w("### Unknown commit hashes")
w("| hash | where |")
w("|---|---|")
for h, loc in unknown_hashes:
w(f"| {h} | {loc} |")
w("")
w("### file:line references out of range or absent")
w("| ref | line | where | why |")
w("|---|---|---|---|")
for f, ln, loc, why in bad_filelines:
w(f"| {f} | {ln} | {loc} | {why} |")
w("")
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text("\n".join(L) + "\n", encoding="ascii", errors="replace")
print(f"record-audit: {len(targets)} files, {len(claims)} claim rows, "
f"{len(contra)} contradiction groups -> {out}")
sys.exit(1 if contra else 0)
if __name__ == "__main__":
main()
