policy-nexus/tools/adr_review_ledger.py

#!/usr/bin/env python3
"""Build the ADR review ledger from source-inventory.json and source files."""

from __future__ import annotations

import argparse
import json
import re
from pathlib import Path
import sys
from typing import Any

from render import split_frontmatter
from source_inventory import _repo_paths, _read_json


BARE_ADR = re.compile(r"(?i)\bADR-0*(\d+)\b")
REQUIRED = ("id", "title", "status", "owner")
FIRST_PUBLISH = {
    ("policy-nexus", "docs/adr/ADR-0001-addressing-and-permanence.md"),
    *(
        (
            "railiance-master",
            f"docs/adr/ADR-000{n}-{slug}.md",
        )
        for n, slug in (
            (1, "repository-prefix-architecture"),
            (2, "rail-kubernetes-wave-1-boundary"),
            (3, "rapp-first-wave-selection"),
            (4, "first-wave-reef-rollout"),
            (5, "derived-rail-composition"),
            (6, "reef-production-admission"),
            (7, "rapp-declaration-contract"),
            (8, "private-by-default-exposure"),
        )
    ),
}
SPECIALS = {
    ("policy-nexus", "docs/adr/ADR-0001-addressing-and-permanence.md"): (
        "First publish set. This repo's addressing contract."
    ),
    ("the-custodian", "canon/architecture/adr-008-multi-tenancy-model.md"): (
        "Relocated to net-kingdom Tenancy Posture. Duplicate-claim candidate (kind 6)."
    ),
}
for n, slug in (
    (1, "repository-prefix-architecture"),
    (2, "rail-kubernetes-wave-1-boundary"),
    (3, "rapp-first-wave-selection"),
    (4, "first-wave-reef-rollout"),
    (5, "derived-rail-composition"),
    (6, "reef-production-admission"),
    (7, "rapp-declaration-contract"),
    (8, "private-by-default-exposure"),
):
    SPECIALS[("railiance-master", f"docs/adr/ADR-000{n}-{slug}.md")] = (
        "First publish set. Publication-ready front-matter."
    )


def _bare_adr(source_path: str, meta_id: str) -> str:
    for text in (meta_id, Path(source_path).name):
        match = BARE_ADR.search(text)
        if match:
            return f"ADR-{int(match.group(1)):04d}"
    return ""


def _missing(meta: dict[str, str]) -> list[str]:
    missing = [field for field in REQUIRED if not meta.get(field)]
    if not (meta.get("revision") or meta.get("version")):
        missing.append("revision|version")
    if not ((meta.get("last_reviewed") or meta.get("updated")) and meta.get("review_interval")):
        missing.append("review")
    return missing


def _frontmatter(path: Path) -> dict[str, str]:
    if not path.is_file():
        return {}
    meta, _body = split_frontmatter(path.read_text(encoding="utf-8", errors="replace"))
    return {key: value for key, value in meta.items() if value}


def _load_rulings(path: Path | None) -> dict[tuple[str, str], dict[str, Any]]:
    if path is None or not path.exists():
        return {}
    payload = _read_json(path)
    rulings: dict[tuple[str, str], dict[str, Any]] = {}
    for ruling in payload.get("rulings", []):
        key = (ruling["source_repo"], ruling["source_path"])
        if key in rulings:
            raise ValueError(f"duplicate ruling: {key[0]}/{key[1]}")
        rulings[key] = ruling
    return rulings


def build_ledger(
    inventory_path: Path,
    config_path: Path,
    *,
    policy_root: Path,
    source_root: Path,
    rulings_path: Path | None = None,
) -> dict[str, Any]:
    inventory = _read_json(inventory_path)
    config = _read_json(config_path)
    rulings = _load_rulings(rulings_path)
    paths = _repo_paths(config, policy_root=policy_root, source_root=source_root)
    rows: list[dict[str, Any]] = []
    for source in inventory.get("sources", []):
        key = (source["source_repo"], source["source_path"])
        repo = paths.get(source["source_repo"])
        file_path = (repo / source["source_path"]) if repo is not None else None
        present = bool(file_path and file_path.is_file())
        meta = _frontmatter(file_path) if present and file_path is not None else {}
        rows.append(
            {
                "source_repo": source["source_repo"],
                "source_path": source["source_path"],
                "inventory_disposition": source.get("disposition", ""),
                "inventory_reason": source.get("reason", ""),
                "file_present": present,
                "frontmatter": {
                    field: meta.get(field, "")
                    for field in (
                        "id",
                        "title",
                        "status",
                        "owner",
                        "revision",
                        "version",
                        "last_reviewed",
                        "updated",
                        "review_interval",
                    )
                },
                "missing_fields": _missing(meta) if present else ["file-missing"],
                "bare_adr": _bare_adr(source["source_path"], meta.get("id", "")),
                "id_collisions": [],
                "bare_adr_collisions": [],
                "notes": SPECIALS.get(key, ""),
                "proposed_disposition": "publish" if key in FIRST_PUBLISH else "unreviewed",
                "conflict_kinds": [],
                "successor": "",
                "review_notes": "",
            }
        )

    by_id: dict[str, list[str]] = {}
    by_bare: dict[str, list[str]] = {}
    for row in rows:
        label = f"{row['source_repo']}/{row['source_path']}"
        meta_id = row["frontmatter"]["id"]
        if meta_id:
            by_id.setdefault(meta_id, []).append(label)
        if row["bare_adr"]:
            by_bare.setdefault(row["bare_adr"], []).append(label)
    for row in rows:
        label = f"{row['source_repo']}/{row['source_path']}"
        meta_id = row["frontmatter"]["id"]
        if meta_id:
            row["id_collisions"] = [other for other in by_id.get(meta_id, []) if other != label]
        if row["bare_adr"]:
            row["bare_adr_collisions"] = [
                other for other in by_bare.get(row["bare_adr"], []) if other != label
            ]
        key = (row["source_repo"], row["source_path"])
        ruling = rulings.get(key)
        if ruling:
            for field in (
                "proposed_disposition",
                "conflict_kinds",
                "successor",
                "review_notes",
            ):
                if field in ruling:
                    row[field] = ruling[field]
        if row["id_collisions"] and 1 not in row["conflict_kinds"]:
            row["conflict_kinds"] = [*row["conflict_kinds"], 1]
            if not row["review_notes"]:
                row["review_notes"] = (
                    f"Front-matter id {row['frontmatter']['id']!r} is shared; "
                    "pick a repo-prefixed publication id before publish. "
                    "Who rules: owning repo."
                )

    return {
        "schema_version": 1,
        "protocol": "docs/adr-review/protocol.md",
        "source_inventory": "source-inventory.json",
        "rulings": str(rulings_path) if rulings_path else "",
        "rows": rows,
    }


def _summary(ledger: dict[str, Any]) -> str:
    rows = ledger["rows"]
    dispositions = {}
    inventory = {}
    for row in rows:
        dispositions[row["proposed_disposition"]] = (
            dispositions.get(row["proposed_disposition"], 0) + 1
        )
        inventory[row["inventory_disposition"]] = inventory.get(row["inventory_disposition"], 0) + 1
    missing_files = [row for row in rows if not row["file_present"]]
    id_hits = [row for row in rows if row["id_collisions"]]
    bare_hits = [row for row in rows if row["bare_adr_collisions"]]
    lines = [
        "# ADR review ledger summary",
        "",
        f"Rows: {len(rows)}",
        "",
        "## Inventory dispositions",
        "",
    ]
    for name, count in sorted(inventory.items()):
        lines.append(f"- `{name}`: {count}")
    lines.extend(["", "## Proposed dispositions", ""])
    for name, count in sorted(dispositions.items()):
        lines.append(f"- `{name}`: {count}")
    lines.extend(["", "## Front-matter `id` collisions", ""])
    if not id_hits:
        lines.append("None.")
    else:
        seen = set()
        for row in id_hits:
            key = row["frontmatter"]["id"]
            if key in seen:
                continue
            seen.add(key)
            others = ", ".join([f"{row['source_repo']}/{row['source_path']}", *row["id_collisions"]])
            lines.append(f"- `{key}`: {others}")
    lines.extend(["", "## Bare ADR-NNNN collisions", ""])
    if not bare_hits:
        lines.append("None.")
    else:
        seen = set()
        for row in bare_hits:
            key = row["bare_adr"]
            if key in seen:
                continue
            seen.add(key)
            others = ", ".join(
                [f"{row['source_repo']}/{row['source_path']}", *row["bare_adr_collisions"]]
            )
            lines.append(f"- `{key}`: {others}")
    conflicts = [row for row in rows if row["proposed_disposition"] == "conflict"]
    lines.extend(["", "## Conflict rows", ""])
    if not conflicts:
        lines.append("None.")
    else:
        for row in conflicts:
            kinds = ",".join(str(kind) for kind in row["conflict_kinds"]) or "-"
            lines.append(
                f"- `{row['source_repo']}/{row['source_path']}` "
                f"(kinds {kinds}): {row['review_notes']}"
            )
    lines.extend(["", "## Missing files", ""])
    if not missing_files:
        lines.append("None.")
    else:
        for row in missing_files:
            lines.append(f"- `{row['source_repo']}/{row['source_path']}`")
    lines.append("")
    return "\n".join(lines)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", type=Path, default=Path("source-inventory.config.json"))
    parser.add_argument("--inventory", type=Path, default=Path("source-inventory.json"))
    parser.add_argument("--output", type=Path, default=Path("docs/adr-review/ledger.json"))
    parser.add_argument("--summary", type=Path, default=Path("docs/adr-review/SUMMARY.md"))
    parser.add_argument("--rulings", type=Path, default=Path("docs/adr-review/rulings.json"))
    parser.add_argument("--policy-root", type=Path)
    parser.add_argument("--source-root", type=Path)
    args = parser.parse_args(argv)
    config_path = args.config.resolve()
    policy_root = (args.policy_root or config_path.parent).resolve()
    source_root = (args.source_root or policy_root.parent).resolve()
    try:
        ledger = build_ledger(
            args.inventory.resolve(),
            config_path,
            policy_root=policy_root,
            source_root=source_root,
            rulings_path=args.rulings.resolve() if args.rulings else None,
        )
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(
            json.dumps(ledger, indent=2, sort_keys=True) + "\n", encoding="utf-8"
        )
        args.summary.write_text(_summary(ledger), encoding="utf-8")
        print(f"{args.output}: {len(ledger['rows'])} row(s)")
        return 0
    except (OSError, ValueError) as exc:
        print(f"adr review ledger failed: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())