Files
ersatztv/scripts/decisions_validate.py
T
timothy 990d32f31a fix(609): anchor and bound the git version probe
Both round-4 findings, both in the version regex introduced in round 3. Both
reproduced against the old code and confirmed closed.

1. The regex was UNANCHORED, so the first dotted number anywhere in the output
   won. `wrapper 2026.1; git version 2.20.1` read as 2026.1 -> True, enabling
   trailer parsing on a git that cannot expand the atom, whose verbatim echo then
   reads as a non-empty trailer and FALSELY ARMS. Now anchored to the canonical
   `git version X.Y` prefix.

2. Digits were unbounded, so a pathological version string raised ValueError
   instead of returning the documented safe False -- Python refuses int()
   conversion of a literal over 4300 digits. Digits are now bounded to 5 each,
   plus a try/except that the bounded regex should make unreachable.

Adds seven probe cases: the wrapper-prefix and multiline-shim strings, Apple git,
an rc suffix, a three-digit major, and the 5000-digit pathological input.
2026-07-25 18:25:26 +02:00

502 lines
21 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. `[decisions-edit]` is kept ONLY for rationale-prose edits (see body-diff below); routine
lifecycle metadata writes are token-free. Fail-open on tooling trouble (missing refs, git errors,
parse issues), matching the old guard.
"""
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")
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 the {EDIT_TOKEN} token")
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
_TRAILER_KEY = "Decisions-Edit"
# NUL framing, not \x1f/\x1e. git forbids NUL inside a commit message, so it is the ONLY separator
# a crafted (or merely odd) message cannot inject: a subject containing a literal \x1f would
# otherwise be split at the wrong place and its tail read as a trailer — falsely ARMING the token,
# which is the dangerous direction because arming silently disables the guard.
_NUL = "\x00" # what we SPLIT the output on
_NUL_FMT = "%x00" # what goes in the --format string; a literal NUL in argv raises ValueError
_FIELDS_PER_COMMIT = 3
# git < 2.22 cannot expand %(trailers:key=…,valueonly) and echoes the atom verbatim with exit 0.
# Detect that by VERSION, not by sniffing the output for the atom text: a value-based sentinel can
# collide with a legitimate trailer whose value happens to be that exact string, silently discarding
# a real token (#609 round 3).
_MIN_TRAILER_GIT = (2, 22)
# ANCHORED to the canonical `git version X.Y` prefix, and digits are BOUNDED. Unanchored, the first
# dotted number anywhere wins — `wrapper 2026.1; git version 2.20.1` would read as 2026.1 and enable
# trailers on a git that cannot expand them, which is the falsely-arming direction. Unbounded digits
# are also unsafe: Python raises ValueError converting an int literal over 4300 digits.
_GIT_VERSION_RE = re.compile(r"^git version (\d{1,5})\.(\d{1,5})")
def token_armed_in(subject: str, trailer: str) -> bool:
"""Whether ONE commit legitimately carries the edit token.
Recognized in exactly two places:
* the commit SUBJECT (first line) — the established form; all 38 historical uses carry the
token in the subject (37 appended, `docs(434)` mid-subject), so this stays compatible.
* a `Decisions-Edit:` git trailer — the forward-looking form, which can carry a reason.
Deliberately NOT the rest of the body. The old check was a bare substring over the whole
message, so a commit that merely *described* the token armed it and silently suppressed every
rationale-rewrite comparison for the PR — a gate reporting green while doing nothing, and it
bit hardest in a PR that hand-resolved a conflict inside the corpus the guard protects (#609).
"""
if EDIT_TOKEN.lower() in subject.lower():
return True
return bool(trailer.strip())
def subject_of(body: str) -> str:
"""The commit's true first line — literally line 1, even when line 1 is blank.
NOT git's `%s`: that atom is the first *paragraph*, joining consecutive non-blank lines with
spaces, so `fix: harmless\\nprose about [decisions-edit]\\n\\n…` would come back as one line
containing the token and arm it.
And NOT "the first non-blank line": `git commit --cleanup=verbatim` accepts a message beginning
with a blank line, and `%B` returns it raw. Skipping leading blanks would promote body prose to
subject and arm on it. An empty first line yields "", which arms nothing — the safe direction.
"""
return body.split("\n", 1)[0].strip()
def _git_supports_trailer_atom() -> bool:
"""Whether this git expands `%(trailers:key=…,valueonly)` (2.22+). Unknown → False.
Unknown resolving to False means trailers are ignored and matching falls back to the subject,
which is the safe direction: a genuine trailer-only token silently not arming is an annoyance,
whereas treating an unexpanded atom as a value would falsely arm and disable the guard.
"""
out = _run(["git", "--version"])
if not out:
return False
m = _GIT_VERSION_RE.match(out.strip())
if not m:
return False
try:
return (int(m.group(1)), int(m.group(2))) >= _MIN_TRAILER_GIT
except ValueError: # belt and braces; the bounded regex should make this unreachable
return False
def _token_armed(mb: str, head: str) -> bool:
"""True if ANY commit in mb..head carries the token. Anything unexpected → False (guard runs)."""
out = _run(
[
"git",
"log",
f"--format=%H{_NUL_FMT}%B{_NUL_FMT}%(trailers:key={_TRAILER_KEY},valueonly){_NUL_FMT}",
f"{mb}..{head}",
]
)
if out is None:
# On git trouble the token is NOT armed, so the guard still runs. Failing the other way
# would silently disable it — the bug this whole function exists to fix.
return False
trailers_usable = _git_supports_trailer_atom()
parts = out.split(_NUL)
if parts and not parts[-1].strip():
parts.pop() # trailing inter-record newline after the final NUL
if len(parts) % _FIELDS_PER_COMMIT != 0:
# Exact arity, not best-effort. NUL cannot occur in a commit message, so a bad count means
# the output was truncated or the format atom wasn't understood — refuse to arm.
return False
for i in range(0, len(parts), _FIELDS_PER_COMMIT):
body, trailer = parts[i + 1], parts[i + 2]
# On a git too old to expand the atom, ignore the trailer field entirely and fall back to
# subject-only matching. The SUBJECT is always honoured either way.
if token_armed_in(subject_of(body), trailer if trailers_usable else ""):
return True
return False
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 _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 = _token_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.glob("*.md"):
for r in dl.parse_file(f):
if r.key:
keys.add(r.key)
return keys
def _budget_total() -> int:
total = 0
for f in dl.active_files():
if f.exists():
total += len(f.read_text(encoding="utf-8").splitlines())
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.glob("*.md"):
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())