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 _parse_frontmatter(text: str, source: Path) -> list[Record]:
    import yaml  # local import: only the new form needs it

    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
    try:
        meta = yaml.safe_load("\n".join(lines[1:end])) or {}
    except yaml.YAMLError:
        return []
    if not isinstance(meta, dict):
        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
        val = meta[fm_key]
        # `None` is what YAML gives for `key:` with no value — the empty-vs-absent distinction
        # that `stale-after` depends on (#603). Preserve it as "" rather than collapsing to None.
        setattr(rec, field, "" if val is None else str(val).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