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
from typing import NamedTuple

import scripts.decisions_lib as dl  # (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

# The coarse, non-ratcheting calibration bound (#688): the ceiling must flag a MEANINGFUL MINORITY
# of records. Below the floor it is parked among the outliers and names almost nobody; above the cap
# it is cutting into the bulk rather than marking a tail. See `ceiling_calibration` for why the fine
# percentile claim is reported instead of asserted.
#
# The floor is NOT "at least one record" — that is nearly unfalsifiable:
# measured on the live corpus it accepted every ceiling from 39 to 229, including the ceiling of 200
# this module's own docstring offered as the case it catches (one 230-line record keeps the count
# nonzero). A 2% floor rejects 200/229/230 and still leaves ~5x headroom below today's 9.8%.
CEILING_MINORITY_MIN = 0.02
CEILING_MINORITY_MAX = 0.25


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 — excluding any
    wing-root `README.md` "since no such file exists today" is simply false, and would additionally
    exempt a future `records/README.md`, i.e. reintroduce 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 _frontmatter_block(text: str) -> str | None:
    """The raw YAML between the opening `---` and the next `---`, or None if there isn't one."""
    if not dl.has_frontmatter(text):
        return None
    lines = text.splitlines()
    end = next((i for i, ln in enumerate(lines[1:], start=1) if ln.rstrip() == "---"), None)
    if end is None:
        return None
    return "\n".join(lines[1:end])


def pyyaml_frontmatter_faults(files) -> tuple[list[str], bool]:
    """Faults where PyYAML disagrees with the dependency-free reader. Returns (faults, ran).

    #674: the two known hazards are a bare apostrophe inside a single-quoted scalar
    (`rule: 'SQLite's LOWER()'`) and an unquoted ` #` (`rule: use --flag #2`). Before this check the
    validator reported OK on both, because `dl._read_frontmatter` is a hand parser that cannot see
    either. It was hit TWICE in one session by two independent agents, which is what makes it worth
    a guard rather than a note.

    The two hazards fail DIFFERENTLY, and catching only the first would have missed half of it:

      * the apostrophe makes PyYAML **reject** the document outright (`ParserError`);
      * the unquoted ` #` parses fine and **silently truncates** the value — PyYAML reads
        `use --flag`, the hand parser reads `use --flag #2`. No exception, a wrong value.

    So this compares the parsed RESULT and does not merely try/except the load. That is also why it
    generalizes past the two known characters, which is the property #674 asked for: any future
    construct where the writer's library and our reader disagree shows up as a diff, without anyone
    enumerating it first.

    DIRECTION MATTERS: PyYAML is the WRITER (`migrate_decisions_split.render_record` emits these
    files with `yaml.safe_dump`), so it is the authority on what the on-disk bytes mean. The hand
    reader is the permissive one, and a disagreement is a defect in the FILE, not in either parser.

    `ran` is False when PyYAML is not importable. The read path is deliberately dependency-free —
    `decisions-guard`, the Husky hooks and every contributor machine install nothing — so this check
    is strictly additive: it must never be the reason the validator cannot run. main() announces the
    skip rather than passing quietly, because a check that reports success while doing nothing is
    the exact defect class this corpus keeps re-learning (#603's `stale-after`, #609's marker).
    """
    try:
        import yaml  # pyright: ignore[reportMissingImports]
    except Exception:
        return [], False

    faults: list[str] = []
    for p in files:
        try:
            text = p.read_text(encoding="utf-8")
        except Exception:  # noqa: S112 (record_wing_faults already reports the unreadable file by name)
            continue
        block = _frontmatter_block(text)
        if block is None:
            continue  # no/unterminated frontmatter: reported by record_wing_faults
        try:
            theirs_raw = yaml.safe_load(block)
        except Exception as exc:
            # DELIBERATELY broad. `yaml.YAMLError` alone is too narrow: PyYAML's timestamp
            # constructor raises a BARE ValueError for a well-shaped but impossible date
            # (`stale-after: 2026-06-31` -> "day is out of range for month"), which would escape as
            # a traceback. This check is meant to be strictly additive — it must never be the
            # reason the validator cannot run, so every failure to load becomes a reported fault.
            first = str(exc).splitlines()[0] if str(exc).strip() else exc.__class__.__name__
            faults.append(
                f"{p}: PyYAML REJECTS this frontmatter, though the dependency-free reader accepted "
                f"it ({exc.__class__.__name__}: {first}). PyYAML is what WROTE these files, so its "
                f"verdict is authoritative. Quote the offending value and, inside single quotes, "
                f"double any literal apostrophe — as `yaml.safe_dump` does. Usual causes: a bare "
                f"apostrophe inside a single-quoted value, an unquoted `: ` or leading backtick, or "
                f"an impossible date."
            )
            continue
        if theirs_raw is None:
            theirs_raw = {}
        if not isinstance(theirs_raw, dict):
            faults.append(f"{p}: frontmatter parses as {type(theirs_raw).__name__}, not a mapping.")
            continue
        mine = dl._read_frontmatter(block)
        if mine is None:
            continue  # the reader bailed: reported by record_wing_faults as parse-to-0
        theirs = {k: ("" if v is None else str(v)) for k, v in theirs_raw.items()}
        # `key=str`: PyYAML returns TYPED mapping keys, so a stray `1: x` yields an int key while the
        # hand reader yields "1", and sorting that mixed set raises TypeError — an uncaught traceback
        # replacing what `_unknown_frontmatter_keys` used to report as an actionable error.
        for k in sorted(set(mine) | set(theirs), key=str):
            if mine.get(k) != theirs.get(k):
                faults.append(
                    f"{p}: frontmatter key {k!r} means different things to the two parsers — "
                    f"reader={mine.get(k)!r} but PyYAML={theirs.get(k)!r}. PyYAML wrote this file, "
                    f"so its reading is the real value and the record is silently corrupt. Common "
                    f"cause: an unquoted ` #`, which YAML treats as a comment and truncates there."
                )
    return faults, True


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 ran 2..59 prose lines (median 26, p90 52) and then jumped 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])


class CeilingCalibration(NamedTuple):
    n: int
    p90: int
    p95: int
    n_over: int
    fraction_over: float
    marks_tail: bool  # the FINE claim: p90 <= ceiling <= p95
    flags_minority: bool  # the COARSE claim: MINORITY_MIN <= fraction_over <= MINORITY_MAX


def ceiling_calibration(records, ceiling: int) -> CeilingCalibration:
    """How well `ceiling` still marks the start of the corpus's tail (#688).

    Two claims of DIFFERENT robustness, deliberately separated, because conflating them is what
    made the previous guard a ratchet:

    `marks_tail` — `p90 <= ceiling <= p95`. Correct as a definition of "start of the tail", but an
    order statistic over a SPARSE distribution is a STEP function: the lengths climb to the ceiling
    and then jump straight to 81 with nothing in between AS MEASURED TODAY (the gap's width moves
    with the corpus — this is the shape, not a constant), so ONE new record could move p90 by 21
    lines, and the only remedy the assertion admitted was to raise the ceiling. It is real signal,
    but it is signal about the CONSTANT drifting, not a defect in the commit under test — the same
    shape as `stale_records`, and it is reported the same way: a notice, never a failure.

    `flags_minority` — `CEILING_MINORITY_MIN <= fraction_over <= CEILING_MINORITY_MAX`. Deliberately
    coarse, and what the blocking test asserts. Each added record moves a fraction by at most 1/N, so
    NO SINGLE ordinary addition can cross it — this is measured headroom, not immunity. From a live
    18/183 (9.8%), BREACHING the 25% cap takes 38
    consecutive over-ceiling additions (37 lands exactly on 0.25, which still passes under `<=`),
    or 718 short ones to dilute below the floor — against ONE record to break `marks_tail`.

    The THIRD arm is the tightest and is stated here because it is the easy one to forget:
    CONSOLIDATION. Taking 15 of today's 18 over-ceiling records out of the over-set drops below the
    2% floor — trimming them to <=60 leaves 3/183 = 1.64%, archiving them outright leaves 3/168 =
    1.79% (the denominator moves too); either way, under the floor. That is a real tension with
    `test_oversized_records_can_go_green` — the ceiling is allowed to go green — and it is accepted
    rather than papered over: at 3/183 the
    constant genuinely IS mis-calibrated and saying so is the signal working. A consolidation PR
    large enough to hit it should re-derive the ceiling in the same change.

    It still catches genuine mis-calibration in both directions, measured on the real corpus: a
    ceiling of 20 flags 60% of records (cutting into the bulk, so every author learns to ignore it),
    and a ceiling of 200 flags 0.5% — one record — which is below the floor and rejected. Note that
    "flags NOBODY" is the wrong way to state the upper failure: at a ceiling of 200 the count is
    still nonzero because one 230-line record exists, which is exactly why the floor is a fraction
    and not `> 0`.

    Why not simply re-derive the constant instead: re-deriving fixes the instance and keeps the
    mechanism. The v3 fraction band and the v4 percentile containment both failed the same way, one
    faster than the other, and picking a new number would queue up the fifth version.
    """
    lengths = sorted(record_prose_lines(r) for r in records if r.key)
    n = len(lengths)
    if n == 0:
        return CeilingCalibration(0, 0, 0, 0, 0.0, False, False)

    def pct(q: float) -> int:
        return lengths[min(int(n * q), n - 1)]

    n_over = sum(1 for v in lengths if v > ceiling)
    frac = n_over / n
    p90, p95 = pct(0.90), pct(0.95)
    return CeilingCalibration(
        n=n,
        p90=p90,
        p95=p95,
        n_over=n_over,
        fraction_over=frac,
        marks_tail=p90 <= ceiling <= p95,
        flags_minority=CEILING_MINORITY_MIN <= frac <= CEILING_MINORITY_MAX,
    )


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)

    # #674: cross-check the dependency-free reader against the library that WROTE these files.
    # Strictly additive — absent PyYAML skips the check (and SAYS so) rather than failing the run.
    yaml_faults, yaml_ran = pyyaml_frontmatter_faults(record_wing_files())
    if not yaml_ran:
        print(
            "::notice::decisions-validate: PyYAML is not importable, so the frontmatter cross-check "
            "was SKIPPED — every other check ran. This is the expected state on the dependency-free "
            "read path (decisions-guard, the Husky hooks); CI's script-tests job runs it with PyYAML "
            "present.",
            file=sys.stderr,
        )

    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() + yaml_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,
        )

    # Ceiling calibration drift (#688): a NOTICE, never a failure. The ceiling drifting away from
    # the tail boundary is the passage of corpus growth, not a defect in the commit under test — the
    # same reasoning `stale_records` is built on. Asserting it in the blocking `script-tests` job
    # made the next author of a substantial record pay for an unrelated constant going out of date.
    cal = ceiling_calibration(records, args.record_ceiling)
    if cal.n and not cal.marks_tail:
        print(
            f"::notice::decisions-validate: the {args.record_ceiling}-line ceiling has drifted from "
            f"the tail boundary of the distribution (p90={cal.p90}, p95={cal.p95}, "
            f"{cal.n_over}/{cal.n} records over it). Re-derive it when convenient — this is a "
            f"maintenance signal about the constant, not a problem with this change.",
            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())