ersatztv/scripts/decisions_validate.py

#!/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}$")

# Per-record prose ceiling (#620). A module constant, not a bare argparse default, so the test that
# guards the calibration claim asserts against the SAME value the CLI uses and the two cannot drift.
RECORD_CEILING_DEFAULT = 60


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,
    removed,
    rewritten,
    archive_records=None,
    demoted=(),
    wing_faults=(),
) -> list[str]:
    errs: list[str] = []
    archive_records = archive_records or []
    # #621: structural "one keyed record per file". ERRORS, not warnings — a file under the record
    # wings that is not a record is a mistake by definition. Reported first because a file that
    # failed to parse is absent from `records`, so every downstream check below is silently
    # evaluating an incomplete corpus.
    errs += list(wing_faults)
    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")
    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 record_wing_files(records_dir: Path | None = None, archive_dir: Path | None = None) -> list[Path]:
    """Every file that MUST be exactly one keyed record.

    Scope is EVERY `*.md` under `records/**` and `archive/**` — including the files directly in
    `archive/`, which are scanned but may qualify for the stripped-index exemption applied in
    `record_wing_faults` (see there). Nothing is excluded by BASENAME.

    Excluding by BASENAME was a real hole: `_NON_DECISION_FILES` is `{README.md, migration-map.md,
    retrieval-eval.md}` — three TOPIC-dir names — and applying it to the wings meant a genuine
    record at `records/docs/retrieval-eval.md` was silently skipped. That path is not hypothetical:
    the path<->key rule forces key `docs.retrieval-eval` to live at exactly that filename, and
    `docs/decisions/retrieval-eval.md` is a real unmigrated file, i.e. a plausible migration target.
    Worse, `dl.active_files()` applies that filter only to the TOPIC_DIR glob, not to
    `RECORDS_DIR.rglob` — so such a file IS a corpus source while being exempt from the guard.

    So the exemption is by exact RELATIVE PATH, never by basename. The only entry is
    `archive/README.md`, a hand-written directory README that really does exist — an earlier
    version of this function excluded any wing-root `README.md` "since no such file exists today",
    which was simply false and would additionally have exempted a future `records/README.md`, i.e.
    reintroduced the very hole one directory over.
    """
    records_dir = dl.RECORDS_DIR if records_dir is None else records_dir
    archive_dir = dl.ARCHIVE_DIR if archive_dir is None else archive_dir
    exempt_paths = {archive_dir / "README.md"}
    files: list[Path] = []
    for wing in (records_dir, archive_dir):
        if wing.exists():
            files += sorted(p for p in wing.rglob("*.md") if p not in exempt_paths)
    return files


def _is_stripped_index(path: Path, archive_dir: Path, recs: list) -> bool:
    """True for a #610 stripped legacy topic file sitting DIRECTLY in `archive/`.

    Those five files (`api.md`, `scan.md`, `spa.md`, `startup.md`, `release-ci-governance.md`) are
    generated "Records formerly in this file" indexes — keyless by construction, and what keeps
    older date-based pointers resolvable. They are indexes, not records.

    The exemption is by IDENTITY, not by location. Exempting everything directly in `archive/`
    would leave that one directory unguarded: a new unparseable `archive/foo.md` would vanish
    silently — the same defect this guard exists to close, in the last place it isn't watched. So a
    file there is exempt only if it actually LOOKS like a stripped index: exactly one keyless
    record whose heading is one of the known generated ones (`dl.SKIP_HEADINGS`).
    """
    return (
        path.parent == archive_dir
        and len(recs) == 1
        and not recs[0].key
        and recs[0].heading in dl.SKIP_HEADINGS
    )


def record_wing_faults(records_dir: Path | None = None, archive_dir: Path | None = None) -> list[str]:
    """Structural check (#621): every file in the record wings parses to exactly ONE keyed record.

    Without this, a file the frontmatter reader cannot parse yields `[]` and simply VANISHES — no
    error, no warning, validator OK, catalog "up to date", record absent from the corpus. That is
    the corpus's own failure mode turned on itself: the one thing worse than a missing record is a
    missing record that reports success (cf. the #609 marker that printed OK while doing nothing and
    the #603 `stale-after` that silently never fired).

    An EXISTING record disappearing was already loud — the no-vanish diff check catches it. This
    closes the case the diff check structurally cannot see: a NEWLY ADDED record, where the author's
    own PR looks clean because there is no prior state to diff against.

    Making it path-driven rather than record-driven is the point: it converts a whole CLASS of
    reader limitations — parse-to-zero, parse-to-many, keyless — from silent to loud in one move,
    instead of enumerating the constructs we happen to know about today.

    It does NOT catch parse-to-WRONG: frontmatter that yields one keyed record with corrupted
    values. `rule: >-` followed by an UNINDENTED continuation containing a colon parses to a valid
    record whose `rule` is literally `>-`, plus a junk key from the continuation line. The junk-key
    check below is what catches that case; a mis-parse producing only known fields would still slip
    through, so this is a strong guard, not a total one.
    """
    records_dir = dl.RECORDS_DIR if records_dir is None else records_dir
    archive_dir = dl.ARCHIVE_DIR if archive_dir is None else archive_dir
    faults: list[str] = []
    files = record_wing_files(records_dir, archive_dir)

    # A wing that is absent or empty must be LOUD, not silently clean. Otherwise a partial checkout,
    # a renamed directory, or a bad monkeypatch turns the whole guard into a no-op that reports
    # success — which is the defect this function exists to close, applied to itself.
    # DELIBERATELY asymmetric: only the ACTIVE wing must be non-empty. A corpus with zero active
    # records is definitionally broken (and means the scan is measuring nothing, so a clean result
    # would be vacuous). An empty ARCHIVE is a perfectly normal state — it just means nothing has
    # been superseded or retired yet, which is true of any young repo and of a fresh clone before
    # the first supersession. Faulting on it would fail a correct corpus. Review asked for symmetry
    # here; the semantics genuinely differ, so this stays asymmetric with the reason stated.
    if not records_dir.exists() or not any(records_dir.rglob("*.md")):
        faults.append(
            f"{records_dir}: the active record wing is missing or contains no *.md files — refusing "
            f"to report a clean corpus from an empty scan."
        )

    for p in files:
        try:
            recs = dl.parse_file(p)
        except Exception as exc:  # unreadable/undecodable file is a fault, not a crash
            faults.append(f"{p}: could not be read as a decision record ({exc})")
            continue
        if _is_stripped_index(p, archive_dir, recs):
            continue
        if len(recs) != 1:
            faults.append(
                f"{p}: parsed to {len(recs)} records, expected exactly 1 — a file under the record "
                f"wings must be one record. Common cause: YAML the dependency-free frontmatter "
                f"reader does not accept (a block scalar such as `rule: >-` or `rule: |`, any "
                f"indented/nested structure, or unterminated `---` frontmatter). Put the value on "
                f"ONE line."
            )
            continue
        if not recs[0].key:
            faults.append(f"{p}: parsed to a record with no `key` — every record in the wings must be keyed.")
            continue
        # Junk keys: the reader accepted a line it should not have. The realistic source is a block
        # scalar whose continuation was read as its own `k: v` pair, which silently truncates the
        # real value (`rule` becomes `>-`). PyYAML REJECTS that input, so the hand reader is more
        # permissive than the writer and nothing else in the pipeline notices.
        unknown = _unknown_frontmatter_keys(p)
        if unknown:
            faults.append(
                f"{p}: frontmatter has unrecognized key(s) {sorted(unknown)} — a value was probably "
                f"split across lines (a block scalar continuation read as its own key), which "
                f"silently truncates the real value. Put each value on ONE line."
            )
    return faults


def _unknown_frontmatter_keys(path: Path) -> set[str]:
    """Frontmatter keys outside the known schema. Empty on any read/parse failure (reported elsewhere)."""
    try:
        text = path.read_text(encoding="utf-8")
    except Exception:
        return set()
    if not dl.has_frontmatter(text):
        return set()
    lines = text.splitlines()
    end = next((i for i, ln in enumerate(lines[1:], start=1) if ln.rstrip() == "---"), None)
    if end is None:
        return set()
    meta = dl._read_frontmatter("\n".join(lines[1:end]))
    if not meta:
        return set()
    known = set(dl._FM_TO_FIELD) | {"title"}
    return {k for k in meta if k not in known}


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.

    Since #620 this is a TREND figure with no threshold attached; the name is historical. It counts
    prose rather than metadata because the question it answers is how much narrative a reader/agent
    must get through. 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.

    Since #620 this is reported as an informational TREND only — it no longer carries a threshold.
    The GENERATED catalog (`docs/decisions/README.md`) is also no longer counted: it used to be
    added on top, which quietly made this partly a record-COUNT metric wearing a line-count
    costume, since the catalog gains exactly one row per active record and no amount of
    consolidating prose can shrink it (189 of the 5658 lines it last reported were that file).
    Counting un-prunable generated output in a number whose stated remedy was "schedule a
    consolidation" pointed the reader at work that cannot be done.
    """
    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)
    return total


def record_prose_lines(rec) -> int:
    """Prose lines in one record's body (frontmatter already stripped by the parser)."""
    return len((rec.body or "").splitlines())


def oversized_records(records, ceiling: int) -> list[tuple[str, int]]:
    """Active records whose prose exceeds `ceiling`, longest first (#620).

    This REPLACES the aggregate line budget as the corpus's actionable size signal. The aggregate
    measured a monotonically growing quantity: a healthy project's decision corpus only gets
    bigger, so any fixed total is a ratchet that must periodically be raised — which is exactly the
    "permanently red, therefore no signal at all" state #542 re-baselined away from and #620 was
    filed about. Re-baselining it again would only restart that treadmill: the growth is not a smooth
    rate to set a threshold against — the corpus FELL from 5089 to 5042 across four days, then gained
    427 in a single evening as two large records landed.

    A per-record ceiling is NOT monotonic. It measures the shape of individual records rather than
    the size of the corpus, so it can go red and green again, and it names a file the reader can
    act on instead of asserting that "the corpus" is too big.

    The ceiling sits at a natural gap in the real distribution rather than a round number: at #620
    the records run 0..59 prose lines (median 26, p90 52) and then jump straight to 83, with
    nothing in between. 60 separates the bulk from the tail without splitting a cluster.

    IMPORTANT — a prompt for judgement, not a target. Length is a PROXY for "grown past what a
    reader can absorb", and the proxy is demonstrably wrong sometimes: the largest record in the
    corpus (`scan.libraryfolder-unique-identity`, 230 lines) is thirteen distinct hard-won traps,
    and shortening it would delete the only copy of most of them. Appearing in this list is an
    invitation to check for REDUNDANCY, never an instruction to cut.
    """
    out = [(r.key, record_prose_lines(r)) for r in records if r.key]
    return sorted([kv for kv in out if kv[1] > ceiling], key=lambda kv: -kv[1])


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")
    # Per-record prose ceiling — the corpus's size signal since #620, replacing the aggregate
    # budget. See `oversized_records` for why a total was the wrong instrument (it ratchets on a
    # monotonically growing quantity) and why 60 (a natural gap in the distribution, not a round
    # number). `--budget` is still ACCEPTED so an existing caller keeps working, but it no longer
    # does anything — and it SAYS SO when passed (below) rather than no-opping quietly. A flag that
    # takes a threshold and silently enforces nothing is the same "reports success while doing
    # nothing" defect as #603's `stale-after` and #609's marker; retiring one signal must not
    # introduce another.
    ap.add_argument("--record-ceiling", type=int, default=RECORD_CEILING_DEFAULT)
    ap.add_argument("--budget", type=int, default=None, help=argparse.SUPPRESS)
    args = ap.parse_args(argv)

    if args.budget is not None:
        print(
            f"::warning::decisions-validate: --budget {args.budget} is RETIRED and was IGNORED "
            f"(#620) — the aggregate is now an unthresholded trend. Use --record-ceiling for the "
            f"enforceable per-record signal.",
            file=sys.stderr,
        )

    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 ([], [], [])
    oversized = oversized_records(records, args.record_ceiling)
    errs = validate(
        records,
        archive_keys=_archive_keys(),
        catalog_ok=_catalog_ok(),
        removed=removed,
        rewritten=rewritten,
        archive_records=archive_records,
        demoted=demoted,
        wing_faults=record_wing_faults(),
    )

    # Aggregate: an unthresholded TREND, not a gate (#620). Printed every run so the number stays
    # visible, with no pass/fail attached — a total over a monotonically growing corpus can only
    # ratchet, and a permanently-tripped warning is indistinguishable from no warning at all.
    keyed = [r for r in records if r.key]
    record_prose = sum(record_prose_lines(r) for r in keyed)
    total_prose = _budget_total()
    print(
        f"::notice::decisions-validate: {record_prose} prose lines across {len(keyed)} records "
        f"(mean {record_prose // max(len(keyed), 1)}), plus {total_prose - record_prose} lines of "
        f"non-record scaffolding = {total_prose} total. Trend only — no threshold; the per-record "
        f"ceiling below is the actionable signal.",
        file=sys.stderr,
    )

    # Per-record ceiling: NON-BLOCKING by design (#520 — a size condition must never turn an
    # unrelated PR red). Names the files, so the reader can act instead of being told the corpus is
    # "too big". Being listed is an invitation to check for redundancy, NOT an instruction to cut:
    # the longest record in the corpus is 13 distinct traps and is a legitimate decline (#620).
    if oversized:
        listed = "; ".join(f"{k} ({n} lines)" for k, n in oversized)
        print(
            f"::warning::decisions-validate: {len(oversized)} record(s) exceed the "
            f"{args.record_ceiling}-line prose ceiling — check each for redundancy against its "
            f"siblings (a long record that is all distinct findings is fine, say so and move on): "
            f"{listed}",
            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())