fix(assessment): establish independent release trust

docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md

@@ -330,6 +330,46 @@ assumptions, gaps, risks, recommendation and proof check. Reassessment must pin
 new composition and review any row whose code, evidence, configuration or target
 requirement changed.
 
+### Repeatable release rerun
+
+The release-aware rerun tool validates the machine-readable assessment against
+its JSON Schema, verifies the catalog's Ed25519 signature against an
+independently pinned local trust keyring, and separately verifies the canonical
+hash of the published or candidate keyring pinned by the signed catalog. The
+downloaded keyring is therefore never its own sole trust root. The tool compares
+catalog sequence and module versions and—when local checkouts are available—checks
+annotated release tags against assessed commits. For repositories represented in
+signed `release.selected_units`, both the peeled commit and annotated tag-object
+ID must match exactly; re-annotating an unchanged commit requires review.
+Release drift produces a machine-readable list of affected capability and
+infrastructure conclusions instead of silently carrying their prior status
+forward:
+
+```bash
+./.venv/bin/python tools/assessments/capability-fit.py \
+  --public \
+  --trusted-keyring /srv/govoplan/trust/catalog-keyring.json
+```
+
+Use `--catalog CATALOG --keyring PUBLISHED_OR_CANDIDATE_KEYRING
+--trusted-keyring PINNED_LOCAL_KEYRING` for an offline or release-candidate
+rerun. `GOVOPLAN_MODULE_PACKAGE_CATALOG_TRUSTED_KEYS_FILE` may provide the
+independently managed local trust-root path instead. During key rotation this
+local file may contain both current and `next` keys within their validity
+windows; revoked, disabled and retired keys are rejected. Do not establish this
+trust file by downloading it in the same rerun. Public JSON responses are
+bounded to 16 MiB. Use `--json` or `--output PATH` for automation.
+
+Exit status `0` means the assessed release metadata is current, `2` means
+explicit review is required, and `1` means the schema or trust checks are
+blocked. The machine report distinguishes independent signature trust,
+published-keyring hash agreement, and local-tag proof. Local-tag proof is only
+marked checked when every composition entry is available for comparison and was
+attempted. A successful rerun proves only the assessment schema, signed catalog
+metadata, published-keyring integrity and optional local tag provenance; it does
+not prove installed artifacts, target providers, target infrastructure, or
+production fitness. Those remain separate proof checks above.
+
 ## Evidence used in this slice
 
 - [Production-like profile](../dev/production-like/README.md) and
@@ -343,7 +383,8 @@ requirement changed.
   Files, Mail, Audit and Addresses tests
 - [Campaign delivery runbook](https://git.add-ideas.de/add-ideas/govoplan-campaign/src/branch/main/docs/CAMPAIGN_DELIVERY_RUNBOOK.md)
 - [Live signed stable catalog](https://govoplan.add-ideas.de/catalogs/v1/channels/stable.json)
-  and [trusted keyring](https://govoplan.add-ideas.de/catalogs/v1/keyring.json)
+  and [published keyring](https://govoplan.add-ideas.de/catalogs/v1/keyring.json),
+  verified against a separately provisioned local trust keyring
 - Annotated source tags `govoplan-core/v0.1.13` and
   `govoplan-campaign/v0.1.10`, including their catalogued Python and WebUI refs
 

docs/RELEASE_CONSOLE.md

@@ -39,6 +39,22 @@ shows the dry-run commands for the selected rows, and `Generate Candidate`
 creates a signed catalog candidate that advances only selected repositories that
 already have a catalog entry.
 
+`Build Plan` also returns structured release-gate findings for each selected
+repository. The plan names the recommended next action and gives an explicit
+remediation for source-version, lockfile, Core WebUI composition, Git state, and
+worktree findings. A target version that has not yet been applied consistently
+to the selected source tree is therefore explained before the source-tag
+preflight rather than appearing only as an error after the operator tries to
+tag. `source_preflight_ready` means that the plan-visible source gates pass; the
+non-mutating `Preview Tag + Publish` remains mandatory for remote, manifest, and
+immutable-tag checks.
+
+Dashboard collection is also fail closed. Unreadable Core version metadata or a
+malformed module contract is returned as a bounded `collection_errors` entry
+with a remediation, marks the dashboard blocked, and becomes a structured
+blocker in both full and selective release plans. The console does not silently
+omit a contract or turn these source errors into an HTTP 500.
+
 The release-control area above the repository table is read-only and is meant
 to become the central release cockpit. It shows:
 

requirements-dev.txt

@@ -29,6 +29,7 @@ httpx==0.28.1
 httpx2>=2.5,<3
 filelock>=3.20.3
 idna>=3.15
+jsonschema>=4,<5
 pip>=26.1.2
 pip-audit>=2.9,<3
 python-multipart>=0.0.31

tests/test_capability_fit_review.py

@@ -0,0 +1,664 @@
+from __future__ import annotations
+
+import base64
+from copy import deepcopy
+from datetime import UTC, datetime, timedelta
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+from unittest import mock
+import unittest
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+
+META_ROOT = Path(__file__).resolve().parents[1]
+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.capability_fit import (  # noqa: E402
+    local_tag_provenance,
+    render_review,
+    review_capability_fit,
+    trusted_keys_from_keyring,
+)
+
+
+class CapabilityFitReviewTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.assessment = json.loads(
+            (META_ROOT / "docs" / "capability-fit-current.json").read_text(
+                encoding="utf-8"
+            )
+        )
+        cls.schema = json.loads(
+            (META_ROOT / "docs" / "capability-fit.schema.json").read_text(
+                encoding="utf-8"
+            )
+        )
+
+    def test_matching_signed_catalog_is_current(self) -> None:
+        catalog, keyring = signed_catalog(self.assessment)
+
+        report = review_capability_fit(
+            assessment=deepcopy(self.assessment),
+            schema=self.schema,
+            catalog=catalog,
+            published_keyring=keyring,
+            trusted_keyring=keyring,
+        )
+
+        self.assertEqual("current", report["status"])
+        self.assertEqual([], report["changes"])
+        self.assertFalse(report["proof_scope"]["installed_artifacts"]["checked"])
+        self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
+        self.assertFalse(report["proof_scope"]["local_tag_provenance"]["checked"])
+        repeated = review_capability_fit(
+            assessment=deepcopy(self.assessment),
+            schema=self.schema,
+            catalog=catalog,
+            published_keyring=keyring,
+            trusted_keyring=keyring,
+        )
+        self.assertEqual(
+            json.dumps(report, sort_keys=True), json.dumps(repeated, sort_keys=True)
+        )
+
+    def test_release_drift_identifies_affected_conclusions(self) -> None:
+        catalog, keyring = signed_catalog(
+            self.assessment, versions={"campaigns": "0.1.11"}, sequence=202607230001
+        )
+
+        report = review_capability_fit(
+            assessment=deepcopy(self.assessment),
+            schema=self.schema,
+            catalog=catalog,
+            published_keyring=keyring,
+            trusted_keyring=keyring,
+        )
+
+        self.assertEqual("review_required", report["status"])
+        target_ids = {item["id"] for item in report["review_targets"]}
+        self.assertIn("campaign.journey", target_ids)
+        self.assertIn("composition.campaigns", target_ids)
+        self.assertIn("assessment.release", target_ids)
+        self.assertIn(
+            "composition_version_changed", {item["code"] for item in report["findings"]}
+        )
+
+    def test_contradictory_signed_selected_unit_metadata_blocks_rerun(self) -> None:
+        core = next(
+            item
+            for item in self.assessment["composition"]
+            if item["module_id"] == "core"
+        )
+        catalog, keyring = signed_catalog(
+            self.assessment,
+            selected_units=[
+                {
+                    "repo": core["repository"],
+                    "version": core["manifest_version"],
+                    "tag": "v999.0.0",
+                    "commit_sha": "0" * 40,
+                    "tag_object_sha": "1" * 40,
+                }
+            ],
+        )
+
+        report = review_capability_fit(
+            assessment=deepcopy(self.assessment),
+            schema=self.schema,
+            catalog=catalog,
+            published_keyring=keyring,
+            trusted_keyring=keyring,
+        )
+
+        self.assertEqual("blocked", report["status"])
+        codes = {item["code"] for item in report["findings"]}
+        self.assertIn("catalog_release_metadata", codes)
+        self.assertIn("composition_commit_changed", codes)
+        self.assertTrue(report["proof_scope"]["catalog_signature_and_keyring"]["valid"])
+        self.assertFalse(report["proof_scope"]["release_metadata"]["valid"])
+
+    def test_signed_selected_unit_commit_drift_requires_review(self) -> None:
+        core = next(
+            item
+            for item in self.assessment["composition"]
+            if item["module_id"] == "core"
+        )
+        catalog, keyring = signed_catalog(
+            self.assessment,
+            selected_units=[
+                {
+                    "repo": core["repository"],
+                    "version": core["manifest_version"],
+                    "tag": f"v{core['manifest_version']}",
+                    "commit_sha": "0" * 40,
+                    "tag_object_sha": "1" * 40,
+                }
+            ],
+        )
+
+        report = review_capability_fit(
+            assessment=deepcopy(self.assessment),
+            schema=self.schema,
+            catalog=catalog,
+            published_keyring=keyring,
+            trusted_keyring=keyring,
+        )
+
+        self.assertEqual("review_required", report["status"])
+        self.assertIn(
+            "composition_commit_changed", {item["code"] for item in report["findings"]}
+        )
+        self.assertIn(
+            "composition.core", {item["id"] for item in report["review_targets"]}
+        )
+        self.assertFalse(report["proof_scope"]["release_metadata"]["valid"])
+
+    def test_missing_duplicate_and_malformed_selected_unit_provenance_blocks(
+        self,
+    ) -> None:
+        core = next(
+            item
+            for item in self.assessment["composition"]
+            if item["module_id"] == "core"
+        )
+        valid_unit = {
+            "repo": core["repository"],
+            "version": core["manifest_version"],
+            "tag": f"v{core['manifest_version']}",
+            "commit_sha": (str(core["commit"]) + "0" * 40)[:40],
+            "tag_object_sha": "1" * 40,
+        }
+        cases = (
+            ("missing", None, False, "catalog_source_provenance"),
+            (
+                "duplicate",
+                [valid_unit, deepcopy(valid_unit)],
+                True,
+                "catalog_release_metadata",
+            ),
+            (
+                "malformed",
+                [{**valid_unit, "commit_sha": "not-a-git-object"}],
+                True,
+                "catalog_source_provenance",
+            ),
+        )
+        for label, selected_units, include_selected_units, expected_code in cases:
+            with self.subTest(label=label):
+                catalog, keyring = signed_catalog(
+                    self.assessment,
+                    selected_units=selected_units,
+                    include_selected_units=include_selected_units,
+                )
+
+                report = review_capability_fit(
+                    assessment=deepcopy(self.assessment),
+                    schema=self.schema,
+                    catalog=catalog,
+                    published_keyring=keyring,
+                    trusted_keyring=keyring,
+                )
+
+                self.assertEqual("blocked", report["status"])
+                self.assertIn(
+                    expected_code, {item["code"] for item in report["findings"]}
+                )
+                self.assertFalse(report["proof_scope"]["release_metadata"]["valid"])
+
+    def test_catalog_validation_ignores_and_preserves_installer_replay_state(
+        self,
+    ) -> None:
+        catalog, keyring = signed_catalog(self.assessment)
+        with tempfile.TemporaryDirectory() as temp_dir:
+            state_path = Path(temp_dir) / "catalog-sequence.json"
+            state_path.write_text(
+                json.dumps(
+                    {"channels": {"stable": {"last_sequence": catalog["sequence"]}}}
+                ),
+                encoding="utf-8",
+            )
+            replay_environment = {
+                "GOVOPLAN_MODULE_PACKAGE_CATALOG_SEQUENCE_STATE": str(state_path),
+                "GOVOPLAN_MODULE_PACKAGE_CATALOG_ENFORCE_SEQUENCE": "true",
+            }
+            with mock.patch.dict(os.environ, replay_environment, clear=False):
+                report = review_capability_fit(
+                    assessment=deepcopy(self.assessment),
+                    schema=self.schema,
+                    catalog=catalog,
+                    published_keyring=keyring,
+                    trusted_keyring=keyring,
+                )
+                self.assertEqual(
+                    replay_environment,
+                    {key: os.environ[key] for key in replay_environment},
+                )
+
+        self.assertEqual("current", report["status"])
+        self.assertTrue(report["proof_scope"]["catalog_signature_and_keyring"]["valid"])
+
+    def test_self_consistent_substitution_fails_independent_trust_root(self) -> None:
+        _, trusted_keyring = signed_catalog(self.assessment)
+        catalog, published_keyring = signed_catalog(self.assessment)
+
+        report = review_capability_fit(
+            assessment=deepcopy(self.assessment),
+            schema=self.schema,
+            catalog=catalog,
+            published_keyring=published_keyring,
+            trusted_keyring=trusted_keyring,
+        )
+
+        self.assertEqual("blocked", report["status"])
+        codes = {item["code"] for item in report["findings"]}
+        self.assertIn("catalog_trust", codes)
+        self.assertNotIn("catalog_keyring_hash_mismatch", codes)
+        self.assertTrue(report["proof_scope"]["assessment_schema"]["valid"])
+        self.assertFalse(
+            report["proof_scope"]["catalog_signature_and_trusted_keyring"]["valid"]
+        )
+        self.assertTrue(report["proof_scope"]["published_keyring_hash"]["valid"])
+
+    def test_locally_pinned_rotation_keys_follow_core_status_semantics(self) -> None:
+        keys = trusted_keys_from_keyring(
+            {
+                "keys": [
+                    {"key_id": "active", "status": "active", "public_key": "a"},
+                    {"key_id": "next", "status": "next", "public_key": "b"},
+                    {"key_id": "retired", "status": "retired", "public_key": "c"},
+                ]
+            }
+        )
+
+        self.assertEqual({"active": "a", "next": "b"}, keys)
+        self.assertEqual(
+            {"legacy": "base64-key"},
+            trusted_keys_from_keyring({"legacy": "base64-key"}),
+        )
+
+    def test_schema_error_blocks_before_comparison(self) -> None:
+        assessment = deepcopy(self.assessment)
+        assessment.pop("scope")
+        catalog, keyring = signed_catalog(self.assessment)
+
+        report = review_capability_fit(
+            assessment=assessment,
+            schema=self.schema,
+            catalog=catalog,
+            published_keyring=keyring,
+            trusted_keyring=keyring,
+        )
+
+        self.assertEqual("blocked", report["status"])
+        self.assertEqual(
+            {"assessment_schema"}, {item["code"] for item in report["findings"]}
+        )
+        self.assertFalse(report["proof_scope"]["assessment_schema"]["valid"])
+        self.assertFalse(
+            report["proof_scope"]["catalog_signature_and_keyring"]["checked"]
+        )
+        self.assertIsNone(
+            report["proof_scope"]["catalog_signature_and_keyring"]["valid"]
+        )
+
+    def test_human_report_states_proof_limit(self) -> None:
+        catalog, keyring = signed_catalog(self.assessment)
+        report = review_capability_fit(
+            assessment=deepcopy(self.assessment),
+            schema=self.schema,
+            catalog=catalog,
+            published_keyring=keyring,
+            trusted_keyring=keyring,
+        )
+
+        rendered = render_review(report)
+
+        self.assertIn("Capability fit rerun: current", rendered)
+        self.assertIn(
+            "No installed-artifact, target-provider, or production proof", rendered
+        )
+
+    def test_public_fetch_failure_is_generic_and_blocking(self) -> None:
+        script = META_ROOT / "tools" / "assessments" / "capability-fit.py"
+        spec = importlib.util.spec_from_file_location("capability_fit_cli", script)
+        assert spec is not None and spec.loader is not None
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            trusted_path = Path(temp_dir) / "trusted-keyring.json"
+            trusted_path.write_text('{"keys": []}\n', encoding="utf-8")
+            with mock.patch.object(
+                module,
+                "fetch_json",
+                return_value={
+                    "ok": False,
+                    "error": "Authorization: secret at https://internal.invalid",
+                },
+            ):
+                with self.assertRaisesRegex(
+                    SystemExit, r"^Could not fetch fixed public catalog endpoint[.]$"
+                ):
+                    module.main(
+                        [
+                            "--public",
+                            "--trusted-keyring",
+                            str(trusted_path),
+                            "--skip-tag-provenance",
+                        ]
+                    )
+
+    def test_cli_requires_an_independent_local_trust_root(self) -> None:
+        script = META_ROOT / "tools" / "assessments" / "capability-fit.py"
+        spec = importlib.util.spec_from_file_location(
+            "capability_fit_cli_trust", script
+        )
+        assert spec is not None and spec.loader is not None
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+
+        with mock.patch.dict(
+            os.environ,
+            {module.TRUSTED_KEYRING_FILE_ENV: ""},
+            clear=False,
+        ):
+            with self.assertRaisesRegex(SystemExit, r"--trusted-keyring .* required"):
+                module.main(["--public", "--skip-tag-provenance"])
+
+    def test_local_selected_tag_commit_and_object_must_match_exactly(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            workspace = Path(temp_dir)
+            commit_sha, tag_object_sha = create_tagged_repository(
+                workspace=workspace,
+                repository="govoplan-core",
+                version="0.1.0",
+            )
+            assessment = deepcopy(self.assessment)
+            core = next(
+                deepcopy(item)
+                for item in assessment["composition"]
+                if item["module_id"] == "core"
+            )
+            core["manifest_version"] = "0.1.0"
+            core["commit"] = commit_sha[:12]
+            assessment["composition"] = [core]
+            selected_unit = {
+                "repo": "govoplan-core",
+                "version": "0.1.0",
+                "tag": "v0.1.0",
+                "commit_sha": commit_sha,
+                "tag_object_sha": tag_object_sha,
+            }
+            catalog, keyring = signed_catalog(
+                assessment,
+                selected_units=[selected_unit],
+            )
+
+            baseline = review_capability_fit(
+                assessment=deepcopy(assessment),
+                schema=self.schema,
+                catalog=catalog,
+                published_keyring=keyring,
+                trusted_keyring=keyring,
+                workspace_root=workspace,
+            )
+
+            self.assertEqual("current", baseline["status"])
+            self.assertTrue(baseline["proof_scope"]["local_tag_provenance"]["valid"])
+
+            forged_commit = commit_sha[:12] + ("0" * 28)
+            if forged_commit == commit_sha:
+                forged_commit = commit_sha[:12] + ("f" * 28)
+            forged_catalog, forged_keyring = signed_catalog(
+                assessment,
+                selected_units=[{**selected_unit, "commit_sha": forged_commit}],
+            )
+            commit_mismatch = review_capability_fit(
+                assessment=deepcopy(assessment),
+                schema=self.schema,
+                catalog=forged_catalog,
+                published_keyring=forged_keyring,
+                trusted_keyring=forged_keyring,
+                workspace_root=workspace,
+            )
+
+            self.assertEqual("review_required", commit_mismatch["status"])
+            self.assertIn(
+                "tag_signed_commit_changed",
+                {item["kind"] for item in commit_mismatch["changes"]},
+            )
+
+            repo = workspace / "govoplan-core"
+            git_text(repo, "tag", "-d", "v0.1.0")
+            git_text(repo, "tag", "-a", "v0.1.0", "-m", "Re-annotated release")
+            reannotated = review_capability_fit(
+                assessment=deepcopy(assessment),
+                schema=self.schema,
+                catalog=catalog,
+                published_keyring=keyring,
+                trusted_keyring=keyring,
+                workspace_root=workspace,
+            )
+
+        self.assertEqual("review_required", reannotated["status"])
+        self.assertIn(
+            "tag_object_changed",
+            {item["kind"] for item in reannotated["changes"]},
+        )
+        self.assertTrue(reannotated["proof_scope"]["local_tag_provenance"]["checked"])
+        self.assertFalse(reannotated["proof_scope"]["local_tag_provenance"]["valid"])
+
+    def test_missing_composition_entry_does_not_claim_complete_tag_proof(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            workspace = Path(temp_dir)
+            core_commit, core_tag_object = create_tagged_repository(
+                workspace=workspace,
+                repository="govoplan-core",
+                version="0.1.0",
+            )
+            tenancy_commit, _ = create_tagged_repository(
+                workspace=workspace,
+                repository="govoplan-tenancy",
+                version="0.1.0",
+            )
+            reviewed_assessment = deepcopy(self.assessment)
+            core = next(
+                deepcopy(item)
+                for item in reviewed_assessment["composition"]
+                if item["module_id"] == "core"
+            )
+            tenancy = next(
+                deepcopy(item)
+                for item in reviewed_assessment["composition"]
+                if item["module_id"] == "tenancy"
+            )
+            core.update(manifest_version="0.1.0", commit=core_commit[:12])
+            tenancy.update(manifest_version="0.1.0", commit=tenancy_commit[:12])
+            catalog_assessment = deepcopy(reviewed_assessment)
+            catalog_assessment["composition"] = [core]
+            reviewed_assessment["composition"] = [core, tenancy]
+            catalog, keyring = signed_catalog(
+                catalog_assessment,
+                selected_units=[
+                    {
+                        "repo": "govoplan-core",
+                        "version": "0.1.0",
+                        "tag": "v0.1.0",
+                        "commit_sha": core_commit,
+                        "tag_object_sha": core_tag_object,
+                    }
+                ],
+            )
+
+            report = review_capability_fit(
+                assessment=reviewed_assessment,
+                schema=self.schema,
+                catalog=catalog,
+                published_keyring=keyring,
+                trusted_keyring=keyring,
+                workspace_root=workspace,
+            )
+
+        proof = report["proof_scope"]["local_tag_provenance"]
+        self.assertEqual("review_required", report["status"])
+        self.assertFalse(proof["checked"])
+        self.assertIsNone(proof["valid"])
+        self.assertEqual(2, proof["attempted_count"])
+        self.assertEqual(2, proof["expected_count"])
+        self.assertNotIn(
+            "tag_provenance_changed", {item["code"] for item in report["findings"]}
+        )
+
+    def test_repository_path_traversal_is_not_resolved(self) -> None:
+        with mock.patch("govoplan_assessment.capability_fit.subprocess.run") as run:
+            result = local_tag_provenance(
+                workspace_root=META_ROOT.parent,
+                repository="../outside",
+                version="0.1.0",
+                assessed_commit="0123456",
+            )
+
+        self.assertEqual("tag_provenance_unavailable", result["kind"])
+        self.assertIn("invalid repository identifier", result["message"])
+        run.assert_not_called()
+
+
+def create_tagged_repository(
+    *, workspace: Path, repository: str, version: str
+) -> tuple[str, str]:
+    repo = workspace / repository
+    git_text(workspace, "init", "-b", "main", str(repo))
+    git_text(repo, "config", "user.name", "Capability Fit Test")
+    git_text(repo, "config", "user.email", "capability-fit@example.invalid")
+    (repo / "README.md").write_text(f"{repository}\n", encoding="utf-8")
+    git_text(repo, "add", "README.md")
+    git_text(repo, "commit", "-m", "Initial release fixture")
+    git_text(repo, "tag", "-a", f"v{version}", "-m", f"Release v{version}")
+    return (
+        git_text(repo, "rev-parse", "HEAD"),
+        git_text(repo, "rev-parse", f"refs/tags/v{version}"),
+    )
+
+
+def git_text(cwd: Path, *args: str) -> str:
+    result = subprocess.run(
+        ("git", *args),
+        cwd=cwd,
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        raise AssertionError(result.stderr or result.stdout)
+    return result.stdout.strip()
+
+
+def signed_catalog(
+    assessment: dict[str, object],
+    *,
+    versions: dict[str, str] | None = None,
+    sequence: int = 202607220843,
+    selected_units: list[dict[str, object]] | None = None,
+    include_selected_units: bool = True,
+) -> tuple[dict[str, object], dict[str, object]]:
+    versions = versions or {}
+    private_key = Ed25519PrivateKey.generate()
+    public_key = base64.b64encode(
+        private_key.public_key().public_bytes(
+            encoding=serialization.Encoding.Raw,
+            format=serialization.PublicFormat.Raw,
+        )
+    ).decode("ascii")
+    keyring: dict[str, object] = {
+        "keyring_version": "1",
+        "purpose": "test",
+        "generated_at": "2026-01-01T00:00:00Z",
+        "keys": [{"key_id": "test-key", "status": "active", "public_key": public_key}],
+    }
+    now = datetime.now(tz=UTC)
+    modules: list[dict[str, object]] = []
+    core_release: dict[str, object] | None = None
+    for component in assessment["composition"]:
+        module_id = str(component["module_id"])
+        repository = str(component["repository"])
+        version = versions.get(module_id, str(component["manifest_version"]))
+        entry: dict[str, object] = {
+            "module_id": module_id,
+            "name": module_id,
+            "version": version,
+            "python_package": repository,
+            "python_ref": f"{repository} @ git+ssh://git@example.invalid/example/{repository}.git@v{version}",
+        }
+        if module_id == "core":
+            core_release = entry
+        else:
+            modules.append(entry)
+    assert core_release is not None
+    if selected_units is None:
+        selected_units = [
+            {
+                "repo": component["repository"],
+                "version": versions.get(
+                    str(component["module_id"]), str(component["manifest_version"])
+                ),
+                "tag": "v"
+                + versions.get(
+                    str(component["module_id"]), str(component["manifest_version"])
+                ),
+                "commit_sha": (str(component["commit"]) + "0" * 40)[:40],
+                "tag_object_sha": hashlib.sha256(
+                    f"{component['repository']}:{component['manifest_version']}:tag".encode()
+                ).hexdigest()[:40],
+            }
+            for component in assessment["composition"]
+        ]
+    release: dict[str, object] = {"keyring_sha256": canonical_hash(keyring)}
+    if include_selected_units:
+        release["selected_units"] = selected_units
+    catalog: dict[str, object] = {
+        "catalog_version": "1",
+        "channel": "stable",
+        "sequence": sequence,
+        "generated_at": now.isoformat().replace("+00:00", "Z"),
+        "expires_at": (now + timedelta(days=30)).isoformat().replace("+00:00", "Z"),
+        "core_release": core_release,
+        "modules": modules,
+        "release": release,
+    }
+    signature_payload = json.dumps(
+        catalog, sort_keys=True, separators=(",", ":"), ensure_ascii=False
+    ).encode("utf-8")
+    catalog["signatures"] = [
+        {
+            "algorithm": "ed25519",
+            "key_id": "test-key",
+            "value": base64.b64encode(private_key.sign(signature_payload)).decode(
+                "ascii"
+            ),
+        }
+    ]
+    return catalog, keyring
+
+
+def canonical_hash(payload: object) -> str:
+    encoded = json.dumps(
+        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_release_dashboard_diagnostics.py

@@ -0,0 +1,177 @@
+from __future__ import annotations
+
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from fastapi.testclient import TestClient
+
+
+META_ROOT = Path(__file__).resolve().parents[1]
+RELEASE_ROOT = META_ROOT / "tools" / "release"
+if str(RELEASE_ROOT) not in sys.path:
+    sys.path.insert(0, str(RELEASE_ROOT))
+
+from server.app import create_app  # noqa: E402
+
+
+class ReleaseDashboardDiagnosticsTests(unittest.TestCase):
+    def test_malformed_core_version_is_bounded_and_blocks_api_plans(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            workspace = Path(temp_dir)
+            core = workspace / "govoplan-core"
+            core.mkdir()
+            (core / "pyproject.toml").write_text("[project\n", encoding="utf-8")
+
+            with TestClient(create_app(workspace_root=workspace)) as client:
+                dashboard_response = client.get(
+                    "/api/dashboard", params={"public_catalog": "false"}
+                )
+                selective_response = client.get(
+                    "/api/selective-plan",
+                    params={
+                        "repos": "govoplan-core",
+                        "repo_versions": "govoplan-core:1.2.4",
+                        "public_catalog": "false",
+                    },
+                )
+                full_plan_response = client.get(
+                    "/api/plan", params={"public_catalog": "false"}
+                )
+
+        self.assertEqual(200, dashboard_response.status_code)
+        dashboard = dashboard_response.json()
+        self.assertEqual("blocked", dashboard["summary"]["status"])
+        error = next(
+            item
+            for item in dashboard["collection_errors"]
+            if item["code"] == "core_version_metadata_unreadable"
+        )
+        self.assertEqual("govoplan-core/pyproject.toml", error["source"])
+        self.assertIn("TOMLDecodeError", error["message"])
+        self.assertIn("Repair govoplan-core/pyproject.toml", error["remediation"])
+        self.assertLess(len(error["message"]), 200)
+        self.assertNotIn(str(workspace), error["message"])
+
+        self.assertEqual(200, selective_response.status_code)
+        selective_plan = selective_response.json()
+        self.assertEqual("blocked", selective_plan["status"])
+        self.assertFalse(selective_plan["source_preflight_ready"])
+        finding = next(
+            item
+            for item in selective_plan["gate_findings"]
+            if item["code"] == "core_version_metadata_unreadable"
+        )
+        self.assertEqual(error["remediation"], finding["remediation"])
+        self.assertEqual(
+            "resolve_release_gate", selective_plan["recommended_action"]["id"]
+        )
+
+        self.assertEqual(200, full_plan_response.status_code)
+        full_plan = full_plan_response.json()
+        self.assertEqual("blocked", full_plan["status"])
+        self.assertTrue(
+            any(
+                action["id"].startswith("dashboard:core_version_metadata_unreadable:")
+                for action in full_plan["actions"]
+            )
+        )
+
+    def test_malformed_manifest_is_not_silently_omitted_from_selected_plan(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            workspace = Path(temp_dir)
+            core = workspace / "govoplan-core"
+            core.mkdir()
+            (core / "pyproject.toml").write_text(
+                '[project]\nname = "govoplan-core"\nversion = "0.1.13"\n',
+                encoding="utf-8",
+            )
+            files = workspace / "govoplan-files"
+            manifest = files / "src" / "govoplan_files" / "backend" / "manifest.py"
+            manifest.parent.mkdir(parents=True)
+            (files / "pyproject.toml").write_text(
+                '[project]\nname = "govoplan-files"\nversion = "1.2.4"\n',
+                encoding="utf-8",
+            )
+            manifest.write_text("def broken(:\n", encoding="utf-8")
+            initialize_repository(files)
+
+            with TestClient(create_app(workspace_root=workspace)) as client:
+                dashboard_response = client.get(
+                    "/api/dashboard", params={"public_catalog": "false"}
+                )
+                selective_response = client.get(
+                    "/api/selective-plan",
+                    params={
+                        "repos": "govoplan-files",
+                        "repo_versions": "govoplan-files:1.2.4",
+                        "public_catalog": "false",
+                    },
+                )
+
+        self.assertEqual(200, dashboard_response.status_code)
+        dashboard = dashboard_response.json()
+        error = next(
+            item
+            for item in dashboard["collection_errors"]
+            if item["code"] == "module_contract_unreadable"
+        )
+        self.assertEqual("govoplan-files", error["repo"])
+        self.assertEqual(
+            "govoplan-files/src/govoplan_files/backend/manifest.py",
+            error["source"],
+        )
+        self.assertIn("SyntaxError", error["message"])
+        self.assertIn("Repair the manifest", error["remediation"])
+
+        self.assertEqual(200, selective_response.status_code)
+        plan = selective_response.json()
+        self.assertEqual("blocked", plan["status"])
+        self.assertFalse(plan["source_preflight_ready"])
+        self.assertIn(
+            "module_contract_unreadable",
+            {finding["code"] for finding in plan["gate_findings"]},
+        )
+        self.assertEqual("resolve_release_gate", plan["recommended_action"]["id"])
+
+
+def initialize_repository(path: Path) -> None:
+    subprocess.run(
+        ["git", "init", "--initial-branch=main", str(path)],
+        check=True,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    subprocess.run(
+        ["git", "-C", str(path), "add", "."],
+        check=True,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    subprocess.run(
+        [
+            "git",
+            "-C",
+            str(path),
+            "-c",
+            "user.name=Release Test",
+            "-c",
+            "user.email=release@example.invalid",
+            "-c",
+            "commit.gpgsign=false",
+            "commit",
+            "-m",
+            "Initial fixture",
+        ],
+        check=True,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_release_http_fetch.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+
+META_ROOT = Path(__file__).resolve().parents[1]
+RELEASE_ROOT = META_ROOT / "tools" / "release"
+if str(RELEASE_ROOT) not in sys.path:
+    sys.path.insert(0, str(RELEASE_ROOT))
+
+from govoplan_release.http_fetch import (  # noqa: E402
+    DEFAULT_MAX_RESPONSE_BYTES,
+    fetch_http,
+)
+
+
+class ReleaseHttpFetchTests(unittest.TestCase):
+    def test_content_length_over_default_limit_blocks_before_read(self) -> None:
+        response = FakeResponse(
+            body=b"ignored",
+            headers={"Content-Length": str(DEFAULT_MAX_RESPONSE_BYTES + 1)},
+        )
+
+        with mock.patch(
+            "govoplan_release.http_fetch.urllib.request.urlopen",
+            return_value=response,
+        ):
+            with self.assertRaisesRegex(ValueError, "exceeds the configured size"):
+                fetch_http("https://example.invalid/catalog.json", timeout=1)
+
+        self.assertEqual([], response.read_sizes)
+
+    def test_streamed_body_is_capped_when_length_is_absent(self) -> None:
+        response = FakeResponse(body=b"x" * 9, headers={})
+
+        with mock.patch(
+            "govoplan_release.http_fetch.urllib.request.urlopen",
+            return_value=response,
+        ):
+            with self.assertRaisesRegex(ValueError, "exceeds the configured size"):
+                fetch_http(
+                    "https://example.invalid/catalog.json",
+                    timeout=1,
+                    max_response_bytes=8,
+                )
+
+        self.assertEqual([9], response.read_sizes)
+
+
+class FakeResponse:
+    def __init__(self, *, body: bytes, headers: dict[str, str]) -> None:
+        self.status = 200
+        self.headers = headers
+        self.body = body
+        self.read_sizes: list[int] = []
+
+    def __enter__(self) -> FakeResponse:
+        return self
+
+    def __exit__(self, *args: object) -> None:
+        return None
+
+    def read(self, size: int) -> bytes:
+        self.read_sizes.append(size)
+        return self.body[:size]
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_release_plan_guidance.py

@@ -0,0 +1,203 @@
+from __future__ import annotations
+
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+
+META_ROOT = Path(__file__).resolve().parents[1]
+RELEASE_ROOT = META_ROOT / "tools" / "release"
+if str(RELEASE_ROOT) not in sys.path:
+    sys.path.insert(0, str(RELEASE_ROOT))
+
+from govoplan_release.model import (  # noqa: E402
+    CatalogSnapshot,
+    DashboardSummary,
+    ReleaseDashboard,
+    RepositorySnapshot,
+    RepositorySpec,
+    VersionSnapshot,
+)
+from govoplan_release.selective_planner import build_selective_release_plan  # noqa: E402
+
+
+class ReleasePlanGuidanceTests(unittest.TestCase):
+    def test_unprepared_target_has_structured_remediation_and_recommendation(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            workspace = Path(temp_dir)
+            repo_path = workspace / "govoplan-files"
+            repo_path.mkdir()
+            (repo_path / "pyproject.toml").write_text(
+                '[project]\nname = "govoplan-files"\nversion = "1.2.3"\n',
+                encoding="utf-8",
+            )
+            plan = build_selective_release_plan(
+                dashboard(workspace=workspace, version="1.2.3"),
+                selected_repos=("govoplan-files",),
+                repo_versions={"govoplan-files": "1.2.4"},
+            )
+
+        self.assertEqual("blocked", plan.status)
+        self.assertFalse(plan.source_preflight_ready)
+        finding = next(
+            item
+            for item in plan.gate_findings
+            if item.code == "repository_version_alignment"
+        )
+        self.assertEqual("govoplan-files", finding.repo)
+        self.assertEqual("pyproject.toml", finding.source)
+        self.assertEqual("1.2.4", finding.expected)
+        self.assertEqual("1.2.3", finding.actual)
+        self.assertIn("Regenerate lockfiles", finding.remediation)
+        self.assertEqual("resolve_release_gate", plan.recommended_action.id)
+        self.assertEqual("govoplan-files", plan.recommended_action.repo)
+
+    def test_aligned_target_recommends_non_mutating_tag_preview(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            workspace = Path(temp_dir)
+            repo_path = workspace / "govoplan-files"
+            repo_path.mkdir()
+            (repo_path / "pyproject.toml").write_text(
+                '[project]\nname = "govoplan-files"\nversion = "1.2.4"\n',
+                encoding="utf-8",
+            )
+            plan = build_selective_release_plan(
+                dashboard(workspace=workspace, version="1.2.4"),
+                selected_repos=("govoplan-files",),
+                repo_versions={"govoplan-files": "1.2.4"},
+            )
+
+        self.assertTrue(plan.source_preflight_ready)
+        self.assertEqual("preview_source_release", plan.recommended_action.id)
+        self.assertIn("Preview Tag + Publish", plan.recommended_action.remediation)
+
+    def test_malformed_version_metadata_is_a_structured_blocker(self) -> None:
+        cases = (
+            ("package.json", "{not-json\n", "JSONDecodeError"),
+            ("pyproject.toml", "[project\n", "TOMLDecodeError"),
+        )
+        for relative_path, malformed, error_name in cases:
+            with self.subTest(relative_path=relative_path):
+                with tempfile.TemporaryDirectory() as temp_dir:
+                    workspace = Path(temp_dir)
+                    repo_path = workspace / "govoplan-files"
+                    repo_path.mkdir()
+                    (repo_path / "pyproject.toml").write_text(
+                        '[project]\nname = "govoplan-files"\nversion = "1.2.4"\n',
+                        encoding="utf-8",
+                    )
+                    (repo_path / relative_path).write_text(malformed, encoding="utf-8")
+
+                    plan = build_selective_release_plan(
+                        dashboard(workspace=workspace, version="1.2.4"),
+                        selected_repos=("govoplan-files",),
+                        repo_versions={"govoplan-files": "1.2.4"},
+                    )
+
+                self.assertEqual("blocked", plan.status)
+                finding = next(
+                    item
+                    for item in plan.gate_findings
+                    if item.code == "repository_version_metadata_unreadable"
+                )
+                self.assertEqual("govoplan-files", finding.repo)
+                self.assertIn(error_name, finding.message)
+                self.assertIn("Repair the malformed TOML or JSON", finding.remediation)
+                self.assertEqual("resolve_release_gate", plan.recommended_action.id)
+
+    def test_malformed_webui_metadata_is_a_structured_blocker(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            workspace = Path(temp_dir)
+            repo_path = workspace / "govoplan-files"
+            (repo_path / "webui").mkdir(parents=True)
+            (repo_path / "pyproject.toml").write_text(
+                '[project]\nname = "govoplan-files"\nversion = "1.2.4"\n',
+                encoding="utf-8",
+            )
+            (repo_path / "webui" / "package.json").write_text(
+                "{not-json\n", encoding="utf-8"
+            )
+
+            plan = build_selective_release_plan(
+                dashboard(workspace=workspace, version="1.2.4"),
+                selected_repos=("govoplan-files",),
+                repo_versions={"govoplan-files": "1.2.4"},
+            )
+
+        self.assertEqual("blocked", plan.status)
+        codes = {item.code for item in plan.gate_findings}
+        self.assertIn("repository_version_metadata_unreadable", codes)
+        finding = next(
+            item
+            for item in plan.gate_findings
+            if item.code == "release_webui_metadata_unreadable"
+        )
+        self.assertIn("JSONDecodeError", finding.message)
+        self.assertIn("Repair malformed JSON", finding.remediation)
+        self.assertEqual("resolve_release_gate", plan.recommended_action.id)
+
+    def test_webui_renders_recommendation_and_remediation(self) -> None:
+        webui = (RELEASE_ROOT / "webui" / "index.html").read_text(encoding="utf-8")
+
+        self.assertIn("plan.gate_findings", webui)
+        self.assertIn("plan.recommended_action", webui)
+        self.assertIn("recommended next", webui)
+        self.assertIn('plan.status === "blocked" ? "block"', webui)
+        self.assertIn('pill("recommended next", recommendationKind)', webui)
+        self.assertIn("<strong>Remediation:</strong>", webui)
+
+
+def dashboard(*, workspace: Path, version: str) -> ReleaseDashboard:
+    repo = RepositorySnapshot(
+        spec=RepositorySpec(
+            name="govoplan-files",
+            category="module",
+            subtype="infrastructure",
+            remote="git@example.test:add-ideas/govoplan-files.git",
+            path="govoplan-files",
+        ),
+        absolute_path=str(workspace / "govoplan-files"),
+        exists=True,
+        is_git=True,
+        has_head=True,
+        branch="main",
+        versions=VersionSnapshot(pyproject=version),
+        local_target_tag_exists=False,
+    )
+    return ReleaseDashboard(
+        generated_at="2026-07-22T00:00:00Z",
+        meta_root=str(workspace / "govoplan"),
+        workspace_root=str(workspace),
+        target_version=None,
+        target_tag=None,
+        online=False,
+        include_migrations=False,
+        summary=DashboardSummary(
+            repository_count=1,
+            missing_count=0,
+            dirty_count=0,
+            ahead_count=0,
+            behind_count=0,
+            no_head_count=0,
+            error_count=0,
+            safe_directory_count=0,
+            local_target_tag_missing_count=0,
+            status="ready",
+        ),
+        repositories=(repo,),
+        catalog=CatalogSnapshot(
+            channel="stable",
+            website_path="",
+            catalog_path="",
+            catalog_exists=False,
+            keyring_path="",
+            keyring_exists=False,
+        ),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_release_repository_tag.py

@@ -118,9 +118,11 @@ class ReleaseRepositoryTagTests(unittest.TestCase):
 
     def test_empty_manifest_workspace_blocks_source_tagging(self) -> None:
         workspace = self.root / "empty-manifest-workspace"
+        remote_root = self.root / "empty-manifest-remotes"
         workspace.mkdir()
+        remote_root.mkdir()
         core, _ = create_release_repo(
-            root=self.root,
+            root=remote_root,
             workspace=workspace,
             name="govoplan-core",
             version="0.1.10",

tests/test_version_alignment.py

@@ -251,6 +251,14 @@ class VersionAlignmentTests(unittest.TestCase):
         self.assertEqual("blocked", plan.status)
         self.assertEqual("blocked", plan.units[0].status)
         self.assertTrue(any("Core release WebUI dependency" in item for item in plan.units[0].blockers))
+        finding = next(
+            item
+            for item in plan.gate_findings
+            if item.code == "release_webui_composition_alignment"
+        )
+        self.assertIn("package.release.json", finding.source)
+        self.assertIn("regenerate webui/package-lock.release.json", finding.remediation)
+        self.assertEqual("resolve_release_gate", plan.recommended_action.id)
 
     def test_selected_repository_gate_reports_missing_version_metadata(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:

tools/assessments/capability-fit.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+"""Re-run a capability-fit assessment against a trusted release catalog."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+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
+
+
+TRUSTED_KEYRING_FILE_ENV = "GOVOPLAN_MODULE_PACKAGE_CATALOG_TRUSTED_KEYS_FILE"
+
+
+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="Published or candidate keyring whose hash is pinned by --catalog.",
+    )
+    parser.add_argument(
+        "--trusted-keyring",
+        type=Path,
+        help=(
+            "Independently pinned local keyring used as the catalog signature trust "
+            f"root; defaults to ${TRUSTED_KEYRING_FILE_ENV}."
+        ),
+    )
+    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")
+    configured_trusted_keyring = os.getenv(TRUSTED_KEYRING_FILE_ENV, "").strip()
+    trusted_keyring_path = args.trusted_keyring or (
+        Path(configured_trusted_keyring) if configured_trusted_keyring else None
+    )
+    if trusted_keyring_path is None:
+        raise SystemExit(
+            f"--trusted-keyring or ${TRUSTED_KEYRING_FILE_ENV} is required"
+        )
+    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",
+        )
+        published_keyring = fetch_public_json(
+            f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/keyring.json",
+            label="published keyring",
+        )
+    else:
+        catalog = read_object(args.catalog, label="catalog")
+        published_keyring = read_object(args.keyring, label="published keyring")
+    trusted_keyring = read_object(
+        trusted_keyring_path,
+        label="trusted keyring",
+    )
+
+    report = review_capability_fit(
+        assessment=assessment,
+        schema=schema,
+        catalog=catalog,
+        published_keyring=published_keyring,
+        trusted_keyring=trusted_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())

tools/assessments/govoplan_assessment/__init__.py

@@ -0,0 +1,5 @@
+"""Repeatable GovOPlaN product-assessment tooling."""
+
+from .capability_fit import review_capability_fit, render_review
+
+__all__ = ("render_review", "review_capability_fit")

tools/assessments/govoplan_assessment/capability_fit.py

@@ -0,0 +1,856 @@
+"""Compare a capability-fit assessment with a signed release catalog.
+
+The result is deliberately limited to release metadata and local tag
+provenance. It never implies installed-artifact, target-provider, or production
+proof.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from datetime import UTC, datetime
+import hashlib
+import json
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+from typing import Any, Iterable
+from unittest import mock
+
+from jsonschema import Draft202012Validator, FormatChecker
+from jsonschema.exceptions import SchemaError
+
+import govoplan_core.core.module_package_catalog as module_package_catalog
+from govoplan_release.source_provenance import catalog_source_selection
+from govoplan_release.version_alignment import candidate_catalog_version_issues
+
+
+RELEASE_REF_PATTERN = re.compile(
+    r"^(?P<channel>[a-z][a-z0-9_-]*)-catalog-(?P<sequence>[0-9]+)$"
+)
+TAG_PATTERN = re.compile(r"(?:[.]git[@#])(?P<tag>[^\s]+)$")
+REPOSITORY_ID_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
+
+
+@dataclass(frozen=True, slots=True)
+class Finding:
+    severity: str
+    code: str
+    message: str
+    assessment_ids: tuple[str, ...] = ()
+
+
+def review_capability_fit(
+    *,
+    assessment: dict[str, Any],
+    schema: dict[str, Any],
+    catalog: dict[str, Any],
+    published_keyring: dict[str, Any],
+    trusted_keyring: dict[str, Any],
+    workspace_root: Path | None = None,
+) -> dict[str, Any]:
+    """Return a secret-free release-drift report for one fit assessment."""
+
+    findings: list[Finding] = []
+    schema_errors = validate_assessment(assessment=assessment, schema=schema)
+    findings.extend(
+        Finding("blocker", "assessment_schema", message) for message in schema_errors
+    )
+    if schema_errors:
+        return build_report(
+            assessment=assessment,
+            catalog=catalog,
+            findings=findings,
+            changes=[],
+            review_targets=[],
+            catalog_checked=False,
+            local_tag_provenance_attempted=0,
+            local_tag_provenance_expected=0,
+            local_tag_catalog_scope_complete=False,
+        )
+
+    catalog_validation = validate_catalog(
+        catalog=catalog,
+        trusted_keyring=trusted_keyring,
+    )
+    if catalog_validation.get("valid") is not True:
+        findings.append(
+            Finding(
+                "blocker",
+                "catalog_trust",
+                str(catalog_validation.get("error") or "Catalog validation failed."),
+                ("assessment.release",),
+            )
+        )
+    findings.extend(
+        Finding(
+            "blocker",
+            "catalog_release_metadata",
+            (
+                f"{issue.source}: {issue.message}; expected {issue.expected!r}, "
+                f"found {issue.actual!r}."
+            ),
+            ("assessment.release",),
+        )
+        for issue in candidate_catalog_version_issues(catalog)
+    )
+    source_selection = catalog_source_selection(catalog)
+    findings.extend(
+        Finding(
+            "blocker",
+            "catalog_source_provenance",
+            issue.message,
+            ("assessment.release",),
+        )
+        for issue in source_selection.issues
+    )
+    expected_keyring_hash = _catalog_keyring_hash(catalog)
+    actual_keyring_hash = canonical_hash(published_keyring)
+    if expected_keyring_hash is None:
+        findings.append(
+            Finding(
+                "blocker",
+                "catalog_keyring_hash_missing",
+                "Catalog release metadata does not pin the published keyring hash.",
+                ("assessment.release",),
+            )
+        )
+    elif expected_keyring_hash != actual_keyring_hash:
+        findings.append(
+            Finding(
+                "blocker",
+                "catalog_keyring_hash_mismatch",
+                "The published keyring does not match the hash pinned by the catalog.",
+                ("assessment.release",),
+            )
+        )
+
+    release = assessment["release"]
+    ref_match = RELEASE_REF_PATTERN.fullmatch(str(release["ref"]))
+    if ref_match is None:
+        findings.append(
+            Finding(
+                "blocker",
+                "assessment_release_ref",
+                "Assessment release.ref must have the form '<channel>-catalog-<sequence>'.",
+                ("assessment.release",),
+            )
+        )
+        assessed_channel = None
+        assessed_sequence = None
+    else:
+        assessed_channel = ref_match.group("channel")
+        assessed_sequence = int(ref_match.group("sequence"))
+
+    catalog_channel = _text(catalog.get("channel"))
+    catalog_sequence = _integer(catalog.get("sequence"))
+    if assessed_channel is not None and catalog_channel != assessed_channel:
+        findings.append(
+            Finding(
+                "blocker",
+                "catalog_channel_mismatch",
+                f"Assessment pins channel {assessed_channel!r}, but the supplied catalog is {catalog_channel!r}.",
+                ("assessment.release",),
+            )
+        )
+    if assessed_sequence is not None and catalog_sequence != assessed_sequence:
+        findings.append(
+            Finding(
+                "review",
+                "catalog_sequence_changed",
+                f"Assessment pins catalog sequence {assessed_sequence}; supplied catalog sequence is {catalog_sequence!r}.",
+                ("assessment.release",),
+            )
+        )
+
+    catalog_entries, entry_findings = catalog_entries_by_id(catalog)
+    findings.extend(entry_findings)
+    changes: list[dict[str, Any]] = []
+    changed_repositories: set[str] = set()
+    assessed_module_ids: set[str] = set()
+    composition = assessment["composition"]
+    local_tag_provenance_expected = len(composition)
+    local_tag_provenance_attempted = 0
+    local_tag_catalog_scope_complete = True
+
+    for module in composition:
+        module_id = str(module["module_id"])
+        repository = str(module["repository"])
+        assessed_version = str(module["manifest_version"])
+        selected_version = source_selection.selected_versions.get(repository)
+        selected_commit = source_selection.selected_commits.get(repository)
+        selected_tag_object = source_selection.selected_tag_objects.get(repository)
+        assessed_module_ids.add(module_id)
+        if workspace_root is not None:
+            local_tag_provenance_attempted += 1
+            provenance = local_tag_provenance(
+                workspace_root=workspace_root,
+                repository=repository,
+                version=assessed_version,
+                assessed_commit=str(module["commit"]),
+                selected_commit=(
+                    selected_commit if selected_version == assessed_version else None
+                ),
+                selected_tag_object=(
+                    selected_tag_object
+                    if selected_version == assessed_version
+                    else None
+                ),
+            )
+            if provenance is not None:
+                changes.append(
+                    {
+                        "module_id": module_id,
+                        "repository": repository,
+                        "kind": provenance["kind"],
+                        "assessed_commit": module["commit"],
+                        "tag_commit": provenance.get("tag_commit"),
+                        "tag_object": provenance.get("tag_object"),
+                    }
+                )
+                changed_repositories.add(repository)
+                findings.append(
+                    Finding(
+                        "review",
+                        "tag_provenance_changed",
+                        str(provenance["message"]),
+                        (f"composition.{module_id}",),
+                    )
+                )
+        entry = catalog_entries.get(module_id)
+        if entry is None:
+            local_tag_catalog_scope_complete = False
+            changes.append(
+                {
+                    "module_id": module_id,
+                    "repository": repository,
+                    "kind": "missing_from_catalog",
+                    "assessed_version": module["manifest_version"],
+                    "catalog_version": None,
+                }
+            )
+            changed_repositories.add(repository)
+            findings.append(
+                Finding(
+                    "review",
+                    "composition_entry_missing",
+                    f"Assessed module {module_id!r} is not present in the supplied catalog.",
+                    (f"composition.{module_id}",),
+                )
+            )
+            continue
+
+        catalog_repository = entry["repository"]
+        if catalog_repository != repository:
+            changes.append(
+                {
+                    "module_id": module_id,
+                    "repository": repository,
+                    "kind": "repository_changed",
+                    "assessed_repository": repository,
+                    "catalog_repository": catalog_repository,
+                }
+            )
+            changed_repositories.add(repository)
+            findings.append(
+                Finding(
+                    "review",
+                    "composition_repository_changed",
+                    f"Module {module_id!r} maps to {catalog_repository!r} in the catalog, not {repository!r}.",
+                    (f"composition.{module_id}",),
+                )
+            )
+
+        catalog_version = entry["version"]
+        if assessed_version != catalog_version:
+            changes.append(
+                {
+                    "module_id": module_id,
+                    "repository": repository,
+                    "kind": "version_changed",
+                    "assessed_version": assessed_version,
+                    "catalog_version": catalog_version,
+                }
+            )
+            changed_repositories.add(repository)
+            findings.append(
+                Finding(
+                    "review",
+                    "composition_version_changed",
+                    f"Module {module_id!r} changed from {assessed_version!r} to {catalog_version!r}.",
+                    (f"composition.{module_id}",),
+                )
+            )
+
+        assessed_commit = str(module["commit"]).lower()
+        if (
+            selected_version == assessed_version
+            and selected_commit is not None
+            and not selected_commit.lower().startswith(assessed_commit)
+        ):
+            changes.append(
+                {
+                    "module_id": module_id,
+                    "repository": repository,
+                    "kind": "commit_changed",
+                    "assessed_commit": module["commit"],
+                    "catalog_commit": selected_commit,
+                }
+            )
+            changed_repositories.add(repository)
+            findings.append(
+                Finding(
+                    "review",
+                    "composition_commit_changed",
+                    (
+                        f"Signed selected-unit provenance for {repository!r} at "
+                        f"{assessed_version!r} points to commit {selected_commit[:12]}, "
+                        f"not assessed commit {module['commit']}."
+                    ),
+                    (f"composition.{module_id}",),
+                )
+            )
+
+        expected_tag = f"v{catalog_version}" if catalog_version else None
+        catalog_tags = entry["tags"]
+        if (
+            expected_tag is None
+            or not catalog_tags
+            or any(tag != expected_tag for tag in catalog_tags)
+        ):
+            findings.append(
+                Finding(
+                    "blocker",
+                    "catalog_ref_version_mismatch",
+                    f"Catalog refs for {module_id!r} do not consistently select expected tag {expected_tag!r}: {catalog_tags!r}.",
+                    (f"composition.{module_id}",),
+                )
+            )
+
+    extras = sorted(set(catalog_entries) - assessed_module_ids)
+    if extras:
+        findings.append(
+            Finding(
+                "info",
+                "catalog_entries_outside_scope",
+                "Catalog entries outside this assessment composition: "
+                + ", ".join(extras)
+                + ".",
+            )
+        )
+
+    review_targets = conclusions_needing_review(
+        assessment=assessment,
+        changed_repositories=changed_repositories,
+        release_changed=assessed_sequence != catalog_sequence,
+    )
+    return build_report(
+        assessment=assessment,
+        catalog=catalog,
+        findings=findings,
+        changes=changes,
+        review_targets=review_targets,
+        catalog_checked=True,
+        local_tag_provenance_attempted=local_tag_provenance_attempted,
+        local_tag_provenance_expected=local_tag_provenance_expected,
+        local_tag_catalog_scope_complete=local_tag_catalog_scope_complete,
+    )
+
+
+def validate_assessment(
+    *, assessment: dict[str, Any], schema: dict[str, Any]
+) -> list[str]:
+    try:
+        Draft202012Validator.check_schema(schema)
+    except SchemaError as exc:
+        return [f"$schema: {exc.message}"]
+    validator = Draft202012Validator(schema, format_checker=FormatChecker())
+    return [
+        f"{_json_path(error.absolute_path)}: {error.message}"
+        for error in sorted(
+            validator.iter_errors(assessment),
+            key=lambda item: (
+                tuple(str(part) for part in item.absolute_path),
+                str(item.validator),
+                item.message,
+            ),
+        )
+    ]
+
+
+def validate_catalog(
+    *, catalog: dict[str, Any], trusted_keyring: dict[str, Any]
+) -> dict[str, object]:
+    trusted_keys = trusted_keys_from_keyring(trusted_keyring)
+    channel = _text(catalog.get("channel"))
+    with tempfile.NamedTemporaryFile(
+        mode="w", suffix=".json", encoding="utf-8"
+    ) as handle:
+        json.dump(catalog, handle)
+        handle.flush()
+        # Installer replay state answers whether a catalog may be accepted again
+        # by one runtime. This read-only assessment instead compares explicit
+        # inputs, so ambient installer state must not affect its result.
+        with (
+            mock.patch.object(
+                module_package_catalog,
+                "_configured_sequence_state_path",
+                return_value=None,
+            ),
+            mock.patch.object(
+                module_package_catalog,
+                "_configured_enforce_sequence",
+                return_value=False,
+            ),
+        ):
+            return module_package_catalog.validate_module_package_catalog(
+                Path(handle.name),
+                require_trusted=True,
+                approved_channels=(channel,) if channel else (),
+                trusted_keys=trusted_keys,
+            )
+
+
+def trusted_keys_from_keyring(payload: dict[str, Any]) -> dict[str, str]:
+    now = datetime.now(tz=UTC)
+    result: dict[str, str] = {}
+    keys = payload.get("keys")
+    if not isinstance(keys, list):
+        return {
+            str(key): str(value)
+            for key, value in payload.items()
+            if str(key).strip() and str(value).strip()
+        }
+    for item in keys:
+        if not isinstance(item, dict):
+            continue
+        if str(item.get("status") or "active").strip().lower() in {
+            "revoked",
+            "disabled",
+            "retired",
+        }:
+            continue
+        raw_not_before = item.get("not_before")
+        raw_not_after = item.get("not_after")
+        not_before = _datetime(raw_not_before)
+        not_after = _datetime(raw_not_after)
+        if raw_not_before is not None and not_before is None:
+            continue
+        if raw_not_after is not None and not_after is None:
+            continue
+        if not_before is not None and now < not_before:
+            continue
+        if not_after is not None and now > not_after:
+            continue
+        key_id = _text(item.get("key_id"))
+        public_key = _text(item.get("public_key")) or _text(
+            item.get("public_key_base64")
+        )
+        if key_id and public_key:
+            result[key_id] = public_key
+    return result
+
+
+def catalog_entries_by_id(
+    catalog: dict[str, Any],
+) -> tuple[dict[str, dict[str, Any]], list[Finding]]:
+    findings: list[Finding] = []
+    entries: dict[str, dict[str, Any]] = {}
+    raw_entries: list[tuple[str, dict[str, Any]]] = []
+    core = catalog.get("core_release")
+    if isinstance(core, dict):
+        raw_entries.append(("core", core))
+    modules = catalog.get("modules")
+    if isinstance(modules, list):
+        raw_entries.extend(
+            (str(item.get("module_id") or ""), item)
+            for item in modules
+            if isinstance(item, dict)
+        )
+    for module_id, item in raw_entries:
+        if not module_id:
+            findings.append(
+                Finding(
+                    "blocker",
+                    "catalog_module_id_missing",
+                    "Catalog entry has no module_id.",
+                )
+            )
+            continue
+        if module_id in entries:
+            findings.append(
+                Finding(
+                    "blocker",
+                    "catalog_module_id_duplicate",
+                    f"Catalog repeats module_id {module_id!r}.",
+                )
+            )
+            continue
+        refs = [
+            value
+            for value in (item.get("python_ref"), item.get("webui_ref"))
+            if isinstance(value, str) and value
+        ]
+        entries[module_id] = {
+            "module_id": module_id,
+            "repository": catalog_repository(item),
+            "version": _text(item.get("version")),
+            "tags": tuple(
+                tag for tag in (catalog_ref_tag(value) for value in refs) if tag
+            ),
+        }
+    return entries, findings
+
+
+def catalog_repository(entry: dict[str, Any]) -> str | None:
+    package = _text(entry.get("python_package"))
+    if package:
+        return package.split("[", 1)[0]
+    for field in ("python_ref", "webui_ref"):
+        value = _text(entry.get(field))
+        if not value:
+            continue
+        match = re.search(r"/([^/@#]+)[.]git(?:[@#]|$)", value)
+        if match:
+            return match.group(1)
+    return None
+
+
+def catalog_ref_tag(value: str) -> str | None:
+    match = TAG_PATTERN.search(value)
+    return match.group("tag") if match else None
+
+
+def local_tag_provenance(
+    *,
+    workspace_root: Path,
+    repository: str,
+    version: str,
+    assessed_commit: str,
+    selected_commit: str | None = None,
+    selected_tag_object: str | None = None,
+) -> dict[str, str] | None:
+    if REPOSITORY_ID_PATTERN.fullmatch(repository) is None:
+        return {
+            "kind": "tag_provenance_unavailable",
+            "message": "Cannot verify tag provenance for an invalid repository identifier.",
+        }
+    resolved_workspace = workspace_root.resolve()
+    repo = (resolved_workspace / repository).resolve()
+    if repo.parent != resolved_workspace:
+        return {
+            "kind": "tag_provenance_unavailable",
+            "message": f"Cannot verify {repository} tag provenance outside the configured workspace.",
+        }
+    if not (repo / ".git").exists():
+        return {
+            "kind": "tag_provenance_unavailable",
+            "message": f"Cannot verify {repository} tag provenance because the local checkout is unavailable.",
+        }
+    try:
+        tag_type = subprocess.run(
+            ["git", "-C", str(repo), "cat-file", "-t", f"refs/tags/v{version}"],
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+        result = subprocess.run(
+            [
+                "git",
+                "-C",
+                str(repo),
+                "rev-parse",
+                "--verify",
+                f"refs/tags/v{version}^{{commit}}",
+            ],
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+        tag_object_result = subprocess.run(
+            [
+                "git",
+                "-C",
+                str(repo),
+                "rev-parse",
+                "--verify",
+                f"refs/tags/v{version}",
+            ],
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+    except OSError:
+        return {
+            "kind": "tag_provenance_unavailable",
+            "message": f"Cannot run the local tag provenance check for {repository}.",
+        }
+    if tag_type.returncode != 0:
+        return {
+            "kind": "tag_provenance_unavailable",
+            "message": f"Cannot inspect {repository} tag v{version} in the local checkout.",
+        }
+    if tag_type.stdout.strip() != "tag":
+        return {
+            "kind": "tag_not_annotated",
+            "message": f"{repository} v{version} is not an annotated release tag.",
+        }
+    if result.returncode != 0:
+        return {
+            "kind": "tag_provenance_unavailable",
+            "message": f"Cannot peel {repository} tag v{version} in the local checkout.",
+        }
+    if tag_object_result.returncode != 0:
+        return {
+            "kind": "tag_provenance_unavailable",
+            "message": f"Cannot resolve {repository} tag object v{version} in the local checkout.",
+        }
+    tag_commit = result.stdout.strip().lower()
+    tag_object = tag_object_result.stdout.strip().lower()
+    if not tag_commit.startswith(assessed_commit.lower()):
+        return {
+            "kind": "tag_commit_changed",
+            "message": f"{repository} v{version} peels to {tag_commit[:12]}, not assessed commit {assessed_commit}.",
+            "tag_commit": tag_commit,
+        }
+    if selected_commit is not None and tag_commit != selected_commit.lower():
+        return {
+            "kind": "tag_signed_commit_changed",
+            "message": (
+                f"{repository} v{version} peels to {tag_commit[:12]}, not signed "
+                f"selected-unit commit {selected_commit[:12]}."
+            ),
+            "tag_commit": tag_commit,
+            "tag_object": tag_object,
+        }
+    if selected_tag_object is not None and tag_object != selected_tag_object.lower():
+        return {
+            "kind": "tag_object_changed",
+            "message": (
+                f"{repository} v{version} resolves to annotated tag object "
+                f"{tag_object[:12]}, not signed selected-unit tag object "
+                f"{selected_tag_object[:12]}."
+            ),
+            "tag_commit": tag_commit,
+            "tag_object": tag_object,
+        }
+    return None
+
+
+def conclusions_needing_review(
+    *,
+    assessment: dict[str, Any],
+    changed_repositories: set[str],
+    release_changed: bool,
+) -> list[dict[str, Any]]:
+    targets: dict[str, dict[str, Any]] = {}
+    if release_changed:
+        targets["assessment.release"] = {
+            "id": "assessment.release",
+            "section": "release",
+            "reason": "Catalog sequence changed; review release-scoped evidence and conclusions.",
+        }
+    for section in ("capabilities", "infrastructure"):
+        for item in assessment.get(section, []):
+            if not isinstance(item, dict):
+                continue
+            locators = "\n".join(
+                str(evidence.get("locator") or "")
+                for evidence in item.get("evidence", [])
+                if isinstance(evidence, dict)
+            ).lower()
+            matched = sorted(
+                repository
+                for repository in changed_repositories
+                if repository.lower() in locators
+            )
+            if not matched:
+                continue
+            item_id = str(item["id"])
+            targets[item_id] = {
+                "id": item_id,
+                "section": section,
+                "current_status": item.get("status"),
+                "repositories": matched,
+                "reason": "Evidence references a repository whose assessed release changed.",
+            }
+    for module in assessment.get("composition", []):
+        if (
+            not isinstance(module, dict)
+            or module.get("repository") not in changed_repositories
+        ):
+            continue
+        module_id = str(module.get("module_id"))
+        target_id = f"composition.{module_id}"
+        targets[target_id] = {
+            "id": target_id,
+            "section": "composition",
+            "repositories": [module.get("repository")],
+            "reason": "Pinned module release or tag provenance changed.",
+        }
+    return [targets[key] for key in sorted(targets)]
+
+
+def build_report(
+    *,
+    assessment: dict[str, Any],
+    catalog: dict[str, Any],
+    findings: Iterable[Finding],
+    changes: list[dict[str, Any]],
+    review_targets: list[dict[str, Any]],
+    catalog_checked: bool,
+    local_tag_provenance_attempted: int,
+    local_tag_provenance_expected: int,
+    local_tag_catalog_scope_complete: bool,
+) -> dict[str, Any]:
+    finding_list = sorted(
+        findings,
+        key=lambda item: (_severity_rank(item.severity), item.code, item.message),
+    )
+    finding_codes = {item.code for item in finding_list}
+    schema_valid = "assessment_schema" not in finding_codes
+    trusted_signature_valid = catalog_checked and "catalog_trust" not in finding_codes
+    published_keyring_valid = catalog_checked and not finding_codes.intersection(
+        {
+            "catalog_keyring_hash_missing",
+            "catalog_keyring_hash_mismatch",
+        }
+    )
+    catalog_valid = trusted_signature_valid and published_keyring_valid
+    release_valid = catalog_valid and not any(
+        item.severity in {"blocker", "review"} for item in finding_list
+    )
+    tag_provenance_checked = (
+        local_tag_provenance_expected > 0
+        and local_tag_provenance_attempted == local_tag_provenance_expected
+        and local_tag_catalog_scope_complete
+    )
+    tag_provenance_valid = (
+        tag_provenance_checked and "tag_provenance_changed" not in finding_codes
+    )
+    if any(item.severity == "blocker" for item in finding_list):
+        status = "blocked"
+    elif any(item.severity == "review" for item in finding_list):
+        status = "review_required"
+    else:
+        status = "current"
+    return {
+        "report_version": "0.1.0",
+        "status": status,
+        "assessment_id": assessment.get("assessment_id"),
+        "assessment_release": assessment.get("release", {}).get("ref")
+        if isinstance(assessment.get("release"), dict)
+        else None,
+        "catalog": {
+            "channel": catalog.get("channel"),
+            "sequence": catalog.get("sequence"),
+            "generated_at": catalog.get("generated_at"),
+        },
+        "proof_scope": {
+            "assessment_schema": {"checked": True, "valid": schema_valid},
+            "catalog_signature_and_keyring": {
+                "checked": catalog_checked,
+                "valid": catalog_valid if catalog_checked else None,
+            },
+            "catalog_signature_and_trusted_keyring": {
+                "checked": catalog_checked,
+                "valid": trusted_signature_valid if catalog_checked else None,
+            },
+            "published_keyring_hash": {
+                "checked": catalog_checked,
+                "valid": published_keyring_valid if catalog_checked else None,
+            },
+            "release_metadata": {
+                "checked": catalog_checked,
+                "valid": release_valid if catalog_checked else None,
+            },
+            "local_tag_provenance": {
+                "checked": tag_provenance_checked,
+                "valid": tag_provenance_valid if tag_provenance_checked else None,
+                "attempted_count": local_tag_provenance_attempted,
+                "expected_count": local_tag_provenance_expected,
+            },
+            "installed_artifacts": {"checked": False, "valid": None},
+            "target_environment": {"checked": False, "valid": None},
+            "external_providers": {"checked": False, "valid": None},
+            "production_approval": {"checked": False, "valid": None},
+        },
+        "findings": [asdict(item) for item in finding_list],
+        "changes": sorted(
+            changes,
+            key=lambda item: (str(item.get("module_id")), str(item.get("kind"))),
+        ),
+        "review_targets": review_targets,
+    }
+
+
+def render_review(report: dict[str, Any]) -> str:
+    lines = [
+        f"Capability fit rerun: {report['status']}",
+        f"Assessment: {report.get('assessment_id') or '-'} ({report.get('assessment_release') or '-'})",
+        f"Catalog: {report['catalog'].get('channel') or '-'} sequence {report['catalog'].get('sequence') or '-'}",
+        "Scope: schema, signed release metadata, and optional local tag provenance only.",
+    ]
+    findings = report.get("findings") or []
+    if findings:
+        lines.append("Findings:")
+        lines.extend(
+            f"- [{item['severity']}] {item['code']}: {item['message']}"
+            for item in findings
+        )
+    else:
+        lines.append("Findings: none")
+    targets = report.get("review_targets") or []
+    if targets:
+        lines.append("Conclusions requiring review:")
+        lines.extend(f"- {item['id']}: {item['reason']}" for item in targets)
+    lines.append(
+        "No installed-artifact, target-provider, or production proof was performed."
+    )
+    return "\n".join(lines) + "\n"
+
+
+def canonical_hash(payload: object) -> str:
+    data = json.dumps(
+        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True
+    ).encode("utf-8")
+    return hashlib.sha256(data).hexdigest()
+
+
+def _catalog_keyring_hash(catalog: dict[str, Any]) -> str | None:
+    release = catalog.get("release")
+    return _text(release.get("keyring_sha256")) if isinstance(release, dict) else None
+
+
+def _datetime(value: object) -> datetime | None:
+    text = _text(value)
+    if text is None:
+        return None
+    try:
+        parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)
+
+
+def _integer(value: object) -> int | None:
+    try:
+        return int(value) if value is not None else None
+    except (TypeError, ValueError):
+        return None
+
+
+def _json_path(parts: Iterable[object]) -> str:
+    path = "$"
+    for part in parts:
+        path += f"[{part}]" if isinstance(part, int) else f".{part}"
+    return path
+
+
+def _severity_rank(severity: str) -> int:
+    return {"blocker": 0, "review": 1, "info": 2}.get(severity, 3)
+
+
+def _text(value: object) -> str | None:
+    return value if isinstance(value, str) and value else None

tools/release/govoplan_release/catalog.py

@@ -7,11 +7,12 @@ import json
 from pathlib import Path
 from urllib.error import HTTPError, URLError
 
-from .http_fetch import fetch_http
+from .http_fetch import DEFAULT_MAX_RESPONSE_BYTES, fetch_http
 from .model import CatalogSnapshot
 from .workspace import website_root
 
 DEFAULT_PUBLIC_BASE_URL = "https://govoplan.add-ideas.de"
+MAX_PUBLIC_JSON_RESPONSE_BYTES = DEFAULT_MAX_RESPONSE_BYTES
 
 
 def collect_catalog_snapshot(
@@ -22,9 +23,13 @@ def collect_catalog_snapshot(
     check_public: bool = False,
 ) -> CatalogSnapshot:
     web_root = website_root(workspace_root)
-    catalog_path = web_root / "public" / "catalogs" / "v1" / "channels" / f"{channel}.json"
+    catalog_path = (
+        web_root / "public" / "catalogs" / "v1" / "channels" / f"{channel}.json"
+    )
     keyring_path = web_root / "public" / "catalogs" / "v1" / "keyring.json"
-    public_catalog_url = f"{public_base_url.rstrip('/')}/catalogs/v1/channels/{channel}.json"
+    public_catalog_url = (
+        f"{public_base_url.rstrip('/')}/catalogs/v1/channels/{channel}.json"
+    )
     public_keyring_url = f"{public_base_url.rstrip('/')}/catalogs/v1/keyring.json"
 
     catalog_version: str | None = None
@@ -129,8 +134,12 @@ def collect_catalog_snapshot(
         public_keyring_hash=public_keyring_hash,
         public_catalog_error=public_catalog_error,
         public_keyring_error=public_keyring_error,
-        catalog_matches_public=compare_hashes(local_catalog_hash, public_catalog_hash) if check_public else None,
-        keyring_matches_public=compare_hashes(local_keyring_hash, public_keyring_hash) if check_public else None,
+        catalog_matches_public=compare_hashes(local_catalog_hash, public_catalog_hash)
+        if check_public
+        else None,
+        keyring_matches_public=compare_hashes(local_keyring_hash, public_keyring_hash)
+        if check_public
+        else None,
     )
 
 
@@ -149,7 +158,9 @@ def parse_catalog_payload(payload: object) -> dict[str, str | int | None]:
     signatures = payload.get("signatures")
     return {
         "catalog_version": string_or_none(payload.get("catalog_version")),
-        "core_release_version": string_or_none(core_release.get("version")) if isinstance(core_release, dict) else None,
+        "core_release_version": string_or_none(core_release.get("version"))
+        if isinstance(core_release, dict)
+        else None,
         "module_count": len(modules) if isinstance(modules, list) else 0,
         "signature_count": len(signatures) if isinstance(signatures, list) else 0,
         "generated_at": string_or_none(payload.get("generated_at")),
@@ -164,6 +175,7 @@ def fetch_json(url: str) -> dict[str, object]:
             timeout=10,
             label="Public release catalog URL",
             headers={"User-Agent": "GovOPlaN release-console"},
+            max_response_bytes=MAX_PUBLIC_JSON_RESPONSE_BYTES,
         )
         return {"ok": True, "payload": json.loads(response.text())}
     except (HTTPError, URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc:
@@ -171,7 +183,9 @@ def fetch_json(url: str) -> dict[str, object]:
 
 
 def canonical_hash(payload: object) -> str:
-    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
+    canonical = json.dumps(
+        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True
+    ).encode("utf-8")
     return hashlib.sha256(canonical).hexdigest()
 
 

tools/release/govoplan_release/contracts.py

@@ -13,6 +13,7 @@ from pathlib import Path
 
 from .model import (
     CompatibilityIssue,
+    DashboardCollectionError,
     InterfaceProviderSnapshot,
     InterfaceRequirementSnapshot,
     ModuleContractSnapshot,
@@ -33,6 +34,43 @@ def collect_contracts(repositories: tuple[RepositorySnapshot, ...]) -> tuple[Mod
     return tuple(contracts)
 
 
+def collect_contracts_with_diagnostics(
+    repositories: tuple[RepositorySnapshot, ...],
+) -> tuple[tuple[ModuleContractSnapshot, ...], tuple[DashboardCollectionError, ...]]:
+    """Collect readable contracts and retain bounded parse diagnostics for the UI."""
+
+    contracts: list[ModuleContractSnapshot] = []
+    diagnostics: list[DashboardCollectionError] = []
+    for repo in repositories:
+        if not repo.exists or not repo.is_git:
+            continue
+        root = Path(repo.absolute_path)
+        for manifest in sorted(root.glob("src/**/backend/manifest.py")):
+            try:
+                contract = parse_manifest_contract(manifest, repo_name=repo.spec.name)
+            except Exception as exc:  # noqa: BLE001 - fail-closed UI diagnostic.
+                relative_source = manifest.relative_to(root).as_posix()
+                diagnostics.append(
+                    DashboardCollectionError(
+                        code="module_contract_unreadable",
+                        message=(
+                            "Module contract metadata could not be read or parsed "
+                            f"({type(exc).__name__})."
+                        ),
+                        remediation=(
+                            "Repair the manifest's Python syntax or read permissions, "
+                            "then refresh the dashboard before planning a release."
+                        ),
+                        repo=repo.spec.name,
+                        source=f"{repo.spec.name}/{relative_source}",
+                    )
+                )
+                continue
+            if contract is not None:
+                contracts.append(contract)
+    return tuple(contracts), tuple(diagnostics)
+
+
 def validate_contracts(contracts: tuple[ModuleContractSnapshot, ...]) -> tuple[CompatibilityIssue, ...]:
     """Validate the statically collected module interface graph.
 

tools/release/govoplan_release/dashboard.py

@@ -6,10 +6,10 @@ from datetime import UTC, datetime
 from pathlib import Path
 
 from .catalog import collect_catalog_snapshot
-from .contracts import collect_contracts
+from .contracts import collect_contracts_with_diagnostics
 from .git_state import collect_repository_snapshot, read_pyproject_version
 from .migrations import collect_migration_snapshots
-from .model import DashboardSummary, ReleaseDashboard, RepositorySnapshot
+from .model import DashboardCollectionError, DashboardSummary, ReleaseDashboard, RepositorySnapshot
 from .workspace import META_ROOT, core_root, load_repository_specs, resolve_workspace_root
 
 
@@ -26,10 +26,25 @@ def build_dashboard(
     channel: str = "stable",
 ) -> ReleaseDashboard:
     resolved_workspace = resolve_workspace_root(workspace_root)
+    collection_errors: list[DashboardCollectionError] = []
     if check_public_catalog is None:
         check_public_catalog = online
     if target_version is None:
-        target_version = read_pyproject_version(core_root(resolved_workspace))
+        try:
+            target_version = read_pyproject_version(core_root(resolved_workspace))
+        except Exception as exc:  # noqa: BLE001 - bounded fail-closed diagnostic.
+            collection_errors.append(
+                DashboardCollectionError(
+                    code="core_version_metadata_unreadable",
+                    message=f"Core version metadata could not be read or parsed ({type(exc).__name__}).",
+                    remediation=(
+                        "Repair govoplan-core/pyproject.toml, refresh the dashboard, "
+                        "then rebuild the release plan."
+                    ),
+                    repo="govoplan-core",
+                    source="govoplan-core/pyproject.toml",
+                )
+            )
     target_tag = f"v{target_version}" if target_version else None
 
     repositories = tuple(
@@ -38,7 +53,11 @@ def build_dashboard(
     )
     catalog = collect_catalog_snapshot(workspace_root=resolved_workspace, channel=channel, check_public=check_public_catalog)
     migrations = collect_migration_snapshots() if include_migrations else ()
-    contracts = collect_contracts(repositories) if include_contracts else ()
+    if include_contracts:
+        contracts, contract_errors = collect_contracts_with_diagnostics(repositories)
+        collection_errors.extend(contract_errors)
+    else:
+        contracts = ()
     return ReleaseDashboard(
         generated_at=datetime.now(tz=UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
         meta_root=str(META_ROOT),
@@ -47,28 +66,34 @@ def build_dashboard(
         target_tag=target_tag,
         online=online or check_remote_tags or check_public_catalog,
         include_migrations=include_migrations,
-        summary=summarize(repositories, target_tag=target_tag),
+        summary=summarize(repositories, target_tag=target_tag, collection_errors=tuple(collection_errors)),
         repositories=repositories,
         catalog=catalog,
         migrations=migrations,
         contracts=contracts,
+        collection_errors=tuple(collection_errors),
     )
 
 
-def summarize(repositories: tuple[RepositorySnapshot, ...], *, target_tag: str | None) -> DashboardSummary:
+def summarize(
+    repositories: tuple[RepositorySnapshot, ...],
+    *,
+    target_tag: str | None,
+    collection_errors: tuple[DashboardCollectionError, ...] = (),
+) -> DashboardSummary:
     missing_count = sum(1 for repo in repositories if not repo.exists or not repo.is_git)
     dirty_count = sum(1 for repo in repositories if repo.dirty)
     ahead_count = sum(1 for repo in repositories if repo.ahead)
     behind_count = sum(1 for repo in repositories if repo.behind)
     no_head_count = sum(1 for repo in repositories if repo.exists and repo.is_git and not repo.has_head)
-    error_count = sum(1 for repo in repositories if repo.errors)
+    error_count = sum(1 for repo in repositories if repo.errors) + len(collection_errors)
     safe_directory_count = sum(1 for repo in repositories if repo.safe_directory_required)
     local_target_tag_missing_count = (
         sum(1 for repo in repositories if repo.local_target_tag_exists is False and repo.versions.primary == target_tag.removeprefix("v"))
         if target_tag
         else 0
     )
-    if missing_count or behind_count or safe_directory_count:
+    if missing_count or behind_count or safe_directory_count or collection_errors:
         status = "blocked"
     elif dirty_count or ahead_count or no_head_count or error_count or local_target_tag_missing_count:
         status = "attention"

tools/release/govoplan_release/git_state.py

@@ -70,7 +70,7 @@ def collect_repository_snapshot(
     try:
         versions = collect_versions(path)
     except Exception as exc:  # noqa: BLE001 - surfaced in the dashboard.
-        errors.append(str(exc))
+        errors.append(f"version metadata could not be read or parsed ({type(exc).__name__})")
         versions = VersionSnapshot()
 
     return RepositorySnapshot(

tools/release/govoplan_release/http_fetch.py

@@ -8,6 +8,9 @@ from dataclasses import dataclass
 from typing import Mapping
 
 
+DEFAULT_MAX_RESPONSE_BYTES = 16 * 1024 * 1024
+
+
 @dataclass(frozen=True, slots=True)
 class HttpFetchResponse:
     status: int
@@ -34,11 +37,29 @@ def fetch_http(
     label: str = "URL",
     method: str = "GET",
     headers: Mapping[str, str] | None = None,
+    max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES,
 ) -> HttpFetchResponse:
-    request = urllib.request.Request(validate_http_url(url, label=label), headers=dict(headers or {}), method=method)
+    if max_response_bytes <= 0:
+        raise ValueError("max_response_bytes must be positive.")
+    request = urllib.request.Request(
+        validate_http_url(url, label=label),
+        headers=dict(headers or {}),
+        method=method,
+    )
     with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310 - URL is validated by validate_http_url.  # nosec B310  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
+        raw_length = response.headers.get("Content-Length")
+        if raw_length is not None:
+            try:
+                declared_length = int(raw_length)
+            except (TypeError, ValueError):
+                declared_length = None
+            if declared_length is not None and declared_length > max_response_bytes:
+                raise ValueError(f"{label} response exceeds the configured size limit.")
+        body = response.read(max_response_bytes + 1)
+        if len(body) > max_response_bytes:
+            raise ValueError(f"{label} response exceeds the configured size limit.")
         return HttpFetchResponse(
             status=int(getattr(response, "status", 0)),
             headers=dict(response.headers.items()),
-            body=response.read(),
+            body=body,
         )

tools/release/govoplan_release/model.py

@@ -111,6 +111,17 @@ class MigrationTrackSnapshot:
     error: str | None = None
 
 
+@dataclass(frozen=True, slots=True)
+class DashboardCollectionError:
+    """Bounded, actionable failure from read-only dashboard collection."""
+
+    code: str
+    message: str
+    remediation: str
+    repo: str | None = None
+    source: str | None = None
+
+
 @dataclass(frozen=True, slots=True)
 class DashboardSummary:
     repository_count: int
@@ -139,6 +150,7 @@ class ReleaseDashboard:
     catalog: CatalogSnapshot
     migrations: tuple[MigrationTrackSnapshot, ...] = ()
     contracts: tuple[ModuleContractSnapshot, ...] = ()
+    collection_errors: tuple[DashboardCollectionError, ...] = ()
 
 
 @dataclass(frozen=True, slots=True)
@@ -186,6 +198,18 @@ class ModuleContractSnapshot:
     requires_interfaces: tuple[InterfaceRequirementSnapshot, ...] = ()
 
 
+@dataclass(frozen=True, slots=True)
+class ReleaseGateFinding:
+    code: str
+    severity: str
+    message: str
+    remediation: str
+    repo: str | None = None
+    source: str | None = None
+    expected: str | None = None
+    actual: str | None = None
+
+
 @dataclass(frozen=True, slots=True)
 class ReleasePlanUnit:
     repo: str
@@ -200,6 +224,8 @@ class ReleasePlanUnit:
     warnings: tuple[str, ...] = ()
     provides_interfaces: tuple[InterfaceProviderSnapshot, ...] = ()
     requires_interfaces: tuple[InterfaceRequirementSnapshot, ...] = ()
+    gate_findings: tuple[ReleaseGateFinding, ...] = ()
+    source_preflight_ready: bool = False
 
 
 @dataclass(frozen=True, slots=True)
@@ -224,6 +250,15 @@ class ReleasePlanStep:
     status: str = "planned"
 
 
+@dataclass(frozen=True, slots=True)
+class ReleaseRecommendedAction:
+    id: str
+    title: str
+    detail: str
+    remediation: str
+    repo: str | None = None
+
+
 @dataclass(frozen=True, slots=True)
 class SelectiveReleasePlan:
     generated_at: str
@@ -233,6 +268,9 @@ class SelectiveReleasePlan:
     compatibility: tuple[CompatibilityIssue, ...]
     dry_run_steps: tuple[ReleasePlanStep, ...]
     notes: tuple[str, ...] = ()
+    gate_findings: tuple[ReleaseGateFinding, ...] = ()
+    recommended_action: ReleaseRecommendedAction | None = None
+    source_preflight_ready: bool = False
 
 
 @dataclass(frozen=True, slots=True)

tools/release/govoplan_release/planner.py

@@ -18,6 +18,17 @@ def build_release_plan(dashboard: ReleaseDashboard) -> ReleasePlan:
     target_version = dashboard.target_version
     target_tag = dashboard.target_tag
 
+    for index, error in enumerate(dashboard.collection_errors, start=1):
+        actions.append(
+            PlanAction(
+                id=f"dashboard:{error.code}:{index}",
+                title=f"Release metadata is unreadable in {error.repo or 'the workspace'}",
+                detail=f"{error.message} {error.remediation}",
+                severity="blocker",
+                repo=error.repo,
+            )
+        )
+
     for repo in dashboard.repositories:
         if not repo.exists:
             actions.append(

tools/release/govoplan_release/selective_planner.py

@@ -12,14 +12,19 @@ from .model import (
     CompatibilityIssue,
     InterfaceProviderSnapshot,
     InterfaceRequirementSnapshot,
+    ReleaseGateFinding,
     ReleaseDashboard,
     ReleasePlanStep,
     ReleasePlanUnit,
+    ReleaseRecommendedAction,
     RepositorySnapshot,
     SelectiveReleasePlan,
     ModuleContractSnapshot,
 )
-from .version_alignment import selected_release_webui_bundle_issues
+from .version_alignment import (
+    selected_release_webui_bundle_issues,
+    selected_repository_version_issues,
+)
 
 
 def build_selective_release_plan(
@@ -34,22 +39,130 @@ def build_selective_release_plan(
     repositories = selected_repositories(dashboard, selected_repos=selected_repos)
     contracts_by_repo = {contract.repo: contract for contract in dashboard.contracts}
     units = tuple(
-        build_unit(repo, target_version=repo_versions.get(repo.spec.name) or target_version, contracts=contracts_by_repo.get(repo.spec.name))
+        build_unit(
+            repo,
+            target_version=repo_versions.get(repo.spec.name) or target_version,
+            contracts=contracts_by_repo.get(repo.spec.name),
+        )
         for repo in repositories
     )
-    units = apply_release_webui_bundle_gate(units, workspace=Path(dashboard.workspace_root))
+    units = apply_repository_version_gate(
+        units, workspace=Path(dashboard.workspace_root)
+    )
+    units = apply_release_webui_bundle_gate(
+        units, workspace=Path(dashboard.workspace_root)
+    )
     compatibility = compatibility_issues(dashboard)
     steps = dry_run_steps(units=units, dashboard=dashboard, channel=channel)
     notes = release_notes(dashboard)
-    status = plan_status(units=units, compatibility=compatibility)
+    collection_findings = dashboard_collection_findings(dashboard)
+    findings = (*collection_findings, *(finding for unit in units for finding in unit.gate_findings))
+    status = plan_status(units=units, compatibility=compatibility, findings=findings)
     return SelectiveReleasePlan(
-        generated_at=datetime.now(tz=UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
+        generated_at=datetime.now(tz=UTC)
+        .replace(microsecond=0)
+        .isoformat()
+        .replace("+00:00", "Z"),
         target_channel=channel,
         status=status,
         units=units,
         compatibility=compatibility,
         dry_run_steps=steps,
         notes=notes,
+        gate_findings=findings,
+        recommended_action=recommended_action(
+            units=units,
+            findings=findings,
+            compatibility=compatibility,
+        ),
+        source_preflight_ready=not collection_findings
+        and bool(units)
+        and all(unit.source_preflight_ready for unit in units),
+    )
+
+
+def dashboard_collection_findings(dashboard: ReleaseDashboard) -> tuple[ReleaseGateFinding, ...]:
+    return tuple(
+        ReleaseGateFinding(
+            code=error.code,
+            severity="blocker",
+            message=error.message,
+            remediation=error.remediation,
+            repo=error.repo,
+            source=error.source,
+            expected="readable release metadata",
+            actual="unreadable",
+        )
+        for error in dashboard.collection_errors
+    )
+
+
+def apply_repository_version_gate(
+    units: tuple[ReleasePlanUnit, ...],
+    *,
+    workspace: Path,
+) -> tuple[ReleasePlanUnit, ...]:
+    issues_by_repo: dict[str, list[ReleaseGateFinding]] = {}
+    for unit in units:
+        try:
+            issues = selected_repository_version_issues(
+                repo_versions={unit.repo: unit.target_version},
+                workspace=workspace,
+            )
+        except Exception as exc:  # noqa: BLE001 - source errors become release findings.
+            issues_by_repo.setdefault(unit.repo, []).append(
+                ReleaseGateFinding(
+                    code="repository_version_metadata_unreadable",
+                    severity="blocker",
+                    message=(
+                        "Version-bearing source metadata could not be read or parsed "
+                        f"({type(exc).__name__})."
+                    ),
+                    remediation=(
+                        "Repair the malformed TOML or JSON version metadata, refresh the "
+                        "repository dashboard, then rebuild this plan."
+                    ),
+                    repo=unit.repo,
+                    source="repository version metadata",
+                    expected=unit.target_version,
+                    actual="unreadable",
+                )
+            )
+            continue
+        for issue in issues:
+            issues_by_repo.setdefault(issue.repo, []).append(
+                ReleaseGateFinding(
+                    code="repository_version_alignment",
+                    severity="blocker",
+                    message=issue.message,
+                    remediation=(
+                        f"Align {issue.source} and every other version-bearing source declaration to "
+                        f"{issue.expected!r}. Regenerate lockfiles instead of editing resolved entries by hand, "
+                        "commit the reviewed changes, then rebuild this plan."
+                    ),
+                    repo=issue.repo,
+                    source=issue.source,
+                    expected=issue.expected,
+                    actual=issue.actual,
+                )
+            )
+    return tuple(
+        replace(
+            unit,
+            status="blocked",
+            blockers=(
+                *unit.blockers,
+                *(
+                    f"{finding.source}={finding.actual!r}, expected {finding.expected!r} ({finding.message})"
+                    for finding in issues_by_repo[unit.repo]
+                ),
+            ),
+            gate_findings=(*unit.gate_findings, *issues_by_repo[unit.repo]),
+            source_preflight_ready=False,
+        )
+        if unit.repo in issues_by_repo
+        else unit
+        for unit in units
     )
 
 
@@ -58,19 +171,63 @@ def apply_release_webui_bundle_gate(
     *,
     workspace: Path,
 ) -> tuple[ReleasePlanUnit, ...]:
-    issues_by_repo: dict[str, list[str]] = {}
-    for issue in selected_release_webui_bundle_issues(
-        repo_versions={unit.repo: unit.target_version for unit in units},
-        workspace=workspace,
-    ):
+    issues_by_repo: dict[str, list[ReleaseGateFinding]] = {}
+    try:
+        issues = selected_release_webui_bundle_issues(
+            repo_versions={unit.repo: unit.target_version for unit in units},
+            workspace=workspace,
+        )
+    except Exception as exc:  # noqa: BLE001 - source errors become release findings.
+        for unit in units:
+            issues_by_repo.setdefault(unit.repo, []).append(
+                ReleaseGateFinding(
+                    code="release_webui_metadata_unreadable",
+                    severity="blocker",
+                    message=(
+                        "Release WebUI composition metadata could not be read or parsed "
+                        f"({type(exc).__name__})."
+                    ),
+                    remediation=(
+                        "Repair malformed JSON in the selected module WebUI package or "
+                        "Core's release package/lockfile, refresh the repository "
+                        "dashboard, then rebuild this plan."
+                    ),
+                    repo=unit.repo,
+                    source="release WebUI composition metadata",
+                    expected=unit.target_version,
+                    actual="unreadable",
+                )
+            )
+        issues = ()
+    for issue in issues:
         issues_by_repo.setdefault(issue.repo, []).append(
-            f"{issue.source}={issue.actual!r}, expected {issue.expected!r} ({issue.message})"
+            ReleaseGateFinding(
+                code="release_webui_composition_alignment",
+                severity="blocker",
+                message=issue.message,
+                remediation=(
+                    "Update Core's webui/package.release.json dependency to the selected module tag, "
+                    "regenerate webui/package-lock.release.json, commit both files, then rebuild this plan."
+                ),
+                repo=issue.repo,
+                source=issue.source,
+                expected=issue.expected,
+                actual=issue.actual,
+            )
         )
     return tuple(
         replace(
             unit,
             status="blocked",
-            blockers=(*unit.blockers, *issues_by_repo[unit.repo]),
+            blockers=(
+                *unit.blockers,
+                *(
+                    f"{finding.source}={finding.actual!r}, expected {finding.expected!r} ({finding.message})"
+                    for finding in issues_by_repo[unit.repo]
+                ),
+            ),
+            gate_findings=(*unit.gate_findings, *issues_by_repo[unit.repo]),
+            source_preflight_ready=False,
         )
         if unit.repo in issues_by_repo
         else unit
@@ -78,33 +235,87 @@ def apply_release_webui_bundle_gate(
     )
 
 
-def selected_repositories(dashboard: ReleaseDashboard, *, selected_repos: tuple[str, ...]) -> tuple[RepositorySnapshot, ...]:
+def selected_repositories(
+    dashboard: ReleaseDashboard, *, selected_repos: tuple[str, ...]
+) -> tuple[RepositorySnapshot, ...]:
     if selected_repos:
         wanted = set(selected_repos)
-        return tuple(repo for repo in dashboard.repositories if repo.spec.name in wanted)
+        return tuple(
+            repo for repo in dashboard.repositories if repo.spec.name in wanted
+        )
     return tuple(repo for repo in dashboard.repositories if release_candidate(repo))
 
 
 def release_candidate(repo: RepositorySnapshot) -> bool:
-    return repo.dirty or bool(repo.ahead) or (repo.exists and repo.is_git and not repo.has_head)
+    return (
+        repo.dirty
+        or bool(repo.ahead)
+        or (repo.exists and repo.is_git and not repo.has_head)
+    )
 
 
-def build_unit(repo: RepositorySnapshot, *, target_version: str | None, contracts: ModuleContractSnapshot | None) -> ReleasePlanUnit:
+def build_unit(
+    repo: RepositorySnapshot,
+    *,
+    target_version: str | None,
+    contracts: ModuleContractSnapshot | None,
+) -> ReleasePlanUnit:
     current_version = repo.versions.primary
     resolved_target = (target_version or current_version or "0.1.0").removeprefix("v")
     target_tag = f"v{resolved_target}"
     blockers: list[str] = []
     warnings: list[str] = []
+    findings: list[ReleaseGateFinding] = []
     if not repo.exists:
         blockers.append("repository is missing locally")
+        findings.append(
+            gate_finding(
+                repo=repo.spec.name,
+                code="repository_missing",
+                message="Repository is missing locally.",
+                remediation="Clone or bootstrap the registered repository, then refresh the dashboard.",
+            )
+        )
     elif not repo.is_git:
         blockers.append("path is not a Git repository")
+        findings.append(
+            gate_finding(
+                repo=repo.spec.name,
+                code="repository_not_git",
+                message="Registered path is not a Git repository.",
+                remediation="Restore the registered checkout as a Git repository, then refresh the dashboard.",
+            )
+        )
     elif repo.safe_directory_required:
         blockers.append("Git safe.directory approval is required")
+        findings.append(
+            gate_finding(
+                repo=repo.spec.name,
+                code="git_safe_directory",
+                message="Git safe.directory approval is required.",
+                remediation="Review checkout ownership and apply Git safe.directory approval only if this path is trusted.",
+            )
+        )
     elif repo.behind:
         blockers.append(f"repository is behind {repo.upstream}")
+        findings.append(
+            gate_finding(
+                repo=repo.spec.name,
+                code="repository_behind",
+                message=f"Repository is behind {repo.upstream}.",
+                remediation="Fetch and integrate the upstream changes, then rebuild the plan from the reviewed HEAD.",
+            )
+        )
     if repo.exists and repo.is_git and not repo.has_head:
         blockers.append("repository has no initial commit")
+        findings.append(
+            gate_finding(
+                repo=repo.spec.name,
+                code="repository_no_head",
+                message="Repository has no initial commit.",
+                remediation="Create and publish a reviewed initial commit before planning its first release.",
+            )
+        )
     local_versions = tuple(
         value
         for value in (
@@ -117,14 +328,67 @@ def build_unit(repo: RepositorySnapshot, *, target_version: str | None, contract
         if value
     )
     if len(set(local_versions)) > 1:
-        blockers.append("backend, manifest, and frontend version metadata is not aligned")
+        blockers.append(
+            "backend, manifest, and frontend version metadata is not aligned"
+        )
+        findings.append(
+            gate_finding(
+                repo=repo.spec.name,
+                code="repository_version_alignment",
+                message="Backend, manifest, and frontend version metadata is not aligned.",
+                remediation=f"Align every version-bearing source declaration to {resolved_target!r}, regenerate lockfiles, and rebuild the plan.",
+            )
+        )
+    if repo.exists and repo.is_git and repo.has_head and not repo.branch:
+        blockers.append("repository is not on a named branch")
+        findings.append(
+            gate_finding(
+                repo=repo.spec.name,
+                code="repository_detached_head",
+                message="Repository is not on a named branch.",
+                remediation="Check out the intended release branch, then rebuild the plan.",
+            )
+        )
     if repo.dirty:
         warnings.append("uncommitted changes will need review before release")
+        findings.append(
+            gate_finding(
+                repo=repo.spec.name,
+                code="worktree_dirty",
+                severity="warning",
+                message="Uncommitted changes prevent source tagging.",
+                remediation="Preview Prepare Changes, review the exact paths, then commit or discard them before rebuilding the plan.",
+            )
+        )
     if repo.ahead:
         warnings.append("unpushed commits should be pushed before catalog publication")
+        findings.append(
+            gate_finding(
+                repo=repo.spec.name,
+                code="repository_ahead",
+                severity="warning",
+                message="The branch contains unpushed commits.",
+                remediation="Preview the source release; guarded publication can push the branch and annotated tag atomically.",
+            )
+        )
     if current_version is None:
-        warnings.append("no local version metadata found; initial release needs version metadata and catalog entry synthesis")
-    if current_version and current_version == resolved_target and repo.local_target_tag_exists:
+        warnings.append(
+            "no local version metadata found; initial release needs version metadata and catalog entry synthesis"
+        )
+        findings.append(
+            gate_finding(
+                repo=repo.spec.name,
+                code="version_metadata_missing",
+                severity="warning",
+                message="No local version metadata was discovered.",
+                remediation=f"Add the required package and manifest version declarations for {resolved_target!r}, then rebuild the plan.",
+            )
+        )
+    if (
+        current_version
+        and current_version == resolved_target
+        and repo.local_target_tag_exists
+    ):
         warnings.append(f"local tag {target_tag} already exists")
 
     provides: tuple[InterfaceProviderSnapshot, ...] = ()
@@ -146,6 +410,85 @@ def build_unit(repo: RepositorySnapshot, *, target_version: str | None, contract
         warnings=tuple(warnings),
         provides_interfaces=provides,
         requires_interfaces=requires,
+        gate_findings=tuple(findings),
+        source_preflight_ready=(
+            not blockers
+            and not repo.dirty
+            and repo.has_head
+            and bool(repo.branch)
+            and current_version == resolved_target
+        ),
+    )
+
+
+def gate_finding(
+    *,
+    repo: str,
+    code: str,
+    message: str,
+    remediation: str,
+    severity: str = "blocker",
+) -> ReleaseGateFinding:
+    return ReleaseGateFinding(
+        code=code,
+        severity=severity,
+        message=message,
+        remediation=remediation,
+        repo=repo,
+    )
+
+
+def recommended_action(
+    *,
+    units: tuple[ReleasePlanUnit, ...],
+    findings: tuple[ReleaseGateFinding, ...],
+    compatibility: tuple[CompatibilityIssue, ...],
+) -> ReleaseRecommendedAction:
+    blocker = next(
+        (finding for finding in findings if finding.severity == "blocker"), None
+    )
+    if blocker is not None:
+        return ReleaseRecommendedAction(
+            id="resolve_release_gate",
+            title=f"Resolve {blocker.repo or 'release'} gate",
+            detail=blocker.message,
+            remediation=blocker.remediation,
+            repo=blocker.repo,
+        )
+    contract_blocker = next(
+        (issue for issue in compatibility if issue.severity == "blocker"), None
+    )
+    if contract_blocker is not None:
+        return ReleaseRecommendedAction(
+            id="resolve_compatibility_gate",
+            title="Resolve interface compatibility gate",
+            detail=contract_blocker.message,
+            remediation="Align the selected provider and consumer interface contracts, then rebuild the plan before publication.",
+            repo=contract_blocker.repo,
+        )
+    worktree = next(
+        (finding for finding in findings if finding.code == "worktree_dirty"), None
+    )
+    if worktree is not None:
+        return ReleaseRecommendedAction(
+            id="prepare_changes",
+            title=f"Prepare {worktree.repo} changes",
+            detail=worktree.message,
+            remediation=worktree.remediation,
+            repo=worktree.repo,
+        )
+    if not units:
+        return ReleaseRecommendedAction(
+            id="select_release_targets",
+            title="Select release targets",
+            detail="No repositories are selected for this release plan.",
+            remediation="Select one or more repositories, confirm their target versions, then build the plan again.",
+        )
+    return ReleaseRecommendedAction(
+        id="preview_source_release",
+        title="Preview source release tags",
+        detail="All plan-visible source preflight gates pass for the selected repositories.",
+        remediation="Run Preview Tag + Publish to verify manifest scope, configured remotes, and immutable local/remote tag state before mutation.",
     )
 
 
@@ -153,7 +496,9 @@ def compatibility_issues(dashboard: ReleaseDashboard) -> tuple[CompatibilityIssu
     return validate_contracts(dashboard.contracts)
 
 
-def dry_run_steps(*, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashboard, channel: str) -> tuple[ReleasePlanStep, ...]:
+def dry_run_steps(
+    *, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashboard, channel: str
+) -> tuple[ReleasePlanStep, ...]:
     steps: list[ReleasePlanStep] = []
     snapshots = {repo.spec.name: repo for repo in dashboard.repositories}
     for unit in units:
@@ -162,7 +507,8 @@ def dry_run_steps(*, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashb
                 id=f"{unit.repo}:preflight",
                 title=f"Inspect {unit.repo}",
                 detail="Verify local branch, dirty state, current version, and target tag before release.",
-                command="git status --short --branch && git tag --list " + shlex.quote(unit.target_tag),
+                command="git status --short --branch && git tag --list "
+                + shlex.quote(unit.target_tag),
                 cwd=f"{dashboard.workspace_root}/{unit.repo}",
                 repo=unit.repo,
             )
@@ -181,7 +527,9 @@ def dry_run_steps(*, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashb
                 )
             )
         snapshot = snapshots.get(unit.repo)
-        if unit.current_version != unit.target_version or (snapshot is not None and snapshot.dirty):
+        if unit.current_version != unit.target_version or (
+            snapshot is not None and snapshot.dirty
+        ):
             steps.append(
                 ReleasePlanStep(
                     id=f"{unit.repo}:commit",
@@ -247,10 +595,10 @@ def dry_run_steps(*, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashb
                         "and updates only selected release units."
                     ),
                     command=(
-                        "KEY_DIR=\"$HOME/.config/govoplan/release-keys\" "
+                        'KEY_DIR="$HOME/.config/govoplan/release-keys" '
                         f"tools/release/release-catalog.py selective --channel {shlex.quote(channel)} "
                         f"{repo_version_args} "
-                        "--catalog-signing-key \"release-key-1=$KEY_DIR/release-key-1.pem\""
+                        '--catalog-signing-key "release-key-1=$KEY_DIR/release-key-1.pem"'
                     ),
                     cwd=dashboard.meta_root,
                     mutating=True,
@@ -267,7 +615,7 @@ def dry_run_steps(*, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashb
                     ),
                     command=(
                         "tools/release/release-catalog.py publish-candidate "
-                        f"--candidate-dir \"$CANDIDATE_DIR\" --channel {shlex.quote(channel)}"
+                        f'--candidate-dir "$CANDIDATE_DIR" --channel {shlex.quote(channel)}'
                     ),
                     cwd=dashboard.meta_root,
                     status="planned",
@@ -283,12 +631,21 @@ def release_notes(dashboard: ReleaseDashboard) -> tuple[str, ...]:
         "Compatibility is checked from currently discovered manifest interface providers and requirements.",
     ]
     if dashboard.catalog.public_checked:
-        notes.append("Local catalog/keyring hashes are compared with the published channel and keyring.")
+        notes.append(
+            "Local catalog/keyring hashes are compared with the published channel and keyring."
+        )
     return tuple(notes)
 
 
-def plan_status(*, units: tuple[ReleasePlanUnit, ...], compatibility: tuple[CompatibilityIssue, ...]) -> str:
-    if any(unit.blockers for unit in units) or any(issue.severity == "blocker" for issue in compatibility):
+def plan_status(
+    *,
+    units: tuple[ReleasePlanUnit, ...],
+    compatibility: tuple[CompatibilityIssue, ...],
+    findings: tuple[ReleaseGateFinding, ...] = (),
+) -> str:
+    if any(finding.severity == "blocker" for finding in findings) or any(unit.blockers for unit in units) or any(
+        issue.severity == "blocker" for issue in compatibility
+    ):
         return "blocked"
     if units or compatibility:
         return "attention"
@@ -299,9 +656,17 @@ def version_satisfies(version: str, requirement: InterfaceRequirementSnapshot) -
     parsed = parse_version(version)
     if parsed is None:
         return False
-    if requirement.version_min and (minimum := parse_version(requirement.version_min)) is not None and parsed < minimum:
+    if (
+        requirement.version_min
+        and (minimum := parse_version(requirement.version_min)) is not None
+        and parsed < minimum
+    ):
         return False
-    if requirement.version_max_exclusive and (maximum := parse_version(requirement.version_max_exclusive)) is not None and parsed >= maximum:
+    if (
+        requirement.version_max_exclusive
+        and (maximum := parse_version(requirement.version_max_exclusive)) is not None
+        and parsed >= maximum
+    ):
         return False
     return True
 

tools/release/webui/index.html

@@ -409,6 +409,11 @@
         padding: 10px;
       }
 
+      .action.recommended {
+        border-left: 4px solid var(--accent);
+        background: var(--panel-alt);
+      }
+
       .action h3 {
         margin: 0 0 5px;
         font-size: 13px;
@@ -963,6 +968,7 @@
           metric("Ahead", s.ahead_count),
           metric("Behind", s.behind_count),
           metric("No HEAD", s.no_head_count),
+          metric("Errors", s.error_count),
           metric("Contracts", data.contracts.length),
         ].join("");
         releaseState.dashboard = data;
@@ -976,7 +982,7 @@
         const selected = elements.releaseUnits ? elements.releaseUnits.querySelectorAll("[data-release-check]:checked").length : 0;
         const gitBlockers = (s.missing_count || 0) + (s.safe_directory_count || 0);
         const branchDrift = (s.ahead_count || 0) + (s.behind_count || 0);
-        const releaseBlockers = gitBlockers + (s.behind_count || 0) + (s.no_head_count || 0) + (s.dirty_count || 0);
+        const releaseBlockers = gitBlockers + (s.behind_count || 0) + (s.no_head_count || 0) + (s.dirty_count || 0) + (s.error_count || 0);
         const catalog = data.catalog || {};
         const hasCandidate = Boolean(elements.candidateDir?.value.trim());
         const catalogPublished = catalog.catalog_matches_public === true && catalog.keyring_matches_public === true;
@@ -1409,10 +1415,28 @@
         const units = plan.units || [];
         const compatibility = plan.compatibility || [];
         const steps = plan.dry_run_steps || [];
-        if (!units.length && !compatibility.length && !steps.length) {
+        const findings = plan.gate_findings || [];
+        const recommendation = plan.recommended_action || null;
+        if (!units.length && !compatibility.length && !steps.length && !recommendation) {
           elements.actions.innerHTML = `<div class="muted">No selective release candidates. Select repositories above to plan a release.</div>`;
           return;
         }
+        const recommendationKind = plan.status === "blocked" ? "block" : plan.source_preflight_ready ? "ok" : "warn";
+        const recommendationHtml = recommendation ? `<div class="action recommended">
+          <h3>${pill("recommended next", recommendationKind)} ${escapeHtml(recommendation.title)}</h3>
+          <p>${escapeHtml(recommendation.detail)}</p>
+          <p class="action-meta"><strong>How to continue:</strong> ${escapeHtml(recommendation.remediation)}</p>
+        </div>` : "";
+        const findingHtml = findings.map((finding) => {
+          const kind = finding.severity === "blocker" ? "block" : finding.severity === "warning" ? "warn" : "ok";
+          const source = finding.source ? `${finding.source}: ${finding.actual || "-"} -> ${finding.expected || "-"}` : "";
+          return `<div class="action">
+            <h3>${pill(finding.severity, kind)} ${escapeHtml(finding.repo || "release")} · ${escapeHtml(finding.code)}</h3>
+            <p>${escapeHtml(finding.message)}</p>
+            ${source ? `<p class="action-meta">${escapeHtml(source)}</p>` : ""}
+            <p class="action-meta"><strong>Remediation:</strong> ${escapeHtml(finding.remediation)}</p>
+          </div>`;
+        }).join("");
         const unitHtml = units.map((unit) => {
           const blockers = unit.blockers.length ? `<p>${escapeHtml(unit.blockers.join("; "))}</p>` : "";
           const warnings = unit.warnings.length ? `<p>${escapeHtml(unit.warnings.join("; "))}</p>` : "";
@@ -1439,6 +1463,8 @@
           </div>`;
         }).join("");
         elements.actions.innerHTML = `
+          ${recommendationHtml}
+          ${findingHtml}
           ${unitHtml}
           ${compatibilityHtml}
           ${stepHtml}