ersatztv/scripts/decisions_lib.py

#!/usr/bin/env python3
"""Parse ErsatzTV decision records from docs/decisions.md and docs/decisions/*.md.

A decision record is a Markdown H2 section. A *migrated* record carries a visible metadata
block as its first non-blank content:

    ## 2026-07-17 — Title … (#406)
    `key: ci.runner-placement` · `status: active` · `since: 2026-07-17` · `supersedes: none` · `superseded-by: none`
    **Rule:** one-line current rule.
    **Signals:** concept · paths: a/b.yml · issues: #406
    **Mechanics:** docs/ci-cd.md → CI lanes
    <rationale prose …>

An H2 with no metadata line is treated as status `legacy-unmigrated` (migration target).

Two OPTIONAL fields (ersatztv#603, adopted from OKF v0.2's lifecycle/provenance families) may also
appear — `stale-after: YYYY-MM-DD` on the metadata line, and a `**Sources:**` line in the metadata
block. Absence is never an error; see decisions_validate for how they are checked.
"""

from __future__ import annotations

import re
from dataclasses import dataclass
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
DECISIONS_MD = REPO_ROOT / "docs" / "decisions.md"
TOPIC_DIR = REPO_ROOT / "docs" / "decisions"
ARCHIVE_DIR = TOPIC_DIR / "archive"
RECORDS_DIR = TOPIC_DIR / "records"  # split layout (#610): one YAML-frontmatter record per file

# H2 headings that are document structure, not decision records. "Records formerly in this file"
# is the generated where-did-it-go index the #610 split appends to each stripped file.
SKIP_HEADINGS = {"Index", "Active catalog", "Contents", "Records formerly in this file"}

STATUSES = {"active", "superseded", "retired", "legacy-unmigrated"}
KEY_RE = re.compile(r"^[a-z0-9]+(\.[a-z0-9-]+)+$")
_HEADING_RE = re.compile(r"^##\s+(.*\S)\s*$")
_META_FIELD_RE = re.compile(r"`([a-z-]+):\s*([^`]*)`")


@dataclass
class Record:
    heading: str
    source: Path
    lineno: int
    key: str | None = None
    status: str = "legacy-unmigrated"
    since: str | None = None
    supersedes: str | None = None
    superseded_by: str | None = None
    rule: str | None = None
    signals: str | None = None
    mechanics: str | None = None
    # Optional (ersatztv#603). `stale_after` is an absolute date, no TTL — the record is due for
    # re-confirmation once today >= it. `sources` cites the evidence the record rests on.
    stale_after: str | None = None
    sources: str | None = None
    body: str = ""


def _parse_meta_line(line: str) -> dict[str, str]:
    return {m.group(1): m.group(2).strip() for m in _META_FIELD_RE.finditer(line)}


def parse_text(text: str, source: Path) -> list[Record]:
    if has_frontmatter(text):
        return _parse_frontmatter(text, source)
    lines = text.splitlines()
    records: list[Record] = []
    i = 0
    while i < len(lines):
        m = _HEADING_RE.match(lines[i])
        if not m:
            i += 1
            continue
        rec = Record(heading=m.group(1), source=source, lineno=i + 1)
        j = i + 1
        body_lines: list[str] = []
        while j < len(lines) and not _HEADING_RE.match(lines[j]):
            body_lines.append(lines[j])
            j += 1
        for k, bl in enumerate(body_lines):
            if not bl.strip():
                continue
            meta = _parse_meta_line(bl)
            if "key" in meta and "status" in meta:
                rec.key = meta.get("key") or None
                rec.status = meta.get("status") or "legacy-unmigrated"
                rec.since = meta.get("since") or None
                rec.supersedes = meta.get("supersedes") or None
                rec.superseded_by = meta.get("superseded-by") or None
                # NOT `or None`: a present-but-empty `stale-after:` must stay "" so the validator can
                # tell it from an absent field and reject it. Collapsing the two would let a blank or
                # truncated value through as "absent" — a field that silently never fires, which is
                # exactly what the blocking format check exists to prevent.
                rec.stale_after = meta.get("stale-after")
                # The metadata block is contiguous: scan only until the first blank line, so a
                # bolded **Rule:** appearing later inside rationale prose can't overwrite the real one.
                for bl2 in body_lines[k + 1 :]:
                    if not bl2.strip():
                        break
                    if bl2.startswith("**Rule:**"):
                        rec.rule = bl2[len("**Rule:**") :].strip()
                    elif bl2.startswith("**Signals:**"):
                        rec.signals = bl2[len("**Signals:**") :].strip()
                    elif bl2.startswith("**Mechanics:**"):
                        rec.mechanics = bl2[len("**Mechanics:**") :].strip()
                    elif bl2.startswith("**Sources:**"):
                        rec.sources = bl2[len("**Sources:**") :].strip()
            break
        rec.body = "\n".join(body_lines).strip()
        records.append(rec)
        i = j
    return records


# --- YAML-frontmatter form (ersatztv#610): one record per file ------------------------------
#
# The corpus is migrating from "many `##` records per file, metadata on a backtick line" to "one
# record per file, metadata in YAML frontmatter". BOTH forms parse here, dispatching on a leading
# `---`. That is what lets the migration validate itself: the body-diff guard parses the OLD form
# at the merge-base and the NEW form at head, both yielding Records keyed on `key`, so no one-time
# bypass is needed for the commit that moves 166 records.
#
# Frontmatter keys are the SAME vocabulary as the backtick line (`superseded-by`, `stale-after`),
# so the on-disk names don't shift under anyone mid-migration.

_FM_DELIM = "---"
_FM_TO_FIELD = {
    "key": "key",
    "status": "status",
    "since": "since",
    "supersedes": "supersedes",
    "superseded-by": "superseded_by",
    "stale-after": "stale_after",
    "rule": "rule",
    "signals": "signals",
    "mechanics": "mechanics",
    "sources": "sources",
}


def has_frontmatter(text: str) -> bool:
    """True if `text` opens with a YAML frontmatter fence. Must be the very first line."""
    lines = text.splitlines()
    return bool(lines) and lines[0].rstrip() == _FM_DELIM


def _unquote(raw: str) -> str:
    """Undo the quoting `yaml.safe_dump` applies. Single-line scalars only.

    The frontmatter this reads is MACHINE-GENERATED by `migrate_decisions_split.render_record`
    with `default_flow_style=False` and `width=10**9`, so every value is a single-line scalar,
    either plain or quoted. That narrowness is what makes a dependency-free reader safe.
    """
    raw = raw.strip()
    if len(raw) >= 2 and raw[0] == raw[-1] == "'":
        return raw[1:-1].replace("''", "'")
    if len(raw) >= 2 and raw[0] == raw[-1] == '"':
        body = raw[1:-1]
        out, i = [], 0
        while i < len(body):
            if body[i] == "\\" and i + 1 < len(body):
                nxt = body[i + 1]
                out.append({"n": "\n", "t": "\t", '"': '"', "\\": "\\"}.get(nxt, nxt))
                i += 2
            else:
                out.append(body[i])
                i += 1
        return "".join(out)
    return raw


def _read_frontmatter(block: str) -> dict[str, str] | None:
    """Parse the frontmatter block WITHOUT PyYAML. None if it doesn't look like our format.

    Deliberately dependency-free: this read path runs in CI's `decisions lifecycle` job, in the
    Husky pre-commit hook, and on every contributor's machine. Requiring PyYAML there made the
    validator crash with ModuleNotFoundError on a runner that installs nothing — the split's own
    records became unparseable. Writing still uses PyYAML (`migrate_decisions_split`), because that
    is a one-shot run by a human who can install it.

    `test_frontmatter_reader_matches_pyyaml` asserts this agrees with PyYAML on every real record.
    """
    meta: dict[str, str] = {}
    for line in block.splitlines():
        if not line.strip() or line.lstrip().startswith("#"):
            continue
        if line[:1].isspace():
            return None  # nested/continued structure — not the flat form we emit
        k, sep, v = line.partition(":")
        if not sep:
            return None
        meta[k.strip()] = _unquote(v)
    return meta


def _parse_frontmatter(text: str, source: Path) -> list[Record]:
    lines = text.splitlines()
    end = None
    for i, ln in enumerate(lines[1:], start=1):
        if ln.rstrip() == _FM_DELIM:
            end = i
            break
    if end is None:
        return []  # unterminated frontmatter — malformed; validator reports the missing record
    meta = _read_frontmatter("\n".join(lines[1:end]))
    if not meta:
        return []

    rec = Record(heading=str(meta.get("title") or "").strip(), source=source, lineno=1)
    for fm_key, field in _FM_TO_FIELD.items():
        if fm_key not in meta:
            continue
        # "" is preserved rather than collapsed to None — the empty-vs-absent distinction
        # `stale-after` depends on (#603).
        setattr(rec, field, meta[fm_key].strip())
    if not rec.status:
        rec.status = "legacy-unmigrated"
    rec.body = "\n".join(lines[end + 1 :]).strip()
    return [rec]


def parse_file(path: Path) -> list[Record]:
    return parse_text(path.read_text(encoding="utf-8"), path)


def metadata_line_count(rec: Record) -> int:
    """Count how many lines in `rec.body` look like a metadata line (a line starting with
    `` `key: `` after stripping). A well-formed record has exactly 1; more indicates a
    duplicate/stacked metadata block left behind by a botched migration."""
    count = 0
    for line in rec.body.splitlines():
        if line.strip().startswith("`key:"):
            count += 1
    return count


_NON_DECISION_FILES = {"README.md", "migration-map.md", "retrieval-eval.md"}


def active_files() -> list[Path]:
    """Every file holding ACTIVE records, in both the legacy and split layouts.

    Legacy: `docs/decisions.md` + the flat topic files. Split (#610): `docs/decisions/records/**`.
    Both are listed so the two layouts can coexist during the migration and so a ref on either side
    of it parses correctly.
    """
    files = [DECISIONS_MD]
    files += sorted(p for p in TOPIC_DIR.glob("*.md") if p.name not in _NON_DECISION_FILES)
    files += sorted(RECORDS_DIR.rglob("*.md"))
    return files


def all_active_records() -> list[Record]:
    recs: list[Record] = []
    for f in active_files():
        if f.exists():
            recs += parse_file(f)
    return recs