#!/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? *** DELIBERATELY NOT DERIVED. *** WHAT THIS TOOL REFUSES TO DERIVE (and why) ------------------------------------------ PROOF-OF-PASS IS NOT DERIVABLE FROM REPO STATE. Nothing tracked in git records a test result: `conformance/_artifacts/` and `results/` are git-ignored, and the only prose record of a green run lives in DECISIONS-LOG.md — hand-maintained, unstructured, and exactly the kind of source the operator has ruled out. The project's own culture is that "merged" is never "done" (agents/definition-of-done.md, R-TEST-03), so this report will NOT infer, guess, or default a requirement to green. Every requirement's proof state is reported as `not-derivable`, and the headline says so. If an untracked JUnit artifact happens to be lying in the workspace it is surfaced ONLY as a quarantined advisory with its mtime — never counted, never mixed into a status. The honest fix is a committed, machine-readable run ledger; until that exists, 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 clusters// 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 ` - ` 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[^#\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/` 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)] 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())) # ========================================================================== 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) -> dict: 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, }, # NEVER derived. See the module docstring. "proof": {"state": "not-derivable", "reason": "no committed, machine-readable record of a test run " "exists in the repo (_artifacts/ and results/ are " "git-ignored); DECISIONS-LOG.md prose is not a signal"}, "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": 0, "proven_derivable": False, "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']} |") o.append(f"| **PROVEN (a test actually passed)** | **{s['proven']} — NOT DERIVABLE** |") o.append("") o.append("> **Proof is deliberately not derived.** Nothing tracked in git records a " "test result: `conformance/_artifacts/` and `results/` are git-ignored, and " "the only record of a green run is DECISIONS-LOG.md prose — hand-maintained, " "which is 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*. Closing the gap needs a committed, machine-readable " "run ledger; until then, 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 "" o.append(f"| {rid}{sc} | **{v['status']}** | {icell} | {vcell} | not-derivable |") 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//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** | **NOT DERIVED** — nothing in git records a " "run result | refused |") 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## 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. It is shown " "for orientation only and contributes to NO status above.") 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") 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/); 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("--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)) rep = build(repo, registry, tests, impl, applied, gitops_roots, spec_ids, args.env) 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']} | PROVEN: not derivable") 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())