Files
home-network-k3s-impl/conformance/lib/status.py
T
jakman85andClaude Opus 5 e9a4b669d4 feat(conformance): committed run ledger — the only thing that can move PROVEN off zero
status.py has always reported PROVEN: 0 for all 51 requirements, and named the fix in its
own header: "a committed, machine-readable run ledger". There was none — _artifacts/ is
gitignored, the root .gitignore kills *.junit.xml, and a stray JUnit was surfaced only as
a quarantined mtime advisory. Built BEFORE the first real L3 execution so that run is
captured rather than reconstructed from memory.

THE TRAP THIS EXISTS TO AVOID. These tests exit 0 and Chainsaw records `passed` on a
SELF-SKIP exactly as on a real run, and Chainsaw's JUnit carries no stdout — only
properties. So committing the JUnit as proof would let DR_RESTORE_SKIPPED_OK be recorded
as proof of R-STATE-03: authoritative-looking and wrong, which is worse than no ledger.
The sentinel token is the ONLY thing distinguishing proven from skipped, and it lives
solely in the run log. The ledger therefore records the LOG alongside the JUnit and
derives proof from the sentinel, never from the JUnit result.

Sentinel tokens are DERIVED, not hand-listed: record_run.py parses every
contains($stdout,'TOKEN') out of each test's own chainsaw-test.yaml, so the list cannot
drift from the tests. 23 of 35 tests declare no sentinel at all.

decide() is the single place `proven` is emitted, ordered WORST-FIRST:
  JUnit failure/error -> failed
  JUnit <skipped/>    -> junit-skipped
  ANY declared skip token in the log -> skipped   (beats a green JUnit AND beats a pass
                                                   token present in the same log)
  no sentinel declared -> proven ONLY for L0/L1 (whose JUnit is truthful on its own, since
                          run_l1_offline.py maps exit 3 to <skipped/>); written as an
                          explicit exempt-set so an unknown/blank layer fails CLOSED
  EVERY declared pass token present -> proven
  otherwise -> unknown-sentinel-absent
Log attribution is a state machine over Chainsaw's per-test headers; a log with no
recognisable header falls back to whole-log scanning restricted to GLOBALLY-UNIQUE tokens,
so DNS_OK (owned by two tests) can never prove either. record_run.py exits 3 if the JUnit
or the log is missing — nothing recorded, never green.

STALENESS: a proven row is honoured only while `git merge-base --is-ancestor` holds
between the requirement's last implementing/test change and the row's git_sha; otherwise
it degrades to stale-proof. A -dirty sha, an unknown rev, or an uncommitted test path all
fail ancestry the same way. A requirement reads proven only when EVERY verifying test
does (worst-of), and the latest row per test wins — not best-ever.

REFUSAL POSTURE PRESERVED: with an empty ledger the report is unchanged in intent — every
requirement not-derivable, headline "PROVEN: 0 — NOT DERIVABLE". This replaces "not
derivable" with real evidence only where evidence exists, never with optimism.

--record is explicit, never implicit (a run that silently records itself is how a bad run
becomes permanent false proof), is rejected for --status/--coverage-only/--lint, and
re-execs under tee returning ${PIPESTATUS[0]} so recording can never soften a gate. It
rm -f's a stale _artifacts/report.junit.xml first — a hazard hit live during development,
where a stale JUnit got paired with a fresh log.

Selftests: record_run.py 23 checks (incl. end-to-end Chainsaw-passed + NODE_LOSS_SKIPPED_OK
-> proof "skipped"), status.py +16 against a throwaway git repo (self-skip -> skipped;
proven-at-older-sha -> stale-proof; dirty/unresolvable -> stale-proof; worst-of;
latest-row-wins; malformed lines dropped). No regressions: --coverage-only, --status,
--lint, --strict-coverage and the merge-gate shape all rc=0 and unchanged; nothing in ci/
references status.py.

Root .gitignore needed no change — conformance/runs/ is not itself excluded, so the deeper
negations override the root's global *.junit.xml. Verified with git check-ignore.

Serves: R-TEST-01, R-TEST-02, R-TEST-03, R-TEST-04.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
2026-09-01 18:49:01 -05:00

1350 lines
68 KiB
Python
Executable File

#!/usr/bin/env python3
"""Requirement STATUS report — derived from repository state, not from a task board.
WHY THIS EXISTS
---------------
`lib/coverage.py` answers exactly one question: *does a requirement-tagged test exist
for R-x?* (R-TEST-04). That is necessary but not sufficient: a requirement can have a
tagged test that is a pure TODO scaffold, and can have implementing manifests that are
authored in git but never reconciled by Flux. Both read as "covered"/"delivered" while
nothing is actually true of the cluster.
This generator derives, from the REPOSITORY ALONE, three orthogonal axes per requirement:
1. IMPLEMENTATION — is there implementing config in the repo, and is it reachable
from a Flux Kustomization entrypoint (i.e. actually reconciled)?
2. VERIFICATION — is there a requirement-tagged test, at which layer, and is it an
EXECUTABLE test or a TODO scaffold?
3. PROOF — has that test actually passed? Derived ONLY from the committed
run ledger, and ONLY as far as that ledger's evidence reaches.
PROOF — WHAT CHANGED, AND WHAT DID NOT
--------------------------------------
This report used to say `not-derivable` for every requirement, because nothing tracked in
git recorded a test result and the only prose record of a green run lived in
DECISIONS-LOG.md — hand-maintained, unstructured, exactly the source the operator ruled
out. The fix this docstring asked for now exists: `conformance/runs/ledger.jsonl`, an
append-only ledger written by `lib/record_run.py` from a run's JUnit **and its run log**
(see conformance/.gitignore for the negations that keep both in git).
THE REFUSAL POSTURE IS UNCHANGED. Evidence replaced "not derivable" only where evidence
exists; nowhere is optimism substituted for it:
* A ledger row reaches `proven` only if `record_run.py` saw that test's own PASS
sentinel token in the run log. The L2/L3 tests SELF-SKIP GREEN — Chainsaw records
`passed` on `DR_RESTORE_SKIPPED_OK` exactly as on a real run, and its JUnit carries no
stdout — so a skip token maps to `skipped` and can never become proof.
* A proof is honoured only while it still describes the current code: if any of the
requirement's implementing manifests or verifying tests changed in a commit that is
NOT an ancestor of the run's `git_sha`, the row degrades to `stale-proof`, which is
not green. A run recorded from a dirty tree (`<sha>-dirty`) is never honoured.
* A requirement with no ledger row, or whose verifying tests did not ALL report proven,
stays `not-derivable` / `unknown` — never rounded up.
* The untracked `_artifacts/report.junit.xml` is STILL only a quarantined advisory with
its mtime. It is not the ledger, it is not evidence, and it feeds no status.
With an empty or absent ledger this report behaves exactly as it did before: every
requirement `not-derivable`, headline PROVEN 0, "the most this report can say is wired and
testable, never proven".
READS (all read-only)
requirements/registry.yaml the authoritative R-* mirror (via lib/coverage.py)
tests/**/chainsaw-test.yaml requirement tags + layer + executable annotation
runs/ledger.jsonl COMMITTED proof-of-pass rows (via lib/record_run.py)
clusters/<env>/ Flux entrypoint -> the reconciliation graph
`git ls-files` TRACKED files only; untracked scratch is not repo state
../home-network-k3s/docs/00-vision-and-scope.md OPTIONAL registry<->spec parity check
WRITES (under --out, default _artifacts/status/, git-ignored like coverage/)
status.json machine-readable, one record per requirement + per-spec rollup
status.md human-readable report
EXIT CODES (ci/lib.sh convention: 0 pass / 1 fail / 3 skipped-not-green)
0 report generated.
1 a derivation failed, or --strict found a defect this tool owns (registry/spec
drift, a requirement-tag hygiene problem, a safety-critical requirement whose only
L3 test is a scaffold).
3 a requested check could not run (e.g. --spec-root given but absent). A SKIP IS
NEVER COUNTED AS GREEN.
Pure standard library (no PyYAML), matching the harness's no-pip-dependency posture
(lib/coverage.py); the file shapes parsed here are line-constrained and documented at
each parser. `catalog/render.py` owns the one PyYAML dependency; the harness does not.
"""
from __future__ import annotations
import argparse
import json
import os
import re
import subprocess
import sys
import time
from pathlib import Path
_LIB = Path(__file__).resolve().parent
if str(_LIB) not in sys.path:
sys.path.insert(0, str(_LIB))
# Importing a sibling module would otherwise litter conformance/lib/__pycache__/ into the
# working tree; the lint stage scans untracked files, so keep the tree clean.
sys.dont_write_bytecode = True
import coverage as cov # noqa: E402 — reuse the registry/test parsers, don't fork them
CONF_ROOT = _LIB.parent
REPO_ROOT = CONF_ROOT.parent
DEFAULT_SPEC_ROOT = REPO_ROOT.parent / "home-network-k3s"
RID_RE = cov.RID_RE
LABEL_PREFIX = cov.LABEL_PREFIX
# Flux Kustomization identity. A `kind: Kustomization` doc is only a *Flux* entrypoint if
# it also carries the kustomize.toolkit apiVersion — plain kustomize overlays share the kind.
FLUX_API = "kustomize.toolkit.fluxcd.io"
# Files excluded from the IMPLEMENTATION scan, with the same reasoning ci/touched-
# requirements.sh applies to its content scan:
# *.md — an R-ID in prose is a REFERENCE, not an implementing artifact.
# (DECISIONS-LOG.md alone names nearly every requirement.)
# conformance/tests — those are the TESTS; they are the VERIFICATION axis, counted there.
# ci/tests — gate unit-test fixtures embed example R-IDs as DATA.
# registry.yaml — the requirement mirror itself, not an implementation of anything.
IMPL_EXCLUDE_PREFIXES = ("conformance/tests/", "ci/tests/")
IMPL_EXCLUDE_PATHS = ("conformance/requirements/registry.yaml",)
# ============================================================================ git
def tracked_files(root: Path) -> list[str]:
"""Repo state == TRACKED files. Untracked scratch must never count as delivery."""
out = subprocess.run(["git", "-C", str(root), "ls-files"],
capture_output=True, text=True, check=True).stdout
return [l for l in out.splitlines() if l]
# ================================================================ kustomize walk
# Parser contract: a kustomization.yaml in this repo uses ONLY a block `resources:` list
# of ` - <path>` entries (verified: `resources` is the sole reference key across all 30
# kustomization.yaml files). Entries are plain scalars, optionally followed by a `#`
# comment. A COMMENTED-OUT entry is not a list item and is therefore correctly EXCLUDED —
# that is the whole point: `platform/kustomization.yaml` comments out velero and
# image-automation, so those directories are authored but not reconciled through it.
_RES_KEY_RE = re.compile(r"^resources:\s*$")
_TOP_KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*:")
# NOTE: list items may be indented (` - x`) or flush-left (`- x`, which is what
# `catalog/render.py --emit-waves` generates for apps/wave-*/kustomization.yaml).
_RES_ITEM_RE = re.compile(r"^\s*-\s+(?P<v>[^#\s]+)\s*(?:#.*)?$")
def kustomize_resources(kfile: Path) -> list[str]:
"""Return the literal `resources:` entries of a kustomization.yaml."""
items: list[str] = []
in_res = False
for raw in kfile.read_text().splitlines():
if _RES_KEY_RE.match(raw):
in_res = True
continue
if in_res:
m = _RES_ITEM_RE.match(raw)
if m:
items.append(m.group("v"))
continue
if raw.strip() == "" or raw.lstrip().startswith("#"):
continue
if _TOP_KEY_RE.match(raw):
in_res = False
return items
# Flux Kustomization `spec.path` extraction. Constraint: `path:` appears at spec-level
# indent inside a doc that carries the Flux apiVersion. Docs are split on a lone `---`.
# Inline `patch: |-` blocks (clusters/dev) carry Flux Kustomizations WITHOUT a path, so
# they contribute nothing — correct, they are patches, not entrypoints.
_DOC_SPLIT_RE = re.compile(r"^---\s*$", re.M)
_PATH_RE = re.compile(r"^\s{2,6}path:\s*(\S+)\s*$", re.M)
_NAME_RE = re.compile(r"^\s{2,6}name:\s*(\S+)\s*$", re.M)
def flux_paths(yfile: Path) -> list[tuple[str, str]]:
"""[(flux-kustomization-name, spec.path)] declared by this file."""
found: list[tuple[str, str]] = []
try:
text = yfile.read_text()
except (OSError, UnicodeDecodeError):
return found
if FLUX_API not in text:
return found
for doc in _DOC_SPLIT_RE.split(text):
if FLUX_API not in doc or not re.search(r"^kind:\s*Kustomization\s*$", doc, re.M):
continue
pm = _PATH_RE.search(doc)
if not pm:
continue
nm = _NAME_RE.search(doc)
found.append((nm.group(1) if nm else "?", pm.group(1)))
return found
def reconciled_files(repo: Path, entrypoint: str) -> tuple[set[str], list[dict]]:
"""Walk the Flux reconciliation graph from `clusters/<env>` and return
(set of repo-relative files Flux actually applies, list of Flux Kustomization nodes)
Algorithm: kustomize-walk a directory's `resources:`; a directory entry recurses, a
file entry is applied. Every applied file is then scanned for Flux Kustomization
objects; each one's `spec.path` is a NEW kustomize-walk root (that is how Flux
expands the graph). Cycles are guarded (clusters/dev/sync.yaml points at its own dir).
This is what makes the difference between AUTHORED and RECONCILED derivable: a
manifest that no walk reaches is in git but is not applied to any cluster.
"""
applied: set[str] = set()
nodes: list[dict] = []
seen_dirs: set[str] = set()
pending: list[tuple[str, str]] = [("<entrypoint>", entrypoint)]
def walk_dir(rel: str) -> None:
if rel in seen_dirs:
return
seen_dirs.add(rel)
kfile = repo / rel / "kustomization.yaml"
if not kfile.is_file():
return
applied.add(os.path.relpath(kfile, repo))
for item in kustomize_resources(kfile):
target = (repo / rel / item).resolve()
try:
trel = os.path.relpath(target, repo)
except ValueError:
continue
if trel.startswith(".."):
continue # outside the repo: not repo state
if target.is_dir():
walk_dir(trel)
elif target.is_file():
applied.add(trel)
for name, p in flux_paths(target):
pending.append((name, p))
while pending:
owner, p = pending.pop()
rel = os.path.normpath(p.lstrip("./")) if p not in (".", "./") else "."
node = {"kustomization": owner, "path": p, "resolved": rel,
"exists": (repo / rel).is_dir()}
nodes.append(node)
if node["exists"]:
walk_dir(rel)
return applied, nodes
# ====================================================== implementation evidence
_COMMENT_LINE_RE = re.compile(r"^\s*#")
def scan_implementation(repo: Path, files: list[str]) -> dict[str, list[dict]]:
"""Map R-ID -> the tracked, non-doc files that name it.
Two strengths are recorded per file, mirroring ci/touched-requirements.sh rule 3:
* a hit on a REAL line -> the artifact carries the ID (label/annotation/field);
* a hit only in a `#` comment -> declared INTENT (a header saying what it serves).
Comment-only evidence is kept (it is the dominant convention in this repo's
manifests) but reported as such, so a reader can tell an implementation from a claim.
"""
out: dict[str, list[dict]] = {}
for rel in files:
if rel.endswith(".md") or rel in IMPL_EXCLUDE_PATHS:
continue
if rel.startswith(IMPL_EXCLUDE_PREFIXES):
continue
path = repo / rel
try:
text = path.read_text()
except (OSError, UnicodeDecodeError):
continue
if "R-" not in text:
continue
# A file is RECONCILABLE only if it is a Kubernetes manifest: a *.yaml with a
# top-level `kind:`. Shell scripts / values files / READMEs under a GitOps root
# are never applied by Flux, so counting them as "authored but not wired" would
# manufacture a false gap.
manifest = (rel.endswith((".yaml", ".yml"))
and re.search(r"^kind:\s*\S", text, re.M) is not None)
hits: dict[str, bool] = {} # rid -> seen on a non-comment line
for line in text.splitlines():
ids = RID_RE.findall(line)
if not ids:
continue
real = not _COMMENT_LINE_RE.match(line)
for rid in ids:
hits[rid] = hits.get(rid, False) or real
for rid, real in hits.items():
out.setdefault(rid, []).append({"path": rel, "comment_only": not real,
"manifest": manifest})
for rid in out:
out[rid].sort(key=lambda h: h["path"])
return out
# ================================================================ verification
_EXECUTABLE_RE = re.compile(
r"^\s*" + re.escape(LABEL_PREFIX) + r"/executable:\s*[\"']?true", re.M | re.I)
_BLOCKED_RE = re.compile(
r"^\s*" + re.escape(LABEL_PREFIX) + r"/blocked-by:\s*(.+)$", re.M)
def classify_test(t: dict) -> dict:
"""Decide EXECUTABLE vs SCAFFOLD for one discovered test, conservatively.
A tagged test is only worth counting if it can actually assert something. Rules, in
order, chosen so ambiguity NEVER upgrades a test:
scaffold the body contains TODO markers (a placeholder), *even if* it
is annotated executable — an annotation must never outrank the
body. That mismatch is also reported as a hygiene defect.
executable annotated `conformance.home.k3s/executable: "true"` and no TODOs.
executable-untagged no TODOs and a real runnable body, but the annotation is
missing (true of the L1 render tests, which execute a
render-check.sh + golden diff, of the L0 static checks, and
of the META self-check).
Counted as executable, and flagged as an annotation gap.
scaffold anything else.
"""
p = Path(t["path"])
text = p.read_text() if p.is_file() else ""
# The offline assertion script a test dir may carry: `render-check.sh` (L1 rendering)
# or `check.sh` (L0 static). BOTH are read, so a placeholder body in either one still
# downgrades the test to a scaffold — an annotation must never outrank the body,
# whichever file the body lives in.
extra = ""
for script in ("render-check.sh", "check.sh"):
sp = p.parent / script
if sp.is_file():
extra += sp.read_text()
declared = bool(_EXECUTABLE_RE.search(text))
has_todo = "TODO" in text or "TODO" in extra
bm = _BLOCKED_RE.search(text)
has_body = bool(extra) or bool(re.search(r"^\s+- (script|assert|apply):", text, re.M))
if has_todo:
state = "scaffold"
elif declared:
state = "executable"
elif has_body:
state = "executable-untagged"
else:
state = "scaffold"
try:
rel = os.path.relpath(p, REPO_ROOT)
except ValueError:
rel = t["path"]
return {
"name": t["name"], "path": rel, "layer": t["layer"],
"state": state,
"declared_executable": declared,
"annotation_conflict": bool(declared and has_todo),
"blocked_by": bm.group(1).strip().strip('"') if bm else "",
}
# ===================================================================== spec parity
def spec_requirement_ids(spec_root: Path) -> set[str] | None:
"""R-* IDs defined by the READ-ONLY spec (docs/00-vision-and-scope.md).
Read-only by construction (R-TEST-06): this tool never writes under the spec tree.
Returns None when the spec tree is not available (a sibling checkout is optional).
"""
doc = spec_root / "docs" / "00-vision-and-scope.md"
if not doc.is_file():
return None
return set(RID_RE.findall(doc.read_text()))
# ========================================================================== ledger
# PROOF axis. Everything here reads `conformance/runs/ledger.jsonl` (written ONLY by an
# explicit `lib/record_run.py` / `run.sh --record` invocation) and downgrades relentlessly.
# Nothing in this section may ever turn absent, ambiguous or aged evidence into green.
DEFAULT_LEDGER = CONF_ROOT / "runs" / "ledger.jsonl"
# Severity ranking, WORST FIRST. A requirement's proof is the MINIMUM over its verifying
# tests, so one failed/skipped/missing test drags the whole requirement down and a single
# proven test can never carry the others. `stale-proof` ranks below `proven` on purpose:
# it is a real past pass that no longer describes the current code, and it is NOT green.
PROOF_ORDER = {
"failed": 0, # a ledger row recorded a JUnit failure/error
"unknown": 1, # ran, but the log could not evidence pass vs self-skip
"not-derivable": 2, # no ledger row for this test at all (the pre-ledger default)
"junit-skipped": 3, # JUnit <skipped/> — did not execute
"skipped": 4, # SELF-SKIPPED green: exited 0 without exercising the behaviour
"stale-proof": 5, # proven at a sha the implementation/test has since moved past
"proven": 6, # the ONLY green state
}
# record_run.py's two "we cannot tell" states collapse to one reported state here; the
# per-test detail in status.json keeps the distinction for whoever has to fix it.
_PROOF_ALIASES = {
"unknown-no-sentinel-declared": "unknown",
"unknown-sentinel-absent": "unknown",
}
def read_ledger(path: Path) -> list[dict]:
"""Parse the append-only JSONL ledger. A malformed line is SKIPPED, never guessed at.
Absent or empty file -> [] , which reproduces this report's pre-ledger behaviour
exactly (every requirement `not-derivable`, headline PROVEN 0).
"""
if not path.is_file():
return []
rows: list[dict] = []
for line in path.read_text(errors="replace").splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
try:
obj = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(obj, dict) and obj.get("test"):
rows.append(obj)
return rows
def latest_by_test(rows: list[dict]) -> dict[str, dict]:
"""Most recent ledger row per test name (ts_utc, then file order as the tiebreak).
Latest wins rather than best-ever: a test that has since gone back to self-skipping
must stop reading as proven.
"""
latest: dict[str, dict] = {}
for i, r in enumerate(rows):
key = r["test"]
prev = latest.get(key)
if prev is None or (r.get("ts_utc", ""), i) >= (prev.get("ts_utc", ""),
prev["_i"]):
latest[key] = {**r, "_i": i}
return latest
class _GitOracle:
"""Small cached wrapper for the two git questions the staleness rule asks."""
def __init__(self, repo: Path):
self.repo = repo
self._last: dict[tuple[str, ...], str | None] = {}
self._anc: dict[tuple[str, str], bool] = {}
def _git(self, *args: str) -> subprocess.CompletedProcess:
return subprocess.run(["git", "-C", str(self.repo), *args],
capture_output=True, text=True)
def last_change(self, paths: tuple[str, ...]) -> str | None:
"""Commit that last touched any of `paths`; None if git knows none of them."""
if not paths:
return None
if paths in self._last:
return self._last[paths]
p = self._git("log", "-1", "--format=%H", "--", *paths)
sha = p.stdout.strip() if p.returncode == 0 else ""
self._last[paths] = sha or None
return self._last[paths]
def is_ancestor(self, older: str, newer: str) -> bool:
"""True iff `older` is an ancestor of (or equal to) `newer`.
False whenever either rev is unresolvable — an unknown or `-dirty` sha (a run
recorded from a modified working tree) can never satisfy the freshness test, which
is the intended fail-closed direction.
"""
key = (older, newer)
if key in self._anc:
return self._anc[key]
ok = self._git("merge-base", "--is-ancestor", older, newer).returncode == 0
self._anc[key] = ok
return ok
def derive_proof(git: _GitOracle, rid: str, verifiers: list[dict],
impl_files: list[dict], latest: dict[str, dict]) -> dict:
"""PROOF for one requirement, from the ledger, with the staleness rule applied.
STALENESS RULE — a proof is a claim about a specific tree, not a permanent badge:
relevant paths = every implementing artifact this report found for the requirement
+ every verifying test's directory
last_change = `git log -1 -- <relevant paths>`
A row's `proven` is honoured only if `last_change` is an ANCESTOR of the row's
`git_sha`, i.e. the run happened at or after the newest change to the code it
claims to prove. Otherwise it degrades to `stale-proof` (not green). A `-dirty`
sha, a sha this clone does not have (rebased/dropped), or a path git has never
seen (an uncommitted test) all fail the ancestry test and degrade the same way.
The requirement's state is the MINIMUM over its verifying tests: every verifying test
must be proven and fresh before the requirement reads `proven`. A test with no ledger
row contributes `not-derivable`, so a partially-recorded run cannot manufacture green.
"""
paths = sorted({f["path"] for f in impl_files}
| {os.path.dirname(v["path"]) for v in verifiers if v.get("path")})
last_change = git.last_change(tuple(paths))
per_test: list[dict] = []
for v in verifiers:
row = latest.get(v["name"])
if row is None:
per_test.append({"test": v["name"], "state": "not-derivable",
"reason": "no ledger row — this test has never been "
"recorded by run.sh --record / record_run.py"})
continue
base = _PROOF_ALIASES.get(row.get("proof", ""), row.get("proof", "unknown"))
if base not in PROOF_ORDER:
base = "unknown"
state, reason = base, row.get("note", "")
if base == "proven":
sha = row.get("git_sha", "")
if last_change and not git.is_ancestor(last_change, sha):
state = "stale-proof"
reason = (f"proven at {sha[:12]}, but the implementing manifests / test "
f"changed in {last_change[:12]}, which that run predates")
per_test.append({"test": v["name"], "state": state, "reason": reason,
"run_id": row.get("run_id"), "ts_utc": row.get("ts_utc"),
"git_sha": row.get("git_sha"), "env": row.get("env"),
"sentinel": row.get("sentinel"),
"junit_result": row.get("junit_result")})
if not per_test:
state = "not-derivable"
reason = "no requirement-tagged test exists to be proven"
else:
worst = min(per_test, key=lambda d: PROOF_ORDER[d["state"]])
state, reason = worst["state"], worst["reason"]
if state == "not-derivable" and not reason:
reason = "no committed ledger row for this requirement's verifying test(s)"
return {"state": state, "reason": reason, "last_change_sha": last_change,
"evidence_paths": paths, "per_test": per_test}
# ========================================================================== build
IMPL_ORDER = {"none": 0, "off-cluster": 1, "unwired": 2, "reconciled-partial": 3,
"reconciled": 4}
VERIFY_ORDER = {"none": 0, "scaffold-only": 1, "executable": 2}
LABELS = {
("reconciled-partial", "executable"): "PARTLY-WIRED+TESTABLE",
("reconciled-partial", "scaffold-only"): "PARTLY-WIRED, TEST-IS-SCAFFOLD",
("reconciled-partial", "none"): "PARTLY-WIRED, UNTESTED",
("reconciled", "executable"): "WIRED+TESTABLE",
("reconciled", "scaffold-only"): "WIRED, TEST-IS-SCAFFOLD",
("reconciled", "none"): "WIRED, UNTESTED",
("unwired", "executable"): "AUTHORED-NOT-WIRED",
("unwired", "scaffold-only"): "AUTHORED-NOT-WIRED",
("unwired", "none"): "AUTHORED-NOT-WIRED",
("off-cluster", "executable"): "TOOLING+TESTABLE",
("off-cluster", "scaffold-only"): "TOOLING, TEST-IS-SCAFFOLD",
("off-cluster", "none"): "TOOLING, UNTESTED",
("none", "executable"): "TEST-ONLY",
("none", "scaffold-only"): "SPEC-ONLY (scaffold test)",
("none", "none"): "SPEC-ONLY",
}
def build(repo: Path, registry: list[dict], tests: list[dict], impl: dict[str, list[dict]],
applied: set[str], gitops_roots: set[str], spec_ids: set[str] | None,
env: str, ledger_rows: list[dict] | None = None) -> dict:
# PROOF axis inputs. `ledger_rows=None` / `[]` reproduces the pre-ledger behaviour:
# `latest` is empty, so every requirement derives `not-derivable` exactly as before.
ledger_rows = ledger_rows or []
latest = latest_by_test(ledger_rows)
git = _GitOracle(repo)
classified = [classify_test(t) for t in tests]
by_req_tests: dict[str, list[dict]] = {r["id"]: [] for r in registry}
for t, c in zip(tests, classified):
for rid in t["labels"]:
if rid in by_req_tests:
by_req_tests[rid].append(c)
records: dict[str, dict] = {}
for r in registry:
rid = r["id"]
files = impl.get(rid, [])
ev = [{**f,
"reconciled": f["path"] in applied,
"gitops": (f["path"].split("/", 1)[0] in gitops_roots
and f.get("manifest", False))}
for f in files]
gitops_ev = [e for e in ev if e["gitops"]]
rec_ev = [e for e in gitops_ev if e["reconciled"]]
unrec_ev = [e for e in gitops_ev if not e["reconciled"]]
if not ev:
impl_state = "none"
elif rec_ev and unrec_ev:
# Some citing manifests ARE applied and some are NOT. Never round this up to
# "reconciled": e.g. R-UPD-01's only applied citation is a PrometheusRule that
# merely mentions the ID, while platform/image-automation/ — the component
# that actually implements it — is commented out of platform/kustomization.yaml.
impl_state = "reconciled-partial"
elif rec_ev:
impl_state = "reconciled"
elif gitops_ev:
impl_state = "unwired"
else:
impl_state = "off-cluster"
ts = by_req_tests[rid]
exec_ts = [t for t in ts if t["state"].startswith("executable")]
if not ts:
verify_state = "none"
elif exec_ts:
verify_state = "executable"
else:
verify_state = "scaffold-only"
safety_l3_ok = None
if r["safety_critical"]:
safety_l3_ok = any(t["layer"] == "L3" and t["state"].startswith("executable")
for t in ts)
records[rid] = {
"doc": r["doc"],
"summary": r["summary"],
"safety_critical": r["safety_critical"],
"implementation": {
"state": impl_state,
"files_total": len(ev),
"files_gitops": len(gitops_ev),
"files_reconciled": len(rec_ev),
"comment_only_evidence_only": bool(ev) and all(e["comment_only"] for e in ev),
"files": ev,
},
"verification": {
"state": verify_state,
"executable_tests": len(exec_ts),
"scaffold_tests": len(ts) - len(exec_ts),
"layers": sorted({t["layer"] for t in exec_ts}),
"safety_critical_l3_executable": safety_l3_ok,
"blocked_by": sorted({t["blocked_by"] for t in ts if t["blocked_by"]}),
"tests": ts,
},
# Derived ONLY from the committed ledger, and only as far as it reaches.
# With no ledger row this stays `not-derivable`, as it always was.
"proof": derive_proof(git, rid, ts, ev, latest),
"status": LABELS[(impl_state, verify_state)],
}
# --- hygiene defects this tool owns -------------------------------------------
hygiene = {
"annotation_conflicts": [t["path"] for t in classified if t["annotation_conflict"]],
"executable_untagged": [t["path"] for t in classified
if t["state"] == "executable-untagged"],
"registry_not_in_spec": [],
"spec_not_in_registry": [],
"spec_parity_checked": spec_ids is not None,
}
if spec_ids is not None:
reg_ids = {r["id"] for r in registry}
hygiene["registry_not_in_spec"] = sorted(reg_ids - spec_ids)
hygiene["spec_not_in_registry"] = sorted(spec_ids - reg_ids)
safety_scaffold = sorted(rid for rid, v in records.items()
if v["safety_critical"]
and v["verification"]["safety_critical_l3_executable"] is False)
def count(pred) -> int:
return sum(1 for v in records.values() if pred(v))
summary = {
"environment": env,
"generated_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"total_requirements": len(registry),
"impl_reconciled": count(lambda v: v["implementation"]["state"] == "reconciled"),
"impl_reconciled_partial": count(
lambda v: v["implementation"]["state"] == "reconciled-partial"),
"impl_unwired": count(lambda v: v["implementation"]["state"] == "unwired"),
"impl_off_cluster": count(lambda v: v["implementation"]["state"] == "off-cluster"),
"impl_none": count(lambda v: v["implementation"]["state"] == "none"),
"verify_executable": count(lambda v: v["verification"]["state"] == "executable"),
"verify_scaffold_only": count(lambda v: v["verification"]["state"] == "scaffold-only"),
"verify_none": count(lambda v: v["verification"]["state"] == "none"),
"wired_and_testable": count(
lambda v: v["implementation"]["state"] == "reconciled"
and v["verification"]["state"] == "executable"),
"highest_derivable_state_reached": count(
lambda v: v["implementation"]["state"] in ("reconciled", "off-cluster")
and v["verification"]["state"] == "executable"),
"wired_or_partial_and_testable": count(
lambda v: v["implementation"]["state"] in ("reconciled", "reconciled-partial")
and v["verification"]["state"] == "executable"),
"safety_critical_total": count(lambda v: v["safety_critical"]),
"safety_critical_without_executable_l3": len(safety_scaffold),
# PROVEN is now a real count — but only of requirements whose EVERY verifying
# test has a committed ledger row that saw its pass sentinel, at a sha the
# implementation has not moved past. `stale-proof` and `skipped` are counted
# separately and are NOT green.
"proven": count(lambda v: v["proof"]["state"] == "proven"),
"proof_stale": count(lambda v: v["proof"]["state"] == "stale-proof"),
"proof_skipped": count(lambda v: v["proof"]["state"] == "skipped"),
"proof_failed": count(lambda v: v["proof"]["state"] == "failed"),
"proof_unknown": count(lambda v: v["proof"]["state"] == "unknown"),
"proof_not_derivable": count(
lambda v: v["proof"]["state"] in ("not-derivable", "junit-skipped")),
"ledger_rows": len(ledger_rows),
"proven_derivable": bool(ledger_rows),
"coverage_style_covered": count(lambda v: v["verification"]["state"] != "none"),
}
summary["uncited_by_any_reconciled_manifest"] = count(
lambda v: v["implementation"]["state"] in ("none", "off-cluster"))
return {"summary": summary, "by_requirement": records,
"safety_critical_without_executable_l3": safety_scaffold,
"hygiene": hygiene}
def spec_rollup(rep: dict) -> dict:
"""Per-spec-document rollup — 'the status of each spec', the operator's actual ask."""
docs: dict[str, dict] = {}
for rid, v in rep["by_requirement"].items():
d = docs.setdefault(v["doc"] or "(unmapped)", {
"requirements": [], "reconciled": 0, "partly_reconciled": 0,
"executable": 0, "wired_and_testable": 0})
d["requirements"].append(rid)
if v["implementation"]["state"] == "reconciled":
d["reconciled"] += 1
elif v["implementation"]["state"] == "reconciled-partial":
d["partly_reconciled"] += 1
if v["verification"]["state"] == "executable":
d["executable"] += 1
if (v["implementation"]["state"] in ("reconciled", "off-cluster")
and v["verification"]["state"] == "executable"):
d["wired_and_testable"] += 1
for d in docs.values():
d["requirements"].sort()
d["total"] = len(d["requirements"])
return dict(sorted(docs.items()))
# ========================================================================= render
def render_md(rep: dict, nodes: list[dict], advisory: dict | None) -> str:
s = rep["summary"]
o: list[str] = []
o.append("# Requirement status — derived from repository state\n")
o.append(f"Environment overlay: `clusters/{s['environment']}` · "
f"generated {s['generated_utc']} · **no cluster was contacted**.\n")
o.append("\n## Headline\n")
o.append(f"| Axis | Count (of {s['total_requirements']}) |")
o.append("|---|---|")
o.append(f"| Implementation fully reconciled by Flux | **{s['impl_reconciled']}** |")
o.append(f"| Implementation only PARTLY reconciled (some citing manifests not "
f"applied) | **{s['impl_reconciled_partial']}** |")
o.append(f"| Implementation authored but NOT wired into the Flux graph | "
f"**{s['impl_unwired']}** |")
o.append(f"| Implementation is off-cluster tooling (CI / harness / IaC) | "
f"{s['impl_off_cluster']} |")
o.append(f"| No implementing artifact in the repo | **{s['impl_none']}** |")
o.append(f"| Has an EXECUTABLE requirement-tagged test | **{s['verify_executable']}** |")
o.append(f"| Tagged test exists but is a TODO SCAFFOLD | **{s['verify_scaffold_only']}** |")
o.append(f"| No requirement-tagged test at all | **{s['verify_none']}** |")
o.append(f"| Reconciled **and** executably tested (highest derivable state) | "
f"**{s['highest_derivable_state_reached']}** |")
o.append(f"| Safety-critical requirements whose L3 test is a scaffold | "
f"**{s['safety_critical_without_executable_l3']}** of "
f"{s['safety_critical_total']} |")
if s["proven_derivable"]:
o.append(f"| **PROVEN (every verifying test passed, evidenced)** | "
f"**{s['proven']}** |")
o.append(f"| Proof exists but is STALE (code changed after the run) | "
f"{s['proof_stale']} |")
o.append(f"| Verifying test SELF-SKIPPED green (not proof) | {s['proof_skipped']} |")
o.append(f"| Verifying test FAILED in the recorded run | {s['proof_failed']} |")
o.append(f"| Ran, but the log could not evidence pass vs self-skip | "
f"{s['proof_unknown']} |")
o.append(f"| No committed ledger row yet | {s['proof_not_derivable']} |")
o.append("")
o.append(f"> **Proof comes from `conformance/runs/ledger.jsonl`** "
f"({s['ledger_rows']} row(s)), written only by an explicit "
f"`run.sh --record` / `lib/record_run.py` invocation. A row reaches "
f"`proven` ONLY when the run log carried that test's own pass sentinel "
f"token — the L2/L3 tests self-skip GREEN (Chainsaw records `passed` on "
f"`DR_RESTORE_SKIPPED_OK` exactly as on a real run, and its JUnit carries "
f"no stdout), so a skip token maps to `skipped` and can never become "
f"proof. A proof is honoured only while the implementing manifests and "
f"the test itself have not changed since the recorded `git_sha`; "
f"otherwise it degrades to `stale-proof`, which is not green.\n")
else:
o.append(f"| **PROVEN (a test actually passed)** | **{s['proven']} — NOT DERIVABLE** |")
o.append("")
o.append("> **No proof is derivable: `conformance/runs/ledger.jsonl` is absent or "
"empty.** The ledger is the committed, machine-readable run record; until "
"a run is captured with `run.sh --record`, nothing tracked in git records "
"a test result (`conformance/_artifacts/` and `results/` are git-ignored, "
"and DECISIONS-LOG.md prose is hand-maintained — the very thing this "
"report replaces). Per `agents/definition-of-done.md` a requirement is "
"done only when its tagged test **passes**, so the strongest claim this "
"report can make is *wired and testable*; treat every row below as "
"unproven.\n")
o.append(f"> For contrast, `coverage.md` counts **{s['coverage_style_covered']}** "
f"requirements as covered — that count includes the "
f"{s['verify_scaffold_only']} whose only verifying test is a TODO scaffold.\n")
o.append("\n## Status legend\n")
o.append("| Label | Meaning |")
o.append("|---|---|")
o.append("| `WIRED+TESTABLE` | implementing config is reconciled by Flux **and** an "
"executable tagged test exists (still unproven) |")
o.append("| `WIRED, TEST-IS-SCAFFOLD` | reconciled, but its only tagged test is a "
"TODO placeholder — coverage counts this green; it is not |")
o.append("| `WIRED, UNTESTED` | reconciled with no tagged test at all |")
o.append("| `PARTLY-WIRED…` | some citing manifests are applied, others are not — "
"check whether the unapplied one is the real implementation |")
o.append("| `AUTHORED-NOT-WIRED` | manifests exist under a GitOps root but no Flux "
"Kustomization reaches them — in git, not on any cluster |")
o.append("| `TOOLING…` | implemented by off-cluster tooling (ci/, conformance/, "
"infra/, bootstrap/, schemas/) — nothing for Flux to reconcile |")
o.append("| `TEST-ONLY` / `SPEC-ONLY` | a test but no implementation / neither |")
o.append("\n## Per-spec rollup\n")
o.append("| Spec document | Reqs | Reconciled | Partly | Executable test | "
"Wired+Testable |")
o.append("|---|---|---|---|---|---|")
for doc, d in spec_rollup(rep).items():
o.append(f"| `{doc}` | {d['total']} | {d['reconciled']} | "
f"{d['partly_reconciled']} | {d['executable']} | "
f"**{d['wired_and_testable']}/{d['total']}** |")
o.append("\n## Status by requirement\n")
o.append("| Requirement | Status | Implementation | Verification | Proof |")
o.append("|---|---|---|---|---|")
for rid, v in rep["by_requirement"].items():
i, ver = v["implementation"], v["verification"]
if i["state"] == "none":
icell = "—"
else:
icell = (f"{i['state']} ({i['files_reconciled']}/{i['files_gitops']} gitops "
f"files applied, {i['files_total']} refs)")
if i["comment_only_evidence_only"]:
icell += " *(comment-only evidence)*"
if ver["state"] == "none":
vcell = "—"
else:
layers = ",".join(ver["layers"]) or "-"
vcell = (f"{ver['state']} [{layers}] "
f"exec={ver['executable_tests']} scaffold={ver['scaffold_tests']}")
if ver["blocked_by"]:
vcell += f" · blocked-by: {'; '.join(ver['blocked_by'])}"
sc = " ⚠safety" if v["safety_critical"] else ""
pr = v["proof"]
if pr["state"] == "not-derivable":
pcell = "not-derivable"
else:
runs = sorted({d["run_id"] for d in pr["per_test"] if d.get("run_id")})
pcell = f"**{pr['state']}**" if pr["state"] == "proven" else pr["state"]
if runs:
pcell += f" ({', '.join(runs)})"
o.append(f"| {rid}{sc} | **{v['status']}** | {icell} | {vcell} | {pcell} |")
if rep["safety_critical_without_executable_l3"]:
o.append("\n## Safety-critical requirements whose L3 fault test is a SCAFFOLD "
"(R-TEST-05)\n")
o.append("`coverage.md` reports zero safety-critical gaps because a tagged L3 "
"test file exists. These L3 tests have TODO bodies and assert nothing:\n")
for rid in rep["safety_critical_without_executable_l3"]:
v = rep["by_requirement"][rid]
bb = "; ".join(v["verification"]["blocked_by"]) or "—"
o.append(f"- **{rid}** — {v['summary']} · blocked-by: {bb}")
partial = [r for r, v in rep["by_requirement"].items()
if v["implementation"]["state"] == "reconciled-partial"]
if partial:
o.append("\n## Partly wired — citing manifests that Flux does NOT apply\n")
o.append("At least one manifest naming these requirements is reconciled, but "
"others are not. Check whether the unapplied file is the one that "
"actually implements the requirement.\n")
for rid in partial:
v = rep["by_requirement"][rid]
miss = [f"`{f['path']}`" for f in v["implementation"]["files"]
if f["gitops"] and not f["reconciled"]]
o.append(f"- **{rid}** — not applied: {', '.join(miss)}")
unwired = [r for r, v in rep["by_requirement"].items()
if v["implementation"]["state"] == "unwired"]
if unwired:
o.append("\n## Authored but NOT reconciled (in git, not on any cluster)\n")
for rid in unwired:
v = rep["by_requirement"][rid]
fs = ", ".join(f"`{f['path']}`" for f in v["implementation"]["files"]
if f["gitops"])[:400]
o.append(f"- **{rid}** — {v['summary']}\n - {fs}")
uncited = [r for r, v in rep["by_requirement"].items()
if v["implementation"]["state"] in ("none", "off-cluster")]
if uncited:
o.append("\n## No reconciled manifest cites these requirements (traceability, "
"R-TEST-04)\n")
o.append("Implementation evidence here is **a tracked, non-Markdown file that "
"names the R-ID**. For these requirements no reconciled Kubernetes "
"manifest names the ID, which means EITHER the requirement is genuinely "
"unimplemented on the cluster, OR the implementing manifest exists but "
"is missing its requirement annotation. This report cannot tell the two "
"apart and deliberately does not guess — annotate the manifest and the "
"row self-corrects.\n")
for rid in uncited:
v = rep["by_requirement"][rid]
o.append(f"- **{rid}** ({v['implementation']['state']}) — {v['summary']}")
o.append("\n## Flux reconciliation graph walked\n")
o.append("| Flux Kustomization | spec.path | present |")
o.append("|---|---|---|")
for n in sorted(nodes, key=lambda n: (n["resolved"], n["kustomization"])):
o.append(f"| `{n['kustomization']}` | `{n['path']}` | "
f"{'yes' if n['exists'] else '**MISSING**'} |")
h = rep["hygiene"]
o.append("\n## Harness hygiene\n")
if h["spec_parity_checked"]:
drift = h["registry_not_in_spec"] + h["spec_not_in_registry"]
o.append(f"- registry ↔ spec parity: {'**DRIFT**: ' + str(drift) if drift else 'in sync'}")
else:
o.append("- registry ↔ spec parity: **NOT CHECKED** (spec tree not available; "
"this is a skip, not a pass)")
o.append(f"- tests annotated executable but containing TODO bodies: "
f"{h['annotation_conflicts'] or 'none'}")
o.append(f"- executable tests missing the `{LABEL_PREFIX}/executable` annotation: "
f"{h['executable_untagged'] or 'none'}")
o.append("\n## Method — what is derived, and what is refused\n")
o.append("| Signal | How it is derived | Trust |")
o.append("|---|---|---|")
o.append("| requirement set | `conformance/requirements/registry.yaml`, "
"cross-checked against the read-only spec `docs/00-vision-and-scope.md` | "
"exact |")
o.append("| implementing artifact | a TRACKED, non-Markdown file that names the R-ID "
"(comments counted, and marked as such) | heuristic — under-reports "
"un-annotated manifests, never over-reports |")
o.append("| reconciled | the file is reachable by walking `resources:` from "
"`clusters/<env>/kustomization.yaml`, expanding every Flux Kustomization's "
"`spec.path` | exact — a commented-out `resources:` entry is correctly NOT "
"reconciled |")
o.append("| verifying test | `conformance.home.k3s/R-*` label on a Chainsaw test "
"(the authoritative tag, same source as `coverage.md`) | exact |")
o.append("| executable vs scaffold | TODO markers in the test body outrank the "
"`executable` annotation; a real body without the annotation still counts "
"and is flagged | conservative — ambiguity never upgrades |")
o.append("| **test actually passed** | the committed `runs/ledger.jsonl` row for "
"every verifying test: JUnit result **plus** that test's own pass sentinel "
"token grepped from the run log | exact where a run was recorded, refused "
"everywhere else — a self-skip token or a missing/unrecognised sentinel "
"NEVER yields `proven` |")
o.append("| proof still current | the row's `git_sha` must be a descendant of the "
"last commit touching the requirement's implementing manifests or its test "
"| conservative — an unresolvable or `-dirty` sha degrades to `stale-proof` |")
o.append("| cluster liveness (pods/CRs healthy) | **NOT DERIVED** — no cluster is "
"contacted; `reconciled` means Flux *would* apply it, not that it is "
"healthy | refused |")
o.append("| phase/board status | **NOT DERIVED** — `agents/task-board.md` is stale "
"and is not a signal | refused |")
o.append("\n## Run ledger — the committed proof-of-pass evidence\n")
if not s["proven_derivable"]:
o.append("- `conformance/runs/ledger.jsonl` holds no rows. Capture a run with "
"`./run.sh --record` (recording is never implicit: a run that silently "
"records itself is how a bad run becomes a permanent false proof), then "
"commit `conformance/runs/<run-id>/` **with its `run.log`** and the "
"ledger. The log is the load-bearing half — Chainsaw's JUnit carries no "
"stdout, so without it a self-skip is indistinguishable from a pass.")
else:
o.append(f"- {s['ledger_rows']} row(s) in `conformance/runs/ledger.jsonl`. "
f"Per-test evidence for every non-`not-derivable` requirement below; "
f"full rows (env, chainsaw/kube versions, sentinel, note) are in "
f"`status.json` under `by_requirement.<R-ID>.proof.per_test`.\n")
o.append("| Requirement | Proof | Test | Ledger state | Sentinel | Run |")
o.append("|---|---|---|---|---|---|")
for rid, v in rep["by_requirement"].items():
pr = v["proof"]
if pr["state"] == "not-derivable":
continue
for d in pr["per_test"]:
o.append(f"| {rid} | {pr['state']} | `{d['test']}` | {d['state']} | "
f"`{d.get('sentinel') or '—'}` | {d.get('run_id') or '—'} |")
stale = [(rid, v) for rid, v in rep["by_requirement"].items()
if v["proof"]["state"] == "stale-proof"]
if stale:
o.append("\n### Stale proofs — a real pass that no longer describes this tree\n")
o.append("These requirements DID pass, but their implementing manifests or "
"their verifying test changed in a commit the recorded run predates. "
"A stale proof is not green; re-run with `--record`.\n")
for rid, v in stale:
pr = v["proof"]
o.append(f"- **{rid}** — {pr['reason']}")
skipped = [(rid, v) for rid, v in rep["by_requirement"].items()
if v["proof"]["state"] == "skipped"]
if skipped:
o.append("\n### SELF-SKIPPED green — recorded, and deliberately NOT proof\n")
o.append("Chainsaw reported `passed` for these, because the test exited 0 "
"after emitting its skip token: a precondition (Longhorn "
"StorageClass, Velero CRDs, >=3 Ready nodes) was absent, so the "
"behaviour was never exercised. This is exactly the row a "
"JUnit-only ledger would have recorded as green.\n")
for rid, v in skipped:
toks = ", ".join(sorted({d.get("sentinel") or "?"
for d in v["proof"]["per_test"]
if d["state"] == "skipped"}))
o.append(f"- **{rid}** — skip token(s): `{toks}`")
o.append("\n## Advisory — untracked local run artifact (NOT counted)\n")
if advisory:
o.append(f"- `{advisory['path']}` (untracked, git-ignored), last written "
f"{advisory['mtime']}: {advisory['tests']} testcase(s), "
f"{advisory['failures']} failure(s), {advisory['skipped']} skipped.")
o.append(f"- requirement IDs it names: {', '.join(advisory['requirements']) or '—'}")
o.append("- This is one developer's local workspace, not repo state, and it is NOT "
"the ledger. It is shown for orientation only and contributes to NO "
"status above — a JUnit alone cannot tell a self-skip from a pass.")
else:
o.append("- none present.")
return "\n".join(o) + "\n"
def read_advisory(junit: Path) -> dict | None:
"""Quarantined read of an untracked local JUnit. Never feeds a status."""
if not junit.is_file():
return None
import xml.etree.ElementTree as ET
try:
root = ET.parse(junit).getroot()
except ET.ParseError:
return None
cases = root.iter("testcase")
n = f = sk = 0
reqs: set[str] = set()
for c in cases:
n += 1
if c.find("failure") is not None or c.find("error") is not None:
f += 1
if c.find("skipped") is not None:
sk += 1
reqs |= set(RID_RE.findall(c.get("classname") or ""))
for p in c.iter("property"):
if p.get("name") == "requirements":
reqs |= set(RID_RE.findall(p.get("value") or ""))
return {"path": os.path.relpath(junit, REPO_ROOT), "tests": n, "failures": f,
"skipped": sk, "requirements": sorted(reqs),
"mtime": time.strftime("%Y-%m-%dT%H:%M:%SZ",
time.gmtime(junit.stat().st_mtime))}
# ======================================================================= selftest
def _selftest() -> int:
"""Deterministic, no-cluster self-test of every derivation rule (house pattern:
`catalog/render.py --selftest`). Uses a throwaway tree, never the real repo."""
import tempfile
ok = True
def check(label: str, cond: bool) -> None:
nonlocal ok
if not cond:
sys.stderr.write(f"status.py selftest: FAIL — {label}\n")
ok = False
with tempfile.TemporaryDirectory() as td:
root = Path(td)
# A kustomization whose second entry is COMMENTED OUT must not be reconciled.
(root / "clusters" / "dev").mkdir(parents=True)
(root / "platform" / "on").mkdir(parents=True)
(root / "platform" / "off").mkdir(parents=True)
(root / "clusters" / "dev" / "kustomization.yaml").write_text(
"resources:\n - platform-ks.yaml # comment\n")
(root / "clusters" / "dev" / "platform-ks.yaml").write_text(
"apiVersion: kustomize.toolkit.fluxcd.io/v1\n"
"kind: Kustomization\nmetadata:\n name: platform\nspec:\n"
" path: ./platform\n")
(root / "platform" / "kustomization.yaml").write_text(
"resources:\n - on\n # - off # deliberately disabled\n")
# flush-left items (the `render.py --emit-waves` style) must parse too.
(root / "platform" / "on" / "kustomization.yaml").write_text(
"resources:\n- r.yaml\npatches: []\n")
(root / "platform" / "on" / "r.yaml").write_text("# serves R-NET-01\nkind: X\n")
(root / "platform" / "off" / "kustomization.yaml").write_text(
"resources:\n - r.yaml\n")
(root / "platform" / "off" / "r.yaml").write_text("# serves R-UPD-01\nkind: X\n")
applied, nodes = reconciled_files(root, "clusters/dev")
check("commented-out resource is NOT reconciled",
"platform/on/r.yaml" in applied and "platform/off/r.yaml" not in applied)
check("flux spec.path expands the graph",
any(n["path"] == "./platform" for n in nodes))
files = ["platform/on/r.yaml", "platform/off/r.yaml", "docs/x.md",
"conformance/tests/t/chainsaw-test.yaml"]
(root / "docs").mkdir(); (root / "docs" / "x.md").write_text("R-NET-01 R-TLS-01\n")
(root / "conformance" / "tests" / "t").mkdir(parents=True)
(root / "conformance" / "tests" / "t" / "chainsaw-test.yaml").write_text("R-DNS-01\n")
impl = scan_implementation(root, files)
check("markdown is excluded from implementation evidence", "R-TLS-01" not in impl)
check("conformance/tests is excluded from implementation evidence",
"R-DNS-01" not in impl)
check("comment-only evidence is recorded as such",
impl["R-NET-01"][0]["comment_only"] is True)
(root / "platform" / "on" / "real.yaml").write_text(
"kind: ClusterPolicy\n conformance.home.k3s/R-SEC-01: \"true\"\n")
impl2 = scan_implementation(root, ["platform/on/real.yaml"])
check("non-comment evidence is recorded as such",
impl2["R-SEC-01"][0]["comment_only"] is False)
check("a yaml with a top-level kind: is a reconcilable manifest",
impl2["R-SEC-01"][0]["manifest"] is True)
(root / "platform" / "on" / "s.sh").write_text("# R-SEC-02\n")
check("a shell script is never a reconcilable manifest",
scan_implementation(root, ["platform/on/s.sh"])["R-SEC-02"][0]["manifest"]
is False)
# classify_test: TODO body outranks an `executable: true` annotation.
td2 = root / "conformance" / "tests" / "s"
td2.mkdir(parents=True)
f = td2 / "chainsaw-test.yaml"
f.write_text("metadata:\n annotations:\n"
" conformance.home.k3s/executable: \"true\"\n"
"spec:\n steps:\n - script:\n"
" content: 'echo \"TODO: write me\"'\n")
c = classify_test({"name": "s", "path": str(f), "layer": "L2"})
check("TODO body downgrades a test annotated executable",
c["state"] == "scaffold" and c["annotation_conflict"])
f.write_text("metadata:\n annotations:\n"
" conformance.home.k3s/executable: \"true\"\n"
"spec:\n steps:\n - script:\n content: 'true'\n")
check("annotated + no TODO == executable",
classify_test({"name": "s", "path": str(f), "layer": "L2"})["state"]
== "executable")
f.write_text("spec:\n steps:\n - script:\n content: 'true'\n")
check("real body without the annotation == executable-untagged",
classify_test({"name": "s", "path": str(f), "layer": "L2"})["state"]
== "executable-untagged")
f.write_text("spec: {}\n")
check("empty body == scaffold",
classify_test({"name": "s", "path": str(f), "layer": "L2"})["state"]
== "scaffold")
# ---------------------------------------------------------------- PROOF / ledger
# These are the assertions that stand between this report and a fake green.
check("empty ledger parses to no rows (pre-ledger behaviour is preserved)",
read_ledger(Path("/nonexistent/ledger.jsonl")) == [])
def row(**kw):
base = {"ts_utc": "2026-09-01T12:00:00Z", "git_sha": "deadbeef", "env": "dev",
"chainsaw_version": "v0.2.12", "kube_version": "v1.36.4", "layer": "L3",
"test": "dr-restore", "requirements": ["R-STATE-03"],
"junit_result": "passed", "sentinel": None, "proof": "skipped",
"run_id": "20260901T120000Z-deadbee", "note": ""}
base.update(kw)
return base
verifiers = [{"name": "dr-restore",
"path": "conformance/tests/l3-fault-injection/dr-restore/"
"chainsaw-test.yaml"}]
with tempfile.TemporaryDirectory() as td:
gr = Path(td)
env = {**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t",
"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t"}
def git(*a):
return subprocess.run(["git", "-C", str(gr), *a], capture_output=True,
text=True, env=env)
git("init", "-q", "-b", "main")
(gr / "platform").mkdir()
(gr / "platform" / "velero.yaml").write_text("kind: X # R-STATE-03\n")
git("add", "-A"); git("commit", "-qm", "one")
sha_a = git("rev-parse", "HEAD").stdout.strip()
(gr / "platform" / "velero.yaml").write_text("kind: X # R-STATE-03 v2\n")
git("add", "-A"); git("commit", "-qm", "two")
sha_b = git("rev-parse", "HEAD").stdout.strip()
oracle = _GitOracle(gr)
impl = [{"path": "platform/velero.yaml"}]
# *** THE CASE THIS WHOLE FEATURE EXISTS FOR ***
pr = derive_proof(oracle, "R-STATE-03", verifiers, impl,
latest_by_test([row(proof="skipped", junit_result="passed",
sentinel="DR_RESTORE_SKIPPED_OK",
git_sha=sha_b)]))
check("a SELF-SKIP row on a GREEN JUnit is 'skipped', NEVER 'proven'",
pr["state"] == "skipped")
pr = derive_proof(oracle, "R-STATE-03", verifiers, impl,
latest_by_test([row(proof="proven", git_sha=sha_b,
sentinel="DR_RESTORED_OK")]))
check("a proven row at the CURRENT sha is honoured", pr["state"] == "proven")
pr = derive_proof(oracle, "R-STATE-03", verifiers, impl,
latest_by_test([row(proof="proven", git_sha=sha_a,
sentinel="DR_RESTORED_OK")]))
check("STALENESS: proven at a sha the implementation has moved past degrades "
"to stale-proof", pr["state"] == "stale-proof")
pr = derive_proof(oracle, "R-STATE-03", verifiers, impl,
latest_by_test([row(proof="proven", git_sha=f"{sha_b}-dirty",
sentinel="DR_RESTORED_OK")]))
check("a run recorded from a DIRTY tree is never honoured as proof",
pr["state"] == "stale-proof")
pr = derive_proof(oracle, "R-STATE-03", verifiers, impl,
latest_by_test([row(proof="proven", git_sha="0" * 40)]))
check("a git_sha this clone cannot resolve degrades, never upgrades",
pr["state"] == "stale-proof")
for bad, expect in (("failed", "failed"),
("junit-skipped", "junit-skipped"),
("unknown-sentinel-absent", "unknown"),
("unknown-no-sentinel-declared", "unknown"),
("something-new-from-the-future", "unknown")):
pr = derive_proof(oracle, "R-STATE-03", verifiers, impl,
latest_by_test([row(proof=bad, git_sha=sha_b)]))
check(f"ledger state {bad!r} maps to {expect!r}, never to proven",
pr["state"] == expect)
pr = derive_proof(oracle, "R-STATE-03", verifiers, impl, {})
check("no ledger row at all stays not-derivable",
pr["state"] == "not-derivable")
# Two verifying tests, only one proven: the requirement must NOT read proven.
two = verifiers + [{"name": "data-survival",
"path": "conformance/tests/l3-fault-injection/"
"data-survival/chainsaw-test.yaml"}]
pr = derive_proof(oracle, "R-STATE-03", two, impl,
latest_by_test([row(proof="proven", git_sha=sha_b)]))
check("a requirement is proven only when EVERY verifying test is",
pr["state"] == "not-derivable")
pr = derive_proof(oracle, "R-STATE-03", two, impl,
latest_by_test([row(proof="proven", git_sha=sha_b),
row(test="data-survival", proof="skipped",
git_sha=sha_b)]))
check("worst-of wins across verifying tests (skip drags proven down)",
pr["state"] == "skipped")
# Latest row wins, not best-ever: a test that regressed to self-skipping must
# stop reading as proven.
pr = derive_proof(oracle, "R-STATE-03", verifiers, impl,
latest_by_test([row(ts_utc="2026-09-01T10:00:00Z",
proof="proven", git_sha=sha_b),
row(ts_utc="2026-09-02T10:00:00Z",
proof="skipped", git_sha=sha_b)]))
check("the LATEST ledger row wins, not the best one ever recorded",
pr["state"] == "skipped")
# A malformed ledger line is skipped, never guessed at.
lf = gr / "ledger.jsonl"
lf.write_text('{"test":"a","proof":"proven"}\nnot json at all\n\n'
'{"proof":"proven"}\n')
rows = read_ledger(lf)
check("malformed / test-less ledger lines are dropped, not guessed at",
len(rows) == 1 and rows[0]["test"] == "a")
if not ok:
return 1
sys.stderr.write("status.py selftest: PASS\n")
return 0
# =========================================================================== main
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--repo", type=Path, default=REPO_ROOT)
ap.add_argument("--registry", type=Path,
default=CONF_ROOT / "requirements" / "registry.yaml")
ap.add_argument("--tests", type=Path, default=CONF_ROOT / "tests")
ap.add_argument("--out", type=Path, default=CONF_ROOT / "_artifacts" / "status")
ap.add_argument("--env", default="dev",
help="cluster overlay to walk (clusters/<env>); default dev")
ap.add_argument("--spec-root", type=Path, default=None,
help=f"read-only spec checkout for the registry parity check "
f"(default {DEFAULT_SPEC_ROOT} if present)")
ap.add_argument("--strict", action="store_true",
help="exit 1 on defects this tool owns: registry/spec drift, "
"executable-annotation conflicts, or a safety-critical "
"requirement whose only L3 test is a scaffold")
ap.add_argument("--ledger", type=Path, default=DEFAULT_LEDGER,
help=f"committed proof-of-pass ledger written by lib/record_run.py "
f"(default {DEFAULT_LEDGER}). Absent or empty => every "
f"requirement stays `not-derivable`, as before the ledger existed")
ap.add_argument("--selftest", action="store_true", help="run the built-in self-test")
args = ap.parse_args()
if args.selftest:
return _selftest()
repo = args.repo.resolve()
entry = f"clusters/{args.env}"
if not (repo / entry / "kustomization.yaml").is_file():
sys.stderr.write(f"[status] no Flux entrypoint at {entry}/kustomization.yaml — "
f"cannot derive reconciliation. SKIPPED (not green).\n")
return 3
explicit_spec = args.spec_root is not None
spec_root = (args.spec_root or DEFAULT_SPEC_ROOT).resolve()
spec_ids = spec_requirement_ids(spec_root)
if spec_ids is None and explicit_spec:
sys.stderr.write(f"[status] --spec-root {spec_root} has no "
f"docs/00-vision-and-scope.md. SKIPPED (not green).\n")
return 3
registry = cov.parse_registry(args.registry)
if not registry:
sys.stderr.write("[status] registry parsed to zero requirements — refusing to "
"emit a report that would read as 100% of nothing.\n")
return 1
tests = cov.discover_tests(args.tests)
applied, nodes = reconciled_files(repo, entry)
gitops_roots = {n["resolved"].split("/", 1)[0] for n in nodes if n["exists"]}
impl = scan_implementation(repo, tracked_files(repo))
ledger_rows = read_ledger(args.ledger)
rep = build(repo, registry, tests, impl, applied, gitops_roots, spec_ids, args.env,
ledger_rows)
rep["ledger"] = {"path": os.path.relpath(args.ledger, REPO_ROOT),
"rows": len(ledger_rows),
"present": args.ledger.is_file()}
rep["flux_graph"] = nodes
rep["gitops_roots"] = sorted(gitops_roots)
rep["reconciled_file_count"] = len(applied)
advisory = read_advisory(CONF_ROOT / "_artifacts" / "report.junit.xml")
rep["advisory_untracked_junit"] = advisory
args.out.mkdir(parents=True, exist_ok=True)
(args.out / "status.json").write_text(json.dumps(rep, indent=2) + "\n")
(args.out / "status.md").write_text(render_md(rep, nodes, advisory))
s = rep["summary"]
print(f"[status] {s['total_requirements']} requirements | reconciled "
f"{s['impl_reconciled']} (+{s['impl_reconciled_partial']} partial) | "
f"executable test {s['verify_executable']} | "
f"wired+testable {s['highest_derivable_state_reached']} "
f"(+{s['wired_or_partial_and_testable'] - s['wired_and_testable']} partial) | "
f"scaffold-only {s['verify_scaffold_only']} | "
f"safety-critical w/o executable L3 "
f"{s['safety_critical_without_executable_l3']} | "
+ (f"PROVEN: {s['proven']} (stale {s['proof_stale']}, self-skipped "
f"{s['proof_skipped']}, failed {s['proof_failed']}, unknown "
f"{s['proof_unknown']}) from {s['ledger_rows']} ledger row(s)"
if s["proven_derivable"] else "PROVEN: not derivable (empty run ledger)"))
print(f"[status] wrote {args.out/'status.md'} and {args.out/'status.json'}")
rc = 0
if args.strict:
h = rep["hygiene"]
if h["registry_not_in_spec"] or h["spec_not_in_registry"]:
sys.stderr.write(f"[status] STRICT: registry/spec drift "
f"{h['registry_not_in_spec']} {h['spec_not_in_registry']}\n")
rc = 1
if h["annotation_conflicts"]:
sys.stderr.write(f"[status] STRICT: tests annotated executable with TODO "
f"bodies: {h['annotation_conflicts']}\n")
rc = 1
if rep["safety_critical_without_executable_l3"]:
sys.stderr.write(f"[status] STRICT: safety-critical requirement(s) whose L3 "
f"test is a scaffold: "
f"{rep['safety_critical_without_executable_l3']}\n")
rc = 1
return rc
if __name__ == "__main__":
raise SystemExit(main())