jakman85andClaude Opus 5
6a8953e288
Minimal prod delivery, authored directly on main (= prod's live revision d1a6628)
rather than by merging work/prod-standup-campaign. That branch carries the DNS apex
inversion, which would have set prod's BASE_DOMAIN to home.keithjackson.dev; prod
runs external-dns with policy: sync and txtOwnerId: prod, so it would have begun
rewriting apex records dev owns while the non-k3s prod still serves 443 there.
Operator sequencing is "dev migrates first". Nothing under clusters/ is touched
here at all, which is what makes this safe to land now.
Also deliberately EXCLUDED, because prod's live cluster-vars has no CNPG_BACKUP_PATH
key and StrictPostBuildSubstitutions would fail the whole tier: the R2 rework of
platform/velero/ and the CNPG destinationPath changes. Those land as
feat/r2-backup-prod together with the endpoint and bucket knobs.
WHAT THIS FIXES
1. Nothing was watching CNPG. All three prod databases have run since bootstrap with
ContinuousArchiving=False and firstRecoverabilityPoint=null, and no alert existed
anywhere. Because archive_mode=on and the archive command fails, Postgres refuses
to recycle WAL: keycloak-db holds every segment ever written (257 MiB on a 5 GiB
volume, ~6.8 MiB/h) — roughly 29 days to a disk-full PANIC on the cluster's SSO
database. Found by hand, 38h late. CNPGWALArchiveBacklog would have fired within
the hour, and evaluates to 3 firing series on prod today.
Rules key on cnpg_collector_* rather than cnpg_pg_stat_archiver_*, because
nextcloud-db's exporter cannot log in to its own database (permission denied for
database "nextcloud", both clusters) so the pg_stat_archiver family does not exist
for it — the obvious metric would have been blind to one database in three.
2. No PVC outside namespace=monitoring had any fill alert. New rules are partitioned
by namespace so they cannot double-fire with MonitoringVolumeFillingUp, pinned to
job="kubelet" (the apiserver copy of the same series reports namespace=default),
and guarded by inodes > 0 — block-mode PVCs such as unifi-rootdisk report
available=0 forever and would otherwise have fired critical permanently.
3. VPNTunnelDown could never clear on a cluster with no gateway node; it has been
firing critical on prod since 2026-09-05. Now guarded so prod self-arms the moment
a node carries the egress-gateway label, with dev's behaviour unchanged today.
4. The Keycloak realm import fails OPEN: an unresolved placeholder is stored
LITERALLY and the import still reports success, then sits at Done=True and never
re-runs. On 2026-09-06 all 12 prod OIDC clients held the literal
$(env:..._CLIENT_SECRET) string and every SSO login failed, while every signal was
green. The CronJob is the level-triggered correction; two alerts make its silence
observable. Repaired live already — this stops it recurring.
Also: the R-NET-04 conformance scan descended into nested agent worktrees under
.claude/, counting 4 egress policies where 1 is authored.
Co-Authored-By: Claude Opus 5 <[email protected]>
1551 lines
80 KiB
Python
Executable File
1551 lines
80 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.
|
|
# `conformance/runs/` is excluded for a subtler reason than the test dirs: the ledger and the
|
|
# preserved run logs CONTAIN the requirement IDs they are evidence FOR. Counting them as
|
|
# implementing evidence makes `last_change` move every time proof is recorded, so a row's
|
|
# git_sha always predates it and EVERY proof degrades to stale-proof the instant it is
|
|
# committed — recording the proof destroys it. The ledger is the proof record; it must not
|
|
# participate in the freshness computation about itself. (Found live 2026-09-01 on the first
|
|
# PROVEN row.)
|
|
IMPL_EXCLUDE_PREFIXES = ("conformance/tests/", "ci/tests/", "conformance/runs/")
|
|
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)
|
|
|
|
# ---------------------------------------------------------- scaffold-marker detection
|
|
# WHY THIS IS NOT A WHOLE-FILE `"TODO" in text`
|
|
# ---------------------------------------------------
|
|
# It used to be, and that mis-reported a SAFETY-CRITICAL requirement. A test whose SUBJECT
|
|
# is placeholder detection has to name the idioms it hunts for:
|
|
# `l0-static/placeholder-guard-coupling/check.sh` carries
|
|
# `SEED = ["TBD", "TODO", "FIXME", "PLACEHOLDER", ...]` — its detection vocabulary, as data.
|
|
# A whole-file substring test read that as the test declaring itself unfinished, classified
|
|
# a 32-assertion executable check as a scaffold, raised a permanent `--strict` annotation
|
|
# conflict, and (because the same file is tagged R-STATE-03) reported R-STATE-03 as
|
|
# `exec=1 scaffold=1`. A previous engineer worked around it by rewording prose.
|
|
#
|
|
# The rule below is derived from the corpus (43 tests), not invented. Every genuine
|
|
# scaffold in this tree — `l2-behavioral/onboarding-completeness`, `l3-fault-injection/
|
|
# alerting`, `l3-fault-injection/gitops-drift`, and no others — declares itself in BOTH of
|
|
# two structural positions, and no executable test does either:
|
|
#
|
|
# 1. a `# STATUS: ... scaffold` line in the file's documentation header. This is the
|
|
# repo's DEDICATED declaration slot (tests/README.md): 3 of 43 tests say
|
|
# `STATUS: scaffold` and they are exactly the three scaffolds; 19 others declare
|
|
# `STATUS: EXECUTABLE` / `IMPLEMENTED` / `RUNNABLE`; the rest declare nothing.
|
|
# 2. a TODO/FIXME marker in the BODY that CARRIES A MESSAGE — `echo "TODO: assert PVC
|
|
# bound ..."`, `TODO(T-4.1): hermetic SMTP capture sink`. A marker that carries a
|
|
# message is a note about work that is missing. A marker that is the ENTIRE content of
|
|
# a quoted literal (`"TODO",` in a vocabulary list) carries no message: it is the
|
|
# marker as a *datum*, which is what a placeholder DETECTOR necessarily contains.
|
|
#
|
|
# Both are still evaluated over the body of `chainsaw-test.yaml` AND of any
|
|
# `render-check.sh` / `check.sh`, and either one still OUTRANKS an
|
|
# `executable: "true"` annotation (see classify_test) — the invariant that an annotation
|
|
# must never outrank the body is unchanged. Ambiguity still only ever DOWNGRADES: an
|
|
# unrecognised marker shape, a marker in body code outside a quoted literal, or a marker
|
|
# carrying any trailing text all classify as scaffold.
|
|
#
|
|
# KNOWN, ACCEPTED IMPRECISION (stated rather than hidden): a test that embeds a marker with
|
|
# text attached but as data — e.g. `re.compile(r"TODO|FIXME")`, or a single string holding
|
|
# several idioms — is still called a scaffold. That is the conservative direction, and the
|
|
# author's remedy is the same structural discipline the corpus already uses: keep each
|
|
# idiom a standalone literal in a list.
|
|
|
|
# Vocabulary: only the idioms that DECLARE unfinished work. Deliberately excludes `TBD`,
|
|
# `PLACEHOLDER`, `XXX`, `REPLACE-ME` — tests in this tree legitimately carry those in
|
|
# executable code as data (`l0-static/egress-paths-selected-by-rules/check.sh` compares
|
|
# authored values against `"TBD"`; `l0-static/no-plaintext-secrets/check.sh` matches
|
|
# `placeholder.*`), and none of them is used here to mark a placeholder body.
|
|
# Case-insensitive: no English word contains either token, and the corpus carries no
|
|
# lower-case occurrence, so folding case only ever ADDS detection (`echo "todo: ..."`).
|
|
_MARKER_RE = re.compile(r"\b(?:TODO|FIXME)\b", re.I)
|
|
|
|
# The dedicated declaration slot, on a comment line, case-insensitive. The declared state
|
|
# must be the FIRST token after `STATUS:` — a `STATUS:` line declares exactly one state,
|
|
# and all three real scaffolds read `# STATUS: scaffold — <what is missing>`. Anchoring it
|
|
# there is what keeps `# STATUS: EXECUTABLE ... this is not a scaffold` from matching; a
|
|
# loose `.*scaffold` would turn any prose that merely uses the word into a declaration.
|
|
_STATUS_SCAFFOLD_RE = re.compile(
|
|
r"^[ \t]*#[ \t]*STATUS:[ \t]*(?:scaffold|placeholder)\b", re.M | re.I)
|
|
|
|
# A quoted literal whose entire content is one marker word: the marker as data.
|
|
_MARKER_AS_DATUM_RE = re.compile(r"(['\"])[ \t]*(?:TODO|FIXME)[ \t]*\1", re.I)
|
|
|
|
|
|
def _split_doc_header(text: str) -> tuple[str, str]:
|
|
"""Split a test file into (documentation header, body).
|
|
|
|
Both file kinds in this tree open with a banner comment block that is PROSE about the
|
|
test: the Given/When/Then, the incident being closed, the mandatory "what this does NOT
|
|
prove" list (tests/README.md, L0 authoring rule 1). Prose necessarily discusses
|
|
placeholder idioms when the test hunts them, so an incidental mention there is
|
|
documentation, not a placeholder body.
|
|
|
|
The body starts at the first line that is neither blank nor a `#` comment:
|
|
*.sh -> `set -euo pipefail` (the shebang is a comment)
|
|
*.yaml -> `apiVersion: chainsaw.kyverno.io/v1alpha1`
|
|
Comments BELOW that point are body, deliberately: a scaffold marks a missing fixture
|
|
with an inline comment (`l3-fault-injection/alerting`: `file: fixtures/smtp-sink.yaml
|
|
# TODO(T-4.1): hermetic SMTP capture sink`), and that must keep counting.
|
|
"""
|
|
lines = text.splitlines(keepends=True)
|
|
for i, ln in enumerate(lines):
|
|
s = ln.strip()
|
|
if s and not s.startswith("#"):
|
|
return "".join(lines[:i]), "".join(lines[i:])
|
|
return text, "" # no executable line at all: it is all header, body is empty
|
|
|
|
|
|
def scaffold_marker(text: str) -> str:
|
|
"""Why `text` declares itself a placeholder body, or "" if it does not.
|
|
|
|
Returns a short human-readable reason (used verbatim in the hygiene report) so the
|
|
defect is actionable rather than a bare path.
|
|
"""
|
|
head, body = _split_doc_header(text)
|
|
m = _STATUS_SCAFFOLD_RE.search(head) or _STATUS_SCAFFOLD_RE.search(body)
|
|
if m:
|
|
return m.group(0).strip().lstrip("#").strip()[:80]
|
|
# Blank out markers held as pure data before looking for a marker that has a message.
|
|
m2 = _MARKER_RE.search(_MARKER_AS_DATUM_RE.sub('""', body))
|
|
if m2:
|
|
line = body[:m2.start()].count("\n")
|
|
snippet = body.splitlines()[line].strip()
|
|
return f"placeholder marker in body: {snippet[:80]}"
|
|
return ""
|
|
|
|
|
|
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 test DECLARES itself a placeholder — a `# STATUS: scaffold`
|
|
header line, or a TODO/FIXME marker carrying a message in the
|
|
body (see `scaffold_marker` for the rule and the corpus it was
|
|
derived from) — *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
|
|
scaffold declaration.
|
|
executable-untagged no scaffold declaration 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 = ""
|
|
scripts: list[tuple[str, str]] = []
|
|
for script in ("render-check.sh", "check.sh"):
|
|
sp = p.parent / script
|
|
if sp.is_file():
|
|
body = sp.read_text()
|
|
extra += body
|
|
scripts.append((script, body))
|
|
declared = bool(_EXECUTABLE_RE.search(text))
|
|
# Each file is classified on ITS OWN structure — the yaml and the script have different
|
|
# header blocks, so they must not be concatenated before the header/body split.
|
|
reason = scaffold_marker(text)
|
|
for script, body in scripts:
|
|
if reason:
|
|
break
|
|
r = scaffold_marker(body)
|
|
reason = f"{script}: {r}" if r else ""
|
|
bm = _BLOCKED_RE.search(text)
|
|
has_body = bool(extra) or bool(re.search(r"^\s+- (script|assert|apply):", text, re.M))
|
|
|
|
if reason:
|
|
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 reason),
|
|
"scaffold_reason": reason,
|
|
"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"]],
|
|
# Same set, with WHY each one was called a placeholder body, so the fix is obvious
|
|
# from the report instead of requiring the reader to re-derive the rule.
|
|
"annotation_conflict_reasons": {t["path"]: t["scaffold_reason"]
|
|
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("- tests annotated executable but declaring a placeholder body: "
|
|
+ (", ".join(f"`{k}` ({v})"
|
|
for k, v in sorted(h["annotation_conflict_reasons"].items()))
|
|
if h["annotation_conflicts"] else "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 | a `# STATUS: scaffold` header line, or a "
|
|
"TODO/FIXME marker CARRYING A MESSAGE in the body (a marker held as a bare "
|
|
"quoted datum — a placeholder detector's own vocabulary — is not one); "
|
|
"either outranks the `executable` annotation, and 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")
|
|
|
|
# === THE BUG THIS DETECTOR WAS REWRITTEN FOR ==============================
|
|
# A whole-file `"TODO" in text` mis-classified `l0-static/placeholder-guard-
|
|
# coupling` — a 32-assertion executable check whose SUBJECT is placeholder
|
|
# detection, so it must name the idioms it hunts — as a scaffold, and through it
|
|
# reported the SAFETY-CRITICAL R-STATE-03 as `exec=1 scaffold=1`. These
|
|
# assertions pin BOTH directions: a marker held as DATA must not downgrade, and a
|
|
# real placeholder body must still be caught. Both halves are load-bearing; the
|
|
# second is what stops this fix from degrading into "always executable".
|
|
yml = ("metadata:\n annotations:\n"
|
|
" conformance.home.k3s/executable: \"true\"\n"
|
|
"spec:\n steps:\n - script:\n"
|
|
" content: bash ./check.sh\n")
|
|
f.write_text(yml)
|
|
detector = td2 / "check.sh"
|
|
# Shaped exactly like the real check: a prose header that discusses the idioms,
|
|
# then executable code holding them as a vocabulary list.
|
|
detector.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
"# L0 static check — placeholder vocabulary\n"
|
|
"# * SEED — loud-placeholder idioms (TBD, TODO, FIXME, PLACEHOLDER)\n"
|
|
"# STATUS: EXECUTABLE. A pure offline assertion script.\n"
|
|
"set -euo pipefail\n"
|
|
"python3 - <<'EOF'\n"
|
|
"SEED = [\"TBD\", \"TODO\", \"FIXME\", \"PLACEHOLDER\"]\n"
|
|
"assert any(w in open('/etc/hostname').read().upper() for w in SEED) is False\n"
|
|
"EOF\n")
|
|
c = classify_test({"name": "s", "path": str(f), "layer": "L0"})
|
|
check("a marker in HEADER PROSE does not downgrade an executable check",
|
|
c["state"] == "executable" and not c["annotation_conflict"])
|
|
check("a marker held as a bare quoted DATUM does not downgrade (the R-STATE-03 bug)",
|
|
c["scaffold_reason"] == "")
|
|
|
|
# ... and the protection is intact: a real placeholder body is still caught, in
|
|
# each of the three shapes the corpus actually uses.
|
|
detector.write_text("#!/usr/bin/env bash\nset -euo pipefail\n"
|
|
"echo \"TODO: assert the PVC is bound\"\n")
|
|
c = classify_test({"name": "s", "path": str(f), "layer": "L0"})
|
|
check("a TODO carrying a message in check.sh IS still a scaffold",
|
|
c["state"] == "scaffold" and c["annotation_conflict"])
|
|
check("the scaffold reason names the file and the offending line",
|
|
c["scaffold_reason"].startswith("check.sh:")
|
|
and "TODO: assert the PVC is bound" in c["scaffold_reason"])
|
|
detector.write_text("#!/usr/bin/env bash\nset -euo pipefail\n"
|
|
"# TODO(T-4.1) write the hermetic sink fixture\ntrue\n")
|
|
check("a TODO(T-x.y) roadmap marker in an inline body comment IS still a scaffold",
|
|
classify_test({"name": "s", "path": str(f), "layer": "L0"})["state"]
|
|
== "scaffold")
|
|
detector.write_text("#!/usr/bin/env bash\nset -euo pipefail\n"
|
|
"echo \"FIXME the extraction before trusting this\"\n")
|
|
check("an unpunctuated FIXME carrying trailing prose IS still a scaffold",
|
|
classify_test({"name": "s", "path": str(f), "layer": "L0"})["state"]
|
|
== "scaffold")
|
|
# The dedicated declaration slot outranks a body that would otherwise look real —
|
|
# this is how all three genuine scaffolds in the tree declare themselves.
|
|
detector.unlink()
|
|
f.write_text("# L3 FAULT-INJECTION SCAFFOLD — x\n"
|
|
"# STATUS: scaffold — needs T-1.4 (Flux bootstrapped).\n" + yml)
|
|
c = classify_test({"name": "s", "path": str(f), "layer": "L3"})
|
|
check("a `# STATUS: scaffold` header declaration IS a scaffold, body notwithstanding",
|
|
c["state"] == "scaffold" and c["annotation_conflict"])
|
|
check("`# STATUS: EXECUTABLE` is not read as a scaffold declaration",
|
|
scaffold_marker("# STATUS: EXECUTABLE. `check.sh` is a pure offline "
|
|
"assertion script; covers no scaffold.\nset -e\n") == "")
|
|
|
|
# ---------------------------------------------------------------- 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(
|
|
"[status] STRICT: tests annotated executable that declare a placeholder "
|
|
"body: " + "; ".join(f"{k} ({v})" for k, v in
|
|
sorted(h["annotation_conflict_reasons"].items()))
|
|
+ "\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())
|