#!/usr/bin/env python3
"""Security boundary for one-approval silent release execution.

The signed plan binds immutable release evidence.  Each capability is consumed
independently so runtime deployment, tag publication, and release finalization
remain separate controllers and separately recoverable Saga steps.
"""
from __future__ import annotations

import base64
import hashlib
import json
import os
import re
import tempfile
import time
import uuid
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
from typing import Any, Callable, Iterable, Mapping, Protocol

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

from silent_release_controller import (
    CANDIDATE_ID_RE,
    OID_RE,
    SHA256_RE,
    ReleaseCandidate,
    ReleaseCandidateStore,
    ReleaseState,
    _atomic_json,
    _parse_utc,
    _sanitize,
    utc_now,
)


IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{2,255}$")
NONCE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{15,255}$")
ALLOWED_CAPABILITIES = frozenset(
    {
        "runtime_switch",
        "tag_publish",
        "baseline_registry_finalize",
    }
)
FORBIDDEN_CAPABILITIES = frozenset(
    {
        "force_push",
        "tag_rewrite",
        "direct_main_push",
        "baseline_mutation",
        "registry_mutation",
    }
)


class CapabilityAuthorizationError(RuntimeError):
    """The signed plan cannot authorize the requested operation."""


class StaleApprovalError(CapabilityAuthorizationError):
    """Live release evidence no longer matches the signed approval."""


class SagaConflict(RuntimeError):
    """A stale controller attempted a non-current Saga transition."""


class JournalIntegrityError(RuntimeError):
    """The durable Saga journal is truncated, reordered, or modified."""


class ClaimProbeResult(str, Enum):
    NOT_APPLIED = "NOT_APPLIED"
    APPLIED_VERIFIED = "APPLIED_VERIFIED"
    INDETERMINATE = "INDETERMINATE"
    CONFLICT = "CONFLICT"


@dataclass(frozen=True)
class ProbeEvidence:
    """Fresh, operation-bound evidence for a side-effect recovery decision."""

    result: ClaimProbeResult
    operation_digest: str
    observed_at: float
    evidence: Mapping[str, Any] = field(default_factory=dict)

    def validate(
        self,
        *,
        expected_operation_digest: str,
        now: float,
        maximum_age_seconds: float,
    ) -> "ProbeEvidence":
        if self.operation_digest != expected_operation_digest:
            raise CapabilityAuthorizationError(
                "probe evidence operation digest mismatch"
            )
        age = float(now) - float(self.observed_at)
        if age < 0 or age > float(maximum_age_seconds):
            raise CapabilityAuthorizationError(
                "probe evidence is outside its freshness window"
            )
        return self

    def to_dict(self) -> dict[str, Any]:
        return {
            "evidence": _sanitize(dict(self.evidence)),
            "evidence_sha256": hashlib.sha256(
                _canonical_json(_sanitize(dict(self.evidence)))
            ).hexdigest(),
            "observed_at": float(self.observed_at),
            "operation_digest": self.operation_digest,
            "result": ClaimProbeResult(self.result).value,
        }


@dataclass(frozen=True)
class TagPublicationProbeEvidence:
    """Independent Git/Gitea observations used to classify TAG_PENDING."""

    repository_identity: str | None
    remote_reachable: bool
    gitea_api_reachable: bool
    gitea_listing_reachable: bool
    remote_tag_oid: str | None = None
    annotated_tag_object_oid: str | None = None
    remote_peeled_target_oid: str | None = None
    gitea_api_tag_oid: str | None = None
    gitea_api_peeled_target_oid: str | None = None
    gitea_listing_visible: bool | None = None
    gitea_listing_tag_oid: str | None = None
    gitea_listing_peeled_target_oid: str | None = None
    observed_at: float = field(default_factory=time.time)

    def to_dict(self) -> dict[str, Any]:
        return _sanitize(asdict(self))


def classify_tag_publication(
    evidence: TagPublicationProbeEvidence,
    plan: "TagPlan",
) -> ClaimProbeResult:
    """Classify a tag without mutating, deleting, or rewriting any ref."""

    if (
        not evidence.remote_reachable
        or not evidence.gitea_api_reachable
        or not evidence.gitea_listing_reachable
        or evidence.repository_identity is None
    ):
        return ClaimProbeResult.INDETERMINATE
    if evidence.repository_identity != plan.repository_identity:
        return ClaimProbeResult.CONFLICT

    tag_oids = (
        evidence.remote_tag_oid,
        evidence.gitea_api_tag_oid,
        evidence.gitea_listing_tag_oid,
    )
    target_oids = (
        evidence.remote_peeled_target_oid,
        evidence.gitea_api_peeled_target_oid,
        evidence.gitea_listing_peeled_target_oid,
    )
    if (
        all(value is None for value in tag_oids)
        and all(value is None for value in target_oids)
        and evidence.annotated_tag_object_oid is None
        and evidence.gitea_listing_visible is False
    ):
        return ClaimProbeResult.NOT_APPLIED

    observed_tag_oids = tuple(value for value in tag_oids if value is not None)
    observed_targets = tuple(
        value for value in target_oids if value is not None
    )
    if (
        any(value != plan.expected_tag_oid for value in observed_tag_oids)
        or any(value != plan.expected_target_oid for value in observed_targets)
        or (
            plan.tag_type == "annotated"
            and evidence.annotated_tag_object_oid is not None
            and evidence.annotated_tag_object_oid != plan.expected_tag_oid
        )
    ):
        return ClaimProbeResult.CONFLICT

    exact = (
        tag_oids
        == (
            plan.expected_tag_oid,
            plan.expected_tag_oid,
            plan.expected_tag_oid,
        )
        and target_oids
        == (
            plan.expected_target_oid,
            plan.expected_target_oid,
            plan.expected_target_oid,
        )
        and evidence.gitea_listing_visible is True
        and (
            plan.tag_type != "annotated"
            or evidence.annotated_tag_object_oid == plan.expected_tag_oid
        )
    )
    return (
        ClaimProbeResult.APPLIED_VERIFIED
        if exact
        else ClaimProbeResult.INDETERMINATE
    )


JOURNAL_SCHEMA_VERSION = 1
JOURNAL_GENESIS_HASH = "0" * 64


def _fsync_directory(path: Path) -> None:
    descriptor = os.open(Path(path), os.O_RDONLY)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def _durable_json(path: Path, payload: Mapping[str, Any]) -> None:
    _atomic_json(path, dict(payload))
    _fsync_directory(path.parent)


def _validate_identifier(name: str, value: str) -> str:
    normalized = str(value or "")
    if not IDENTIFIER_RE.fullmatch(normalized):
        raise ValueError(f"invalid {name}")
    return normalized


def _validate_oid(name: str, value: str) -> str:
    normalized = str(value or "").lower()
    if not OID_RE.fullmatch(normalized):
        raise ValueError(f"{name} must be a full 40-character OID")
    return normalized


def _validate_sha(name: str, value: str) -> str:
    normalized = str(value or "").lower()
    if not SHA256_RE.fullmatch(normalized):
        raise ValueError(f"{name} must be a SHA-256 digest")
    return normalized


def _canonical_json(payload: Mapping[str, Any]) -> bytes:
    return json.dumps(
        payload,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=True,
    ).encode("utf-8")


def _safe_child(root: Path, name: str, suffix: str = ".json") -> Path:
    _validate_identifier("storage key", name)
    resolved_root = Path(root).resolve()
    candidate = (resolved_root / f"{name}{suffix}").resolve()
    if candidate.parent != resolved_root:
        raise ValueError("storage path escapes root")
    return candidate


@dataclass(frozen=True)
class DeploymentPlan:
    service: str
    release_path: str
    release_root: str
    dropin_path: str
    dropin_sha256: str
    rollback_path: str
    rollback_sha256: str
    source_commit: str
    artifact_sha256: str
    manifest_sha256: str
    policy_sha256: str
    health_contract: dict[str, Any]
    cutover_link: str = ""
    systemd_unit_sha256: str = ""
    tree_oid: str = ""
    package_closure_sha256: str = ""
    authoritative_git_dir: str = ""
    repository_identity: str = ""
    writer_expected: bool = False
    baseline_before_sha256: str = ""
    registry_before_sha256: str = ""
    production_tag_refset_before_sha256: str = ""
    health_contract_version: str = ""
    timeout_policy: dict[str, Any] = field(default_factory=dict)
    retry_policy: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        _validate_identifier("service", self.service)
        _validate_oid("source_commit", self.source_commit)
        for name in (
            "dropin_sha256",
            "rollback_sha256",
            "artifact_sha256",
            "manifest_sha256",
            "policy_sha256",
        ):
            _validate_sha(name, getattr(self, name))
        if self.tree_oid:
            _validate_oid("tree_oid", self.tree_oid)
        for name in (
            "systemd_unit_sha256",
            "package_closure_sha256",
            "baseline_before_sha256",
            "registry_before_sha256",
            "production_tag_refset_before_sha256",
        ):
            if getattr(self, name):
                _validate_sha(name, getattr(self, name))
        if not isinstance(self.health_contract, dict):
            raise ValueError("health_contract must be an object")
        if not isinstance(self.timeout_policy, dict):
            raise ValueError("timeout_policy must be an object")
        if not isinstance(self.retry_policy, dict):
            raise ValueError("retry_policy must be an object")
        if self.writer_expected is not False:
            raise ValueError("production writer must remain disabled")

    @property
    def canonical_payload(self) -> dict[str, Any]:
        return {
            "artifact_sha256": self.artifact_sha256,
            "authoritative_git_dir": self.authoritative_git_dir,
            "baseline_before_sha256": self.baseline_before_sha256,
            "cutover_link": self.cutover_link,
            "dropin_path": self.dropin_path,
            "dropin_sha256": self.dropin_sha256,
            "health_contract": self.health_contract,
            "health_contract_version": self.health_contract_version,
            "manifest_sha256": self.manifest_sha256,
            "package_closure_sha256": self.package_closure_sha256,
            "policy_sha256": self.policy_sha256,
            "production_tag_refset_before_sha256": (
                self.production_tag_refset_before_sha256
            ),
            "release_path": self.release_path,
            "release_root": self.release_root,
            "repository_identity": self.repository_identity,
            "retry_policy": self.retry_policy,
            "rollback_path": self.rollback_path,
            "rollback_sha256": self.rollback_sha256,
            "service": self.service,
            "source_commit": self.source_commit,
            "systemd_unit_sha256": self.systemd_unit_sha256,
            "timeout_policy": self.timeout_policy,
            "tree_oid": self.tree_oid,
            "writer_expected": self.writer_expected,
            "registry_before_sha256": self.registry_before_sha256,
        }

    @property
    def sha256(self) -> str:
        return hashlib.sha256(
            _canonical_json(self.canonical_payload)
        ).hexdigest()

    def validate(
        self,
        *,
        service_allowlist: Iterable[str],
        dropin_allowlist: Iterable[str],
    ) -> "DeploymentPlan":
        if self.service not in frozenset(service_allowlist):
            raise ValueError("service is not allowlisted")
        release_root = Path(self.release_root).resolve()
        release_path = Path(self.release_path).resolve()
        try:
            release_path.relative_to(release_root)
        except ValueError as exc:
            raise ValueError("release path escapes release root") from exc
        if release_path == release_root:
            raise ValueError("release path must be a release child")
        dropin = Path(self.dropin_path).resolve()
        try:
            dropin.relative_to(release_path)
        except ValueError as exc:
            raise ValueError("drop-in escapes release path") from exc
        allowed_dropins = {
            str(Path(value).resolve()) for value in dropin_allowlist
        }
        if str(dropin) not in allowed_dropins:
            raise ValueError("drop-in is not allowlisted")
        return self

    def validate_complete(self) -> "DeploymentPlan":
        for name in (
            "cutover_link",
            "authoritative_git_dir",
            "repository_identity",
            "health_contract_version",
        ):
            if not str(getattr(self, name) or ""):
                raise ValueError(f"{name} is required")
        _validate_oid("tree_oid", self.tree_oid)
        for name in (
            "systemd_unit_sha256",
            "package_closure_sha256",
            "baseline_before_sha256",
            "registry_before_sha256",
            "production_tag_refset_before_sha256",
        ):
            _validate_sha(name, getattr(self, name))
        if not self.timeout_policy or not self.retry_policy:
            raise ValueError("timeout and retry policies are required")
        return self

    def to_dict(self) -> dict[str, Any]:
        return _sanitize({**self.canonical_payload, "sha256": self.sha256})

    @classmethod
    def from_dict(cls, payload: Mapping[str, Any]) -> "DeploymentPlan":
        values = dict(payload)
        values.pop("sha256", None)
        return cls(**values)


@dataclass(frozen=True)
class TagPlan:
    version: str
    tag_type: str
    expected_tag_oid: str
    expected_target_oid: str
    evidence_closure_oid: str
    approval_plan_id: str
    repository_identity: str
    gitea_owner: str
    gitea_repository: str
    current_remote_tag_oid: str | None
    remote_absent_evidence_sha256: str | None
    verification_contract: str
    force_push_allowed: bool = False
    tag_rewrite_allowed: bool = False

    def __post_init__(self) -> None:
        _validate_identifier("version", self.version)
        if self.tag_type not in {"annotated", "lightweight"}:
            raise ValueError("unsupported tag type")
        for name in (
            "expected_tag_oid",
            "expected_target_oid",
            "evidence_closure_oid",
        ):
            _validate_oid(name, getattr(self, name))
        _validate_identifier("approval_plan_id", self.approval_plan_id)
        _validate_identifier("gitea_owner", self.gitea_owner)
        _validate_identifier("gitea_repository", self.gitea_repository)
        present = self.current_remote_tag_oid is not None
        absent = self.remote_absent_evidence_sha256 is not None
        if present == absent:
            raise ValueError(
                "tag plan must bind either current tag or signed absence"
            )
        if present:
            _validate_oid(
                "current_remote_tag_oid", self.current_remote_tag_oid or ""
            )
        if absent:
            _validate_sha(
                "remote_absent_evidence_sha256",
                self.remote_absent_evidence_sha256 or "",
            )
        if not self.repository_identity or not self.verification_contract:
            raise ValueError("tag repository and verification are required")
        if self.force_push_allowed or self.tag_rewrite_allowed:
            raise ValueError("force push and tag rewrite are forbidden")

    @property
    def canonical_payload(self) -> dict[str, Any]:
        return asdict(self)

    @property
    def sha256(self) -> str:
        return hashlib.sha256(
            _canonical_json(self.canonical_payload)
        ).hexdigest()

    def to_dict(self) -> dict[str, Any]:
        return _sanitize({**self.canonical_payload, "sha256": self.sha256})

    @classmethod
    def from_dict(cls, payload: Mapping[str, Any]) -> "TagPlan":
        values = dict(payload)
        values.pop("sha256", None)
        return cls(**values)


@dataclass(frozen=True)
class FinalizationPlan:
    version: str
    expected_tag_oid: str
    expected_target_oid: str
    expected_previous_baseline_sha256: str
    registry_expected_record: dict[str, Any]
    baseline_expected_record: dict[str, Any]
    final_consistency_contract: str

    def __post_init__(self) -> None:
        _validate_identifier("version", self.version)
        _validate_oid("expected_tag_oid", self.expected_tag_oid)
        _validate_oid("expected_target_oid", self.expected_target_oid)
        _validate_sha(
            "expected_previous_baseline_sha256",
            self.expected_previous_baseline_sha256,
        )
        if not isinstance(self.registry_expected_record, dict):
            raise ValueError("registry expected record must be an object")
        if not isinstance(self.baseline_expected_record, dict):
            raise ValueError("baseline expected record must be an object")
        if not self.final_consistency_contract:
            raise ValueError("final consistency contract is required")

    @property
    def canonical_payload(self) -> dict[str, Any]:
        return asdict(self)

    @property
    def sha256(self) -> str:
        return hashlib.sha256(
            _canonical_json(self.canonical_payload)
        ).hexdigest()

    def to_dict(self) -> dict[str, Any]:
        return _sanitize({**self.canonical_payload, "sha256": self.sha256})

    @classmethod
    def from_dict(cls, payload: Mapping[str, Any]) -> "FinalizationPlan":
        values = dict(payload)
        values.pop("sha256", None)
        return cls(**values)


@dataclass(frozen=True)
class ReleasePlan:
    deployment: DeploymentPlan
    tag: TagPlan
    finalization: FinalizationPlan
    policy_sha256: str

    def __post_init__(self) -> None:
        _validate_sha("policy_sha256", self.policy_sha256)

    def validate(self) -> "ReleasePlan":
        self.deployment.validate_complete()
        if self.deployment.policy_sha256 != self.policy_sha256:
            raise ValueError("deployment policy does not match release plan")
        if self.deployment.repository_identity != self.tag.repository_identity:
            raise ValueError("release repository identity is inconsistent")
        if self.deployment.source_commit != self.tag.expected_target_oid:
            raise ValueError("deployment commit and tag target are inconsistent")
        if self.tag.version != self.finalization.version:
            raise ValueError("release version is inconsistent")
        if self.tag.expected_tag_oid != self.finalization.expected_tag_oid:
            raise ValueError("release tag OID is inconsistent")
        if (
            self.tag.expected_target_oid
            != self.finalization.expected_target_oid
        ):
            raise ValueError("release target OID is inconsistent")
        if (
            self.deployment.baseline_before_sha256
            != self.finalization.expected_previous_baseline_sha256
        ):
            raise ValueError("previous baseline is inconsistent")
        return self

    @property
    def canonical_payload(self) -> dict[str, Any]:
        return {
            "deployment": self.deployment.canonical_payload,
            "finalization": self.finalization.canonical_payload,
            "policy_sha256": self.policy_sha256,
            "tag": self.tag.canonical_payload,
        }

    @property
    def sha256(self) -> str:
        return hashlib.sha256(
            _canonical_json(self.canonical_payload)
        ).hexdigest()

    def to_dict(self) -> dict[str, Any]:
        return _sanitize({**self.canonical_payload, "sha256": self.sha256})

    @classmethod
    def from_dict(cls, payload: Mapping[str, Any]) -> "ReleasePlan":
        values = dict(payload)
        values.pop("sha256", None)
        return cls(
            deployment=DeploymentPlan.from_dict(values["deployment"]),
            tag=TagPlan.from_dict(values["tag"]),
            finalization=FinalizationPlan.from_dict(
                values["finalization"]
            ),
            policy_sha256=values["policy_sha256"],
        )


@dataclass(frozen=True)
class TrustedReleasePolicy:
    services: frozenset[str]
    release_roots: frozenset[str]
    cutover_links: frozenset[str]
    dropin_paths: frozenset[str]
    authoritative_git_dirs: frozenset[str]
    repository_identities: frozenset[str]

    def validate_deployment(self, plan: DeploymentPlan) -> DeploymentPlan:
        plan.validate_complete()
        checks = (
            (plan.service, self.services, "service"),
            (plan.release_root, self.release_roots, "release root"),
            (plan.cutover_link, self.cutover_links, "cutover link"),
            (plan.dropin_path, self.dropin_paths, "drop-in"),
            (
                plan.authoritative_git_dir,
                self.authoritative_git_dirs,
                "authoritative Git directory",
            ),
            (
                plan.repository_identity,
                self.repository_identities,
                "repository identity",
            ),
        )
        for value, allowlist, label in checks:
            if value not in allowlist:
                raise ValueError(f"{label} is not allowlisted")
        plan.validate(
            service_allowlist=self.services,
            dropin_allowlist=self.dropin_paths,
        )
        return plan


class ReleasePlanStore:
    def __init__(self, root: Path):
        self.root = Path(root)

    def save(self, candidate_id: str, plan: ReleasePlan) -> Path:
        path = _safe_child(self.root, candidate_id)
        if path.exists():
            existing = self.load(candidate_id)
            if existing.sha256 != plan.sha256:
                raise SagaConflict(
                    "candidate is bound to another release plan"
                )
            return path
        _durable_json(path, plan.to_dict())
        return path

    def load(self, candidate_id: str) -> ReleasePlan:
        return ReleasePlan.from_dict(
            json.loads(
                _safe_child(self.root, candidate_id).read_text(
                    encoding="utf-8"
                )
            )
        )


class SignedApprovalStore:
    def __init__(self, root: Path):
        self.root = Path(root)

    def save(
        self, candidate_id: str, approval: "SignedReleasePlanApproval"
    ) -> tuple[Path, bool]:
        path = _safe_child(self.root, candidate_id)
        if path.exists():
            existing = self.load(candidate_id)
            if existing.fingerprint != approval.fingerprint:
                raise CapabilityAuthorizationError(
                    "candidate has a conflicting owner approval"
                )
            return path, False
        _durable_json(path, approval.to_dict())
        return path, True

    def load(self, candidate_id: str) -> "SignedReleasePlanApproval":
        return SignedReleasePlanApproval.from_dict(
            json.loads(
                _safe_child(self.root, candidate_id).read_text(
                    encoding="utf-8"
                )
            )
        )


@dataclass(frozen=True)
class BuilderAttestation:
    builder_attestation_id: str
    builder_key_id: str
    source_commit: str
    tree_oid: str
    artifact_sha256: str
    manifest_sha256: str
    package_closure_sha256: str
    test_evidence_sha256: str
    canary_evidence_sha256: str
    issued_at: str
    signature: str

    def __post_init__(self) -> None:
        _validate_identifier(
            "builder_attestation_id", self.builder_attestation_id
        )
        _validate_identifier("builder_key_id", self.builder_key_id)
        _validate_oid("source_commit", self.source_commit)
        _validate_oid("tree_oid", self.tree_oid)
        for name in (
            "artifact_sha256",
            "manifest_sha256",
            "package_closure_sha256",
            "test_evidence_sha256",
            "canary_evidence_sha256",
        ):
            _validate_sha(name, getattr(self, name))
        _parse_utc(self.issued_at)

    @property
    def canonical_payload(self) -> dict[str, Any]:
        values = asdict(self)
        values.pop("signature")
        return values

    def canonical_bytes(self) -> bytes:
        return _canonical_json(self.canonical_payload)

    @property
    def fingerprint(self) -> str:
        return hashlib.sha256(self.canonical_bytes()).hexdigest()

    def to_dict(self) -> dict[str, Any]:
        return {
            **self.canonical_payload,
            "fingerprint": self.fingerprint,
            "signature": self.signature,
        }

    @classmethod
    def from_dict(cls, payload: Mapping[str, Any]) -> "BuilderAttestation":
        values = dict(payload)
        values.pop("fingerprint", None)
        return cls(**values)


class BuilderAttestationVerifier:
    """Public-key-only verifier; no builder private material is accepted."""

    def __init__(self, public_keys: Mapping[str, bytes]):
        self.public_keys = {
            _validate_identifier("builder_key_id", key_id): bytes(value)
            for key_id, value in public_keys.items()
        }

    def verify(
        self,
        attestation: BuilderAttestation,
        candidate: ReleaseCandidate,
        release_plan: ReleasePlan,
    ) -> BuilderAttestation:
        expected = (
            candidate.source_commit,
            candidate.tree_oid,
            candidate.artifact_sha256,
            candidate.manifest_sha256,
            release_plan.deployment.package_closure_sha256,
        )
        actual = (
            attestation.source_commit,
            attestation.tree_oid,
            attestation.artifact_sha256,
            attestation.manifest_sha256,
            attestation.package_closure_sha256,
        )
        if actual != expected:
            raise CapabilityAuthorizationError(
                "builder attestation does not bind candidate evidence"
            )
        key = self.public_keys.get(attestation.builder_key_id)
        if key is None:
            raise CapabilityAuthorizationError(
                "builder attestation key is not trusted"
            )
        try:
            signature = base64.b64decode(
                attestation.signature, validate=True
            )
            Ed25519PublicKey.from_public_bytes(key).verify(
                signature, attestation.canonical_bytes()
            )
        except (InvalidSignature, ValueError, TypeError) as exc:
            raise CapabilityAuthorizationError(
                "builder attestation signature is invalid"
            ) from exc
        return attestation


@dataclass(frozen=True)
class TrustedCandidateBundle:
    candidate: ReleaseCandidate
    release_plan: ReleasePlan
    checks: Mapping[str, bool]
    attestation: BuilderAttestation | None = None

    REQUIRED_CHECKS = frozenset(
        {
            "authoritative_git",
            "artifact",
            "canary",
            "gitea",
            "manifest",
            "package_closure",
            "rollback",
            "secret_scan",
            "tests",
        }
    )

    def validate(
        self,
        attestation_verifier: BuilderAttestationVerifier | None = None,
    ) -> "TrustedCandidateBundle":
        missing = self.REQUIRED_CHECKS.difference(self.checks)
        if missing or any(self.checks.get(name) is not True for name in self.REQUIRED_CHECKS):
            raise CapabilityAuthorizationError(
                "trusted builder acceptance is incomplete"
            )
        candidate = self.candidate
        if (
            candidate.state != ReleaseState.READY_FOR_APPROVAL
            or candidate.tests.get("passed") is not True
            or candidate.canary.get("passed") is not True
            or candidate.rollback_ready is not True
            or candidate.git_consistency.get("consistent") is not True
            or candidate.gitea_consistency.get("consistent") is not True
        ):
            raise CapabilityAuthorizationError(
                "candidate is not trusted release evidence"
            )
        plan = self.release_plan.validate()
        if (
            plan.deployment.source_commit != candidate.source_commit
            or plan.deployment.tree_oid != candidate.tree_oid
            or plan.deployment.artifact_sha256
            != candidate.artifact_sha256
            or plan.deployment.manifest_sha256
            != candidate.manifest_sha256
        ):
            raise CapabilityAuthorizationError(
                "trusted candidate and release plan are inconsistent"
            )
        if attestation_verifier is not None:
            if self.attestation is None:
                raise CapabilityAuthorizationError(
                    "trusted builder attestation is required"
                )
            attestation_verifier.verify(
                self.attestation, candidate, self.release_plan
            )
        return self

    def acceptance_record(self) -> dict[str, Any]:
        return {
            "attestation": (
                self.attestation.to_dict() if self.attestation else None
            ),
            "candidate_id": self.candidate.candidate_id,
            "checks": _sanitize(dict(self.checks)),
            "release_plan_sha256": self.release_plan.sha256,
            "status": "ACCEPTED",
        }


class TrustedCandidateBuilder(Protocol):
    def build(self, intent: Mapping[str, Any]) -> TrustedCandidateBundle:
        ...


class DeploymentPlanStore:
    def __init__(self, root: Path):
        self.root = Path(root)

    def save(self, candidate_id: str, plan: DeploymentPlan) -> Path:
        path = _safe_child(self.root, candidate_id)
        _atomic_json(path, plan.to_dict())
        return path

    def load(self, candidate_id: str) -> DeploymentPlan:
        return DeploymentPlan.from_dict(
            json.loads(
                _safe_child(self.root, candidate_id).read_text(
                    encoding="utf-8"
                )
            )
        )


@dataclass(frozen=True)
class SignedReleasePlanApproval:
    release_plan_id: str
    candidate_id: str
    source_commit: str
    tree_oid: str
    artifact_sha256: str
    manifest_sha256: str
    deployment_plan_sha256: str
    policy_sha256: str
    approval_plan_id: str
    evidence_closure_oid: str
    owner_approval_record_oid: str
    capabilities: tuple[str, ...]
    expires_at: str
    nonce: str
    key_id: str
    signature: str
    owner_identity: str = ""
    trust_domain: str = ""
    issued_at: str = "1970-01-01T00:00:00Z"
    not_before: str = "1970-01-01T00:00:00Z"
    release_plan_sha256: str = ""

    def __post_init__(self) -> None:
        for name in (
            "release_plan_id",
            "candidate_id",
            "approval_plan_id",
            "key_id",
        ):
            _validate_identifier(name, getattr(self, name))
        if not NONCE_RE.fullmatch(str(self.nonce or "")):
            raise ValueError("invalid nonce")
        for name in (
            "source_commit",
            "tree_oid",
            "evidence_closure_oid",
            "owner_approval_record_oid",
        ):
            _validate_oid(name, getattr(self, name))
        for name in (
            "artifact_sha256",
            "manifest_sha256",
            "deployment_plan_sha256",
            "policy_sha256",
        ):
            _validate_sha(name, getattr(self, name))
        if self.release_plan_sha256:
            _validate_sha(
                "release_plan_sha256", self.release_plan_sha256
            )
        capabilities = tuple(str(item) for item in self.capabilities)
        object.__setattr__(self, "capabilities", capabilities)
        if (
            not capabilities
            or len(capabilities) != len(set(capabilities))
            or not set(capabilities).issubset(ALLOWED_CAPABILITIES)
        ):
            raise ValueError("approval contains invalid capabilities")
        _parse_utc(self.expires_at)
        _parse_utc(self.issued_at)
        _parse_utc(self.not_before)

    @property
    def canonical_payload(self) -> dict[str, Any]:
        return {
            "approval_plan_id": self.approval_plan_id,
            "artifact_sha256": self.artifact_sha256,
            "candidate_id": self.candidate_id,
            "capabilities": list(self.capabilities),
            "deployment_plan_sha256": self.deployment_plan_sha256,
            "evidence_closure_oid": self.evidence_closure_oid,
            "expires_at": self.expires_at,
            "key_id": self.key_id,
            "manifest_sha256": self.manifest_sha256,
            "nonce": self.nonce,
            "not_before": self.not_before,
            "owner_identity": self.owner_identity,
            "owner_approval_record_oid": self.owner_approval_record_oid,
            "policy_sha256": self.policy_sha256,
            "release_plan_sha256": self.release_plan_sha256,
            "release_plan_id": self.release_plan_id,
            "source_commit": self.source_commit,
            "tree_oid": self.tree_oid,
            "trust_domain": self.trust_domain,
            "issued_at": self.issued_at,
        }

    def canonical_bytes(self) -> bytes:
        return _canonical_json(self.canonical_payload)

    @property
    def fingerprint(self) -> str:
        return hashlib.sha256(self.canonical_bytes()).hexdigest()

    def to_dict(self) -> dict[str, Any]:
        return {
            **self.canonical_payload,
            "fingerprint": self.fingerprint,
            "signature": self.signature,
        }

    @classmethod
    def from_dict(
        cls, payload: Mapping[str, Any]
    ) -> "SignedReleasePlanApproval":
        values = dict(payload)
        values.pop("fingerprint", None)
        values["capabilities"] = tuple(values.get("capabilities") or ())
        return cls(**values)


class ApprovalKeyRing:
    """Public-only Ed25519 key ring with in-process rotation/revocation."""

    def __init__(self, public_keys: Mapping[str, bytes]):
        self._keys: dict[str, bytes] = {}
        self._revoked: set[str] = set()
        for key_id, public_key in public_keys.items():
            self.add(key_id, public_key)

    def add(self, key_id: str, public_key: bytes) -> None:
        key_id = _validate_identifier("key_id", key_id)
        raw = bytes(public_key)
        if len(raw) != 32:
            raise ValueError("Ed25519 public key must contain 32 bytes")
        Ed25519PublicKey.from_public_bytes(raw)
        self._keys[key_id] = raw
        self._revoked.discard(key_id)

    def revoke(self, key_id: str) -> None:
        if key_id not in self._keys:
            raise KeyError("unknown approval signing key")
        self._revoked.add(key_id)

    def verify(
        self, key_id: str, signature: bytes, message: bytes
    ) -> None:
        if key_id not in self._keys or key_id in self._revoked:
            raise CapabilityAuthorizationError(
                "approval signing key is unavailable or revoked"
            )
        try:
            Ed25519PublicKey.from_public_bytes(
                self._keys[key_id]
            ).verify(signature, message)
        except InvalidSignature as exc:
            raise CapabilityAuthorizationError(
                "owner approval signature is invalid"
            ) from exc

    def save_public(self, path: Path) -> None:
        payload = {
            "algorithm": "Ed25519",
            "keys": {
                key_id: base64.b64encode(value).decode("ascii")
                for key_id, value in sorted(self._keys.items())
            },
            "revoked": sorted(self._revoked),
        }
        _atomic_json(Path(path), payload)

    @classmethod
    def from_file(cls, path: Path) -> "ApprovalKeyRing":
        payload = json.loads(Path(path).read_text(encoding="utf-8"))
        if payload.get("algorithm") != "Ed25519":
            raise ValueError("unsupported approval signature algorithm")
        ring = cls(
            {
                key_id: base64.b64decode(encoded, validate=True)
                for key_id, encoded in payload.get("keys", {}).items()
            }
        )
        for key_id in payload.get("revoked", []):
            ring.revoke(key_id)
        return ring


class ApprovalRevocationStore:
    def __init__(self, root: Path):
        self.root = Path(root)

    def revoke(self, approval_plan_id: str, reason: str) -> dict[str, Any]:
        path = _safe_child(self.root, approval_plan_id)
        payload = {
            "approval_plan_id": approval_plan_id,
            "reason": str(reason or "revoked"),
            "revoked_at": utc_now(),
        }
        _atomic_json(path, payload)
        return payload

    def is_revoked(self, approval_plan_id: str) -> bool:
        return _safe_child(self.root, approval_plan_id).is_file()


class SignedApprovalVerifier:
    def __init__(
        self,
        keyring: ApprovalKeyRing,
        revocations: ApprovalRevocationStore,
        *,
        allowed_trust_domains: Iterable[str] = (),
        maximum_approval_lifetime_seconds: int = 3600,
    ):
        self.keyring = keyring
        self.revocations = revocations
        self.allowed_trust_domains = frozenset(allowed_trust_domains)
        self.maximum_approval_lifetime_seconds = int(
            maximum_approval_lifetime_seconds
        )

    def _verify_signature(
        self, approval: SignedReleasePlanApproval
    ) -> None:
        try:
            signature = base64.b64decode(
                approval.signature, validate=True
            )
        except (ValueError, TypeError) as exc:
            raise CapabilityAuthorizationError(
                "owner approval signature encoding is invalid"
            ) from exc
        self.keyring.verify(
            approval.key_id,
            signature,
            approval.canonical_bytes(),
        )

    def verify(
        self,
        approval: SignedReleasePlanApproval,
        candidate: ReleaseCandidate,
        deployment_plan: DeploymentPlan,
        policy_sha256: str,
    ) -> SignedReleasePlanApproval:
        try:
            if _parse_utc(approval.expires_at) <= datetime.now(timezone.utc):
                raise CapabilityAuthorizationError(
                    "owner approval has expired"
                )
            if self.revocations.is_revoked(approval.approval_plan_id):
                raise CapabilityAuthorizationError(
                    "owner approval has been revoked"
                )
            expected = (
                candidate.candidate_id,
                candidate.source_commit,
                candidate.tree_oid,
                candidate.artifact_sha256,
                candidate.manifest_sha256,
                deployment_plan.sha256,
                _validate_sha("policy_sha256", policy_sha256),
            )
            actual = (
                approval.candidate_id,
                approval.source_commit,
                approval.tree_oid,
                approval.artifact_sha256,
                approval.manifest_sha256,
                approval.deployment_plan_sha256,
                approval.policy_sha256,
            )
            if actual != expected:
                raise CapabilityAuthorizationError(
                    "owner approval binding does not match release evidence"
                )
            self._verify_signature(approval)
            return approval
        except CapabilityAuthorizationError:
            raise
        except Exception as exc:
            raise CapabilityAuthorizationError(
                "owner approval verification failed"
            ) from exc

    def verify_release_plan(
        self,
        approval: SignedReleasePlanApproval,
        candidate: ReleaseCandidate,
        release_plan: ReleasePlan,
        policy_sha256: str,
        *,
        now: str | datetime | None = None,
    ) -> SignedReleasePlanApproval:
        try:
            release_plan.validate()
            policy_sha256 = _validate_sha(
                "policy_sha256", policy_sha256
            )
            instant = (
                datetime.now(timezone.utc)
                if now is None
                else _parse_utc(now)
                if isinstance(now, str)
                else now
            )
            if instant.tzinfo is None:
                raise CapabilityAuthorizationError(
                    "verification time must be timezone-aware"
                )
            issued_at = _parse_utc(approval.issued_at)
            not_before = _parse_utc(approval.not_before)
            expires_at = _parse_utc(approval.expires_at)
            if not issued_at <= not_before <= expires_at:
                raise CapabilityAuthorizationError(
                    "owner approval time window is invalid"
                )
            if not not_before <= instant <= expires_at:
                raise CapabilityAuthorizationError(
                    "owner approval is outside its validity window"
                )
            lifetime = (expires_at - issued_at).total_seconds()
            if (
                lifetime <= 0
                or lifetime > self.maximum_approval_lifetime_seconds
            ):
                raise CapabilityAuthorizationError(
                    "owner approval lifetime exceeds policy"
                )
            if (
                not approval.owner_identity
                or not approval.trust_domain
                or (
                    self.allowed_trust_domains
                    and approval.trust_domain
                    not in self.allowed_trust_domains
                )
            ):
                raise CapabilityAuthorizationError(
                    "owner approval trust binding is invalid"
                )
            if self.revocations.is_revoked(approval.approval_plan_id):
                raise CapabilityAuthorizationError(
                    "owner approval has been revoked"
                )
            expected = (
                candidate.candidate_id,
                candidate.source_commit,
                candidate.tree_oid,
                candidate.artifact_sha256,
                candidate.manifest_sha256,
                release_plan.deployment.sha256,
                policy_sha256,
                release_plan.sha256,
                release_plan.tag.approval_plan_id,
            )
            actual = (
                approval.candidate_id,
                approval.source_commit,
                approval.tree_oid,
                approval.artifact_sha256,
                approval.manifest_sha256,
                approval.deployment_plan_sha256,
                approval.policy_sha256,
                approval.release_plan_sha256,
                approval.approval_plan_id,
            )
            if actual != expected:
                raise CapabilityAuthorizationError(
                    "owner approval does not bind the complete release plan"
                )
            self._verify_signature(approval)
            return approval
        except CapabilityAuthorizationError:
            raise
        except Exception as exc:
            raise CapabilityAuthorizationError(
                "owner release plan verification failed"
            ) from exc


class CapabilityLedger:
    """Durable exclusive claims with leases and fail-closed recovery probes."""

    STATES = frozenset(
        {
            "PENDING",
            "CLAIMED",
            "COMPLETED",
            "FAILED_RECOVERABLE",
            "RECONCILIATION_REQUIRED",
            "MANUAL_OWNER_REVIEW_REQUIRED",
        }
    )

    def __init__(
        self,
        root: Path,
        *,
        maximum_probe_age_seconds: float = 60,
    ):
        self.root = Path(root)
        self.maximum_probe_age_seconds = float(maximum_probe_age_seconds)
        if self.maximum_probe_age_seconds <= 0:
            raise ValueError("maximum_probe_age_seconds must be positive")

    def _operation_path(
        self,
        approval: SignedReleasePlanApproval,
        capability: str,
        operation: str,
    ) -> Path:
        key = hashlib.sha256(
            "\0".join(
                (approval.release_plan_id, capability, operation)
            ).encode("utf-8")
        ).hexdigest()
        return _safe_child(self.root, key)

    def _nonce_path(self, nonce: str) -> Path:
        key = hashlib.sha256(nonce.encode("utf-8")).hexdigest()
        return _safe_child(self.root / "nonces", key)

    def _lock(self, path: Path):
        lock_path = self.root / ".locks" / f"{path.stem}.lock"
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        os.chmod(lock_path.parent, 0o700)
        handle = lock_path.open("a+", encoding="utf-8")
        os.chmod(lock_path, 0o600)
        try:
            import fcntl

            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
        except ImportError:
            pass
        return handle

    @staticmethod
    def _unlock(handle) -> None:
        try:
            import fcntl

            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
        except ImportError:
            pass
        handle.close()

    @staticmethod
    def _assert_authorized(
        approval: SignedReleasePlanApproval,
        capability: str,
        operation: str,
    ) -> None:
        if (
            capability in FORBIDDEN_CAPABILITIES
            or capability not in ALLOWED_CAPABILITIES
            or capability not in approval.capabilities
        ):
            raise CapabilityAuthorizationError(
                "capability is not authorized by owner approval"
            )
        _validate_identifier("operation", operation)

    @staticmethod
    def _immutable_payload(
        approval: SignedReleasePlanApproval,
        capability: str,
        operation: str,
    ) -> dict[str, Any]:
        operation_digest = hashlib.sha256(
            _canonical_json(
                {
                    "approval_fingerprint": approval.fingerprint,
                    "capability": capability,
                    "operation": operation,
                    "release_plan_id": approval.release_plan_id,
                }
            )
        ).hexdigest()
        return {
            "approval_fingerprint": approval.fingerprint,
            "approval_plan_id": approval.approval_plan_id,
            "capability": capability,
            "operation": operation,
            "operation_digest": operation_digest,
            "release_plan_id": approval.release_plan_id,
        }

    @staticmethod
    def _assert_record_matches(
        existing: Mapping[str, Any],
        expected: Mapping[str, Any],
    ) -> None:
        if {key: existing.get(key) for key in expected} != dict(expected):
            raise CapabilityAuthorizationError(
                "capability replay conflicts with durable record"
            )

    def _write(self, path: Path, payload: Mapping[str, Any]) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        os.chmod(path.parent, 0o700)
        _durable_json(path, payload)
        os.chmod(path, 0o600)

    def _probe_journal_path(self, operation_path: Path) -> Path:
        return _safe_child(
            self.root / ".probe-journal",
            operation_path.stem,
            ".events.jsonl",
        )

    def _append_probe_event(
        self,
        operation_path: Path,
        *,
        probe: ProbeEvidence,
    ) -> dict[str, Any]:
        path = self._probe_journal_path(operation_path)
        path.parent.mkdir(parents=True, exist_ok=True)
        os.chmod(path.parent, 0o700)
        records = self.verify_probe_journal(operation_path, missing_ok=True)
        previous_hash = (
            records[-1]["event_hash"] if records else JOURNAL_GENESIS_HASH
        )
        core = {
            "sequence": len(records) + 1,
            "previous_hash": previous_hash,
            "payload": probe.to_dict(),
        }
        record = {
            "schema_version": JOURNAL_SCHEMA_VERSION,
            **core,
            "event_hash": hashlib.sha256(_canonical_json(core)).hexdigest(),
        }
        with path.open("a", encoding="utf-8") as output:
            output.write(
                json.dumps(record, sort_keys=True, separators=(",", ":"))
                + "\n"
            )
            output.flush()
            os.fsync(output.fileno())
        os.chmod(path, 0o600)
        _fsync_directory(path.parent)
        return record

    def verify_probe_journal(
        self,
        operation_path: Path,
        *,
        missing_ok: bool = False,
    ) -> list[dict[str, Any]]:
        path = self._probe_journal_path(operation_path)
        if not path.exists():
            if missing_ok:
                return []
            raise JournalIntegrityError("capability probe journal is missing")
        records: list[dict[str, Any]] = []
        previous = JOURNAL_GENESIS_HASH
        for sequence, line in enumerate(
            path.read_text(encoding="utf-8").splitlines(), 1
        ):
            try:
                record = json.loads(line)
            except json.JSONDecodeError as exc:
                raise JournalIntegrityError(
                    "capability probe journal contains invalid JSON"
                ) from exc
            if set(record) != {
                "schema_version",
                "sequence",
                "previous_hash",
                "payload",
                "event_hash",
            }:
                raise JournalIntegrityError(
                    "capability probe journal schema is invalid"
                )
            core = {
                "sequence": record["sequence"],
                "previous_hash": record["previous_hash"],
                "payload": record["payload"],
            }
            expected_hash = hashlib.sha256(_canonical_json(core)).hexdigest()
            if (
                record["schema_version"] != JOURNAL_SCHEMA_VERSION
                or record["sequence"] != sequence
                or record["previous_hash"] != previous
                or record["event_hash"] != expected_hash
            ):
                raise JournalIntegrityError(
                    "capability probe journal hash chain verification failed"
                )
            previous = record["event_hash"]
            records.append(record)
        return records

    @staticmethod
    def _coerce_probe(
        value: ClaimProbeResult | ProbeEvidence | Mapping[str, Any],
        *,
        operation_digest: str,
        now: float,
    ) -> ProbeEvidence:
        if isinstance(value, ProbeEvidence):
            return value
        if isinstance(value, Mapping):
            return ProbeEvidence(
                result=ClaimProbeResult(value["result"]),
                operation_digest=str(value["operation_digest"]),
                observed_at=float(value["observed_at"]),
                evidence=dict(value.get("evidence") or {}),
            )
        return ProbeEvidence(
            result=ClaimProbeResult(value),
            operation_digest=operation_digest,
            observed_at=now,
            evidence={},
        )

    def _run_probe_locked(
        self,
        path: Path,
        record: dict[str, Any],
        side_effect_probe: Callable[
            [], ClaimProbeResult | ProbeEvidence | Mapping[str, Any]
        ],
        *,
        now: float,
    ) -> tuple[dict[str, Any], ProbeEvidence]:
        probe = self._coerce_probe(
            side_effect_probe(),
            operation_digest=str(record["operation_digest"]),
            now=now,
        ).validate(
            expected_operation_digest=str(record["operation_digest"]),
            now=now,
            maximum_age_seconds=self.maximum_probe_age_seconds,
        )
        event = self._append_probe_event(path, probe=probe)
        record.update(
            {
                "last_probe": probe.to_dict(),
                "probe_journal_event_hash": event["event_hash"],
                "probe_result": probe.result.value,
            }
        )
        return record, probe

    def read(
        self,
        approval: SignedReleasePlanApproval,
        capability: str,
        *,
        operation: str,
    ) -> dict[str, Any] | None:
        path = self._operation_path(approval, capability, operation)
        if not path.is_file():
            return None
        record = json.loads(path.read_text(encoding="utf-8"))
        self._assert_record_matches(
            record,
            self._immutable_payload(approval, capability, operation),
        )
        return _sanitize(record)

    def _bind_nonce(self, approval: SignedReleasePlanApproval) -> None:
        path = self._nonce_path(approval.nonce)
        expected = {
            "fingerprint": approval.fingerprint,
            "release_plan_id": approval.release_plan_id,
        }
        path.parent.mkdir(parents=True, exist_ok=True)
        os.chmod(path.parent, 0o700)
        try:
            descriptor = os.open(
                path,
                os.O_WRONLY | os.O_CREAT | os.O_EXCL,
                0o600,
            )
        except FileExistsError:
            actual = json.loads(path.read_text(encoding="utf-8"))
            if actual != expected:
                raise CapabilityAuthorizationError(
                    "approval nonce replay does not match original plan"
                )
            return
        with os.fdopen(descriptor, "w", encoding="utf-8") as output:
            json.dump(expected, output, sort_keys=True)
            output.write("\n")
            output.flush()
            os.fsync(output.fileno())
        _fsync_directory(path.parent)

    def claim(
        self,
        approval: SignedReleasePlanApproval,
        capability: str,
        *,
        operation: str,
        executor_id: str,
        lease_seconds: float,
        now: float | None = None,
        side_effect_probe: Callable[
            [], ClaimProbeResult | ProbeEvidence | Mapping[str, Any]
        ]
        | None = None,
    ) -> dict[str, Any]:
        self._assert_authorized(approval, capability, operation)
        _validate_identifier("executor_id", executor_id)
        if lease_seconds <= 0:
            raise ValueError("lease_seconds must be positive")
        timestamp = time.time() if now is None else float(now)
        path = self._operation_path(approval, capability, operation)
        path.parent.mkdir(parents=True, exist_ok=True)
        self._bind_nonce(approval)
        immutable = self._immutable_payload(
            approval, capability, operation
        )
        lock = self._lock(path)
        try:
            if path.exists():
                record = json.loads(path.read_text(encoding="utf-8"))
                self._assert_record_matches(record, immutable)
            else:
                record = {
                    **immutable,
                    "attempt": 0,
                    "state": "PENDING",
                }
            state = str(record.get("state") or "PENDING")
            if state == "COMPLETED":
                return {**record, "status": "IDEMPOTENT"}
            if state in {
                "RECONCILIATION_REQUIRED",
                "MANUAL_OWNER_REVIEW_REQUIRED",
            }:
                return {**record, "status": state}
            recovery_required = state == "FAILED_RECOVERABLE"
            if state == "CLAIMED":
                lease_until = float(record.get("lease_until") or 0)
                if lease_until > timestamp:
                    if record.get("executor_id") == executor_id:
                        return {**record, "status": "CLAIMED"}
                    return {**record, "status": "BUSY"}
                recovery_required = True
            if recovery_required:
                if side_effect_probe is None:
                    return {**record, "status": "PROBE_REQUIRED"}
                record, probe = self._run_probe_locked(
                    path,
                    record,
                    side_effect_probe,
                    now=timestamp,
                )
                if probe.result == ClaimProbeResult.APPLIED_VERIFIED:
                    record.update(
                        {
                            "completed_at": utc_now(),
                            "state": "COMPLETED",
                        }
                    )
                    self._write(path, record)
                    return {**record, "status": "IDEMPOTENT"}
                if probe.result == ClaimProbeResult.INDETERMINATE:
                    record.update(
                        {
                            "state": "RECONCILIATION_REQUIRED",
                        }
                    )
                    self._write(path, record)
                    return {**record, "status": record["state"]}
                if probe.result == ClaimProbeResult.CONFLICT:
                    record.update(
                        {"state": "MANUAL_OWNER_REVIEW_REQUIRED"}
                    )
                    self._write(path, record)
                    return {**record, "status": record["state"]}
            claim_id = uuid.uuid4().hex
            claimed = {
                **record,
                **immutable,
                "attempt": int(record.get("attempt") or 0) + 1,
                "claim_id": claim_id,
                "claimed_at": timestamp,
                "executor_id": executor_id,
                "lease_until": timestamp + float(lease_seconds),
                "state": "CLAIMED",
            }
            self._write(path, claimed)
            return {**claimed, "status": "CLAIMED"}
        finally:
            self._unlock(lock)

    def complete(
        self,
        approval: SignedReleasePlanApproval,
        capability: str,
        *,
        operation: str,
        claim_id: str | None = None,
        evidence: Mapping[str, Any] | None = None,
    ) -> dict[str, Any]:
        self._assert_authorized(approval, capability, operation)
        path = self._operation_path(approval, capability, operation)
        immutable = self._immutable_payload(
            approval, capability, operation
        )
        lock = self._lock(path)
        try:
            record = json.loads(path.read_text(encoding="utf-8"))
            self._assert_record_matches(record, immutable)
            if record.get("state") == "COMPLETED":
                return {**record, "status": "IDEMPOTENT"}
            if (
                record.get("state") != "CLAIMED"
                or not claim_id
                or record.get("claim_id") != claim_id
            ):
                raise CapabilityAuthorizationError(
                    "capability completion claim_id mismatch"
                )
            record.update(
                {
                    "completed_at": utc_now(),
                    "evidence": _sanitize(dict(evidence or {})),
                    "state": "COMPLETED",
                }
            )
            self._write(path, record)
            return {**record, "status": "COMPLETED"}
        finally:
            self._unlock(lock)

    def fail_recoverable(
        self,
        approval: SignedReleasePlanApproval,
        capability: str,
        *,
        operation: str,
        claim_id: str,
        reason: str,
    ) -> dict[str, Any]:
        self._assert_authorized(approval, capability, operation)
        path = self._operation_path(approval, capability, operation)
        immutable = self._immutable_payload(
            approval, capability, operation
        )
        lock = self._lock(path)
        try:
            record = json.loads(path.read_text(encoding="utf-8"))
            self._assert_record_matches(record, immutable)
            if (
                record.get("state") != "CLAIMED"
                or record.get("claim_id") != claim_id
            ):
                raise CapabilityAuthorizationError(
                    "capability failure claim_id mismatch"
                )
            record.update(
                {
                    "failure_reason": str(reason),
                    "state": "FAILED_RECOVERABLE",
                }
            )
            self._write(path, record)
            return record
        finally:
            self._unlock(lock)

    def reserve(
        self,
        approval: SignedReleasePlanApproval,
        capability: str,
        *,
        operation: str,
    ) -> dict[str, Any]:
        """Compatibility wrapper; production controllers use claim()."""
        path = self._operation_path(approval, capability, operation)
        if path.exists():
            record = json.loads(path.read_text(encoding="utf-8"))
            if record.get("state") == "COMPLETED":
                return {**record, "status": "IDEMPOTENT"}
            if record.get("state") in {
                "CLAIMED",
                "FAILED_RECOVERABLE",
            }:
                return {**record, "status": "RESUME"}
        result = self.claim(
            approval,
            capability,
            operation=operation,
            executor_id="compatibility-reservation",
            lease_seconds=300,
        )
        aliases = {
            "CLAIMED": "RESERVED",
            "BUSY": "RESUME",
            "IDEMPOTENT": "IDEMPOTENT",
        }
        return {**result, "status": aliases.get(result["status"], result["status"])}

    def consume(
        self,
        approval: SignedReleasePlanApproval,
        capability: str,
        *,
        operation: str,
    ) -> dict[str, Any]:
        reservation = self.reserve(
            approval, capability, operation=operation
        )
        if reservation["status"] == "IDEMPOTENT":
            return reservation
        completed = self.complete(
            approval,
            capability,
            operation=operation,
            claim_id=reservation["claim_id"],
        )
        return {**completed, "status": "CONSUMED"}


class SagaState(str, Enum):
    PREPARING = "PREPARING"
    VERIFYING = "VERIFYING"
    READY_FOR_APPROVAL = "READY_FOR_APPROVAL"
    APPROVED = "APPROVED"
    PRE_SWITCH_REVALIDATING = "PRE_SWITCH_REVALIDATING"
    SWITCH_PENDING = "SWITCH_PENDING"
    SWITCHED = "SWITCHED"
    RUNTIME_VERIFYING = "RUNTIME_VERIFYING"
    RUNTIME_VERIFIED = "RUNTIME_VERIFIED"
    TAG_PENDING = "TAG_PENDING"
    TAG_VERIFIED = "TAG_VERIFIED"
    BASELINE_REGISTRY_PENDING = "BASELINE_REGISTRY_PENDING"
    COMPLETED = "COMPLETED"
    ROLLBACK_PENDING = "ROLLBACK_PENDING"
    ROLLED_BACK = "ROLLED_BACK"
    RECONCILIATION_REQUIRED = "RECONCILIATION_REQUIRED"
    MANUAL_OWNER_REVIEW_REQUIRED = "MANUAL_OWNER_REVIEW_REQUIRED"


@dataclass
class ReleaseSaga:
    candidate_id: str
    release_plan_id: str
    state: SagaState = SagaState.PREPARING
    revision: int = 0
    events: list[dict[str, Any]] = field(default_factory=list)
    journal_sequence: int = 0
    journal_head_hash: str = JOURNAL_GENESIS_HASH
    journal_integrity: bool = True
    created_at: str = field(default_factory=utc_now)
    updated_at: str = field(default_factory=utc_now)

    def to_dict(self) -> dict[str, Any]:
        payload = asdict(self)
        payload["state"] = self.state.value
        return _sanitize(payload)

    @classmethod
    def from_dict(cls, payload: Mapping[str, Any]) -> "ReleaseSaga":
        values = dict(payload)
        values["state"] = SagaState(values["state"])
        return cls(**values)


class ReleaseSagaStore:
    """Hash-chained journal is truth; atomic snapshot is a rebuildable cache."""

    RECOVERY_ACTIONS = {
        SagaState.SWITCH_PENDING: "resume_runtime_switch",
        SagaState.SWITCHED: "verify_runtime",
        SagaState.RUNTIME_VERIFYING: "verify_runtime",
        SagaState.TAG_PENDING: "reconcile_tag",
        SagaState.TAG_VERIFIED: "reconcile_baseline_registry",
        SagaState.BASELINE_REGISTRY_PENDING: "reconcile_baseline_registry",
        SagaState.ROLLBACK_PENDING: "resume_rollback",
        SagaState.RECONCILIATION_REQUIRED: "reconcile_release",
        SagaState.MANUAL_OWNER_REVIEW_REQUIRED: "manual_owner_review",
    }

    def __init__(
        self,
        root: Path,
        *,
        after_journal_hook: Callable[[Mapping[str, Any]], None] | None = None,
    ):
        self.root = Path(root)
        self.after_journal_hook = after_journal_hook

    def _path(self, candidate_id: str) -> Path:
        if not CANDIDATE_ID_RE.fullmatch(str(candidate_id or "")):
            raise ValueError("invalid candidate_id")
        return _safe_child(self.root, candidate_id)

    def _journal(self, candidate_id: str) -> Path:
        if not CANDIDATE_ID_RE.fullmatch(str(candidate_id or "")):
            raise ValueError("invalid candidate_id")
        return _safe_child(self.root, candidate_id, ".events.jsonl")

    def _lock(self, candidate_id: str):
        path = _safe_child(self.root / ".locks", candidate_id, ".lock")
        path.parent.mkdir(parents=True, exist_ok=True)
        handle = path.open("a+", encoding="utf-8")
        try:
            import fcntl

            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
        except ImportError:
            pass
        return handle

    @staticmethod
    def _unlock(handle) -> None:
        try:
            import fcntl

            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
        except ImportError:
            pass
        handle.close()

    def create(
        self, candidate_id: str, release_plan_id: str
    ) -> ReleaseSaga:
        _validate_identifier("release_plan_id", release_plan_id)
        path = self._path(candidate_id)
        path.parent.mkdir(parents=True, exist_ok=True)
        lock = self._lock(candidate_id)
        try:
            if path.exists() or self._journal(candidate_id).exists():
                existing = self._load_locked(candidate_id)
                if existing.release_plan_id != release_plan_id:
                    raise SagaConflict(
                        "candidate is bound to a different release plan"
                    )
                return existing
            saga = ReleaseSaga(
                candidate_id=candidate_id,
                release_plan_id=release_plan_id,
            )
            payload = {
                "candidate_id": candidate_id,
                "details": {},
                "event": "saga_created",
                "release_plan_id": release_plan_id,
                "revision": 0,
                "state": saga.state.value,
                "timestamp": saga.created_at,
            }
            record = self._append_event(
                candidate_id,
                payload,
                sequence=1,
                previous_hash=JOURNAL_GENESIS_HASH,
            )
            saga.events.append(payload)
            saga.journal_sequence = 1
            saga.journal_head_hash = record["event_hash"]
            self._after_journal(record)
            _durable_json(path, saga.to_dict())
            return saga
        finally:
            self._unlock(lock)

    def load(self, candidate_id: str) -> ReleaseSaga:
        lock = self._lock(candidate_id)
        try:
            try:
                return self._load_locked(candidate_id)
            except JournalIntegrityError as exc:
                return self._manual_review_snapshot(candidate_id, str(exc))
        finally:
            self._unlock(lock)

    def _load_locked(self, candidate_id: str) -> ReleaseSaga:
        records = self.verify_journal(candidate_id)
        saga = self._rebuild(records)
        path = self._path(candidate_id)
        cached = None
        if path.exists():
            try:
                cached = ReleaseSaga.from_dict(
                    json.loads(path.read_text(encoding="utf-8"))
                )
            except Exception:
                cached = None
        if cached is not None and (
            cached.journal_sequence > saga.journal_sequence
            or (
                cached.journal_sequence == saga.journal_sequence
                and cached.journal_head_hash != saga.journal_head_hash
            )
        ):
            raise JournalIntegrityError(
                "Saga snapshot references a missing or different journal head"
            )
        if cached is None or cached.to_dict() != saga.to_dict():
            _durable_json(path, saga.to_dict())
        return saga

    def compare_and_set(
        self,
        candidate_id: str,
        *,
        expected_revision: int,
        expected_states: Iterable[SagaState],
        target: SagaState,
        event: str,
        details: Mapping[str, Any] | None = None,
    ) -> ReleaseSaga:
        lock = self._lock(candidate_id)
        try:
            try:
                saga = self._load_locked(candidate_id)
            except JournalIntegrityError as exc:
                return self._manual_review_snapshot(candidate_id, str(exc))
            states = {SagaState(value) for value in expected_states}
            if (
                saga.revision != expected_revision
                or saga.state not in states
            ):
                raise SagaConflict("Saga compare-and-set conflict")
            saga.revision += 1
            saga.state = SagaState(target)
            saga.updated_at = utc_now()
            payload = {
                "candidate_id": saga.candidate_id,
                "details": _sanitize(dict(details or {})),
                "event": str(event),
                "release_plan_id": saga.release_plan_id,
                "revision": saga.revision,
                "state": saga.state.value,
                "timestamp": saga.updated_at,
            }
            record = self._append_event(
                candidate_id,
                payload,
                sequence=saga.journal_sequence + 1,
                previous_hash=saga.journal_head_hash,
            )
            saga.events.append(payload)
            saga.journal_sequence = record["sequence"]
            saga.journal_head_hash = record["event_hash"]
            self._after_journal(record)
            _durable_json(self._path(candidate_id), saga.to_dict())
            return saga
        finally:
            self._unlock(lock)

    def force_state_for_recovery_test(
        self, candidate_id: str, state: SagaState
    ) -> ReleaseSaga:
        saga = self.load(candidate_id)
        return self.compare_and_set(
            candidate_id,
            expected_revision=saga.revision,
            expected_states={saga.state},
            target=state,
            event="test_recovery_checkpoint",
        )

    def bind_release_plan(
        self, candidate_id: str, release_plan_id: str
    ) -> ReleaseSaga:
        _validate_identifier("release_plan_id", release_plan_id)
        lock = self._lock(candidate_id)
        try:
            saga = self._load_locked(candidate_id)
            pending = f"pending-{candidate_id}"
            if saga.release_plan_id not in {pending, release_plan_id}:
                raise SagaConflict(
                    "Saga is already bound to another release plan"
                )
            if saga.release_plan_id == release_plan_id:
                return saga
            saga.release_plan_id = release_plan_id
            saga.revision += 1
            saga.updated_at = utc_now()
            payload = {
                "candidate_id": saga.candidate_id,
                "details": {"release_plan_id": release_plan_id},
                "event": "owner_approval_bound",
                "release_plan_id": release_plan_id,
                "revision": saga.revision,
                "state": saga.state.value,
                "timestamp": saga.updated_at,
            }
            record = self._append_event(
                candidate_id,
                payload,
                sequence=saga.journal_sequence + 1,
                previous_hash=saga.journal_head_hash,
            )
            saga.events.append(payload)
            saga.journal_sequence = record["sequence"]
            saga.journal_head_hash = record["event_hash"]
            self._after_journal(record)
            _durable_json(self._path(candidate_id), saga.to_dict())
            return saga
        finally:
            self._unlock(lock)

    def _append_event(
        self,
        candidate_id: str,
        payload: Mapping[str, Any],
        *,
        sequence: int,
        previous_hash: str,
    ) -> dict[str, Any]:
        path = self._journal(candidate_id)
        path.parent.mkdir(parents=True, exist_ok=True)
        core = {
            "sequence": int(sequence),
            "previous_hash": str(previous_hash),
            "payload": _sanitize(dict(payload)),
        }
        record = {
            "schema_version": JOURNAL_SCHEMA_VERSION,
            **core,
            "event_hash": hashlib.sha256(
                _canonical_json(core)
            ).hexdigest(),
        }
        with path.open("a", encoding="utf-8") as output:
            output.write(
                json.dumps(
                    record,
                    sort_keys=True,
                    separators=(",", ":"),
                )
                + "\n"
            )
            output.flush()
            os.fsync(output.fileno())
        _fsync_directory(path.parent)
        return record

    def _after_journal(self, record: Mapping[str, Any]) -> None:
        if self.after_journal_hook is not None:
            self.after_journal_hook(record)

    def verify_journal(
        self, candidate_id: str
    ) -> list[dict[str, Any]]:
        path = self._journal(candidate_id)
        if not path.is_file():
            raise JournalIntegrityError("Saga journal is missing")
        records: list[dict[str, Any]] = []
        previous = JOURNAL_GENESIS_HASH
        try:
            lines = path.read_text(encoding="utf-8").splitlines()
        except OSError as exc:
            raise JournalIntegrityError("Saga journal is unreadable") from exc
        if not lines:
            raise JournalIntegrityError("Saga journal is empty")
        for expected_sequence, line in enumerate(lines, 1):
            try:
                record = json.loads(line)
            except json.JSONDecodeError as exc:
                raise JournalIntegrityError(
                    "Saga journal contains invalid JSON"
                ) from exc
            if set(record) != {
                "schema_version",
                "sequence",
                "previous_hash",
                "payload",
                "event_hash",
            }:
                raise JournalIntegrityError(
                    "Saga journal record schema is invalid"
                )
            core = {
                "sequence": record["sequence"],
                "previous_hash": record["previous_hash"],
                "payload": record["payload"],
            }
            expected_hash = hashlib.sha256(
                _canonical_json(core)
            ).hexdigest()
            if (
                record["schema_version"] != JOURNAL_SCHEMA_VERSION
                or record["sequence"] != expected_sequence
                or record["previous_hash"] != previous
                or record["event_hash"] != expected_hash
            ):
                raise JournalIntegrityError(
                    "Saga journal hash chain verification failed"
                )
            previous = record["event_hash"]
            records.append(record)
        return records

    @staticmethod
    def _rebuild(records: list[Mapping[str, Any]]) -> ReleaseSaga:
        first = dict(records[0]["payload"])
        saga = ReleaseSaga(
            candidate_id=str(first["candidate_id"]),
            release_plan_id=str(first["release_plan_id"]),
            state=SagaState(first["state"]),
            revision=int(first["revision"]),
            events=[],
            created_at=str(first["timestamp"]),
            updated_at=str(first["timestamp"]),
        )
        for record in records:
            payload = dict(record["payload"])
            if payload.get("candidate_id") != saga.candidate_id:
                raise JournalIntegrityError(
                    "Saga journal candidate identity changed"
                )
            saga.release_plan_id = str(payload["release_plan_id"])
            saga.state = SagaState(payload["state"])
            saga.revision = int(payload["revision"])
            saga.updated_at = str(payload["timestamp"])
            saga.events.append(_sanitize(payload))
            saga.journal_sequence = int(record["sequence"])
            saga.journal_head_hash = str(record["event_hash"])
        return saga

    def _manual_review_snapshot(
        self, candidate_id: str, reason: str
    ) -> ReleaseSaga:
        path = self._path(candidate_id)
        try:
            cached = ReleaseSaga.from_dict(
                json.loads(path.read_text(encoding="utf-8"))
            )
        except Exception:
            cached = ReleaseSaga(
                candidate_id=candidate_id,
                release_plan_id=f"corrupt-{candidate_id}",
            )
        cached.state = SagaState.MANUAL_OWNER_REVIEW_REQUIRED
        cached.journal_integrity = False
        cached.updated_at = utc_now()
        cached.events.append(
            {
                "event": "journal_integrity_failure",
                "reason": str(reason),
                "timestamp": cached.updated_at,
            }
        )
        _durable_json(path, cached.to_dict())
        return cached

    @classmethod
    def recovery_action(cls, state: SagaState) -> str:
        return cls.RECOVERY_ACTIONS.get(
            SagaState(state), "no_automatic_action"
        )


class PreSwitchProbe(Protocol):
    def collect(
        self,
        release_candidate: ReleaseCandidate,
        deployment_plan: DeploymentPlan,
    ) -> Mapping[str, Any]:
        ...


class PreSwitchValidator:
    def __init__(self, probe: PreSwitchProbe):
        self.probe = probe

    def validate(
        self,
        candidate: ReleaseCandidate,
        approval: SignedReleasePlanApproval,
        plan: DeploymentPlan,
        *,
        expected_repository: str,
    ) -> dict[str, Any]:
        expected = {
            "source_commit": candidate.source_commit,
            "tree_oid": candidate.tree_oid,
            "artifact_sha256": candidate.artifact_sha256,
            "manifest_sha256": candidate.manifest_sha256,
            "deployment_plan_sha256": approval.deployment_plan_sha256,
            "policy_sha256": approval.policy_sha256,
            "rollback_ready": True,
            "git_refs_consistent": True,
            "repository_identity": expected_repository,
            "writer_disabled": True,
        }
        plan_binding = (
            approval.candidate_id == candidate.candidate_id
            and approval.source_commit == candidate.source_commit
            and approval.tree_oid == candidate.tree_oid
            and approval.artifact_sha256 == candidate.artifact_sha256
            and approval.manifest_sha256 == candidate.manifest_sha256
            and approval.deployment_plan_sha256 == plan.sha256
            and plan.source_commit == candidate.source_commit
            and plan.artifact_sha256 == candidate.artifact_sha256
            and plan.manifest_sha256 == candidate.manifest_sha256
            and plan.policy_sha256 == approval.policy_sha256
        )
        if not plan_binding:
            raise StaleApprovalError(
                "APPROVAL_STALE_REAPPROVAL_REQUIRED: "
                "deployment plan binding changed"
            )
        actual = dict(self.probe.collect(candidate, plan))
        failed = [
            name
            for name, expected_value in expected.items()
            if actual.get(name) != expected_value
        ]
        if failed:
            raise StaleApprovalError(
                "APPROVAL_STALE_REAPPROVAL_REQUIRED: "
                + ", ".join(sorted(failed))
            )
        return {
            "valid": True,
            "checks": {name: True for name in expected},
            "validated_at": utc_now(),
        }


class CapabilityController:
    """One narrow controller consumes one capability and runs one adapter."""

    def __init__(
        self,
        *,
        capability: str,
        ledger: CapabilityLedger,
        adapter: Callable[..., Mapping[str, Any]],
        executor_id: str = "silent-release-controller",
        lease_seconds: float = 300,
        recovery_probe: Callable[[], ClaimProbeResult] | None = None,
    ):
        if capability not in ALLOWED_CAPABILITIES:
            raise ValueError("invalid release capability")
        self.capability = capability
        self.ledger = ledger
        self.adapter = adapter
        self.executor_id = executor_id
        self.lease_seconds = lease_seconds
        self.recovery_probe = recovery_probe

    def execute(
        self,
        approval: SignedReleasePlanApproval,
        *,
        operation: str,
        **kwargs: Any,
    ) -> dict[str, Any]:
        recovery_probe = kwargs.pop("_recovery_probe", self.recovery_probe)
        stop_after_not_applied_probe = bool(
            kwargs.pop("_stop_after_not_applied_probe", False)
        )
        claim = self.ledger.claim(
            approval,
            self.capability,
            operation=operation,
            executor_id=self.executor_id,
            lease_seconds=self.lease_seconds,
            side_effect_probe=recovery_probe,
        )
        if (
            stop_after_not_applied_probe
            and claim["status"] == "CLAIMED"
            and claim.get("probe_result")
            == ClaimProbeResult.NOT_APPLIED.value
        ):
            self.ledger.fail_recoverable(
                approval,
                self.capability,
                operation=operation,
                claim_id=claim["claim_id"],
                reason="verified_not_applied",
            )
            return {
                "capability": self.capability,
                "probe_result": ClaimProbeResult.NOT_APPLIED.value,
                "status": "NOT_APPLIED",
            }
        if claim["status"] == "IDEMPOTENT":
            return {
                "capability": self.capability,
                "status": "IDEMPOTENT",
            }
        if claim["status"] == "BUSY":
            return {
                "capability": self.capability,
                "claim_id": claim.get("claim_id"),
                "status": "BUSY",
            }
        if claim["status"] in {
            "PROBE_REQUIRED",
            "RECONCILIATION_REQUIRED",
            "MANUAL_OWNER_REVIEW_REQUIRED",
        }:
            return {
                "capability": self.capability,
                "probe_result": claim.get("probe_result"),
                "status": claim["status"],
            }
        if claim["status"] != "CLAIMED":
            raise CapabilityAuthorizationError(
                f"capability claim is not executable: {claim['status']}"
            )
        try:
            result = dict(self.adapter(**kwargs))
            self.validate_result(approval, result)
            self.ledger.complete(
                approval,
                self.capability,
                operation=operation,
                claim_id=claim["claim_id"],
                evidence={
                    "result_sha256": hashlib.sha256(
                        _canonical_json(_sanitize(result))
                    ).hexdigest()
                },
            )
        except Exception as exc:
            self.ledger.fail_recoverable(
                approval,
                self.capability,
                operation=operation,
                claim_id=claim["claim_id"],
                reason=type(exc).__name__,
            )
            raise
        return {
            "capability": self.capability,
            "result": _sanitize(result),
            "status": "EXECUTED",
        }

    def validate_result(
        self,
        approval: SignedReleasePlanApproval,
        result: Mapping[str, Any],
    ) -> None:
        if result.get("verified") is not True:
            raise CapabilityAuthorizationError(
                "capability result is not verified"
            )


class RuntimeSwitchController(CapabilityController):
    def __init__(
        self,
        ledger: CapabilityLedger,
        adapter: Callable[..., Any],
        *,
        recovery_probe: Callable[
            [], ClaimProbeResult | ProbeEvidence | Mapping[str, Any]
        ]
        | None = None,
        executor_id: str = "silent-release-runtime-switch",
        lease_seconds: float = 300,
    ):
        super().__init__(
            capability="runtime_switch",
            ledger=ledger,
            adapter=adapter,
            recovery_probe=recovery_probe,
            executor_id=executor_id,
            lease_seconds=lease_seconds,
        )


class HardenedTagController(CapabilityController):
    """Tag adapter must independently verify Git, Gitea API, and listing."""

    REQUIRED_PROOF = frozenset(
        {
            "evidence_closure_oid",
            "approval_plan_id",
            "repository_identity",
            "remote_tag_oid",
            "remote_peeled_target_oid",
            "gitea_api_tag_oid",
            "gitea_api_peeled_target_oid",
            "gitea_listing_visible",
        }
    )

    def __init__(
        self,
        ledger: CapabilityLedger,
        adapter: Callable[..., Any],
        *,
        expected_repository: str = "",
        publication_probe: Callable[
            [TagPlan], TagPublicationProbeEvidence
        ]
        | None = None,
        executor_id: str = "silent-release-tag-publish",
        lease_seconds: float = 300,
    ):
        self.expected_repository = expected_repository
        self.publication_probe = publication_probe
        super().__init__(
            capability="tag_publish",
            ledger=ledger,
            adapter=adapter,
            executor_id=executor_id,
            lease_seconds=lease_seconds,
        )

    def execute(
        self,
        approval: SignedReleasePlanApproval,
        *,
        operation: str,
        tag_plan: TagPlan | None = None,
        **kwargs: Any,
    ) -> dict[str, Any]:
        recovery_probe = None
        if self.publication_probe is not None and tag_plan is not None:
            immutable = self.ledger._immutable_payload(
                approval, self.capability, operation
            )

            def recovery_probe() -> ProbeEvidence:
                evidence = self.publication_probe(tag_plan)
                return ProbeEvidence(
                    result=classify_tag_publication(evidence, tag_plan),
                    operation_digest=immutable["operation_digest"],
                    observed_at=evidence.observed_at,
                    evidence=evidence.to_dict(),
                )

        adapter_kwargs = dict(kwargs)
        if tag_plan is not None:
            adapter_kwargs["tag_plan"] = tag_plan
        result = super().execute(
            approval,
            operation=operation,
            _recovery_probe=recovery_probe,
            _stop_after_not_applied_probe=True,
            **adapter_kwargs,
        )
        if (
            result["status"] == "EXECUTED"
            and result.get("result", {}).get("classification")
            in {member.value for member in ClaimProbeResult}
        ):
            classification = ClaimProbeResult(
                result["result"]["classification"]
            )
            if classification == ClaimProbeResult.CONFLICT:
                raise TagPublicationConflict(
                    "tag publication conflicts with the signed tag plan"
                )
            if classification == ClaimProbeResult.INDETERMINATE:
                return {
                    "capability": self.capability,
                    "probe_result": classification.value,
                    "status": "RECONCILIATION_REQUIRED",
                }
        return result

    def validate_result(self, approval, proof):
        if (
            not self.REQUIRED_PROOF.issubset(proof)
            or proof.get("evidence_closure_oid")
            != approval.evidence_closure_oid
            or proof.get("approval_plan_id")
            != approval.approval_plan_id
            or proof.get("gitea_listing_visible") is not True
            or proof.get("remote_peeled_target_oid")
            != approval.source_commit
            or proof.get("gitea_api_peeled_target_oid")
            != approval.source_commit
            or proof.get("gitea_api_tag_oid")
            != proof.get("remote_tag_oid")
            or (
                self.expected_repository
                and proof.get("repository_identity")
                != self.expected_repository
            )
        ):
            raise CapabilityAuthorizationError(
                "tag publication proof is incomplete"
            )


class BaselineRegistryFinalizeController(CapabilityController):
    def __init__(
        self,
        ledger: CapabilityLedger,
        adapter: Callable[..., Any],
        *,
        recovery_probe: Callable[
            [], ClaimProbeResult | ProbeEvidence | Mapping[str, Any]
        ]
        | None = None,
        executor_id: str = "silent-release-finalization",
        lease_seconds: float = 300,
    ):
        super().__init__(
            capability="baseline_registry_finalize",
            ledger=ledger,
            adapter=adapter,
            recovery_probe=recovery_probe,
            executor_id=executor_id,
            lease_seconds=lease_seconds,
        )

    def validate_result(self, approval, result):
        required = {
            "verified": True,
            "tag_verified": True,
            "remote_peeled_target_oid": approval.source_commit,
            "gitea_api_peeled_target_oid": approval.source_commit,
            "gitea_listing_visible": True,
            "evidence_closure_oid": approval.evidence_closure_oid,
            "approval_plan_id": approval.approval_plan_id,
        }
        if any(result.get(key) != value for key, value in required.items()):
            raise CapabilityAuthorizationError(
                "finalization is blocked until exact tag proof is verified"
            )


class BaselineController(BaselineRegistryFinalizeController):
    """Narrow Baseline half of the governed finalization capability."""

    def validate_result(self, approval, result):
        super().validate_result(approval, result)
        required = {
            "verified": True,
            "baseline_target_oid": approval.source_commit,
            "evidence_closure_oid": approval.evidence_closure_oid,
            "approval_plan_id": approval.approval_plan_id,
        }
        if any(result.get(key) != value for key, value in required.items()):
            raise CapabilityAuthorizationError(
                "Baseline finalization proof is inconsistent"
            )


class RegistryController(BaselineRegistryFinalizeController):
    """Narrow Registry half of the governed finalization capability."""

    def validate_result(self, approval, result):
        super().validate_result(approval, result)
        required = {
            "verified": True,
            "registry_target_oid": approval.source_commit,
            "evidence_closure_oid": approval.evidence_closure_oid,
            "approval_plan_id": approval.approval_plan_id,
        }
        if any(result.get(key) != value for key, value in required.items()):
            raise CapabilityAuthorizationError(
                "Registry finalization proof is inconsistent"
            )


class GovernedFinalizationCoordinator:
    """Registry is verified first; Baseline is the final mutation."""

    def __init__(
        self,
        registry: RegistryController,
        baseline: BaselineController,
    ):
        self.registry = registry
        self.baseline = baseline

    def execute(
        self,
        approval: SignedReleasePlanApproval,
        *,
        operation: str,
        finalization_plan: FinalizationPlan,
    ) -> dict[str, Any]:
        registry = self.registry.execute(
            approval,
            operation=f"{operation}:registry",
            finalization_plan=finalization_plan,
        )
        if registry["status"] in {
            "BUSY",
            "PROBE_REQUIRED",
            "RECONCILIATION_REQUIRED",
            "MANUAL_OWNER_REVIEW_REQUIRED",
        }:
            return registry
        baseline = self.baseline.execute(
            approval,
            operation=f"{operation}:baseline",
            finalization_plan=finalization_plan,
        )
        if baseline["status"] in {
            "BUSY",
            "PROBE_REQUIRED",
            "RECONCILIATION_REQUIRED",
            "MANUAL_OWNER_REVIEW_REQUIRED",
        }:
            return baseline
        return {
            "baseline_status": baseline["status"],
            "registry_status": registry["status"],
            "result": baseline.get("result") or {},
            "status": "EXECUTED",
        }


class RecoverableReleaseError(RuntimeError):
    """A side effect is unknown or temporarily unverifiable."""


class TagPublicationConflict(RuntimeError):
    """An existing tag disagrees with the signed immutable tag plan."""


class ReleaseOrchestrator:
    """Durable one-approval execution with Baseline/Registry finalized last."""

    def __init__(
        self,
        *,
        candidates: ReleaseCandidateStore,
        plans: ReleasePlanStore,
        approvals: SignedApprovalStore,
        sagas: ReleaseSagaStore,
        approval_verifier: SignedApprovalVerifier,
        trusted_policy: TrustedReleasePolicy,
        actual_plan_factory: Callable[[ReleaseCandidate], ReleasePlan],
        pre_switch_validator: PreSwitchValidator,
        runtime_switch: RuntimeSwitchController,
        runtime_verify: Callable[..., Mapping[str, Any]],
        runtime_rollback: Callable[..., Mapping[str, Any]],
        tag_publish: HardenedTagController,
        finalizer: GovernedFinalizationCoordinator,
    ):
        self.candidates = candidates
        self.plans = plans
        self.approvals = approvals
        self.sagas = sagas
        self.approval_verifier = approval_verifier
        self.trusted_policy = trusted_policy
        self.actual_plan_factory = actual_plan_factory
        self.pre_switch_validator = pre_switch_validator
        self.runtime_switch = runtime_switch
        self.runtime_verify = runtime_verify
        self.runtime_rollback = runtime_rollback
        self.tag_publish = tag_publish
        self.finalizer = finalizer

    def _transition(
        self,
        saga: ReleaseSaga,
        target: SagaState,
        event: str,
        details: Mapping[str, Any] | None = None,
    ) -> ReleaseSaga:
        return self.sagas.compare_and_set(
            saga.candidate_id,
            expected_revision=saga.revision,
            expected_states={saga.state},
            target=target,
            event=event,
            details=details,
        )

    def _load_and_reverify(
        self, candidate_id: str
    ) -> tuple[
        ReleaseCandidate,
        ReleasePlan,
        SignedReleasePlanApproval,
        ReleaseSaga,
    ]:
        candidate = self.candidates.load(candidate_id)
        plan = self.plans.load(candidate_id)
        approval = self.approvals.load(candidate_id)
        saga = self.sagas.load(candidate_id)
        if not saga.journal_integrity:
            raise JournalIntegrityError(
                "Saga journal requires manual owner review"
            )
        actual_plan = self.actual_plan_factory(candidate)
        self.trusted_policy.validate_deployment(actual_plan.deployment)
        if actual_plan.sha256 != plan.sha256:
            raise StaleApprovalError(
                "actual deployment configuration changed after approval"
            )
        self.approval_verifier.verify_release_plan(
            approval,
            candidate,
            plan,
            plan.policy_sha256,
        )
        return candidate, plan, approval, saga

    @staticmethod
    def _runtime_proof_matches(
        proof: Mapping[str, Any],
        candidate: ReleaseCandidate,
        plan: ReleasePlan,
    ) -> bool:
        expected = {
            "verified": True,
            "source_commit": candidate.source_commit,
            "tree_oid": candidate.tree_oid,
            "artifact_sha256": candidate.artifact_sha256,
            "manifest_sha256": candidate.manifest_sha256,
            "repository_identity": plan.deployment.repository_identity,
            "authoritative_git_dir": (
                plan.deployment.authoritative_git_dir
            ),
            "writer_disabled": True,
        }
        return all(proof.get(key) == value for key, value in expected.items())

    @staticmethod
    def _finalization_proof_matches(
        proof: Mapping[str, Any], plan: ReleasePlan
    ) -> bool:
        expected = {
            "verified": True,
            "remote_tag_oid": plan.tag.expected_tag_oid,
            "remote_peeled_target_oid": plan.tag.expected_target_oid,
            "registry_record": (
                plan.finalization.registry_expected_record
            ),
            "baseline_record": (
                plan.finalization.baseline_expected_record
            ),
            "baseline_written_last": True,
        }
        return all(proof.get(key) == value for key, value in expected.items())

    def _rollback(
        self,
        saga: ReleaseSaga,
        candidate: ReleaseCandidate,
        plan: ReleasePlan,
        reason: str,
    ) -> dict[str, Any]:
        if saga.state != SagaState.ROLLBACK_PENDING:
            saga = self._transition(
                saga,
                SagaState.ROLLBACK_PENDING,
                "runtime_rollback_pending",
                {"reason": reason},
            )
        proof = dict(
            self.runtime_rollback(candidate=candidate, plan=plan)
        )
        if proof.get("verified") is not True:
            return {
                "candidate_id": candidate.candidate_id,
                "state": saga.state.value,
                "status": "ROLLBACK_PENDING",
            }
        saga = self._transition(
            saga,
            SagaState.ROLLED_BACK,
            "runtime_rollback_verified",
            {"proof": proof},
        )
        return {
            "candidate_id": candidate.candidate_id,
            "state": saga.state.value,
            "status": "ROLLED_BACK",
        }

    def _manual(
        self, saga: ReleaseSaga, reason: str
    ) -> dict[str, Any]:
        if saga.state != SagaState.MANUAL_OWNER_REVIEW_REQUIRED:
            saga = self._transition(
                saga,
                SagaState.MANUAL_OWNER_REVIEW_REQUIRED,
                "manual_owner_review_required",
                {"reason": reason},
            )
        return {
            "candidate_id": saga.candidate_id,
            "state": saga.state.value,
            "status": saga.state.value,
        }

    def _reconciliation(
        self, saga: ReleaseSaga, reason: str
    ) -> dict[str, Any]:
        if saga.state != SagaState.RECONCILIATION_REQUIRED:
            saga = self._transition(
                saga,
                SagaState.RECONCILIATION_REQUIRED,
                "release_reconciliation_required",
                {"reason": reason},
            )
        return {
            "candidate_id": saga.candidate_id,
            "state": saga.state.value,
            "status": saga.state.value,
        }

    def execute(self, candidate_id: str) -> dict[str, Any]:
        candidate, plan, approval, saga = self._load_and_reverify(
            candidate_id
        )
        if saga.state == SagaState.MANUAL_OWNER_REVIEW_REQUIRED:
            return self._manual(saga, "existing manual review gate")
        if saga.state in {SagaState.COMPLETED, SagaState.ROLLED_BACK}:
            return {
                "candidate_id": candidate_id,
                "state": saga.state.value,
                "status": "IDEMPOTENT",
            }
        if saga.state == SagaState.ROLLBACK_PENDING:
            return self._rollback(
                saga, candidate, plan, "resuming rollback"
            )
        if saga.state == SagaState.RECONCILIATION_REQUIRED:
            return self._reconciliation(
                saga, "automatic reconciliation is forbidden"
            )
        runtime_has_switched = saga.state in {
            SagaState.SWITCH_PENDING,
            SagaState.SWITCHED,
            SagaState.RUNTIME_VERIFYING,
            SagaState.RUNTIME_VERIFIED,
            SagaState.TAG_PENDING,
            SagaState.TAG_VERIFIED,
            SagaState.BASELINE_REGISTRY_PENDING,
        }
        try:
            if saga.state in {
                SagaState.APPROVED,
                SagaState.PRE_SWITCH_REVALIDATING,
            }:
                if saga.state == SagaState.APPROVED:
                    saga = self._transition(
                        saga,
                        SagaState.PRE_SWITCH_REVALIDATING,
                        "pre_switch_revalidation_started",
                    )
                self.pre_switch_validator.validate(
                    candidate,
                    approval,
                    plan.deployment,
                    expected_repository=(
                        plan.deployment.repository_identity
                    ),
                )
                saga = self._transition(
                    saga,
                    SagaState.SWITCH_PENDING,
                    "runtime_switch_pending",
                )
            if saga.state == SagaState.SWITCH_PENDING:
                result = self.runtime_switch.execute(
                    approval,
                    operation=f"runtime:{plan.sha256}",
                    candidate=candidate,
                    plan=plan,
                )
                if result["status"] == "BUSY":
                    return {
                        "candidate_id": candidate_id,
                        "state": saga.state.value,
                        "status": "BUSY",
                    }
                if result["status"] == "RECONCILIATION_REQUIRED":
                    return self._reconciliation(
                        saga, "runtime switch probe is indeterminate"
                    )
                if result["status"] == "MANUAL_OWNER_REVIEW_REQUIRED":
                    return self._manual(
                        saga, "runtime switch probe found a conflict"
                    )
                if result["status"] == "PROBE_REQUIRED":
                    return self._reconciliation(
                        saga, "runtime switch probe is required"
                    )
                runtime_has_switched = True
                saga = self._transition(
                    saga,
                    SagaState.SWITCHED,
                    "runtime_switch_completed",
                    {"capability_status": result["status"]},
                )
            if saga.state in {
                SagaState.SWITCHED,
                SagaState.RUNTIME_VERIFYING,
            }:
                if saga.state == SagaState.SWITCHED:
                    saga = self._transition(
                        saga,
                        SagaState.RUNTIME_VERIFYING,
                        "runtime_verification_started",
                    )
                proof = dict(
                    self.runtime_verify(candidate=candidate, plan=plan)
                )
                if not self._runtime_proof_matches(
                    proof, candidate, plan
                ):
                    raise RecoverableReleaseError(
                        "runtime verification failed"
                    )
                saga = self._transition(
                    saga,
                    SagaState.RUNTIME_VERIFIED,
                    "runtime_verification_completed",
                    {"proof": proof},
                )
            if saga.state == SagaState.RUNTIME_VERIFIED:
                saga = self._transition(
                    saga,
                    SagaState.TAG_PENDING,
                    "tag_publication_pending",
                )
            if saga.state == SagaState.TAG_PENDING:
                result = self.tag_publish.execute(
                    approval,
                    operation=f"tag:{plan.tag.sha256}",
                    tag_plan=plan.tag,
                )
                if result["status"] == "BUSY":
                    return {
                        "candidate_id": candidate_id,
                        "state": saga.state.value,
                        "status": "BUSY",
                    }
                if result["status"] == "RECONCILIATION_REQUIRED":
                    return self._reconciliation(
                        saga, "tag publication is indeterminate"
                    )
                if result["status"] == "MANUAL_OWNER_REVIEW_REQUIRED":
                    return self._manual(
                        saga, "tag publication conflicts with signed plan"
                    )
                if result["status"] == "PROBE_REQUIRED":
                    return self._reconciliation(
                        saga, "tag publication probe is required"
                    )
                if result["status"] == "NOT_APPLIED":
                    return self._rollback(
                        saga,
                        candidate,
                        plan,
                        "tag publication was verified absent",
                    )
                saga = self._transition(
                    saga,
                    SagaState.TAG_VERIFIED,
                    "tag_publication_verified",
                    {"capability_status": result["status"]},
                )
            if saga.state == SagaState.TAG_VERIFIED:
                saga = self._transition(
                    saga,
                    SagaState.BASELINE_REGISTRY_PENDING,
                    "baseline_registry_finalization_pending",
                )
            if saga.state == SagaState.BASELINE_REGISTRY_PENDING:
                result = self.finalizer.execute(
                    approval,
                    operation=f"finalize:{plan.finalization.sha256}",
                    finalization_plan=plan.finalization,
                )
                if result["status"] == "BUSY":
                    return {
                        "candidate_id": candidate_id,
                        "state": saga.state.value,
                        "status": "BUSY",
                    }
                if result["status"] == "RECONCILIATION_REQUIRED":
                    return self._reconciliation(
                        saga, "finalization probe is indeterminate"
                    )
                if result["status"] == "MANUAL_OWNER_REVIEW_REQUIRED":
                    return self._manual(
                        saga, "finalization probe found a conflict"
                    )
                if result["status"] == "PROBE_REQUIRED":
                    return self._reconciliation(
                        saga, "finalization probe is required"
                    )
                proof = result.get("result") or {}
                if not self._finalization_proof_matches(proof, plan):
                    raise RecoverableReleaseError(
                        "final consistency verification failed"
                    )
                saga = self._transition(
                    saga,
                    SagaState.COMPLETED,
                    "release_consistency_verified",
                    {"proof": proof},
                )
            return {
                "candidate_id": candidate_id,
                "state": saga.state.value,
                "status": "VERIFIED",
            }
        except TagPublicationConflict as exc:
            return self._manual(saga, str(exc))
        except Exception as exc:
            if saga.state in {
                SagaState.TAG_PENDING,
                SagaState.TAG_VERIFIED,
                SagaState.BASELINE_REGISTRY_PENDING,
            }:
                return self._reconciliation(saga, type(exc).__name__)
            if runtime_has_switched:
                return self._rollback(
                    saga, candidate, plan, type(exc).__name__
                )
            raise


class ReleaseRecoveryWorker:
    def __init__(
        self,
        saga_store: ReleaseSagaStore,
        orchestrator: ReleaseOrchestrator,
        *,
        shadow_mode: bool,
    ):
        self.saga_store = saga_store
        self.orchestrator = orchestrator
        self.shadow_mode = bool(shadow_mode)

    def run_once(self) -> dict[str, Any]:
        reports: list[dict[str, Any]] = []
        root = self.saga_store.root
        for journal in sorted(root.glob("*.events.jsonl")):
            candidate_id = journal.name[: -len(".events.jsonl")]
            saga = self.saga_store.load(candidate_id)
            if saga.state == SagaState.MANUAL_OWNER_REVIEW_REQUIRED:
                reports.append(
                    {
                        "candidate_id": candidate_id,
                        "status": "SKIPPED_MANUAL_REVIEW",
                    }
                )
            elif self.shadow_mode:
                reports.append(
                    {
                        "candidate_id": candidate_id,
                        "state": saga.state.value,
                        "status": "SHADOW_REPORTED",
                    }
                )
            elif saga.state not in {
                SagaState.COMPLETED,
                SagaState.ROLLED_BACK,
            }:
                reports.append(self.orchestrator.execute(candidate_id))
        return {
            "shadow_mode": self.shadow_mode,
            "reports": reports,
        }


class SilentReleaseActionService:
    """Durable control-plane facade; execution is shadow-disabled by default."""

    def __init__(
        self,
        root: Path,
        *,
        execution_enabled: bool = False,
        runtime_executor: Any = None,
        approval_verifier: SignedApprovalVerifier | None = None,
        pre_switch_validator: PreSwitchValidator | None = None,
        candidate_builder: TrustedCandidateBuilder | None = None,
        builder_attestation_verifier: BuilderAttestationVerifier | None = None,
        trusted_policy: TrustedReleasePolicy | None = None,
        orchestrator: ReleaseOrchestrator | None = None,
        allow_legacy_candidate_input: bool = False,
        expected_repository: str = "",
        policy_sha256: str = "",
    ):
        self.root = Path(root)
        self.execution_enabled = bool(execution_enabled)
        self.runtime_executor = runtime_executor
        self.approval_verifier = approval_verifier
        self.pre_switch_validator = pre_switch_validator
        self.candidate_builder = candidate_builder
        self.builder_attestation_verifier = builder_attestation_verifier
        self.trusted_policy = trusted_policy
        self.orchestrator = orchestrator
        self.allow_legacy_candidate_input = bool(
            allow_legacy_candidate_input
        )
        self.expected_repository = expected_repository
        self.policy_sha256 = policy_sha256
        self.candidates = ReleaseCandidateStore(
            self.root / "candidates"
        )
        self.plans = DeploymentPlanStore(self.root / "plans")
        self.release_plans = ReleasePlanStore(
            self.root / "release-plans"
        )
        self.sagas = ReleaseSagaStore(self.root / "sagas")
        self.approvals_root = self.root / "approvals"
        self.approvals = SignedApprovalStore(self.approvals_root)
        self.intents_root = self.root / "intents"
        self.acceptance_root = self.root / "acceptance"
        self.notifications_root = self.root / "notifications"
        self.capabilities_root = self.root / "capabilities"
        if self.execution_enabled:
            missing = [
                name
                for name, component in (
                    ("approval_verifier", approval_verifier),
                    ("candidate_builder", candidate_builder),
                    (
                        "builder_attestation_verifier",
                        builder_attestation_verifier,
                    ),
                    ("orchestrator", orchestrator),
                    ("pre_switch_validator", pre_switch_validator),
                    ("trusted_policy", trusted_policy),
                )
                if component is None
            ]
            if missing:
                raise CapabilityAuthorizationError(
                    "production execution components are missing: "
                    + ", ".join(missing)
                )

    def create_release_intent(
        self, payload: Mapping[str, Any]
    ) -> dict[str, Any]:
        allowed = {"event_id", "source_commit"}
        unknown = set(payload).difference(allowed)
        if unknown:
            raise CapabilityAuthorizationError(
                "release intent contains caller-supplied evidence"
            )
        event_id = _validate_identifier(
            "event_id", str(payload.get("event_id") or "")
        )
        source_commit = _validate_oid(
            "source_commit", str(payload.get("source_commit") or "")
        )
        intent_id = "intent-" + hashlib.sha256(
            f"{event_id}\0{source_commit}".encode("utf-8")
        ).hexdigest()[:24]
        path = _safe_child(self.intents_root, intent_id)
        record = {
            "created_at": utc_now(),
            "event_id": event_id,
            "intent_id": intent_id,
            "source_commit": source_commit,
            "status": (
                "BUILDING"
                if self.candidate_builder is not None
                else "BUILDER_UNAVAILABLE"
            ),
        }
        if path.exists():
            existing = json.loads(path.read_text(encoding="utf-8"))
            if (
                existing.get("event_id") != event_id
                or existing.get("source_commit") != source_commit
            ):
                raise SagaConflict(
                    "release intent id is bound to different evidence"
                )
            return {
                **_sanitize(existing),
                "idempotent": True,
                "execution_enabled": self.execution_enabled,
                "shadow_mode": not self.execution_enabled,
                "side_effects_executed": False,
            }
        _durable_json(path, record)
        result = dict(record)
        if self.candidate_builder is not None:
            candidate = self.create_release_candidate(
                {
                    "event_id": event_id,
                    "intent_id": intent_id,
                    "source_commit": source_commit,
                }
            )
            result.update(
                {
                    "candidate_id": candidate["candidate_id"],
                    "status": candidate["saga_state"],
                }
            )
            _durable_json(path, result)
        return {
            **_sanitize(result),
            "idempotent": False,
            "execution_enabled": self.execution_enabled,
            "shadow_mode": not self.execution_enabled,
            "side_effects_executed": False,
        }

    @classmethod
    def shadow(
        cls, root: Path, *, runtime_executor: Any = None
    ) -> "SilentReleaseActionService":
        return cls(
            root,
            execution_enabled=False,
            runtime_executor=runtime_executor,
            allow_legacy_candidate_input=True,
        )

    def create_release_candidate(
        self, payload: Mapping[str, Any]
    ) -> dict[str, Any]:
        if self.candidate_builder is not None:
            forbidden_evidence = {
                "artifact_sha256",
                "canary",
                "deployment_plan",
                "git_consistency",
                "gitea_consistency",
                "manifest_sha256",
                "release_plan",
                "rollback_ready",
                "tests",
                "tree_oid",
            }
            if forbidden_evidence.intersection(payload):
                raise CapabilityAuthorizationError(
                    "candidate evidence must come from the trusted builder"
                )
            bundle = self.candidate_builder.build(dict(payload)).validate(
                self.builder_attestation_verifier
            )
            if self.trusted_policy is None:
                raise CapabilityAuthorizationError(
                    "trusted release policy is not configured"
                )
            self.trusted_policy.validate_deployment(
                bundle.release_plan.deployment
            )
            if (
                bundle.release_plan.policy_sha256
                != self.policy_sha256
            ):
                raise CapabilityAuthorizationError(
                    "trusted builder used a different release policy"
                )
            candidate = bundle.candidate
            try:
                existing = self.candidates.load(
                    candidate.candidate_id
                )
            except FileNotFoundError:
                self.candidates.save(candidate)
                existing = candidate
            else:
                if existing.fingerprint != candidate.fingerprint:
                    raise SagaConflict(
                        "candidate id is already bound to other evidence"
                    )
            self.release_plans.save(
                existing.candidate_id, bundle.release_plan
            )
            _durable_json(
                _safe_child(self.acceptance_root, existing.candidate_id),
                bundle.acceptance_record(),
            )
            saga = self.sagas.create(
                existing.candidate_id,
                f"pending-{existing.candidate_id}",
            )
            if saga.state == SagaState.PREPARING:
                saga = self.sagas.compare_and_set(
                    existing.candidate_id,
                    expected_revision=saga.revision,
                    expected_states={SagaState.PREPARING},
                    target=SagaState.READY_FOR_APPROVAL,
                    event="trusted_candidate_ready_for_approval",
                    details={
                        "release_plan_sha256": (
                            bundle.release_plan.sha256
                        )
                    },
                )
            return {
                **existing.to_dict(),
                "release_plan_sha256": bundle.release_plan.sha256,
                "saga_state": saga.state.value,
                "builder_attestation_id": (
                    bundle.attestation.builder_attestation_id
                    if bundle.attestation
                    else None
                ),
                "execution_enabled": self.execution_enabled,
                "shadow_mode": not self.execution_enabled,
                "side_effects_executed": False,
            }
        if not self.allow_legacy_candidate_input:
            raise CapabilityAuthorizationError(
                "trusted candidate builder is not configured"
            )
        values = dict(payload)
        raw_plan = values.pop("deployment_plan", None)
        candidate = ReleaseCandidate.from_dict(values)
        try:
            existing = self.candidates.load(candidate.candidate_id)
        except FileNotFoundError:
            self.candidates.save(candidate)
            existing = candidate
        else:
            if existing.fingerprint != candidate.fingerprint:
                raise SagaConflict(
                    "candidate id is already bound to other evidence"
                )
        self.sagas.create(
            existing.candidate_id,
            f"pending-{existing.candidate_id}",
        )
        result = existing.to_dict()
        if raw_plan is not None:
            plan = DeploymentPlan.from_dict(raw_plan)
            if (
                plan.source_commit != existing.source_commit
                or plan.artifact_sha256 != existing.artifact_sha256
                or plan.manifest_sha256 != existing.manifest_sha256
            ):
                raise StaleApprovalError(
                    "deployment plan does not match release candidate"
                )
            try:
                stored = self.plans.load(existing.candidate_id)
            except FileNotFoundError:
                self.plans.save(existing.candidate_id, plan)
            else:
                if stored.sha256 != plan.sha256:
                    raise SagaConflict(
                        "candidate has a different deployment plan"
                    )
            result["deployment_plan_sha256"] = plan.sha256
        return result

    def get_release_candidate(self, candidate_id: str) -> dict[str, Any]:
        return {
            **self.candidates.load(candidate_id).to_dict(),
            "execution_enabled": self.execution_enabled,
            "shadow_mode": not self.execution_enabled,
            "side_effects_executed": False,
        }

    def get_release_saga(self, candidate_id: str) -> dict[str, Any]:
        return {
            **self.sagas.load(candidate_id).to_dict(),
            "execution_enabled": self.execution_enabled,
            "shadow_mode": not self.execution_enabled,
            "side_effects_executed": False,
        }

    def get_release_plan(self, candidate_id: str) -> dict[str, Any]:
        return {
            **self.release_plans.load(candidate_id).to_dict(),
            "candidate_id": candidate_id,
            "execution_enabled": self.execution_enabled,
            "shadow_mode": not self.execution_enabled,
            "side_effects_executed": False,
        }

    def get_acceptance(self, candidate_id: str) -> dict[str, Any]:
        record = json.loads(
            _safe_child(self.acceptance_root, candidate_id).read_text(
                encoding="utf-8"
            )
        )
        return {
            **_sanitize(record),
            "execution_enabled": self.execution_enabled,
            "shadow_mode": not self.execution_enabled,
            "side_effects_executed": False,
        }

    def get_capability_status(
        self, candidate_id: str
    ) -> dict[str, Any]:
        self.candidates.load(candidate_id)
        records: list[dict[str, Any]] = []
        for path in sorted(self.capabilities_root.glob("*.json")):
            try:
                record = json.loads(path.read_text(encoding="utf-8"))
            except (OSError, json.JSONDecodeError):
                continue
            if record.get("release_plan_id") == self.sagas.load(
                candidate_id
            ).release_plan_id:
                records.append(_sanitize(record))
        return {
            "candidate_id": candidate_id,
            "capabilities": records,
            "execution_enabled": self.execution_enabled,
            "shadow_mode": not self.execution_enabled,
            "side_effects_executed": False,
        }

    def get_recovery_status(self) -> dict[str, Any]:
        reports: list[dict[str, Any]] = []
        for journal in sorted(self.sagas.root.glob("*.events.jsonl")):
            candidate_id = journal.name[: -len(".events.jsonl")]
            saga = self.sagas.load(candidate_id)
            reports.append(
                {
                    "candidate_id": candidate_id,
                    "recovery_action": self.sagas.recovery_action(
                        saga.state
                    ),
                    "state": saga.state.value,
                }
            )
        return {
            "execution_enabled": self.execution_enabled,
            "shadow_mode": not self.execution_enabled,
            "side_effects_executed": False,
            "sagas": reports,
        }

    def ensure_owner_notification(
        self, candidate_id: str
    ) -> dict[str, Any]:
        self.candidates.load(candidate_id)
        path = _safe_child(self.notifications_root, candidate_id)
        payload = {
            "candidate_id": candidate_id,
            "created_at": utc_now(),
            "notification_id": "notification-"
            + hashlib.sha256(candidate_id.encode("utf-8")).hexdigest()[:24],
            "status": "OWNER_APPROVAL_REQUESTED",
        }
        path.parent.mkdir(parents=True, exist_ok=True)
        try:
            descriptor = os.open(
                path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600
            )
        except FileExistsError:
            existing = json.loads(path.read_text(encoding="utf-8"))
            return {**_sanitize(existing), "idempotent": True}
        with os.fdopen(descriptor, "w", encoding="utf-8") as output:
            json.dump(payload, output, sort_keys=True)
            output.write("\n")
            output.flush()
            os.fsync(output.fileno())
        _fsync_directory(path.parent)
        return {**payload, "idempotent": False}

    def get_approval_envelope(self, candidate_id: str) -> dict[str, Any]:
        candidate = self.candidates.load(candidate_id)
        try:
            release_plan = self.release_plans.load(candidate_id)
            deployment_plan_sha256 = release_plan.deployment.sha256
            release_plan_sha256 = release_plan.sha256
        except FileNotFoundError:
            try:
                deployment_plan_sha256 = self.plans.load(
                    candidate_id
                ).sha256
            except FileNotFoundError:
                deployment_plan_sha256 = None
            release_plan_sha256 = None
        return {
            "candidate_id": candidate.candidate_id,
            "candidate_fingerprint": candidate.fingerprint,
            "execution_enabled": self.execution_enabled,
            "required_capabilities": sorted(ALLOWED_CAPABILITIES),
            "signature_algorithm": "Ed25519",
            "deployment_plan_sha256": deployment_plan_sha256,
            "release_plan_sha256": release_plan_sha256,
            "policy_sha256": self.policy_sha256 or None,
            "status": "READY_FOR_OWNER_SIGNATURE",
        }

    def submit_owner_approval(
        self, payload: Mapping[str, Any]
    ) -> dict[str, Any]:
        approval = SignedReleasePlanApproval.from_dict(payload)
        if self.approval_verifier is None:
            raise CapabilityAuthorizationError(
                "owner approval verifier is not configured"
            )
        candidate = self.candidates.load(approval.candidate_id)
        try:
            release_plan = self.release_plans.load(
                approval.candidate_id
            )
        except FileNotFoundError:
            plan = self.plans.load(approval.candidate_id)
            self.approval_verifier.verify(
                approval,
                candidate,
                plan,
                self.policy_sha256,
            )
        else:
            self.approval_verifier.verify_release_plan(
                approval,
                candidate,
                release_plan,
                self.policy_sha256,
            )
        _, created = self.approvals.save(
            approval.candidate_id, approval
        )
        saga = self.sagas.bind_release_plan(
            approval.candidate_id, approval.release_plan_id
        )
        if saga.state == SagaState.READY_FOR_APPROVAL:
            saga = self.sagas.compare_and_set(
                approval.candidate_id,
                expected_revision=saga.revision,
                expected_states={SagaState.READY_FOR_APPROVAL},
                target=SagaState.APPROVED,
                event="signed_owner_approval_accepted",
                details={
                    "approval_fingerprint": approval.fingerprint,
                    "release_plan_sha256": (
                        approval.release_plan_sha256 or None
                    ),
                },
            )
        return {
            "fingerprint": approval.fingerprint,
            "saga_state": saga.state.value,
            "status": (
                "VERIFIED_AND_STORED" if created else "IDEMPOTENT"
            ),
            "execution_enabled": self.execution_enabled,
            "shadow_mode": not self.execution_enabled,
            "side_effects_executed": False,
        }

    def start_deployment(
        self,
        candidate_id: str,
        *,
        approval: SignedReleasePlanApproval | None = None,
    ) -> dict[str, Any]:
        self.candidates.load(candidate_id)
        if not self.execution_enabled:
            return {
                "candidate_id": candidate_id,
                "execution_enabled": False,
                "shadow_mode": True,
                "side_effects": [],
                "side_effects_executed": False,
                "status": "SHADOW_COMPLETED",
            }
        if self.orchestrator is None:
            raise CapabilityAuthorizationError(
                "release orchestrator is not configured"
            )
        # Caller-supplied approvals are deliberately ignored. The orchestrator
        # loads the durable approval bound during submit_owner_approval().
        return _sanitize(dict(self.orchestrator.execute(candidate_id)))