Files
ersatztv/scripts/decisions_validate.py
T
timothy 1af65b7bee feat(610): budget counts PROSE, excluding YAML frontmatter
The line budget exists to bound how much narrative a reader or agent must get
through. Under the split each record carries ~11 frontmatter lines plus two
fences -- 1789 lines across 166 records -- which are the structured restatement
of what used to be one dense backtick line. Counting them inflates the metric
without any new knowledge being added.

Stated plainly because it flatters the number: this is a change of METRIC, not a
consolidation. It re-measures the same corpus, it does not shrink it. Whole-file
counting put the migrated corpus at 6837 against a 5600 budget; prose-only puts
the same content at ~5048. The consolidation work is still worth doing -- it is
simply no longer being signalled by a warning that was partly measuring
punctuation.

Inert pre-migration: no legacy file has frontmatter, so the branch is never
taken and today's number is unchanged.
2026-07-25 19:08:58 +02:00

477 lines
22 KiB
Python

#!/usr/bin/env python3
"""Validate ErsatzTV decision-record lifecycle invariants.
Replaces the line-level append-only guard (ersatztv#303 H9) with lifecycle checks that preserve its
spirit — rationale is never silently rewritten or deleted; every history touch is deliberate and
reviewable. A `Decisions-Edit: yes` commit trailer is kept ONLY for rationale-prose edits; routine
lifecycle metadata writes are marker-free. Fail-open on tooling trouble (missing refs, git errors,
parse issues), matching the old guard — with ONE deliberate exception: resolving the edit marker
itself fails CLOSED (`_edit_trailer_armed`), because a fail-open marker lookup is just #609 again.
"""
from __future__ import annotations
import argparse
import re
import subprocess
import sys
from datetime import date
from pathlib import Path
import scripts.decisions_lib as dl # noqa: E402 (run with PYTHONPATH=. or as module)
SKIP_HEADINGS = {"Index", "Active catalog", "Contents"}
# `signals` is required alongside the lifecycle fields: the `**Signals:**` line (plus `key:`) is what
# MemPalace's keyword recall matches on when surfacing a decision, so a record without it ingests with
# weak recall metadata and produces confident false-negatives for the "MemPalace to find, file to
# confirm" workflow (ersatztv#545).
REQUIRED_META = ("key", "status", "since", "supersedes", "superseded_by", "signals")
# The rationale-edit marker is a git TRAILER, not a bare substring anywhere in the commit range.
# ersatztv#609: the original `[decisions-edit]` substring test armed on ANY commit message containing
# the literal string — including prose *about* the mechanism, and including a message explaining why
# no token was needed — which silently disabled all three body-diff comparisons while the job still
# reported green. Git recognises a trailer only in a message's final trailer block, so discussing the
# marker mid-message — which is what actually happened — can no longer arm anything. It is NOT a total
# immunity: a quoted example that IS the final paragraph of a NON-MERGE commit message parses as a
# real trailer and does arm (merge commits are excluded — see `_edit_trailer_armed`). See
# `ci.decisions-edit-trailer` for that residual, stated rather than papered over.
EDIT_TRAILER = "Decisions-Edit"
_EDIT_TRAILER_AFFIRMATIVE = frozenset({"yes", "true", "1"})
_LEGACY_EDIT_TOKEN = "[decisions-edit]" # noqa: S105 (a commit-message marker, not a credential)
# `date.fromisoformat` is NOT a YYYY-MM-DD validator: on Python >= 3.11 it also accepts ISO basic
# format ("20270101") and week dates ("2027-W01-1"), and which of those parse depends on the
# interpreter version — so the same corpus could validate differently on a dev machine and on the
# runner (pr-checks.yml pins only `python-version: '3.x'`). Gate on the extended calendar form first.
# The catalog's Review-due section additionally sorts on the raw STRING, which is only equivalent to
# sorting by date because this regex forces a fixed-width zero-padded form.
_STALE_AFTER_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
def _parse_stale_after(value: str | None) -> date | None:
"""`stale-after` as a date, or None if absent, empty, or malformed.
Callers must NOT use a None return to mean "absent" — an empty or malformed value returns None
too. Distinguish on the raw field with `is not None` (absent stays None; present-but-empty is "").
"""
if not value or not _STALE_AFTER_RE.match(value.strip()):
return None
try:
return date.fromisoformat(value.strip())
except ValueError:
return None # e.g. 2027-02-30 — well-shaped but not a real date
def stale_records(records, today: date) -> list[tuple[str, str]]:
"""(heading, stale-after) for ACTIVE records that are past due — today >= stale-after.
OKF v0.2 semantics: an absolute date, no TTL. Deliberately NOT part of `validate()` — a record
going stale is the passage of time, not a defect in the commit under test, so it must never fail
a PR that didn't touch it. main() reports these as a non-blocking notice.
"""
out = []
for r in records:
if r.status != "active":
continue
d = _parse_stale_after(r.stale_after)
if d is not None and today >= d:
out.append((r.heading, r.stale_after))
return sorted(out)
def _key_of(ref: str | None) -> str | None:
if not ref or ref == "none":
return None
return ref.split("@", 1)[0].strip()
def validate(
records,
*,
archive_keys,
catalog_ok,
budget_ok,
removed,
rewritten,
archive_records=None,
demoted=(),
) -> list[str]:
errs: list[str] = []
archive_records = archive_records or []
decision_recs = [r for r in records if r.heading not in SKIP_HEADINGS and r.status != "legacy-unmigrated"]
active_by_key: dict[str, int] = {}
known_keys = set(archive_keys)
for r in decision_recs:
if r.status not in dl.STATUSES:
errs.append(f"{r.heading}: invalid status {r.status!r}")
if not r.key or not dl.KEY_RE.match(r.key):
errs.append(f"{r.heading}: bad key format {r.key!r}")
continue
known_keys.add(r.key)
for f in REQUIRED_META:
if getattr(r, f) in (None, ""):
errs.append(f"{r.heading}: missing required metadata {f}")
# `stale-after` is OPTIONAL, but a malformed one is a defect in the commit that wrote it —
# it would silently never fire. Absent is fine; unparseable is not.
if r.stale_after is not None and _parse_stale_after(r.stale_after) is None:
errs.append(f"{r.heading}: stale-after {r.stale_after!r} is not a YYYY-MM-DD date")
if r.status == "active":
active_by_key[r.key] = active_by_key.get(r.key, 0) + 1
if r.status in ("superseded", "retired"):
errs.append(
f"{r.heading}: status {r.status} but still in the active set — relocate to docs/decisions/archive/"
)
n_meta = dl.metadata_line_count(r)
if n_meta > 1:
errs.append(f"{r.heading}: {n_meta} metadata blocks found (expected 1) — duplicate metadata block")
for r in archive_records:
if r.status == "active":
errs.append(f"{r.heading}: active record must not live under docs/decisions/archive/")
# An archived record's staleness is moot, but a malformed date there is still a typo worth
# catching — and checking both wings keeps the field's contract symmetric.
if r.stale_after is not None and _parse_stale_after(r.stale_after) is None:
errs.append(f"{r.heading}: stale-after {r.stale_after!r} is not a YYYY-MM-DD date")
for key, n in active_by_key.items():
if n > 1:
errs.append(f"key {key!r}: more than one active record ({n})")
# by_key map across BOTH wings, for reciprocity checks (existence-only checks still use
# known_keys, which additionally includes keys the caller only knows about via archive_keys).
by_key: dict[str, dl.Record] = {}
for r in decision_recs + list(archive_records):
if r.key:
by_key[r.key] = r
# reciprocal supersession (successor/predecessor may live in active OR archive). Runs over the
# union of active + archive records so a twice-reversed decision (an archived record whose
# superseded-by points to ANOTHER archived record) is checked from both sides too.
for r in decision_recs:
if r.status == "active" and r.superseded_by not in (None, "", "none"):
errs.append(f"{r.heading}: active record cannot already be superseded (superseded-by set)")
for r in decision_recs + list(archive_records):
sk = _key_of(r.superseded_by)
if sk:
if sk not in known_keys:
errs.append(f"{r.heading}: superseded-by points to unknown key {sk!r}")
else:
b = by_key.get(sk)
if b is not None and r.key and _key_of(b.supersedes) != r.key:
errs.append(f"{r.heading}: superseded-by {sk} but {sk} does not point back (supersedes)")
pk = _key_of(r.supersedes)
if pk:
if pk not in known_keys:
errs.append(f"{r.heading}: supersedes points to unknown key {pk!r}")
else:
a = by_key.get(pk)
if a is not None and r.key and _key_of(a.superseded_by) != r.key:
errs.append(f"{r.heading}: supersedes {pk} but {pk} does not point back (superseded-by)")
for h in removed:
errs.append(f"record removed from the active set without an archive copy: {h!r}")
for h in rewritten:
errs.append(f"rationale prose of {h!r} changed without a '{EDIT_TRAILER}: yes' commit trailer")
for h in demoted:
errs.append(f"{h}: migrated record demoted to legacy-unmigrated (metadata block removed)")
if not catalog_ok:
errs.append("docs/decisions/README.md active catalog is stale — run build_decisions_catalog.py")
if not budget_ok:
# non-blocking: reported as a warning by the caller (main()), never added here.
pass
return errs
# ---- git helpers (all fail-open: return neutral values on any error) ----
def _run(args: list[str]) -> str | None:
try:
r = subprocess.run(args, capture_output=True, text=True, timeout=30)
except (OSError, subprocess.SubprocessError):
return None
return r.stdout if r.returncode == 0 else None
def _merge_base(base: str, head: str) -> str | None:
out = _run(["git", "merge-base", base, head])
return out.strip() if out else None
def _active_paths_at(ref: str) -> list[str]:
"""Active decision files at a ref: decisions.md + docs/decisions/*.md minus README/archive/migration-map."""
out = _run(["git", "ls-tree", "-r", "--name-only", ref, "docs/decisions.md", "docs/decisions/"])
if out is None:
return []
paths = []
for p in out.splitlines():
if not p.endswith(".md"):
continue
if p.startswith("docs/decisions/archive/") or Path(p).name in dl._NON_DECISION_FILES:
continue
paths.append(p)
return paths
def _records_at(ref: str, paths: list[str]):
"""{heading: Record} across the given paths at a ref."""
by_heading = {}
for p in paths:
blob = _run(["git", "show", f"{ref}:{p}"])
if blob is None:
continue
for rec in dl.parse_text(blob, Path(p)):
if rec.heading not in SKIP_HEADINGS:
by_heading[rec.heading] = rec
return by_heading
def _archive_records_at(ref: str):
out = _run(["git", "ls-tree", "-r", "--name-only", ref, "docs/decisions/archive/"])
paths = [p for p in (out or "").splitlines() if p.endswith(".md")]
by_heading = {}
for p in paths:
blob = _run(["git", "show", f"{ref}:{p}"])
if blob is None:
continue
for rec in dl.parse_text(blob, Path(p)):
by_heading[rec.heading] = rec
return by_heading
def _rationale(rec) -> str:
"""Record body with the contiguous top metadata block stripped, whitespace-normalized.
Only the block from the first non-blank line (when it is the `key:` meta line) up to the first
following blank line is metadata. A **Rule:**/**Signals:**/**Mechanics:**/`key:` line appearing
later in rationale prose is prose, not metadata (mirrors decisions_lib's contiguous-block rule).
"""
lines = rec.body.splitlines()
start = 0
while start < len(lines) and not lines[start].strip():
start += 1
if start < len(lines) and lines[start].strip().startswith("`key:"):
end = start + 1
while end < len(lines) and lines[end].strip():
end += 1
rest = lines[:start] + lines[end:]
else:
rest = lines
return "\n".join(s for s in (ln.strip() for ln in rest) if s)
def _edit_trailer_armed(mb: str, head: str) -> bool:
"""True when some NON-MERGE commit in `mb..head` carries an affirmative `Decisions-Edit:` trailer.
A non-affirmative value (`no`, `not-needed`, …) deliberately does NOT arm the exemption: the
marker must be an explicit yes, so a trailer written to *record that no edit was intended* can
never disable the guard — the ersatztv#609 failure mode, one abstraction level up.
This is the one git helper here that deliberately does NOT fail open (see the module docstring):
on a git error it returns False, leaving the guard ON. A lost marker costs a contributor one
clear error message and a re-push; a lost guard costs a silent history rewrite in the PR that
most needed policing. Failing open here would reintroduce #609 through a different door.
`--no-merges` skips merge commits: on a `pull_request` event `actions/checkout` lands on a
synthetic merge commit whose body is composed by the forge from the PR description, not by the
author, so a trailer parsed out of it was never a deliberate marker.
This does cost a real (if narrow) false negative, and the trade is deliberate. Merging main into
a PR branch is discouraged by convention but NOT mechanically blocked — `prepush-rebase-check.sh`
only refuses a branch that is BEHIND origin/main, and a merge makes origin/main an ancestor, so
the push is allowed. An author who resolves a rationale conflict in such a merge and puts the
ONLY marker on that merge commit gets their legitimate rewrite rejected. That failure is LOUD and
costs one extra commit carrying the trailer; honoring forge-composed merge bodies would instead
disable the guard SILENTLY, which is the #609 failure itself. Prefer the loud one.
`unfold` joins RFC-822 continuation lines before the value is compared. Without it, a folded
`Decisions-Edit: no\\n yes` yields two lines and the continuation ` yes` arms the exemption on
its own — the value the author actually wrote is `no`.
"""
fmt = f"--format=%(trailers:key={EDIT_TRAILER},valueonly,unfold)"
out = _run(["git", "log", "--no-merges", fmt, f"{mb}..{head}"])
if out is not None and any(ln.strip().lower() in _EDIT_TRAILER_AFFIRMATIVE for ln in out.splitlines()):
return True
# Retired-form nudge: without it, a contributor using the pre-#609 habit gets a bare "prose
# changed" failure and no hint that the marker's FORM (not their intent) is what changed.
if _LEGACY_EDIT_TOKEN in (_run(["git", "log", "--no-merges", "--format=%B", f"{mb}..{head}"]) or "").lower():
print(
f"::warning::decisions-validate: the commit range carries the retired "
f"{_LEGACY_EDIT_TOKEN} marker but no '{EDIT_TRAILER}: yes' trailer — since ersatztv#609 "
"the rationale-edit marker is a git trailer, and the substring no longer exempts anything",
file=sys.stderr,
)
return False
def _diff_findings(base: str, head: str) -> tuple[list[str], list[str], list[str]]:
"""(removed, rewritten, demoted) between merge-base(base,head) and head. Fail-open → ([], [], [])."""
mb = _merge_base(base, head)
if not mb:
print(
f"::warning::decisions-validate: could not resolve merge-base({base},{head}); no-vanish/body-diff skipped",
file=sys.stderr,
)
return [], [], []
token = _edit_trailer_armed(mb, head)
base_active = _records_at(mb, _active_paths_at(mb))
head_active = _records_at(head, _active_paths_at(head))
head_archive = _archive_records_at(head)
base_archive = _archive_records_at(mb)
gone = set(base_active) - set(head_active)
removed = sorted(h for h in gone if h not in head_archive)
# an archive record present at base must not vanish entirely (neither wing has it at head)
archive_gone = set(base_archive) - set(head_archive)
removed += sorted(h for h in archive_gone if h not in head_active)
rewritten: list[str] = []
if not token:
# surviving records whose rationale prose changed
for h in set(base_active) & set(head_active):
if _rationale(base_active[h]) != _rationale(head_active[h]):
rewritten.append(h)
# archived records must body-match their base active copy (no laundering rewrites via archive)
for h in gone & set(head_archive):
if _rationale(base_active[h]) != _rationale(head_archive[h]):
rewritten.append(h)
# archive records that survive in the archive wing: rationale must not be rewritten either
for h in set(base_archive) & set(head_archive):
if _rationale(base_archive[h]) != _rationale(head_archive[h]):
rewritten.append(h)
# migrated (had a key) at base, demoted to legacy-unmigrated (no key, or key changed) at head,
# same heading, still present in the active set.
demoted: list[str] = []
for h in set(base_active) & set(head_active):
base_key = base_active[h].key
head_key = head_active[h].key
if base_key and base_key != head_key:
demoted.append(h)
return sorted(set(removed)), sorted(set(rewritten)), sorted(demoted)
def _archive_keys() -> set[str]:
keys: set[str] = set()
if dl.ARCHIVE_DIR.exists():
for f in dl.ARCHIVE_DIR.rglob("*.md"): # rglob: archive is nested by area after #610
for r in dl.parse_file(f):
if r.key:
keys.add(r.key)
return keys
def _budget_total() -> int:
"""Lines of PROSE in the active corpus — YAML frontmatter excluded.
The budget exists to bound how much narrative a reader/agent must get through, so it counts
prose, not metadata. Under the #610 split each record carries ~11 frontmatter lines plus two
fences (1789 lines across 166 records), which are the structured restatement of what used to be
one dense backtick line — counting them would inflate the metric without any new knowledge
being added.
NOTE this is a change of METRIC, not a consolidation: it re-measures the same corpus, it does
not shrink it. At the time of the split, whole-file counting put the corpus a few hundred lines
over budget; prose-only puts the same content comfortably under. The consolidation work is still
worth doing — it is simply no longer being signalled by a warning that was partly measuring
punctuation. Pre-migration this function is inert: no legacy file has frontmatter.
"""
total = 0
for f in dl.active_files():
if not f.exists():
continue
text = f.read_text(encoding="utf-8")
lines = text.splitlines()
if dl.has_frontmatter(text):
end = next((i for i, ln in enumerate(lines[1:], start=1) if ln.rstrip() == "---"), None)
if end is not None:
lines = lines[end + 1 :]
total += len(lines)
cat = dl.TOPIC_DIR / "README.md"
if cat.exists():
total += len(cat.read_text(encoding="utf-8").splitlines())
return total
def _budget_ok(limit: int) -> bool:
return _budget_total() <= limit
def _catalog_ok() -> bool:
try:
import scripts.build_decisions_catalog as bc # pyright: ignore[reportMissingImports]
except Exception:
return True # fail-open; the CI job also runs an independent --check step
want = bc.render_catalog(dl.all_active_records())
cat = dl.TOPIC_DIR / "README.md"
have = cat.read_text(encoding="utf-8") if cat.exists() else ""
return want.strip() == have.strip()
def main(argv=None) -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--base")
ap.add_argument("--head")
# Aggregate active-corpus budget. Re-baselined 2026-07-21 (#542): 4800 -> 5600. The corpus grew
# by ~530 lines because 31 workflow/CI/review rules were MOVED into it from
# docs/handoffs/chicorytv-issue-queue.md, which was their only copy. That is the corpus doing its
# job, not drift — the knowledge existed already, it just wasn't retrievable. Do not raise this
# again to accommodate genuinely new records; that is what the warning is for.
ap.add_argument("--budget", type=int, default=5600)
args = ap.parse_args(argv)
records = dl.all_active_records()
archive_records = []
if dl.ARCHIVE_DIR.exists():
for f in dl.ARCHIVE_DIR.rglob("*.md"): # rglob: archive is nested by area after #610
archive_records += dl.parse_file(f)
removed, rewritten, demoted = _diff_findings(args.base, args.head) if args.base and args.head else ([], [], [])
budget_ok = _budget_ok(args.budget)
errs = validate(
records,
archive_keys=_archive_keys(),
catalog_ok=_catalog_ok(),
budget_ok=budget_ok,
removed=removed,
rewritten=rewritten,
archive_records=archive_records,
demoted=demoted,
)
if not budget_ok:
print(
f"::warning::decisions-validate: aggregate active-corpus is {_budget_total()} lines "
f"(budget {args.budget}) — schedule a consolidation",
file=sys.stderr,
)
stale = stale_records(records, date.today())
if stale:
listed = "; ".join(f"{h} (stale-after {d})" for h, d in stale)
print(
f"::notice::decisions-validate: {len(stale)} active record(s) past their stale-after date "
f"— re-confirm or extend: {listed}",
file=sys.stderr,
)
unmigrated = [r for r in records if r.status == "legacy-unmigrated" and r.heading not in SKIP_HEADINGS]
if unmigrated:
print(f"::notice::{len(unmigrated)} legacy-unmigrated decision record(s) remain (must trend to 0).")
if errs:
for e in errs:
print(f"decisions-validate: {e}", file=sys.stderr)
return 1
print("decisions-validate: OK")
return 0
if __name__ == "__main__":
raise SystemExit(main())