#!/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 = dl.SKIP_HEADINGS # single source of truth # `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") # PATH <-> KEY (#610). In the split layout the filename is derived from the key, which is what # makes one-active-record-per-key a filesystem property rather than a check. That only holds if # the two cannot drift, so enforce the correspondence directly. for r in decision_recs + list(archive_records): if not r.key: continue src = Path(r.source) if dl.RECORDS_DIR not in src.parents and dl.ARCHIVE_DIR not in src.parents: continue # legacy multi-record file; the rule applies to the split layout only area, _, topic = r.key.partition(".") if src.parent.name != area or src.stem != topic: errs.append(f"{r.heading}: key {r.key!r} does not match its path {src.parent.name}/{src.name}") 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]): """{key: Record} across the given paths at a ref. Keyed by `key`, NOT by heading (#610). Heading-keying made a pure RENAME look like a removal plus an unrelated addition, so retitling a record failed CI as an "unlogged removal" — a trap that has bitten before. `key` is the record's identity and is stable across a retitle, a move between files, and the legacy->frontmatter format change, which is also what lets this compare correctly straight through the migration commit. A record with no key (legacy-unmigrated) falls back to its heading, prefixed so it cannot collide with a real dotted key. """ by_key = {} 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 in SKIP_HEADINGS: continue by_key[rec.key or f"heading:{rec.heading}"] = rec return by_key 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_key = {} 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)): # Filter SKIP_HEADINGS exactly as _records_at does. Without it the generated # "Records formerly in this file" section in each stripped archive file parses as a # pseudo-record; they then collide by heading across files, and any later PR that # touches those generated bullets trips a false "rationale prose changed without a # trailer" failure. if rec.heading not in SKIP_HEADINGS: by_key[rec.key or f"heading:{rec.heading}"] = rec return by_key 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 → ([], [], []). Records are MATCHED by `key` and REPORTED by heading. Matching by key means a retitle is a retitle rather than a removal-plus-addition (#610) — heading-keying made renaming a heading fail CI as an "unlogged removal", a trap that has bitten before — and it is what lets the comparison work straight through the legacy→frontmatter migration, since `key` survives both the format change and the move between files. Headings are still what appears in the error, because a key alone is not enough for a human to find the record. """ 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) def name(rec) -> str: return rec.heading gone = set(base_active) - set(head_active) removed = [name(base_active[k]) for k in sorted(gone) if k 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 += [name(base_archive[k]) for k in sorted(archive_gone) if k not in head_active] rewritten: list[str] = [] if not token: # surviving records whose rationale prose changed for k in set(base_active) & set(head_active): if _rationale(base_active[k]) != _rationale(head_active[k]): rewritten.append(name(head_active[k])) # archived records must body-match their base active copy (no laundering rewrites via archive) for k in gone & set(head_archive): if _rationale(base_active[k]) != _rationale(head_archive[k]): rewritten.append(name(head_archive[k])) # archive records that survive in the archive wing: rationale must not be rewritten either for k in set(base_archive) & set(head_archive): if _rationale(base_archive[k]) != _rationale(head_archive[k]): rewritten.append(name(head_archive[k])) # Demotion: a record that HAD a key at base and lost it at head. Under key-matching that shows # up as its key vanishing while a `heading:`-fallback entry appears for the same heading, so it # is detected on the heading axis rather than the key axis. head_by_heading = {r.heading: r for r in head_active.values()} demoted: list[str] = [] for k in gone: rec = base_active[k] if not rec.key: continue successor = head_by_heading.get(rec.heading) if successor is not None and successor.key != rec.key: demoted.append(rec.heading) # a demoted record is not also "removed" — report the specific fault, not both removed = [h for h in removed if h not in set(demoted)] return sorted(set(removed)), sorted(set(rewritten)), sorted(set(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())