policy-nexus/tools/source_inventory.py

#!/usr/bin/env python3
"""Audit the explicit canon/ADR corpus and emit deterministic source evidence."""

from __future__ import annotations

import argparse
import fnmatch
import hashlib
import json
from pathlib import Path
import subprocess
import sys
from typing import Any


SCHEMA_VERSION = "policy-nexus-source-inventory/v1"
DISPOSITIONS = {"published", "metadata-pending", "excluded", "unsupported-format"}


def _read_json(path: Path) -> dict[str, Any]:
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError(f"{path}: expected a JSON object")
    return value


def _tracked_files(repo: Path) -> list[str]:
    if not repo.is_dir():
        raise FileNotFoundError(f"source repository is unavailable: {repo}")
    if (repo / ".git").exists():
        result = subprocess.run(
            ["git", "-C", str(repo), "ls-tree", "-r", "--name-only", "HEAD"],
            check=True,
            capture_output=True,
            text=True,
        )
        return [line for line in result.stdout.splitlines() if line]
    return sorted(
        path.relative_to(repo).as_posix()
        for path in repo.rglob("*")
        if path.is_file() and not path.is_symlink()
    )


def _revision(repo: Path) -> str:
    if not (repo / ".git").exists():
        raise ValueError(f"{repo}: source lock is required for an archive checkout")
    result = subprocess.run(
        ["git", "-C", str(repo), "rev-parse", "HEAD"],
        check=True,
        capture_output=True,
        text=True,
    )
    revision = result.stdout.strip()
    if len(revision) != 40 or any(char not in "0123456789abcdef" for char in revision):
        raise ValueError(f"{repo}: invalid Git revision {revision!r}")
    return revision


def _repo_paths(
    config: dict[str, Any], *, policy_root: Path, source_root: Path
) -> dict[str, Path]:
    paths: dict[str, Path] = {}
    for name, repository in config["repositories"].items():
        paths[name] = policy_root if repository.get("local") else source_root / name
    return paths


def discover(
    config: dict[str, Any], *, policy_root: Path, source_root: Path
) -> tuple[list[dict[str, str]], dict[str, Path]]:
    paths = _repo_paths(config, policy_root=policy_root, source_root=source_root)
    sources: list[dict[str, str]] = []
    for name, repository in sorted(config["repositories"].items()):
        selectors = repository.get("selectors", [])
        if not selectors:
            raise ValueError(f"{name}: at least one source selector is required")
        for path in _tracked_files(paths[name]):
            if any(fnmatch.fnmatchcase(path, selector) for selector in selectors):
                sources.append({"source_repo": name, "source_path": path})
    return sources, paths


def _published_sources(publication_path: Path) -> set[tuple[str, str]]:
    publication = _read_json(publication_path)
    return {
        (document["source_repo"], document["source_path"])
        for document in publication.get("documents", [])
    }


def _new_entry(source: dict[str, str], published: set[tuple[str, str]]) -> dict[str, str]:
    key = (source["source_repo"], source["source_path"])
    path = Path(source["source_path"])
    if key in published:
        disposition = "published"
        reason = "Published through an explicit publication.json document entry."
    elif path.name.lower() == "readme.md":
        disposition = "excluded"
        reason = "Directory index, not an architecture decision record."
    elif path.suffix.lower() != ".md":
        disposition = "unsupported-format"
        reason = "Inventoried governing source is not Markdown and has no renderer yet."
    else:
        disposition = "metadata-pending"
        reason = (
            "In scope; awaits explicit publication addressing and owner/revision/review metadata."
        )
    return source | {"disposition": disposition, "reason": reason}


def refresh(
    config_path: Path,
    inventory_path: Path,
    publication_path: Path,
    *,
    policy_root: Path,
    source_root: Path,
) -> dict[str, Any]:
    config = _read_json(config_path)
    if config.get("schema_version") != 1:
        raise ValueError("source inventory config schema_version must be 1")
    discovered, _paths = discover(config, policy_root=policy_root, source_root=source_root)
    published = _published_sources(publication_path)
    existing: dict[tuple[str, str], dict[str, str]] = {}
    if inventory_path.exists():
        for source in _read_json(inventory_path).get("sources", []):
            existing[(source["source_repo"], source["source_path"])] = source
    sources = []
    for source in discovered:
        key = (source["source_repo"], source["source_path"])
        sources.append(existing.get(key, _new_entry(source, published)))
    inventory = {
        "schema_version": 1,
        "sources": sources,
    }
    inventory_path.write_text(
        json.dumps(inventory, indent=2, sort_keys=True) + "\n", encoding="utf-8"
    )
    return inventory


def _load_revisions(
    config: dict[str, Any], paths: dict[str, Path], lock_path: Path | None
) -> tuple[dict[str, str], str]:
    if lock_path:
        lock = _read_json(lock_path)
        if lock.get("schema_version") != 1:
            raise ValueError("source lock schema_version must be 1")
        repositories = lock.get("repositories", {})
        revisions = {
            name: repositories[name]["revision"] for name in config["repositories"]
        }
    else:
        revisions = {name: _revision(paths[name]) for name in config["repositories"]}
    for name, revision in revisions.items():
        if len(revision) != 40 or any(char not in "0123456789abcdef" for char in revision):
            raise ValueError(f"{name}: invalid locked revision {revision!r}")
    canonical = json.dumps(revisions, sort_keys=True, separators=(",", ":")).encode()
    digest = hashlib.sha256(canonical).hexdigest()
    if lock_path:
        locked_digest = lock.get("source_set_digest")
        if locked_digest != digest:
            raise ValueError(
                f"source lock digest mismatch: recorded {locked_digest!r}, computed {digest}"
            )
    return revisions, digest


def check(
    config_path: Path,
    inventory_path: Path,
    publication_path: Path,
    *,
    policy_root: Path,
    source_root: Path,
    lock_path: Path | None = None,
) -> dict[str, Any]:
    config = _read_json(config_path)
    inventory = _read_json(inventory_path)
    if config.get("schema_version") != 1 or inventory.get("schema_version") != 1:
        raise ValueError("source inventory config and inventory schema_version must be 1")
    discovered, paths = discover(config, policy_root=policy_root, source_root=source_root)
    discovered_keys = {
        (source["source_repo"], source["source_path"]) for source in discovered
    }
    entries = inventory.get("sources", [])
    inventory_keys: set[tuple[str, str]] = set()
    for source in entries:
        key = (source.get("source_repo", ""), source.get("source_path", ""))
        if key in inventory_keys:
            raise ValueError(f"duplicate source inventory entry: {key[0]}/{key[1]}")
        inventory_keys.add(key)
        if source.get("disposition") not in DISPOSITIONS:
            raise ValueError(f"{key[0]}/{key[1]}: invalid disposition")
        if not source.get("reason"):
            raise ValueError(f"{key[0]}/{key[1]}: disposition reason is required")
    missing = sorted(discovered_keys - inventory_keys)
    stale = sorted(inventory_keys - discovered_keys)
    if missing or stale:
        details = []
        if missing:
            details.append("unreviewed sources: " + ", ".join(f"{r}/{p}" for r, p in missing))
        if stale:
            details.append("inventory entries no longer present: " + ", ".join(f"{r}/{p}" for r, p in stale))
        raise ValueError("; ".join(details) + "; run source_inventory.py refresh and review the diff")

    published = _published_sources(publication_path)
    inventory_published = {
        (source["source_repo"], source["source_path"])
        for source in entries
        if source["disposition"] == "published"
    }
    if published != inventory_published:
        raise ValueError(
            "published source inventory must exactly match publication.json: "
            f"manifest_only={sorted(published - inventory_published)}, "
            f"inventory_only={sorted(inventory_published - published)}"
        )

    revisions, source_set_digest = _load_revisions(config, paths, lock_path)
    counts = {disposition: 0 for disposition in sorted(DISPOSITIONS)}
    repository_counts = {name: 0 for name in config["repositories"]}
    for source in entries:
        counts[source["disposition"]] += 1
        repository_counts[source["source_repo"]] += 1
    return {
        "schema_version": SCHEMA_VERSION,
        "source_set_digest": source_set_digest,
        "dispositions": counts,
        "repositories": [
            {
                "name": name,
                "revision": revisions[name],
                "source_count": repository_counts[name],
            }
            for name in sorted(config["repositories"])
        ],
        "excluded_scopes": config.get("excluded_scopes", []),
        "sources": entries,
    }


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("command", choices=("check", "refresh"))
    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("--publication", type=Path, default=Path("publication.json"))
    parser.add_argument("--policy-root", type=Path)
    parser.add_argument("--source-root", type=Path)
    parser.add_argument("--lock", type=Path)
    parser.add_argument("--report", 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:
        if args.command == "refresh":
            inventory = refresh(
                config_path,
                args.inventory.resolve(),
                args.publication.resolve(),
                policy_root=policy_root,
                source_root=source_root,
            )
            print(f"{args.inventory}: recorded {len(inventory['sources'])} source(s)")
            return 0
        report = check(
            config_path,
            args.inventory.resolve(),
            args.publication.resolve(),
            policy_root=policy_root,
            source_root=source_root,
            lock_path=args.lock.resolve() if args.lock else None,
        )
        if args.report:
            args.report.parent.mkdir(parents=True, exist_ok=True)
            args.report.write_text(
                json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8"
            )
        counts = report["dispositions"]
        print(
            f"source inventory ok: {len(report['sources'])} source(s), "
            f"{counts['published']} published, {counts['metadata-pending']} metadata-pending, "
            f"source-set {report['source_set_digest']}"
        )
        return 0
    except (KeyError, OSError, ValueError, subprocess.CalledProcessError) as exc:
        print(f"source inventory failed: {exc}", file=sys.stderr)
        return 1


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