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"

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]:
    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


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]:
    files = [DECISIONS_MD]
    files += sorted(p for p in TOPIC_DIR.glob("*.md") if p.name not in _NON_DECISION_FILES)
    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