govoplan/tools/release/govoplan_release/release_execution.py

"""Narrow durable-run adapters for existing confirmed release executors."""

from __future__ import annotations

from dataclasses import dataclass
import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import signal
import stat
import subprocess
import tempfile
import tomllib
from typing import Any

from .artifact_identity import (
    MAX_WHEEL_BYTES,
    inspect_python_wheel,
    normalize_package_name,
    selected_artifact_identity_issues,
)
from .catalog import DEFAULT_PUBLIC_BASE_URL, canonical_hash
from .candidate_artifact import (
    CandidateArtifactReceipt,
    candidate_output_path,
    harden_private_candidate_tree,
    issue_candidate_receipt,
    validate_release_channel,
)
from .git_state import (
    collect_repository_snapshot,
    git,
    git_text,
    sanitized_git_environment,
)
from .model import to_jsonable
from .module_directory import module_directory_payloads
from .publisher import (
    bind_publication_remote,
    catalog_public_base_url,
    default_tag_name,
    publication_runtime_trust_issues,
    publish_catalog_candidate,
    registered_website_remote,
    remote_publication_identity,
    validate_catalog_payload,
    verify_committed_publication,
)
from .repository_tag import ref_commit, remote_tag_commit, tag_repositories
from .selective_catalog import (
    build_selective_catalog_candidate,
    configured_signer_public_keys,
    load_authenticated_catalog_base,
    parse_signing_key,
    trusted_keys_from_keyring,
)
from .source_provenance import catalog_source_selection
from .workspace import (
    META_ROOT,
    REPOSITORIES_FILE,
    load_repository_specs,
    resolve_repo_path,
    website_root,
)


MAX_EXECUTOR_OUTPUT = 4_000
MAX_SIGNING_KEYS = 8
MAX_REMOTE_METADATA_BYTES = 16 * 1024
MAX_CATALOG_SOURCE_REPOSITORIES = 128
MAX_TRUSTED_RUNTIME_ENTRIES = 10_000
MAX_TRUSTED_GIT_ENTRIES = 500_000
DURABLE_REMOTE = "origin"
FLATPAK_BUILDER_PROXY = Path("/usr/bin/flatpak-spawn")
HOST_BUBBLEWRAP = Path("/usr/bin/bwrap")
FLATPAK_HOST_BUS = Path("/run/flatpak/bus")
_WEBUI_RELEASE_SOURCE = re.compile(
    r"/(?P<repo>govoplan-[a-z0-9-]+)\.git#v(?P<version>[0-9]+\.[0-9]+\.[0-9]+)$"
)


class ReleaseExecutionError(RuntimeError):
    """Base error for a plan step that cannot be executed safely."""


class ReleaseExecutionBlocked(ReleaseExecutionError):
    """The executor proved that no intended external effect was completed."""


class ReleaseExecutionAmbiguous(ReleaseExecutionError):
    """The executor may have produced an effect and requires reconciliation."""


@dataclass(frozen=True, slots=True)
class ExecutorSpec:
    kind: str
    confirmation: str
    result_code: str
    mutating: bool


EXECUTOR_SPECS = {
    "preflight": ExecutorSpec("preflight", "", "preflight_valid", False),
    "tag": ExecutorSpec("tag", "TAG", "tag_created", True),
    "push": ExecutorSpec("push", "PUBLISH", "tag_published", True),
    "catalog:selective-generator": ExecutorSpec(
        "catalog_generate", "GENERATE", "catalog_candidate_ready", True
    ),
    "catalog:validate-sign-publish": ExecutorSpec(
        "catalog_publish", "PUSH", "catalog_published", True
    ),
}


def executor_spec(plan_step: dict[str, Any]) -> ExecutorSpec | None:
    step_id = plan_step.get("id")
    if step_id in EXECUTOR_SPECS:
        spec = EXECUTOR_SPECS[step_id]
    elif isinstance(step_id, str) and ":" in step_id:
        spec = EXECUTOR_SPECS.get(step_id.rsplit(":", 1)[1])
    else:
        spec = None
    if spec is None or plan_step.get("status") != "planned":
        return None
    if plan_step.get("mutating") is not spec.mutating:
        return None
    if spec.kind == "catalog_generate":
        return spec if plan_step.get("repo") is None else None
    if spec.kind == "catalog_publish":
        binding = plan_step.get("source_binding")
        return (
            spec
            if plan_step.get("repo") is None
            and isinstance(binding, dict)
            and binding.get("repo") == "addideas-govoplan-website"
            else None
        )
    repo = plan_step.get("repo")
    binding = plan_step.get("source_binding")
    return (
        spec
        if isinstance(repo, str)
        and repo
        and isinstance(binding, dict)
        and binding.get("kind") == "repository_state"
        else None
    )


def require_trusted_release_runtime(*, workspace_root: Path) -> None:
    """Require an operator-controlled console/config and workspace boundary."""

    dependency_issues = publication_runtime_trust_issues()
    if dependency_issues:
        raise ReleaseExecutionBlocked(
            "Release runtime is not operator-controlled: "
            + "; ".join(dependency_issues)
        )
    _require_operator_checkout(META_ROOT)
    _require_trusted_tree(META_ROOT / "tools" / "release")
    _require_trusted_tree(META_ROOT / "tools" / "checks")
    _require_trusted_regular_file(REPOSITORIES_FILE, label="repository registry")
    migration_baseline = META_ROOT / "docs" / "migration-release-baselines.json"
    if migration_baseline.exists():
        _require_trusted_regular_file(
            migration_baseline,
            label="migration release baseline",
        )
    _require_trusted_directory(workspace_root, label="release workspace")
    _require_trusted_ancestor_chain(workspace_root)


def bind_plan_source_states(
    *,
    plan: dict[str, Any],
    repo_versions: dict[str, str],
    workspace_root: Path,
) -> dict[str, Any]:
    """Freeze creation-time Git identities into the immutable plan snapshot."""

    plan["runtime_binding"] = trusted_release_runtime_receipt(
        workspace_root=workspace_root
    )
    steps = plan.get("dry_run_steps")
    if not isinstance(steps, list):
        raise ReleaseExecutionBlocked("Release plan steps are unavailable.")
    bindings: dict[str, dict[str, Any] | None] = {}
    for repo, version in repo_versions.items():
        try:
            bindings[repo] = repository_state_receipt(
                repo=repo,
                target_tag=f"v{version.removeprefix('v')}",
                workspace_root=workspace_root,
            )
        except ReleaseExecutionBlocked:
            bindings[repo] = None
    try:
        website_binding = repository_state_receipt(
            repo="addideas-govoplan-website",
            target_tag="",
            workspace_root=workspace_root,
            include_website=True,
        )
    except ReleaseExecutionBlocked:
        website_binding = None
    for step in steps:
        if not isinstance(step, dict):
            continue
        repo = step.get("repo")
        if isinstance(repo, str):
            step["source_binding"] = bindings.get(repo)
        elif step.get("id") == "catalog:validate-sign-publish":
            step["source_binding"] = website_binding
    return plan


def trusted_release_runtime_receipt(*, workspace_root: Path) -> dict[str, Any]:
    """Bind trusted release code/config to one clean origin branch commit."""

    require_trusted_release_runtime(workspace_root=workspace_root)
    specs = {item.name: item for item in load_repository_specs(include_website=False)}
    meta_spec = specs.get("govoplan")
    if meta_spec is None or resolve_repo_path(meta_spec, workspace_root) != META_ROOT:
        raise ReleaseExecutionBlocked(
            "Trusted release tooling is not the registered workspace meta repository."
        )
    receipt = repository_state_receipt(
        repo="govoplan",
        target_tag="",
        workspace_root=workspace_root,
    )
    if not receipt["worktree_clean"]:
        raise ReleaseExecutionBlocked(
            "Release tooling checkout must be clean before a durable run."
        )
    path = resolve_repo_path(meta_spec, workspace_root)
    remote_head = _remote_branch_commit(path, branch=str(receipt["branch"]))
    if remote_head != receipt["head"]:
        raise ReleaseExecutionBlocked(
            "Release tooling HEAD must equal its registered origin branch."
        )
    return receipt


def verify_release_runtime_binding(
    *, expected: object, workspace_root: Path
) -> dict[str, Any]:
    current = trusted_release_runtime_receipt(workspace_root=workspace_root)
    _require_receipt_match(current, expected, operation="release execution")
    return current


def execute_repository_step(
    *,
    spec: ExecutorSpec,
    plan_step: dict[str, Any],
    repo_versions: dict[str, str],
    workspace_root: Path,
    remote: str,
    expected_receipt: dict[str, Any] | None,
) -> tuple[dict[str, Any], dict[str, Any]]:
    require_trusted_release_runtime(workspace_root=workspace_root)
    if remote != DURABLE_REMOTE:
        raise ReleaseExecutionBlocked("Durable release execution uses only origin.")
    repo = str(plan_step["repo"])
    version = repo_versions.get(repo)
    if version is None:
        raise ReleaseExecutionBlocked(
            "The frozen plan step has no matching repository version."
        )
    if spec.kind == "preflight":
        result, receipt = repository_preflight(
            repo=repo,
            version=version,
            workspace_root=workspace_root,
        )
        _require_receipt_match(
            receipt,
            plan_step.get("source_binding"),
            operation="repository preflight",
        )
        return result, receipt
    verify_repository_step_precondition(
        spec=spec,
        plan_step=plan_step,
        version=version,
        workspace_root=workspace_root,
        expected_receipt=expected_receipt,
    )
    if spec.kind == "tag":
        result = tag_repositories(
            repos=(repo,),
            repo_versions={repo: version},
            workspace_root=workspace_root,
            remote=remote,
            apply=True,
            push=False,
        )
        payload = _known_or_ambiguous_result(
            result,
            success={"tagged", "noop"},
            operation="release tag creation",
        )
        try:
            receipt = repository_state_receipt(
                repo=repo,
                target_tag=f"v{version.removeprefix('v')}",
                workspace_root=workspace_root,
            )
            if receipt["tag_object"] is None:
                raise ReleaseExecutionBlocked(
                    "Tag executor returned success without a verifiable annotated "
                    "tag object."
                )
            _require_same_repository_base(receipt, expected_receipt)
        except ReleaseExecutionBlocked as exc:
            raise ReleaseExecutionAmbiguous(
                "The tag executor reported success, but its post-effect repository "
                "receipt could not be verified; reconcile the local tag."
            ) from exc
        return payload, receipt
    if spec.kind == "push":
        result = tag_repositories(
            repos=(repo,),
            repo_versions={repo: version},
            workspace_root=workspace_root,
            remote=remote,
            apply=True,
            push=True,
        )
        payload = _known_or_ambiguous_result(
            result,
            success={"published"},
            operation="atomic release tag publication",
        )
        try:
            receipt = verified_repository_publication_receipt(
                repo=repo,
                version=version,
                workspace_root=workspace_root,
                expected_receipt=expected_receipt,
            )
        except ReleaseExecutionBlocked as exc:
            raise ReleaseExecutionAmbiguous(
                "The push executor reported success, but its post-effect remote "
                "receipt could not be verified; reconcile the local and remote tag."
            ) from exc
        return payload, receipt
    raise ReleaseExecutionBlocked("No repository executor matches this plan step.")


def repository_preflight(
    *, repo: str, version: str, workspace_root: Path
) -> tuple[dict[str, Any], dict[str, Any]]:
    specs = {item.name: item for item in load_repository_specs(include_website=False)}
    repository = specs.get(repo)
    if repository is None:
        raise ReleaseExecutionBlocked("Repository is not registered.")
    path = resolve_repo_path(repository, workspace_root)
    _require_operator_checkout(path)
    snapshot = collect_repository_snapshot(
        repository,
        workspace_root=workspace_root,
        target_tag=f"v{version.removeprefix('v')}",
        online=False,
    )
    if not snapshot.exists or not snapshot.is_git:
        raise ReleaseExecutionBlocked("Repository is missing or is not a Git checkout.")
    if snapshot.safe_directory_required:
        raise ReleaseExecutionBlocked("Git safe.directory approval is required.")
    if not snapshot.has_head or not snapshot.branch:
        raise ReleaseExecutionBlocked("Repository has no named-branch HEAD.")
    tag = f"v{version.removeprefix('v')}"
    receipt = repository_state_receipt(
        repo=repo,
        target_tag=tag,
        workspace_root=workspace_root,
    )
    return {
        "status": "inspected",
        "repo": repo,
        "head": receipt["head"],
        "branch": receipt["branch"],
        "ahead": int(snapshot.ahead or 0),
        "behind": int(snapshot.behind or 0),
        "dirty_count": len(snapshot.dirty_entries),
        "target_tag": tag,
        "local_tag_commit": ref_commit(
            resolve_repo_path(repository, workspace_root), f"refs/tags/{tag}"
        ),
    }, receipt


def repository_state_receipt(
    *,
    repo: str,
    target_tag: str,
    workspace_root: Path,
    include_website: bool = False,
) -> dict[str, Any]:
    """Capture only bounded repository identity, never a credential-bearing URL."""

    specs = {
        item.name: item
        for item in load_repository_specs(include_website=include_website)
    }
    repository = specs.get(repo)
    if repository is None:
        raise ReleaseExecutionBlocked("Repository is not registered.")
    path = resolve_repo_path(repository, workspace_root)
    _require_operator_checkout(path)
    snapshot = collect_repository_snapshot(
        repository,
        workspace_root=workspace_root,
        target_tag=target_tag or None,
        online=False,
    )
    if not snapshot.exists or not snapshot.is_git:
        raise ReleaseExecutionBlocked("Repository is missing or is not a Git checkout.")
    if snapshot.safe_directory_required:
        raise ReleaseExecutionBlocked("Git safe.directory approval is required.")
    if not snapshot.has_head or not snapshot.branch:
        raise ReleaseExecutionBlocked("Repository has no named-branch HEAD.")
    head = git_text(path, "rev-parse", "--verify", "HEAD")
    if len(head) not in {40, 64}:
        raise ReleaseExecutionBlocked("Repository HEAD cannot be bound safely.")
    fetch_urls = _bounded_remote_urls(path, push=False)
    push_urls = _bounded_remote_urls(path, push=True)
    expected_remote = repository.remote.strip()
    if fetch_urls != (expected_remote,) or push_urls != (expected_remote,):
        raise ReleaseExecutionBlocked(
            "Repository origin does not match its registered release remote."
        )
    tag_object = (
        git_text(path, "rev-parse", "--verify", f"refs/tags/{target_tag}")
        if target_tag
        else None
    ) or None
    if tag_object is not None and len(tag_object) not in {40, 64}:
        raise ReleaseExecutionBlocked("Repository tag object cannot be bound safely.")
    if tag_object is not None and git_text(
        path, "cat-file", "-t", f"refs/tags/{target_tag}"
    ) != "tag":
        raise ReleaseExecutionBlocked(
            "Existing release tag is not an annotated tag object."
        )
    return {
        "kind": "repository_state",
        "repo": repo,
        "head": head,
        "branch": snapshot.branch,
        "remote": DURABLE_REMOTE,
        "remote_sha256": hashlib.sha256(
            json.dumps(
                {"fetch": fetch_urls, "push": push_urls},
                sort_keys=True,
                separators=(",", ":"),
                ensure_ascii=True,
            ).encode("utf-8")
        ).hexdigest(),
        "worktree_clean": not snapshot.dirty_entries,
        "target_tag": target_tag,
        "tag_object": tag_object,
    }


def _require_operator_checkout(path: Path) -> None:
    _require_trusted_ancestor_chain(path)
    required = (
        (path, "repository", True),
        (path / ".git", "Git metadata", True),
        (path / ".git" / "config", "Git configuration", False),
        (path / ".git" / "HEAD", "Git HEAD", False),
        (path / ".git" / "objects", "Git object database", True),
        (path / ".git" / "refs", "Git references", True),
    )
    optional = (
        (path / ".git" / "index", "Git index", False),
        (path / ".git" / "packed-refs", "packed Git references", False),
    )
    for candidate, label, directory in (*required, *optional):
        try:
            metadata = candidate.lstat()
        except FileNotFoundError:
            if (candidate, label, directory) in optional:
                continue
            raise ReleaseExecutionBlocked(
                f"Release {label} cannot be inspected safely."
            ) from None
        except OSError as exc:
            raise ReleaseExecutionBlocked(
                f"Release {label} cannot be inspected safely."
            ) from exc
        expected_type = stat.S_ISDIR if directory else stat.S_ISREG
        if stat.S_ISLNK(metadata.st_mode) or not expected_type(metadata.st_mode):
            raise ReleaseExecutionBlocked(
                f"Release {label} must be a real operator-controlled path."
            )
        if hasattr(os, "geteuid") and metadata.st_uid != os.geteuid():
            raise ReleaseExecutionBlocked(
                f"Release {label} is not owned by the console operator."
            )
        if metadata.st_mode & 0o022:
            raise ReleaseExecutionBlocked(
                f"Release {label} is group/world writable."
            )
    _require_trusted_git_tree(path / ".git" / "refs")
    _require_trusted_git_tree(path / ".git" / "objects")
    if any(
        candidate.exists()
        for candidate in (
            path / ".git" / "objects" / "info" / "alternates",
            path / ".git" / "info" / "grafts",
        )
    ):
        raise ReleaseExecutionBlocked(
            "Release Git object alternates and grafts are not permitted."
        )
    _require_trusted_tracked_worktree(path)


def _require_trusted_git_tree(root: Path) -> None:
    """Boundedly inspect nested authority-bearing Git metadata."""

    observed = 0
    pending = [root]
    while pending:
        parent = pending.pop()
        _require_trusted_directory(parent, label="Git metadata directory")
        try:
            entries = tuple(os.scandir(parent))
        except OSError as exc:
            raise ReleaseExecutionBlocked(
                "Nested Git metadata cannot be inspected safely."
            ) from exc
        for entry in entries:
            observed += 1
            if observed > MAX_TRUSTED_GIT_ENTRIES:
                raise ReleaseExecutionBlocked(
                    "Nested Git metadata exceeds its trust-validation limit."
                )
            candidate = Path(entry.path)
            try:
                metadata = candidate.lstat()
            except OSError as exc:
                raise ReleaseExecutionBlocked(
                    "Nested Git metadata changed during trust validation."
                ) from exc
            if stat.S_ISDIR(metadata.st_mode):
                pending.append(candidate)
            elif stat.S_ISREG(metadata.st_mode):
                _require_trusted_regular_file(
                    candidate,
                    label="Git metadata file",
                )
            else:
                raise ReleaseExecutionBlocked(
                    "Nested Git metadata contains a symlink or special file."
                )


def _require_trusted_tracked_worktree(path: Path) -> None:
    """Reject tracked metadata inputs that another local user can race."""

    result = git(path, "ls-files", "-z", timeout=30)
    encoded = result.stdout.encode("utf-8", errors="strict")
    if result.returncode != 0 or len(encoded) > 16 * 1024 * 1024:
        raise ReleaseExecutionBlocked(
            "Tracked release worktree paths cannot be inspected safely."
        )
    names = [name for name in result.stdout.split("\0") if name]
    if len(names) > 100_000:
        raise ReleaseExecutionBlocked(
            "Tracked release worktree exceeds its trust-validation limit."
        )
    checked_directories: set[Path] = {path}
    for name in names:
        relative = Path(name)
        if (
            relative.is_absolute()
            or not relative.parts
            or any(part in {"", ".", ".."} for part in relative.parts)
        ):
            raise ReleaseExecutionBlocked(
                "Tracked release worktree contains a non-canonical path."
            )
        candidate = path / relative
        pending_directories: list[Path] = []
        parent = candidate.parent
        while parent != path:
            if path not in parent.parents:
                raise ReleaseExecutionBlocked(
                    "Tracked release worktree path leaves its repository."
                )
            if parent not in checked_directories:
                pending_directories.append(parent)
            parent = parent.parent
        for directory in reversed(pending_directories):
            _require_trusted_directory(
                directory,
                label="tracked repository directory",
            )
            checked_directories.add(directory)
        try:
            metadata = candidate.lstat()
        except OSError as exc:
            raise ReleaseExecutionBlocked(
                "Tracked release worktree changed during trust validation."
            ) from exc
        if (
            stat.S_ISLNK(metadata.st_mode)
            or not stat.S_ISREG(metadata.st_mode)
            or metadata.st_uid != os.geteuid()
            or metadata.st_mode & 0o022
        ):
            raise ReleaseExecutionBlocked(
                "Tracked release files must be operator-owned, non-writable regular files."
            )


def _require_trusted_ancestor_chain(path: Path) -> None:
    """Reject replaceable path ancestors; root/current owners are authorities."""

    resolved = path.absolute()
    for ancestor in resolved.parents:
        try:
            metadata = ancestor.lstat()
        except OSError as exc:
            raise ReleaseExecutionBlocked(
                "Release path ancestry cannot be inspected safely."
            ) from exc
        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
            raise ReleaseExecutionBlocked(
                "Release path ancestry must contain only real directories."
            )
        if hasattr(os, "geteuid") and metadata.st_uid not in {0, os.geteuid()}:
            raise ReleaseExecutionBlocked(
                "Release path ancestry is not owned by root or the console operator."
            )
        if metadata.st_mode & 0o022:
            raise ReleaseExecutionBlocked(
                "Release path ancestry is group/world writable."
            )


def _require_trusted_directory(path: Path, *, label: str) -> None:
    try:
        metadata = path.lstat()
    except OSError as exc:
        raise ReleaseExecutionBlocked(f"{label.capitalize()} is unavailable.") from exc
    if (
        stat.S_ISLNK(metadata.st_mode)
        or not stat.S_ISDIR(metadata.st_mode)
        or (hasattr(os, "geteuid") and metadata.st_uid != os.geteuid())
        or metadata.st_mode & 0o022
    ):
        raise ReleaseExecutionBlocked(
            f"{label.capitalize()} must be an operator-owned, non-writable real directory."
        )


def _require_trusted_regular_file(path: Path, *, label: str) -> None:
    try:
        metadata = path.lstat()
    except OSError as exc:
        raise ReleaseExecutionBlocked(f"{label.capitalize()} is unavailable.") from exc
    if (
        stat.S_ISLNK(metadata.st_mode)
        or not stat.S_ISREG(metadata.st_mode)
        or (hasattr(os, "geteuid") and metadata.st_uid != os.geteuid())
        or metadata.st_mode & 0o022
    ):
        raise ReleaseExecutionBlocked(
            f"{label.capitalize()} must be an operator-owned, non-writable regular file."
        )


def _require_trusted_tree(root: Path) -> None:
    _require_trusted_directory(root, label="release tooling directory")
    observed = 0
    for current, directory_names, file_names in os.walk(root, followlinks=False):
        current_path = Path(current)
        _require_trusted_directory(current_path, label="release tooling directory")
        for name in (*directory_names, *file_names):
            observed += 1
            if observed > MAX_TRUSTED_RUNTIME_ENTRIES:
                raise ReleaseExecutionBlocked(
                    "Release tooling tree exceeds its trust-validation limit."
                )
            child = current_path / name
            metadata = child.lstat()
            if stat.S_ISDIR(metadata.st_mode):
                _require_trusted_directory(child, label="release tooling directory")
            elif stat.S_ISREG(metadata.st_mode):
                _require_trusted_regular_file(child, label="release tooling file")
            else:
                raise ReleaseExecutionBlocked(
                    "Release tooling tree contains a symlink or non-file member."
                )


def _bounded_remote_urls(path: Path, *, push: bool) -> tuple[str, ...]:
    arguments = ["remote", "get-url", "--all"]
    if push:
        arguments.append("--push")
    arguments.append(DURABLE_REMOTE)
    result = git(path, *arguments)
    encoded = result.stdout.encode("utf-8", errors="strict")
    lines = tuple(line.strip() for line in result.stdout.splitlines() if line.strip())
    if (
        result.returncode != 0
        or not lines
        or len(lines) > 16
        or len(encoded) > MAX_REMOTE_METADATA_BYTES
    ):
        kind = "push" if push else "fetch"
        raise ReleaseExecutionBlocked(
            f"Repository origin {kind} URLs cannot be bound safely."
        )
    return lines


def _remote_branch_commit(path: Path, *, branch: str) -> str | None:
    result = git(
        path,
        "ls-remote",
        "--heads",
        DURABLE_REMOTE,
        f"refs/heads/{branch}",
        timeout=15,
    )
    rows = [line.split("\t", 1) for line in result.stdout.splitlines()]
    expected_ref = f"refs/heads/{branch}"
    if (
        result.returncode != 0
        or len(rows) != 1
        or len(rows[0]) != 2
        or rows[0][1] != expected_ref
        or len(rows[0][0]) not in {40, 64}
    ):
        return None
    return rows[0][0]


def verify_repository_step_precondition(
    *,
    spec: ExecutorSpec,
    plan_step: dict[str, Any],
    version: str,
    workspace_root: Path,
    expected_receipt: dict[str, Any] | None,
) -> dict[str, Any]:
    if spec.kind not in {"tag", "push"}:
        raise ReleaseExecutionBlocked("Repository precondition is unsupported.")
    if not isinstance(expected_receipt, dict):
        raise ReleaseExecutionBlocked(
            "Repository mutation requires the preceding durable state receipt."
        )
    current = repository_state_receipt(
        repo=str(plan_step["repo"]),
        target_tag=f"v{version.removeprefix('v')}",
        workspace_root=workspace_root,
    )
    _require_receipt_match(current, expected_receipt, operation=spec.kind)
    if not current["worktree_clean"]:
        raise ReleaseExecutionBlocked(
            "Repository worktree changed after the run was frozen; create a new run."
        )
    if spec.kind == "push" and current["tag_object"] is None:
        raise ReleaseExecutionBlocked("Tag publication requires an annotated local tag.")
    return current


def verify_repository_preflight_binding(
    *, plan_step: dict[str, Any], version: str, workspace_root: Path
) -> dict[str, Any]:
    current = repository_state_receipt(
        repo=str(plan_step["repo"]),
        target_tag=f"v{version.removeprefix('v')}",
        workspace_root=workspace_root,
    )
    _require_receipt_match(
        current,
        plan_step.get("source_binding"),
        operation="repository preflight",
    )
    return current


def verify_catalog_publication_precondition(
    *, plan_step: dict[str, Any], workspace_root: Path
) -> dict[str, Any]:
    current = repository_state_receipt(
        repo="addideas-govoplan-website",
        target_tag="",
        workspace_root=workspace_root,
        include_website=True,
    )
    _require_receipt_match(
        current,
        plan_step.get("source_binding"),
        operation="catalog publication",
    )
    if not current["worktree_clean"]:
        raise ReleaseExecutionBlocked(
            "Website worktree changed after the run was frozen; create a new run."
        )
    return current


def reconciled_repository_receipt(
    *,
    spec: ExecutorSpec,
    plan_step: dict[str, Any],
    version: str,
    workspace_root: Path,
    expected_receipt: dict[str, Any] | None,
) -> dict[str, Any]:
    """Independently prove the repository effect before advancing reconciliation."""

    if spec.kind == "tag":
        if not isinstance(expected_receipt, dict):
            raise ReleaseExecutionBlocked(
                "Tag reconciliation requires the preceding repository receipt."
            )
        receipt = repository_state_receipt(
            repo=str(plan_step["repo"]),
            target_tag=f"v{version.removeprefix('v')}",
            workspace_root=workspace_root,
        )
        _require_same_repository_base(receipt, expected_receipt)
        if receipt["tag_object"] is None:
            raise ReleaseExecutionBlocked(
                "The annotated local tag effect could not be verified."
            )
        return receipt
    if spec.kind == "push":
        return verified_repository_publication_receipt(
            repo=str(plan_step["repo"]),
            version=version,
            workspace_root=workspace_root,
            expected_receipt=expected_receipt,
        )
    raise ReleaseExecutionBlocked(
        "This repository effect has no automatic success reconciliation proof."
    )


def verified_repository_publication_receipt(
    *,
    repo: str,
    version: str,
    workspace_root: Path,
    expected_receipt: dict[str, Any] | None,
) -> dict[str, Any]:
    if not isinstance(expected_receipt, dict):
        raise ReleaseExecutionBlocked(
            "Tag publication requires the preceding tag receipt."
        )
    tag = f"v{version.removeprefix('v')}"
    receipt = repository_state_receipt(
        repo=repo,
        target_tag=tag,
        workspace_root=workspace_root,
    )
    _require_receipt_match(receipt, expected_receipt, operation="tag publication")
    specs = {item.name: item for item in load_repository_specs(include_website=False)}
    path = resolve_repo_path(specs[repo], workspace_root)
    remote = remote_tag_commit(path, remote=DURABLE_REMOTE, tag=tag)
    if remote.error:
        raise ReleaseExecutionBlocked(remote.error)
    if (
        not remote.annotated
        or remote.commit != receipt["head"]
        or remote.tag_object != receipt["tag_object"]
    ):
        raise ReleaseExecutionBlocked(
            "Remote annotated tag does not match the receipt-bound local tag."
        )
    if _remote_branch_commit(path, branch=str(receipt["branch"])) != receipt["head"]:
        raise ReleaseExecutionBlocked(
            "Remote branch does not match the receipt-bound atomic publication."
        )
    return receipt


def _require_receipt_match(
    current: dict[str, Any], expected: object, *, operation: str
) -> None:
    if current != expected:
        raise ReleaseExecutionBlocked(
            f"Repository state drifted before {operation}; create a new release run."
        )


def _require_same_repository_base(
    current: dict[str, Any], expected: dict[str, Any] | None
) -> None:
    if not isinstance(expected, dict):
        raise ReleaseExecutionBlocked("Preceding repository receipt is unavailable.")
    ignored = {"tag_object"}
    if any(
        current.get(key) != value
        for key, value in expected.items()
        if key not in ignored
    ):
        raise ReleaseExecutionBlocked(
            "Repository identity changed while creating the release tag."
        )


def generate_catalog_candidate(
    *,
    candidate_root: Path,
    candidate_id: str,
    channel: str,
    repo_versions: dict[str, str],
    workspace_root: Path,
    signing_keys: tuple[str, ...],
    remote: str,
    check_public: bool,
    source_receipts: dict[str, dict[str, Any]],
    base_catalog: Path,
    base_keyring: Path,
) -> tuple[dict[str, Any], CandidateArtifactReceipt]:
    require_trusted_release_runtime(workspace_root=workspace_root)
    if not signing_keys or len(signing_keys) > MAX_SIGNING_KEYS:
        raise ReleaseExecutionBlocked(
            "Durable catalog generation requires one to eight configured signing keys."
        )
    output = candidate_output_path(candidate_root, candidate_id)
    if output.exists():
        raise ReleaseExecutionAmbiguous(
            "The server-issued candidate handle already exists; reconcile before reuse."
        )
    try:
        frozen_sources = {
            repo: verify_frozen_repository_tag_receipt(
                receipt=source_receipts.get(repo),
                workspace_root=workspace_root,
            )
            for repo in repo_versions
        }
        with tempfile.TemporaryDirectory(
            prefix="govoplan-release-synthesis-",
            dir=candidate_root,
        ) as synthesis_dir:
            synthesis_workspace = Path(synthesis_dir)
            authenticated_base, snapshot_catalog, snapshot_keyring = (
                _snapshot_authenticated_catalog_base(
                    base_catalog=base_catalog,
                    base_keyring=base_keyring,
                    channel=channel,
                    signing_keys=signing_keys,
                    workspace_root=workspace_root,
                    snapshot_root=synthesis_workspace / ".authenticated-base",
                )
            )
            base_selection = catalog_source_selection(authenticated_base)
            source_versions = dict(base_selection.all_versions)
            source_versions.update(repo_versions)
            if (
                not source_versions
                or len(source_versions) > MAX_CATALOG_SOURCE_REPOSITORIES
            ):
                raise ReleaseExecutionBlocked(
                    "Authenticated catalog has no bounded source repository set."
                )
            cloned_sources = _clone_catalog_sources(
                source_versions=source_versions,
                repo_versions=repo_versions,
                source_receipts=frozen_sources,
                workspace_root=workspace_root,
                destination=synthesis_workspace,
            )
            python_artifacts = build_selected_wheels(
                repo_versions=repo_versions,
                workspace_root=synthesis_workspace,
                candidate_output=output,
                source_receipts=frozen_sources,
                frozen_sources=cloned_sources,
            )
            candidate = build_selective_catalog_candidate(
                repo_versions=repo_versions,
                channel=channel,
                workspace_root=synthesis_workspace,
                output_dir=output,
                signing_keys=signing_keys,
                base_catalog=snapshot_catalog,
                base_keyring=snapshot_keyring,
                source_remote=remote,
                check_public=check_public,
                python_artifacts=python_artifacts,
                frozen_source_provenance={
                    repo: {
                        "commit_sha": str(receipt["head"]),
                        "tag_object_sha": str(receipt["tag_object"]),
                    }
                    for repo, receipt in frozen_sources.items()
                },
            )
        harden_private_candidate_tree(output)
        payload = _read_generated_catalog(output, channel=channel)
        identity_issues = selected_artifact_identity_issues(payload)
        if identity_issues:
            raise ReleaseExecutionBlocked(
                "Generated catalog lacks selected release artifact identities: "
                + "; ".join(identity_issues)
            )
        if getattr(candidate, "status", None) != "ready":
            raise ReleaseExecutionBlocked(
                "Generated catalog candidate did not pass signature validation."
            )
        receipt = issue_candidate_receipt(
            root=candidate_root,
            candidate_id=candidate_id,
            channel=channel,
        )
        return to_jsonable(candidate), receipt
    except ReleaseExecutionAmbiguous:
        raise
    except ReleaseExecutionError:
        _remove_partial_candidate(output)
        raise
    except (OSError, subprocess.SubprocessError, ValueError) as exc:
        _remove_partial_candidate(output)
        raise ReleaseExecutionBlocked(
            f"Catalog candidate generation failed before a trusted receipt: {exc}"
        ) from exc


def validated_candidate_receipt(
    *, candidate_root: Path, candidate_id: str, channel: str
) -> CandidateArtifactReceipt:
    candidate = candidate_output_path(candidate_root, candidate_id)
    payload = _read_generated_catalog(candidate, channel=channel)
    issues = selected_artifact_identity_issues(payload)
    if issues:
        raise ReleaseExecutionBlocked(
            "Candidate lacks selected release artifact identities: "
            + "; ".join(issues)
        )
    return issue_candidate_receipt(
        root=candidate_root,
        candidate_id=candidate_id,
        channel=channel,
    )


def _snapshot_authenticated_catalog_base(
    *,
    base_catalog: Path,
    base_keyring: Path,
    channel: str,
    signing_keys: tuple[str, ...],
    workspace_root: Path,
    snapshot_root: Path,
) -> tuple[dict[str, Any], Path, Path]:
    """Authenticate once, then pin the exact JSON pair in private storage."""

    parsed_keys = tuple(parse_signing_key(value) for value in signing_keys)
    authenticated = load_authenticated_catalog_base(
        base_catalog=base_catalog,
        base_keyring=base_keyring,
        web_root=website_root(workspace_root),
        channel=channel,
        public_base_url=DEFAULT_PUBLIC_BASE_URL,
        signer_public_keys=configured_signer_public_keys(parsed_keys),
    )
    snapshot_root.mkdir(mode=0o700, parents=False, exist_ok=False)
    os.chmod(snapshot_root, 0o700, follow_symlinks=False)
    catalog_path = snapshot_root / f"{validate_release_channel(channel)}.json"
    keyring_path = snapshot_root / "keyring.json"
    _write_private_json_snapshot(catalog_path, authenticated.catalog)
    _write_private_json_snapshot(keyring_path, authenticated.keyring)
    return authenticated.catalog, catalog_path, keyring_path


def _write_private_json_snapshot(path: Path, payload: object) -> None:
    encoded = (
        json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=True) + "\n"
    ).encode("utf-8")
    if len(encoded) > 16 * 1024 * 1024:
        raise ReleaseExecutionBlocked(
            "Authenticated catalog trust material exceeds its size limit."
        )
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = os.open(path, flags, 0o600)
    try:
        os.fchmod(descriptor, 0o600)
        with os.fdopen(descriptor, "wb", closefd=False) as handle:
            handle.write(encoded)
            handle.flush()
            os.fsync(handle.fileno())
    finally:
        os.close(descriptor)


def build_selected_wheels(
    *,
    repo_versions: dict[str, str],
    workspace_root: Path,
    candidate_output: Path,
    source_receipts: dict[str, dict[str, Any]],
    frozen_sources: dict[str, Path] | None = None,
) -> dict[str, Path]:
    artifacts = candidate_output / "artifacts"
    artifacts.mkdir(parents=True, exist_ok=False)
    specs = {item.name: item for item in load_repository_specs(include_website=False)}
    if frozen_sources is None:
        with tempfile.TemporaryDirectory(
            prefix="govoplan-release-build-"
        ) as temp_dir:
            cloned = _clone_frozen_sources(
                repo_versions=repo_versions,
                source_receipts=source_receipts,
                workspace_root=workspace_root,
                destination=Path(temp_dir),
            )
            return _build_selected_wheels_from_sources(
                repo_versions=repo_versions,
                candidate_output=candidate_output,
                source_receipts=source_receipts,
                frozen_sources=cloned,
                specs=specs,
                workspace_root=workspace_root,
            )
    return _build_selected_wheels_from_sources(
        repo_versions=repo_versions,
        candidate_output=candidate_output,
        source_receipts=source_receipts,
        frozen_sources=frozen_sources,
        specs=specs,
        workspace_root=workspace_root,
    )


def _build_selected_wheels_from_sources(
    *,
    repo_versions: dict[str, str],
    candidate_output: Path,
    source_receipts: dict[str, dict[str, Any]],
    frozen_sources: dict[str, Path],
    specs: dict[str, Any],
    workspace_root: Path,
) -> dict[str, Path]:
    artifacts = candidate_output / "artifacts"
    if not artifacts.exists():
        artifacts.mkdir(parents=True, exist_ok=False)
    result: dict[str, Path] = {}
    for repo in sorted(repo_versions):
        repository = specs.get(repo)
        receipt = source_receipts.get(repo)
        frozen_path = frozen_sources.get(repo)
        if (
            repository is None
            or not isinstance(receipt, dict)
            or frozen_path is None
        ):
            raise ReleaseExecutionBlocked(
                f"Repository {repo} has no verified frozen source receipt."
            )
        project_name = python_project_name(frozen_path)
        if project_name is None:
            continue
        staging = Path(
            tempfile.mkdtemp(prefix=f".{repo}-build-", dir=artifacts)
        )
        os.chmod(staging, 0o700, follow_symlinks=False)
        source_staging = staging / "source"
        wheel_staging = staging / "out"
        source_staging.mkdir(mode=0o700)
        staged = git(
            frozen_path,
            "checkout-index",
            "--all",
            "--force",
            f"--prefix={source_staging.absolute()}/",
            timeout=120,
        )
        if staged.returncode != 0:
            shutil.rmtree(staging)
            raise ReleaseExecutionBlocked(
                f"Could not stage the verified tracked source tree for {repo}."
            )
        wheel_staging.mkdir(mode=0o700)
        command = isolated_wheel_builder_command(
            source=source_staging,
            output=wheel_staging,
        )
        if _run_isolated_wheel_builder(command) != 0:
            shutil.rmtree(staging)
            raise ReleaseExecutionBlocked(
                f"Server-owned wheel staging failed for {repo}: "
                "the isolated no-network worker did not complete successfully"
            )
        created = _single_isolated_wheel(wheel_staging)
        destination = artifacts / created.name
        if destination.exists() or destination.is_symlink():
            raise ReleaseExecutionBlocked(
                f"Isolated wheel output name collides for {repo}."
            )
        _copy_isolated_wheel(created, destination)
        identity = inspect_python_wheel(destination)
        if (
            identity.package_name != normalize_package_name(project_name)
            or identity.package_version != repo_versions[repo].removeprefix("v")
        ):
            destination.unlink(missing_ok=True)
            raise ReleaseExecutionBlocked(
                f"Isolated wheel identity does not match selected source {repo}."
            )
        os.chmod(destination, 0o600, follow_symlinks=False)
        shutil.rmtree(staging)
        result[repo] = destination
    return result


def isolated_wheel_builder_command(*, source: Path, output: Path) -> tuple[str, ...]:
    """Build one absolute bubblewrap command with no host data or network view."""

    launcher = isolated_builder_launcher()
    return (
        *launcher,
        "--unshare-all",
        "--die-with-parent",
        "--new-session",
        "--cap-drop",
        "ALL",
        "--ro-bind",
        "/usr",
        "/usr",
        "--symlink",
        "usr/bin",
        "/bin",
        "--symlink",
        "usr/lib",
        "/lib",
        "--symlink",
        "usr/lib64",
        "/lib64",
        "--proc",
        "/proc",
        "--dev",
        "/dev",
        "--tmpfs",
        "/tmp",
        "--dir",
        "/tmp/home",
        "--bind",
        str(output.absolute()),
        "/out",
        "--bind",
        str(source.absolute()),
        "/src",
        "--chdir",
        "/tmp",
        "--clearenv",
        "--setenv",
        "PATH",
        "/usr/bin",
        "--setenv",
        "HOME",
        "/tmp/home",
        "--setenv",
        "TMPDIR",
        "/tmp",
        "--setenv",
        "PYTHONNOUSERSITE",
        "1",
        "--setenv",
        "PYTHONHASHSEED",
        "0",
        "--setenv",
        "SOURCE_DATE_EPOCH",
        "315532800",
        "--setenv",
        "PIP_CONFIG_FILE",
        "/dev/null",
        "--setenv",
        "PIP_DISABLE_PIP_VERSION_CHECK",
        "1",
        "--setenv",
        "PIP_NO_CACHE_DIR",
        "1",
        "--setenv",
        "PIP_NO_INDEX",
        "1",
        "/usr/bin/prlimit",
        f"--fsize={MAX_WHEEL_BYTES}",
        "--cpu=600",
        "--nproc=256",
        "--as=4294967296",
        "--",
        "/usr/bin/python3",
        "-m",
        "pip",
        "wheel",
        "--no-deps",
        "--no-build-isolation",
        "--disable-pip-version-check",
        "--no-index",
        "--wheel-dir",
        "/out",
        "/src",
    )


def isolated_builder_launcher() -> tuple[str, ...]:
    if _trusted_flatpak_builder_proxy():
        return (str(FLATPAK_BUILDER_PROXY), "--host", str(HOST_BUBBLEWRAP))
    if _trusted_system_executable(HOST_BUBBLEWRAP):
        return (str(HOST_BUBBLEWRAP),)
    raise ReleaseExecutionBlocked(
        "Catalog generation requires a trusted bubblewrap worker; no supported "
        "isolated builder is available."
    )


def _trusted_system_executable(path: Path) -> bool:
    try:
        metadata = path.lstat()
    except OSError:
        return False
    return (
        stat.S_ISREG(metadata.st_mode)
        and metadata.st_uid == 0
        and metadata.st_mode & 0o022 == 0
        and os.access(path, os.X_OK)
    )


def _trusted_flatpak_builder_proxy() -> bool:
    """Trust only Flatpak's immutable SDK proxy and a root-owned host bwrap."""

    if os.environ.get("container") != "flatpak":
        return False
    try:
        flatpak_info = Path("/.flatpak-info").lstat()
        flatpak_runtime = FLATPAK_HOST_BUS.parent.lstat()
        host_bus = FLATPAK_HOST_BUS.lstat()
        proxy = FLATPAK_BUILDER_PROXY.lstat()
        ancestors = (Path("/usr"), Path("/usr/bin"))
        ancestor_metadata = tuple(path.lstat() for path in ancestors)
        read_only = all(
            os.statvfs(path).f_flag & os.ST_RDONLY
            for path in (*ancestors, FLATPAK_BUILDER_PROXY)
        )
    except (AttributeError, OSError):
        return False
    if not (
        stat.S_ISREG(flatpak_info.st_mode)
        and flatpak_info.st_uid == os.getuid()
        and flatpak_info.st_mode & 0o077 == 0
        and stat.S_ISDIR(flatpak_runtime.st_mode)
        and flatpak_runtime.st_uid == os.getuid()
        and flatpak_runtime.st_mode & 0o077 == 0
        and stat.S_ISSOCK(host_bus.st_mode)
        and host_bus.st_uid == os.getuid()
        and host_bus.st_mode & 0o022 == 0
        and stat.S_ISREG(proxy.st_mode)
        and proxy.st_mode & 0o022 == 0
        and proxy.st_mode & 0o111 != 0
        and all(
            stat.S_ISDIR(metadata.st_mode)
            and metadata.st_mode & 0o022 == 0
            for metadata in ancestor_metadata
        )
        and read_only
    ):
        return False
    return _trusted_host_bubblewrap()


def _trusted_host_bubblewrap() -> bool:
    try:
        check = subprocess.run(  # noqa: S603 - immutable allowlisted proxy.
            (
                str(FLATPAK_BUILDER_PROXY),
                "--host",
                "/usr/bin/stat",
                "-c",
                "%f|%u",
                str(HOST_BUBBLEWRAP),
            ),
            cwd=Path("/"),
            env=_flatpak_proxy_environment(),
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            text=True,
            timeout=5,
            check=False,
            close_fds=True,
        )
    except (OSError, subprocess.SubprocessError):
        return False
    output = check.stdout.strip()
    if check.returncode != 0 or len(output) > 64:
        return False
    try:
        mode_text, uid_text = output.split("|", maxsplit=1)
        mode = int(mode_text, 16)
        uid = int(uid_text, 10)
    except ValueError:
        return False
    return (
        stat.S_ISREG(mode)
        and uid == 0
        and mode & 0o022 == 0
        and mode & 0o111 != 0
    )


def _flatpak_proxy_environment() -> dict[str, str]:
    return {
        "PATH": "/usr/bin:/bin",
        "LANG": "C.UTF-8",
        "LC_ALL": "C.UTF-8",
        "DBUS_SESSION_BUS_ADDRESS": f"unix:path={FLATPAK_HOST_BUS}",
    }


def _run_isolated_wheel_builder(command: tuple[str, ...]) -> int:
    environment = {
        "PATH": "/usr/bin:/bin",
        "LANG": "C.UTF-8",
        "LC_ALL": "C.UTF-8",
    }
    if command[:1] == (str(FLATPAK_BUILDER_PROXY),):
        environment = _flatpak_proxy_environment()
    process = subprocess.Popen(  # noqa: S603 - absolute trusted launcher only.
        command,
        cwd=Path("/"),
        env=environment,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        close_fds=True,
        start_new_session=True,
    )
    try:
        return process.wait(timeout=620)
    except subprocess.TimeoutExpired as exc:
        try:
            os.killpg(process.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        process.wait()
        raise ReleaseExecutionBlocked(
            "Isolated wheel builder exceeded its bounded execution time."
        ) from exc


def _single_isolated_wheel(output: Path) -> Path:
    entries: list[Path] = []
    with os.scandir(output) as iterator:
        for entry in iterator:
            entries.append(Path(entry.path))
            if len(entries) > 1:
                break
    if len(entries) != 1:
        raise ReleaseExecutionBlocked(
            "Isolated builder must return exactly one wheel and no extra output."
        )
    wheel = entries[0]
    metadata = wheel.lstat()
    if (
        stat.S_ISLNK(metadata.st_mode)
        or not stat.S_ISREG(metadata.st_mode)
        or wheel.suffix != ".whl"
        or metadata.st_uid != os.geteuid()
        or metadata.st_nlink != 1
        or not 0 < metadata.st_size <= MAX_WHEEL_BYTES
    ):
        raise ReleaseExecutionBlocked(
            "Isolated builder returned an unsafe or oversized wheel."
        )
    return wheel


def _copy_isolated_wheel(source: Path, destination: Path) -> None:
    """Copy worker output through pinned descriptors into a fresh candidate file."""

    source_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    destination_flags = (
        os.O_WRONLY
        | os.O_CREAT
        | os.O_EXCL
        | getattr(os, "O_NOFOLLOW", 0)
    )
    source_descriptor = os.open(source, source_flags)
    destination_descriptor = -1
    try:
        metadata = os.fstat(source_descriptor)
        if (
            not stat.S_ISREG(metadata.st_mode)
            or metadata.st_uid != os.geteuid()
            or metadata.st_nlink != 1
            or not 0 < metadata.st_size <= MAX_WHEEL_BYTES
        ):
            raise ReleaseExecutionBlocked(
                "Isolated builder wheel changed before trusted staging."
            )
        destination_descriptor = os.open(
            destination,
            destination_flags,
            0o600,
        )
        remaining = metadata.st_size
        while remaining:
            chunk = os.read(source_descriptor, min(1024 * 1024, remaining))
            if not chunk:
                raise ReleaseExecutionBlocked(
                    "Isolated builder wheel was truncated during trusted staging."
                )
            view = memoryview(chunk)
            while view:
                written = os.write(destination_descriptor, view)
                if written <= 0:
                    raise ReleaseExecutionBlocked(
                        "Isolated builder wheel could not be staged safely."
                    )
                view = view[written:]
            remaining -= len(chunk)
        if os.read(source_descriptor, 1):
            raise ReleaseExecutionBlocked(
                "Isolated builder wheel grew beyond its verified size."
            )
        os.fchmod(destination_descriptor, 0o600)
        os.fsync(destination_descriptor)
    except BaseException:
        if destination_descriptor >= 0:
            os.close(destination_descriptor)
            destination_descriptor = -1
        destination.unlink(missing_ok=True)
        raise
    finally:
        os.close(source_descriptor)
        if destination_descriptor >= 0:
            os.close(destination_descriptor)
    directory_descriptor = os.open(
        destination.parent,
        os.O_RDONLY | getattr(os, "O_DIRECTORY", 0),
    )
    try:
        os.fsync(directory_descriptor)
    finally:
        os.close(directory_descriptor)


def _clone_frozen_sources(
    *,
    repo_versions: dict[str, str],
    source_receipts: dict[str, dict[str, Any]],
    workspace_root: Path,
    destination: Path,
) -> dict[str, Path]:
    specs = {item.name: item for item in load_repository_specs(include_website=False)}
    result: dict[str, Path] = {}
    for repo in sorted(repo_versions):
        repository = specs.get(repo)
        receipt = source_receipts.get(repo)
        if repository is None or not isinstance(receipt, dict):
            raise ReleaseExecutionBlocked(
                f"Repository {repo} has no verified frozen source receipt."
            )
        # Recheck the private local receipt and registered origin immediately
        # before fetching the immutable source tag.
        verify_frozen_repository_tag_receipt(
            receipt=receipt,
            workspace_root=workspace_root,
        )
        result[repo] = _checkout_frozen_source(
            repo=repo,
            source_remote=repository.remote,
            commit=str(receipt["head"]),
            tag=str(receipt["target_tag"]),
            tag_object=str(receipt["tag_object"]),
            source_root=destination,
        )
    return result


def _clone_catalog_sources(
    *,
    source_versions: dict[str, str],
    repo_versions: dict[str, str],
    source_receipts: dict[str, dict[str, Any]],
    workspace_root: Path,
    destination: Path,
) -> dict[str, Path]:
    """Clone every authenticated catalog source, overriding selected exact receipts."""

    specs = {item.name: item for item in load_repository_specs(include_website=False)}
    result: dict[str, Path] = {}
    for repo, version in sorted(source_versions.items()):
        repository = specs.get(repo)
        if repository is None:
            raise ReleaseExecutionBlocked(
                f"Authenticated catalog source {repo} is not registered."
            )
        receipt = source_receipts.get(repo) if repo in repo_versions else None
        if repo in repo_versions:
            if not isinstance(receipt, dict):
                raise ReleaseExecutionBlocked(
                    f"Selected repository {repo} has no verified source receipt."
                )
            verify_frozen_repository_tag_receipt(
                receipt=receipt,
                workspace_root=workspace_root,
            )
            expected_tag = f"v{version.removeprefix('v')}"
            if receipt.get("target_tag") != expected_tag:
                raise ReleaseExecutionBlocked(
                    f"Selected repository {repo} receipt has the wrong release tag."
                )
            result[repo] = _checkout_frozen_source(
                repo=repo,
                source_remote=repository.remote,
                commit=str(receipt["head"]),
                tag=expected_tag,
                tag_object=str(receipt["tag_object"]),
                source_root=destination,
            )
        else:
            result[repo] = _checkout_registered_source_tag(
                repo=repo,
                source_remote=repository.remote,
                tag=f"v{version.removeprefix('v')}",
                source_root=destination,
            )
    _clone_release_bundle_sources(
        result=result,
        specs=specs,
        source_root=destination,
    )
    return result


def _clone_release_bundle_sources(
    *,
    result: dict[str, Path],
    specs: dict[str, Any],
    source_root: Path,
) -> None:
    """Add exact Core bundle dependencies needed only for alignment checks."""

    core = result.get("govoplan-core")
    if core is None:
        raise ReleaseExecutionBlocked(
            "Authenticated catalog composition has no Core source."
        )
    package_path = core / "webui" / "package.release.json"
    lock_path = core / "webui" / "package-lock.release.json"
    if not package_path.exists() and not lock_path.exists():
        return
    package = _read_bounded_release_json(
        package_path,
        label="Core release WebUI package",
    )
    lock = _read_bounded_release_json(
        lock_path,
        label="Core release WebUI lock",
    )
    dependencies = package.get("dependencies")
    lock_packages = lock.get("packages")
    if not isinstance(dependencies, dict) or not isinstance(lock_packages, dict):
        raise ReleaseExecutionBlocked(
            "Core release WebUI dependency metadata is incomplete."
        )
    if len(dependencies) > MAX_CATALOG_SOURCE_REPOSITORIES:
        raise ReleaseExecutionBlocked(
            "Core release WebUI dependency set exceeds its bound."
        )
    for package_name, reference in sorted(dependencies.items()):
        if not isinstance(package_name, str) or not package_name.startswith(
            "@govoplan/"
        ):
            continue
        match = (
            _WEBUI_RELEASE_SOURCE.search(reference)
            if isinstance(reference, str)
            else None
        )
        if match is None:
            raise ReleaseExecutionBlocked(
                f"Core release dependency {package_name} has no immutable source tag."
            )
        repo = match.group("repo")
        version = match.group("version")
        repository = specs.get(repo)
        if repository is None:
            raise ReleaseExecutionBlocked(
                f"Core release dependency {repo} is not registered."
            )
        locked = lock_packages.get(f"node_modules/{package_name}")
        resolved = locked.get("resolved") if isinstance(locked, dict) else None
        locked_version = locked.get("version") if isinstance(locked, dict) else None
        commit = (
            resolved.rsplit("#", 1)[-1]
            if isinstance(resolved, str) and "#" in resolved
            else ""
        )
        if (
            locked_version != version
            or not re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", commit)
        ):
            raise ReleaseExecutionBlocked(
                f"Core release dependency {repo} has no exact lock identity."
            )
        if repo not in result:
            if len(result) >= MAX_CATALOG_SOURCE_REPOSITORIES:
                raise ReleaseExecutionBlocked(
                    "Release composition source set exceeds its bound."
                )
            result[repo] = _checkout_registered_source_tag(
                repo=repo,
                source_remote=repository.remote,
                tag=f"v{version}",
                source_root=source_root,
            )
        observed_type = git_text(
            result[repo], "cat-file", "-t", f"refs/tags/v{version}"
        )
        observed = ref_commit(result[repo], f"refs/tags/v{version}")
        if observed_type != "tag" or observed != commit:
            raise ReleaseExecutionBlocked(
                f"Core release dependency {repo} lock does not match its origin tag."
            )


def _read_bounded_release_json(path: Path, *, label: str) -> dict[str, Any]:
    try:
        encoded = path.read_bytes()
    except OSError as exc:
        raise ReleaseExecutionBlocked(f"{label} cannot be read.") from exc
    if len(encoded) > 4 * 1024 * 1024:
        raise ReleaseExecutionBlocked(f"{label} exceeds its size limit.")
    try:
        payload = json.loads(encoded)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ReleaseExecutionBlocked(f"{label} is malformed.") from exc
    if not isinstance(payload, dict):
        raise ReleaseExecutionBlocked(f"{label} must be a JSON object.")
    return payload


def verify_frozen_repository_tag_receipt(
    *, receipt: dict[str, Any] | None, workspace_root: Path
) -> dict[str, Any]:
    """Verify a stored local/remote annotated tag without trusting live HEAD."""

    if not isinstance(receipt, dict) or receipt.get("kind") != "repository_state":
        raise ReleaseExecutionBlocked(
            "Catalog generation requires each source push's durable receipt."
        )
    repo = receipt.get("repo")
    tag = receipt.get("target_tag")
    tag_object = receipt.get("tag_object")
    head = receipt.get("head")
    if not all(isinstance(value, str) and value for value in (repo, tag, tag_object, head)):
        raise ReleaseExecutionBlocked("Frozen source receipt is incomplete.")
    specs = {item.name: item for item in load_repository_specs(include_website=False)}
    repository = specs.get(repo)
    if repository is None:
        raise ReleaseExecutionBlocked("Frozen source repository is not registered.")
    path = resolve_repo_path(repository, workspace_root)
    _require_operator_checkout(path)
    fetch_urls = _bounded_remote_urls(path, push=False)
    push_urls = _bounded_remote_urls(path, push=True)
    if fetch_urls != (repository.remote.strip(),) or push_urls != (
        repository.remote.strip(),
    ):
        raise ReleaseExecutionBlocked(
            "Frozen source origin no longer matches its registered release remote."
        )
    remote_digest = hashlib.sha256(
        json.dumps(
            {"fetch": fetch_urls, "push": push_urls},
            sort_keys=True,
            separators=(",", ":"),
            ensure_ascii=True,
        ).encode("utf-8")
    ).hexdigest()
    if receipt.get("remote") != DURABLE_REMOTE or receipt.get(
        "remote_sha256"
    ) != remote_digest:
        raise ReleaseExecutionBlocked("Frozen source remote binding changed.")
    local_object = git_text(path, "rev-parse", "--verify", f"refs/tags/{tag}")
    local_type = git_text(path, "cat-file", "-t", f"refs/tags/{tag}")
    local_commit = ref_commit(path, f"refs/tags/{tag}")
    remote = remote_tag_commit(path, remote=DURABLE_REMOTE, tag=tag)
    if (
        local_type != "tag"
        or local_object != tag_object
        or local_commit != head
        or remote.error
        or not remote.annotated
        or remote.tag_object != tag_object
        or remote.commit != head
    ):
        raise ReleaseExecutionBlocked(
            "Frozen source tag is not the same annotated object locally and on origin."
        )
    return dict(receipt)


def _checkout_frozen_source(
    *,
    repo: str,
    source_remote: str | Path,
    commit: str,
    tag: str,
    tag_object: str,
    source_root: Path,
) -> Path:
    destination = source_root / repo
    clone = subprocess.run(
        (
            "/usr/bin/git",
            "clone",
            "--quiet",
            "--no-checkout",
            "--no-hardlinks",
            str(source_remote),
            str(destination),
        ),
        cwd=source_root,
        check=False,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=120,
        env=_git_release_environment(),
    )
    if clone.returncode != 0:
        raise ReleaseExecutionBlocked(
            f"Frozen source clone failed for {repo}."
        )
    cloned_tag_object = git_text(
        destination, "rev-parse", "--verify", f"refs/tags/{tag}"
    )
    cloned_tag_commit = ref_commit(destination, f"refs/tags/{tag}")
    cloned_tag_type = git_text(destination, "cat-file", "-t", f"refs/tags/{tag}")
    if (
        cloned_tag_type != "tag"
        or cloned_tag_object != tag_object
        or cloned_tag_commit != commit
    ):
        raise ReleaseExecutionBlocked(
            f"Registered origin returned a different release tag for {repo}."
        )
    checkout = subprocess.run(
        ("/usr/bin/git", "checkout", "--quiet", "--detach", commit),
        cwd=destination,
        check=False,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=120,
        env=_git_release_environment(),
    )
    if checkout.returncode != 0 or git_text(
        destination, "rev-parse", "--verify", "HEAD"
    ) != commit:
        raise ReleaseExecutionBlocked(
            f"Frozen source checkout failed for {repo}: "
            f"{_compact_output(checkout.stderr) or 'commit identity mismatch'}"
        )
    return destination


def _checkout_registered_source_tag(
    *,
    repo: str,
    source_remote: str | Path,
    tag: str,
    source_root: Path,
) -> Path:
    """Clone one unchanged authenticated-base source at its annotated origin tag."""

    destination = source_root / repo
    clone = subprocess.run(
        (
            "/usr/bin/git",
            "clone",
            "--quiet",
            "--no-checkout",
            "--no-hardlinks",
            str(source_remote),
            str(destination),
        ),
        cwd=source_root,
        check=False,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=120,
        env=_git_release_environment(),
    )
    if clone.returncode != 0:
        raise ReleaseExecutionBlocked(
            f"Authenticated catalog source clone failed for {repo}."
        )
    tag_object = git_text(destination, "rev-parse", "--verify", f"refs/tags/{tag}")
    tag_commit = ref_commit(destination, f"refs/tags/{tag}")
    if (
        git_text(destination, "cat-file", "-t", f"refs/tags/{tag}") != "tag"
        or len(tag_object) not in {40, 64}
        or not tag_commit
    ):
        raise ReleaseExecutionBlocked(
            f"Authenticated catalog source tag is not annotated for {repo}."
        )
    checkout = subprocess.run(
        ("/usr/bin/git", "checkout", "--quiet", "--detach", tag_commit),
        cwd=destination,
        check=False,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=120,
        env=_git_release_environment(),
    )
    if checkout.returncode != 0 or git_text(
        destination, "rev-parse", "--verify", "HEAD"
    ) != tag_commit:
        raise ReleaseExecutionBlocked(
            f"Authenticated catalog source checkout failed for {repo}."
        )
    return destination


def _git_release_environment() -> dict[str, str]:
    environment = sanitized_git_environment()
    environment.setdefault(
        "GIT_SSH_COMMAND",
        "ssh -o BatchMode=yes -o ConnectTimeout=8",
    )
    return environment


def python_project_name(repository: Path) -> str | None:
    pyproject = repository / "pyproject.toml"
    try:
        encoded = pyproject.read_bytes()
    except FileNotFoundError:
        return None
    except OSError as exc:
        raise ReleaseExecutionBlocked("Python project metadata cannot be read.") from exc
    if len(encoded) > 1024 * 1024:
        raise ReleaseExecutionBlocked("Python project metadata exceeds its size limit.")
    try:
        payload = tomllib.loads(encoded.decode("utf-8"))
    except (UnicodeDecodeError, tomllib.TOMLDecodeError) as exc:
        raise ReleaseExecutionBlocked("Python project metadata is malformed.") from exc
    project = payload.get("project")
    name = project.get("name") if isinstance(project, dict) else None
    return name.strip() if isinstance(name, str) and name.strip() else None


def publish_received_candidate(
    *,
    candidate_path: Path,
    candidate_receipt: dict[str, Any],
    channel: str,
    workspace_root: Path,
    remote: str,
    expected_website_receipt: dict[str, Any],
) -> tuple[dict[str, Any], dict[str, Any]]:
    require_trusted_release_runtime(workspace_root=workspace_root)
    website_binding = _validated_website_binding(expected_website_receipt)
    if remote != DURABLE_REMOTE:
        raise ReleaseExecutionBlocked("Catalog publication uses only origin.")
    try:
        result = publish_catalog_candidate(
            candidate_dir=candidate_path,
            workspace_root=workspace_root,
            channel=channel,
            apply=True,
            commit=True,
            tag=True,
            push=True,
            remote=remote,
            source_remote=remote,
            expected_website_head=str(website_binding["head"]),
            expected_website_branch=str(website_binding["branch"]),
            expected_remote_sha256=str(website_binding["remote_sha256"]),
        )
    except Exception as exc:  # noqa: BLE001 - partial Git effects are ambiguous.
        raise ReleaseExecutionAmbiguous(
            f"Catalog publication raised {type(exc).__name__}; reconcile website and remote state."
        ) from exc
    payload = to_jsonable(result)
    status = payload.get("status") if isinstance(payload, dict) else None
    if status == "published":
        try:
            receipt = reconciled_catalog_publication_receipt(
                candidate_path=candidate_path,
                candidate_receipt=candidate_receipt,
                channel=channel,
                workspace_root=workspace_root,
                remote=remote,
                expected_website_receipt=website_binding,
            )
            _require_publisher_result_matches_receipt(payload, receipt)
        except (ReleaseExecutionBlocked, OSError, RuntimeError, ValueError) as exc:
            raise ReleaseExecutionAmbiguous(
                "Catalog publisher reported success, but its exact commit, derived "
                "files, annotated tag, and origin state could not be proved."
            ) from exc
        return payload, receipt
    if status == "blocked":
        raise ReleaseExecutionBlocked(_bounded_result_detail(payload, "catalog publication"))
    raise ReleaseExecutionAmbiguous(_bounded_result_detail(payload, "catalog publication"))


def reconciled_catalog_publication_receipt(
    *,
    candidate_path: Path,
    candidate_receipt: dict[str, Any],
    channel: str,
    workspace_root: Path,
    remote: str,
    expected_website_receipt: dict[str, Any],
) -> dict[str, Any]:
    """Independently prove an exact candidate publication after interruption."""

    require_trusted_release_runtime(workspace_root=workspace_root)
    if remote != DURABLE_REMOTE:
        raise ReleaseExecutionBlocked("Catalog reconciliation uses only origin.")
    website_binding = _validated_website_binding(expected_website_receipt)
    candidate_id, expected_catalog_hash = _validated_catalog_candidate_receipt(
        candidate_receipt
    )
    channel = validate_release_channel(channel)
    catalog_payload = _read_generated_catalog(candidate_path, channel=channel)
    catalog_hash = canonical_hash(catalog_payload)
    if catalog_hash != expected_catalog_hash:
        raise ReleaseExecutionBlocked(
            "Catalog candidate changed after its durable generator receipt."
        )
    keyring_payload = _read_candidate_json(
        candidate_path / "keyring.json",
        label="candidate keyring",
    )
    keyring_hash = canonical_hash(keyring_payload)
    release = catalog_payload.get("release")
    if (
        not isinstance(release, dict)
        or release.get("keyring_sha256") != keyring_hash
    ):
        raise ReleaseExecutionBlocked(
            "Signed catalog does not pin the exact candidate keyring."
        )
    artifact_issues = selected_artifact_identity_issues(catalog_payload)
    if artifact_issues:
        raise ReleaseExecutionBlocked(
            "Candidate release artifact identity is incomplete: "
            + "; ".join(artifact_issues)
        )

    web_root = website_root(workspace_root)
    current = repository_state_receipt(
        repo="addideas-govoplan-website",
        target_tag="",
        workspace_root=workspace_root,
        include_website=True,
    )
    if (
        current.get("branch") != website_binding["branch"]
        or current.get("remote") != website_binding["remote"]
        or current.get("remote_sha256") != website_binding["remote_sha256"]
        or current.get("worktree_clean") is not True
    ):
        raise ReleaseExecutionBlocked(
            "Website checkout does not match the frozen publication boundary."
        )
    frozen_remote = bind_publication_remote(
        web_root=web_root,
        remote=remote,
        registered_remote=registered_website_remote(),
    )
    if frozen_remote.sha256 != website_binding["remote_sha256"]:
        raise ReleaseExecutionBlocked(
            "Website origin changed after the publication plan was frozen."
        )
    tag_name = default_tag_name(catalog_payload, channel=channel)
    if not tag_name:
        raise ReleaseExecutionBlocked(
            "Catalog publication tag cannot be reconstructed from the candidate."
        )
    identity = remote_publication_identity(
        web_root=web_root,
        remote_url=frozen_remote.url,
        branch=str(website_binding["branch"]),
        tag_name=tag_name,
    )
    commit = identity["publication_commit_sha"]
    tag_object = identity["publication_tag_object_sha"]
    if (
        identity["publication_tag_commit_sha"] != commit
        or current.get("head") != commit
    ):
        raise ReleaseExecutionBlocked(
            "Remote catalog branch and annotated tag do not identify one local commit."
        )
    if (
        git_text(web_root, "cat-file", "-t", f"refs/tags/{tag_name}") != "tag"
        or git_text(
            web_root,
            "rev-parse",
            "--verify",
            f"refs/tags/{tag_name}",
        )
        != tag_object
        or ref_commit(web_root, f"refs/tags/{tag_name}") != commit
    ):
        raise ReleaseExecutionBlocked(
            "Local catalog publication tag differs from the origin annotation."
        )
    parent_line = git_text(web_root, "rev-list", "--parents", "-n", "1", commit)
    if parent_line.split() != [commit, str(website_binding["head"])]:
        raise ReleaseExecutionBlocked(
            "Catalog publication commit does not have the sole frozen parent."
        )

    previous_keyring = _read_git_json_blob(
        web_root,
        commit=str(website_binding["head"]),
        repository_path="public/catalogs/v1/keyring.json",
        label="frozen website keyring",
    )
    trusted_keys = trusted_keys_from_keyring(previous_keyring)
    if not trusted_keys:
        raise ReleaseExecutionBlocked(
            "Frozen website commit has no catalog trust anchor."
        )
    candidate_keys = trusted_keys_from_keyring(keyring_payload)
    for key_id in set(candidate_keys) & set(trusted_keys):
        if candidate_keys[key_id] != trusted_keys[key_id]:
            raise ReleaseExecutionBlocked(
                "Candidate keyring replaces an existing trusted public key."
            )
    validation = validate_catalog_payload(
        catalog_payload,
        require_trusted=True,
        approved_channels=(channel,),
        trusted_keys=trusted_keys,
    )
    if validation.get("valid") is not True:
        raise ReleaseExecutionBlocked(
            "Candidate does not validate against the frozen website trust anchor."
        )

    expected_blobs, module_root = _expected_catalog_publication_blobs(
        web_root=web_root,
        catalog_payload=catalog_payload,
        keyring_payload=keyring_payload,
        channel=channel,
    )
    _verify_exact_publication_delta(
        web_root=web_root,
        frozen_parent=str(website_binding["head"]),
        commit=commit,
        expected_blobs=expected_blobs,
        module_root=module_root,
    )
    verify_committed_publication(
        web_root=web_root,
        commit_sha=commit,
        expected_blobs=expected_blobs,
        module_root=module_root,
    )
    return {
        "kind": "catalog_publication",
        "candidate_id": candidate_id,
        "catalog_sha256": catalog_hash,
        "keyring_sha256": keyring_hash,
        "publication_commit_sha": commit,
        "publication_tag_object_sha": tag_object,
        "publication_tag_commit_sha": commit,
        "branch": str(website_binding["branch"]),
        "tag_name": tag_name,
        "remote": remote,
        "remote_sha256": str(website_binding["remote_sha256"]),
    }


def _validated_website_binding(value: object) -> dict[str, Any]:
    if (
        not isinstance(value, dict)
        or value.get("kind") != "repository_state"
        or value.get("repo") != "addideas-govoplan-website"
        or value.get("remote") != DURABLE_REMOTE
        or value.get("target_tag") != ""
        or value.get("tag_object") is not None
        or value.get("worktree_clean") is not True
        or not isinstance(value.get("head"), str)
        or len(value["head"]) not in {40, 64}
        or not isinstance(value.get("branch"), str)
        or not value["branch"]
        or not isinstance(value.get("remote_sha256"), str)
        or len(value["remote_sha256"]) != 64
    ):
        raise ReleaseExecutionBlocked(
            "Catalog publication has no valid frozen website receipt."
        )
    return dict(value)


def _validated_catalog_candidate_receipt(value: object) -> tuple[str, str]:
    if (
        not isinstance(value, dict)
        or set(value) != {"kind", "candidate_id", "catalog_sha256"}
        or value.get("kind") != "catalog_candidate"
        or not isinstance(value.get("candidate_id"), str)
        or not value["candidate_id"].startswith("candidate-")
        or len(value["candidate_id"]) != 42
        or not isinstance(value.get("catalog_sha256"), str)
        or len(value["catalog_sha256"]) != 64
    ):
        raise ReleaseExecutionBlocked(
            "Catalog publication has no valid generator receipt."
        )
    return value["candidate_id"], value["catalog_sha256"]


def _expected_catalog_publication_blobs(
    *,
    web_root: Path,
    catalog_payload: dict[str, Any],
    keyring_payload: dict[str, Any],
    channel: str,
) -> tuple[dict[str, bytes], Path]:
    catalog_root = web_root / "public" / "catalogs" / "v1"
    module_root = catalog_root / "modules"
    payloads = module_directory_payloads(
        catalog_payload=catalog_payload,
        keyring_payload=keyring_payload,
        channel=channel,
        public_base_url=catalog_public_base_url(catalog_payload),
    )
    expected = {
        f"public/catalogs/v1/channels/{channel}.json": _public_json_bytes(
            catalog_payload
        ),
        "public/catalogs/v1/keyring.json": _public_json_bytes(keyring_payload),
    }
    for relative_path, payload in payloads:
        repository_path = (Path("public/catalogs/v1") / relative_path).as_posix()
        if repository_path in expected:
            raise ReleaseExecutionBlocked(
                "Derived catalog publication paths are ambiguous."
            )
        expected[repository_path] = _public_json_bytes(payload)
    return expected, module_root


def _verify_exact_publication_delta(
    *,
    web_root: Path,
    frozen_parent: str,
    commit: str,
    expected_blobs: dict[str, bytes],
    module_root: Path,
) -> None:
    """Require the publication commit to change only the derived catalog tree."""

    try:
        module_prefix = module_root.absolute().relative_to(web_root.absolute()).as_posix()
    except ValueError as exc:
        raise ReleaseExecutionBlocked(
            "Catalog module root leaves the website checkout."
        ) from exc
    tree = git(
        web_root,
        "ls-tree",
        "-r",
        "-z",
        frozen_parent,
        "--",
        "public/catalogs/v1/channels",
        "public/catalogs/v1/keyring.json",
        module_prefix,
        timeout=30,
    )
    tree_bytes = tree.stdout.encode("utf-8", errors="strict")
    if tree.returncode != 0 or len(tree_bytes) > 16 * 1024 * 1024:
        raise ReleaseExecutionBlocked(
            "Frozen catalog tree cannot be inspected within its bounds."
        )
    old_blobs: dict[str, str] = {}
    for row in (item for item in tree.stdout.split("\0") if item):
        metadata, separator, repository_path = row.partition("\t")
        fields = metadata.split()
        if (
            separator != "\t"
            or len(fields) != 3
            or fields[0] != "100644"
            or fields[1] != "blob"
            or len(fields[2]) not in {40, 64}
            or repository_path in old_blobs
        ):
            raise ReleaseExecutionBlocked(
                "Frozen catalog tree contains an ambiguous entry."
            )
        old_blobs[repository_path] = fields[2]
    object_format = git_text(web_root, "rev-parse", "--show-object-format")
    if object_format not in {"sha1", "sha256"}:
        raise ReleaseExecutionBlocked(
            "Website Git object format cannot be established."
        )
    expected_delta = {
        path
        for path, encoded in expected_blobs.items()
        if old_blobs.get(path) != _git_blob_identity(encoded, object_format)
    }
    expected_delta.update(
        path
        for path in old_blobs
        if path.startswith(f"{module_prefix}/") and path not in expected_blobs
    )
    observed = git(
        web_root,
        "diff-tree",
        "--no-commit-id",
        "--name-only",
        "--no-renames",
        "--no-ext-diff",
        "-r",
        "-z",
        frozen_parent,
        commit,
        timeout=30,
    )
    observed_bytes = observed.stdout.encode("utf-8", errors="strict")
    if observed.returncode != 0 or len(observed_bytes) > 16 * 1024 * 1024:
        raise ReleaseExecutionBlocked(
            "Catalog publication delta cannot be inspected within its bounds."
        )
    observed_delta = {path for path in observed.stdout.split("\0") if path}
    if observed_delta != expected_delta:
        raise ReleaseExecutionBlocked(
            "Catalog publication commit contains paths outside the exact derived delta."
        )


def _git_blob_identity(encoded: bytes, object_format: str) -> str:
    digest = hashlib.new(object_format)
    digest.update(f"blob {len(encoded)}\0".encode("ascii"))
    digest.update(encoded)
    return digest.hexdigest()


def _public_json_bytes(payload: object) -> bytes:
    encoded = (
        json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=True) + "\n"
    ).encode("utf-8")
    if len(encoded) > 16 * 1024 * 1024:
        raise ReleaseExecutionBlocked(
            "Derived catalog publication object exceeds its size limit."
        )
    return encoded


def _read_git_json_blob(
    repository: Path,
    *,
    commit: str,
    repository_path: str,
    label: str,
) -> dict[str, Any]:
    entry = git_text(
        repository,
        "ls-tree",
        commit,
        "--",
        repository_path,
    )
    metadata, separator, observed_path = entry.partition("\t")
    fields = metadata.split()
    if (
        separator != "\t"
        or observed_path != repository_path
        or len(fields) != 3
        or fields[0] != "100644"
        or fields[1] != "blob"
        or len(fields[2]) not in {40, 64}
    ):
        raise ReleaseExecutionBlocked(f"{label.capitalize()} is not one regular blob.")
    result = git(repository, "cat-file", "blob", fields[2], timeout=30)
    encoded = result.stdout.encode("utf-8", errors="strict")
    if result.returncode != 0 or len(encoded) > 16 * 1024 * 1024:
        raise ReleaseExecutionBlocked(f"{label.capitalize()} cannot be read safely.")
    try:
        payload = json.loads(encoded)
    except json.JSONDecodeError as exc:
        raise ReleaseExecutionBlocked(f"{label.capitalize()} is not valid JSON.") from exc
    if not isinstance(payload, dict):
        raise ReleaseExecutionBlocked(f"{label.capitalize()} is not a JSON object.")
    return payload


def _read_candidate_json(path: Path, *, label: str) -> dict[str, Any]:
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    descriptor = -1
    try:
        descriptor = os.open(path, flags)
        metadata = os.fstat(descriptor)
        if (
            not stat.S_ISREG(metadata.st_mode)
            or metadata.st_uid != os.geteuid()
            or metadata.st_mode & 0o077
        ):
            raise ReleaseExecutionBlocked(
                f"{label.capitalize()} is not a private regular file."
            )
        chunks: list[bytes] = []
        remaining = 16 * 1024 * 1024 + 1
        while remaining:
            chunk = os.read(descriptor, min(1024 * 1024, remaining))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        encoded = b"".join(chunks)
    except OSError as exc:
        raise ReleaseExecutionBlocked(
            f"{label.capitalize()} cannot be read safely."
        ) from exc
    finally:
        if descriptor >= 0:
            os.close(descriptor)
    if len(encoded) > 16 * 1024 * 1024:
        raise ReleaseExecutionBlocked(f"{label.capitalize()} exceeds its size limit.")
    try:
        payload = json.loads(encoded.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ReleaseExecutionBlocked(f"{label.capitalize()} is not valid JSON.") from exc
    if not isinstance(payload, dict):
        raise ReleaseExecutionBlocked(f"{label.capitalize()} is not a JSON object.")
    return payload


def _require_publisher_result_matches_receipt(
    result: dict[str, Any], receipt: dict[str, Any]
) -> None:
    fields = {
        "candidate_catalog_hash": "catalog_sha256",
        "candidate_keyring_hash": "keyring_sha256",
        "publication_commit_sha": "publication_commit_sha",
        "publication_tag_object_sha": "publication_tag_object_sha",
        "publication_tag_commit_sha": "publication_tag_commit_sha",
        "branch": "branch",
        "tag_name": "tag_name",
        "remote": "remote",
    }
    if any(result.get(source) != receipt[target] for source, target in fields.items()):
        raise ReleaseExecutionBlocked(
            "Catalog publisher result differs from independent publication proof."
        )


def _known_or_ambiguous_result(
    result: dict[str, object], *, success: set[str], operation: str
) -> dict[str, Any]:
    payload = to_jsonable(result)
    status = payload.get("status") if isinstance(payload, dict) else None
    if status in success:
        return payload
    if status == "blocked":
        raise ReleaseExecutionBlocked(_bounded_result_detail(payload, operation))
    raise ReleaseExecutionAmbiguous(_bounded_result_detail(payload, operation))


def _bounded_result_detail(payload: dict[str, Any], operation: str) -> str:
    repositories = payload.get("repositories")
    details = [
        str(item.get("detail"))
        for item in repositories
        if isinstance(item, dict) and item.get("detail")
    ] if isinstance(repositories, list) else []
    detail = "; ".join(details) or str(payload.get("status") or "unknown result")
    return _compact_output(f"{operation} did not establish success: {detail}")


def _read_generated_catalog(candidate: Path, *, channel: str) -> dict[str, Any]:
    channel = validate_release_channel(channel)
    path = candidate / "channels" / f"{channel}.json"
    flags = os.O_RDONLY
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = -1
    try:
        descriptor = os.open(path, flags)
        metadata = os.fstat(descriptor)
        if not stat.S_ISREG(metadata.st_mode):
            raise ReleaseExecutionBlocked("Generated catalog is not a regular file.")
        with os.fdopen(descriptor, "rb") as handle:
            descriptor = -1
            encoded = handle.read(16 * 1024 * 1024 + 1)
    except OSError as exc:
        raise ReleaseExecutionBlocked("Generated catalog cannot be read safely.") from exc
    finally:
        if descriptor >= 0:
            os.close(descriptor)
    if len(encoded) > 16 * 1024 * 1024:
        raise ReleaseExecutionBlocked("Generated catalog exceeds its size limit.")
    payload = json.loads(encoded.decode("utf-8"))
    if not isinstance(payload, dict):
        raise ReleaseExecutionBlocked("Generated catalog is not a JSON object.")
    return payload


def _remove_partial_candidate(candidate: Path) -> None:
    if not candidate.exists():
        return
    if candidate.is_symlink() or not candidate.is_dir():
        raise ReleaseExecutionAmbiguous(
            "Partial candidate path is unsafe and requires manual reconciliation."
        )
    shutil.rmtree(candidate)


def _compact_output(value: str) -> str:
    text = value.strip()
    return text if len(text) <= MAX_EXECUTOR_OUTPUT else text[:MAX_EXECUTOR_OUTPUT] + "…"