feat(assessment): gate signed release drift

tools/assessments/capability-fit.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""Re-run a capability-fit assessment against a trusted release catalog."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+import sys
+
+
+META_ROOT = Path(__file__).resolve().parents[2]
+ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments"
+RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release"
+for tools_root in (ASSESSMENT_TOOLS_ROOT, RELEASE_TOOLS_ROOT):
+    if str(tools_root) not in sys.path:
+        sys.path.insert(0, str(tools_root))
+
+from govoplan_assessment import render_review, review_capability_fit  # noqa: E402
+from govoplan_release.catalog import DEFAULT_PUBLIC_BASE_URL, fetch_json  # noqa: E402
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Validate a fit assessment and identify conclusions needing review after release drift."
+    )
+    parser.add_argument(
+        "--assessment",
+        type=Path,
+        default=META_ROOT / "docs" / "capability-fit-current.json",
+    )
+    parser.add_argument(
+        "--schema",
+        type=Path,
+        default=META_ROOT / "docs" / "capability-fit.schema.json",
+    )
+    source = parser.add_mutually_exclusive_group(required=True)
+    source.add_argument(
+        "--public",
+        action="store_true",
+        help="Use the fixed public GovOPlaN stable catalog and keyring URLs.",
+    )
+    source.add_argument(
+        "--catalog", type=Path, help="Read a catalog from a local file."
+    )
+    parser.add_argument(
+        "--keyring", type=Path, help="Trusted local keyring; required with --catalog."
+    )
+    parser.add_argument(
+        "--workspace-root",
+        type=Path,
+        default=META_ROOT.parent,
+        help="Workspace containing repository checkouts for local tag provenance.",
+    )
+    parser.add_argument(
+        "--skip-tag-provenance",
+        action="store_true",
+        help="Compare signed metadata without checking local annotated tags.",
+    )
+    parser.add_argument(
+        "--json", action="store_true", help="Print the machine-readable review report."
+    )
+    parser.add_argument(
+        "--output",
+        type=Path,
+        help="Also write the machine-readable review report to this path.",
+    )
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    if args.catalog is not None and args.keyring is None:
+        raise SystemExit("--keyring is required with --catalog")
+    if args.public and args.keyring is not None:
+        raise SystemExit("--keyring cannot be combined with --public")
+
+    assessment = read_object(args.assessment, label="assessment")
+    schema = read_object(args.schema, label="schema")
+    if args.public:
+        catalog = fetch_public_json(
+            f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/channels/stable.json",
+            label="catalog",
+        )
+        keyring = fetch_public_json(
+            f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/keyring.json", label="keyring"
+        )
+    else:
+        catalog = read_object(args.catalog, label="catalog")
+        keyring = read_object(args.keyring, label="keyring")
+
+    report = review_capability_fit(
+        assessment=assessment,
+        schema=schema,
+        catalog=catalog,
+        keyring=keyring,
+        workspace_root=None
+        if args.skip_tag_provenance
+        else args.workspace_root.resolve(),
+    )
+    encoded = json.dumps(report, indent=2, sort_keys=True) + "\n"
+    if args.output is not None:
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(encoded, encoding="utf-8")
+    sys.stdout.write(encoded if args.json else render_review(report))
+    return {"current": 0, "blocked": 1, "review_required": 2}[report["status"]]
+
+
+def read_object(path: Path | None, *, label: str) -> dict[str, object]:
+    if path is None:
+        raise SystemExit(f"Missing {label} path")
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise SystemExit(f"Could not read {label} {path}: {exc}") from exc
+    if not isinstance(payload, dict):
+        raise SystemExit(f"{label.capitalize()} must be a JSON object: {path}")
+    return payload
+
+
+def fetch_public_json(url: str, *, label: str) -> dict[str, object]:
+    result = fetch_json(url)
+    if result.get("ok") is not True or not isinstance(result.get("payload"), dict):
+        raise SystemExit(f"Could not fetch fixed public {label} endpoint.")
+    return result["payload"]
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())