fix(gitops): add tamper-evident saga and exclusive capability claims

active-runtime/app/silent_release_hardening.py

@@ -14,6 +14,7 @@ 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
@@ -68,6 +69,34 @@ 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"
+
+
+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):
@@ -467,7 +496,17 @@ class SignedApprovalVerifier:
 
 
 class CapabilityLedger:
-    """Durable per-capability replay guard for a signed release plan."""
+    """Durable exclusive claims with leases and fail-closed recovery probes."""
+
+    STATES = frozenset(
+        {
+            "PENDING",
+            "CLAIMED",
+            "COMPLETED",
+            "FAILED_RECOVERABLE",
+            "MANUAL_OWNER_REVIEW_REQUIRED",
+        }
+    )
 
     def __init__(self, root: Path):
         self.root = Path(root)
@@ -480,11 +519,7 @@ class CapabilityLedger:
     ) -> Path:
         key = hashlib.sha256(
             "\0".join(
-                (
-                    approval.release_plan_id,
-                    capability,
-                    operation,
-                )
+                (approval.release_plan_id, capability, operation)
             ).encode("utf-8")
         ).hexdigest()
         return _safe_child(self.root, key)
@@ -493,6 +528,30 @@ class CapabilityLedger:
         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,
@@ -515,11 +574,22 @@ class CapabilityLedger:
         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,
         }
 
@@ -528,14 +598,198 @@ class CapabilityLedger:
         existing: Mapping[str, Any],
         expected: Mapping[str, Any],
     ) -> None:
-        immutable = {
-            key: existing.get(key) for key in expected
-        }
-        if immutable != dict(expected):
+        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 _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] | 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 == "MANUAL_OWNER_REVIEW_REQUIRED":
+                return {**record, "status": state}
+            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"}
+                if side_effect_probe is None:
+                    return {**record, "status": "PROBE_REQUIRED"}
+                probe = ClaimProbeResult(side_effect_probe())
+                if probe == ClaimProbeResult.APPLIED_VERIFIED:
+                    record.update(
+                        {
+                            "completed_at": utc_now(),
+                            "probe_result": probe.value,
+                            "state": "COMPLETED",
+                        }
+                    )
+                    self._write(path, record)
+                    return {**record, "status": "IDEMPOTENT"}
+                if probe in {
+                    ClaimProbeResult.INDETERMINATE,
+                    ClaimProbeResult.CONFLICT,
+                }:
+                    record.update(
+                        {
+                            "probe_result": probe.value,
+                            "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]:
+        path = self._operation_path(approval, capability, operation)
+        lock = self._lock(path)
+        try:
+            record = json.loads(path.read_text(encoding="utf-8"))
+            if 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,
@@ -543,96 +797,30 @@ class CapabilityLedger:
         *,
         operation: str,
     ) -> dict[str, Any]:
-        self._assert_authorized(approval, capability, operation)
-        nonce_path = self._nonce_path(approval.nonce)
-        operation_path = self._operation_path(
-            approval, capability, operation
+        """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,
         )
-        nonce_path.parent.mkdir(parents=True, exist_ok=True)
-        operation_path.parent.mkdir(parents=True, exist_ok=True)
-        nonce_payload = {
-            "fingerprint": approval.fingerprint,
-            "release_plan_id": approval.release_plan_id,
+        aliases = {
+            "CLAIMED": "RESERVED",
+            "BUSY": "RESUME",
+            "IDEMPOTENT": "IDEMPOTENT",
         }
-        if nonce_path.exists():
-            existing_nonce = json.loads(
-                nonce_path.read_text(encoding="utf-8")
-            )
-            if existing_nonce != nonce_payload:
-                raise CapabilityAuthorizationError(
-                    "approval nonce replay does not match original plan"
-                )
-        elif not self._exclusive_json(nonce_path, nonce_payload):
-            existing_nonce = json.loads(
-                nonce_path.read_text(encoding="utf-8")
-            )
-            if existing_nonce != nonce_payload:
-                raise CapabilityAuthorizationError(
-                    "approval nonce replay does not match original plan"
-                )
-        immutable = self._immutable_payload(
-            approval, capability, operation
-        )
-        payload = {
-            **immutable,
-            "reserved_at": utc_now(),
-            "state": "RESERVED",
-        }
-        if operation_path.exists():
-            existing = json.loads(
-                operation_path.read_text(encoding="utf-8")
-            )
-            self._assert_record_matches(existing, immutable)
-            return {
-                **existing,
-                "status": (
-                    "IDEMPOTENT"
-                    if existing.get("state") == "COMPLETED"
-                    else "RESUME"
-                ),
-            }
-        if self._exclusive_json(operation_path, payload):
-            return {**payload, "status": "RESERVED"}
-        existing = json.loads(
-            operation_path.read_text(encoding="utf-8")
-        )
-        self._assert_record_matches(existing, immutable)
-        return {
-            **existing,
-            "status": (
-                "IDEMPOTENT"
-                if existing.get("state") == "COMPLETED"
-                else "RESUME"
-            ),
-        }
-
-    def complete(
-        self,
-        approval: SignedReleasePlanApproval,
-        capability: str,
-        *,
-        operation: str,
-        evidence: Mapping[str, Any] | None = None,
-    ) -> dict[str, Any]:
-        self._assert_authorized(approval, capability, operation)
-        path = self._operation_path(
-            approval, capability, operation
-        )
-        existing = json.loads(path.read_text(encoding="utf-8"))
-        immutable = self._immutable_payload(
-            approval, capability, operation
-        )
-        self._assert_record_matches(existing, immutable)
-        if existing.get("state") == "COMPLETED":
-            return {**existing, "status": "IDEMPOTENT"}
-        payload = {
-            **existing,
-            "completed_at": utc_now(),
-            "evidence": _sanitize(dict(evidence or {})),
-            "state": "COMPLETED",
-        }
-        _atomic_json(path, payload)
-        return {**payload, "status": "CONSUMED"}
+        return {**result, "status": aliases.get(result["status"], result["status"])}
 
     def consume(
         self,
@@ -641,35 +829,18 @@ class CapabilityLedger:
         *,
         operation: str,
     ) -> dict[str, Any]:
-        reserved = self.reserve(
+        reservation = self.reserve(
             approval, capability, operation=operation
         )
-        if reserved["status"] == "IDEMPOTENT":
-            return reserved
-        return self.complete(
+        if reservation["status"] == "IDEMPOTENT":
+            return reservation
+        completed = self.complete(
             approval,
             capability,
             operation=operation,
+            claim_id=reservation["claim_id"],
         )
-
-    @staticmethod
-    def _exclusive_json(
-        path: Path, payload: Mapping[str, Any]
-    ) -> bool:
-        try:
-            descriptor = os.open(
-                path,
-                os.O_WRONLY | os.O_CREAT | os.O_EXCL,
-                0o600,
-            )
-        except FileExistsError:
-            return False
-        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())
-        return True
+        return {**completed, "status": "CONSUMED"}
 
 
 class SagaState(str, Enum):
@@ -681,6 +852,7 @@ class SagaState(str, Enum):
     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"
@@ -698,6 +870,9 @@ class ReleaseSaga:
     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)
 
@@ -714,7 +889,7 @@ class ReleaseSaga:
 
 
 class ReleaseSagaStore:
-    """Atomic Saga snapshots plus append-only recovery journal."""
+    """Hash-chained journal is truth; atomic snapshot is a rebuildable cache."""
 
     RECOVERY_ACTIONS = {
         SagaState.SWITCH_PENDING: "resume_runtime_switch",
@@ -728,8 +903,14 @@ class ReleaseSagaStore:
         SagaState.MANUAL_OWNER_REVIEW_REQUIRED: "manual_owner_review",
     }
 
-    def __init__(self, root: Path):
+    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 "")):
@@ -771,8 +952,8 @@ class ReleaseSagaStore:
         path.parent.mkdir(parents=True, exist_ok=True)
         lock = self._lock(candidate_id)
         try:
-            if path.exists():
-                existing = self.load(candidate_id)
+            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"
@@ -782,23 +963,65 @@ class ReleaseSagaStore:
                 candidate_id=candidate_id,
                 release_plan_id=release_plan_id,
             )
-            event = {
+            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,
             }
-            saga.events.append(event)
-            _atomic_json(path, saga.to_dict())
-            self._append_event(candidate_id, event)
+            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:
-        return ReleaseSaga.from_dict(
-            json.loads(self._path(candidate_id).read_text(encoding="utf-8"))
-        )
+        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,
@@ -812,7 +1035,10 @@ class ReleaseSagaStore:
     ) -> ReleaseSaga:
         lock = self._lock(candidate_id)
         try:
-            saga = self.load(candidate_id)
+            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
@@ -822,16 +1048,26 @@ class ReleaseSagaStore:
             saga.revision += 1
             saga.state = SagaState(target)
             saga.updated_at = utc_now()
-            record = {
+            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,
             }
-            saga.events.append(record)
-            _atomic_json(self._path(candidate_id), saga.to_dict())
-            self._append_event(candidate_id, record)
+            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)
@@ -854,7 +1090,7 @@ class ReleaseSagaStore:
         _validate_identifier("release_plan_id", release_plan_id)
         lock = self._lock(candidate_id)
         try:
-            saga = self.load(candidate_id)
+            saga = self._load_locked(candidate_id)
             pending = f"pending-{candidate_id}"
             if saga.release_plan_id not in {pending, release_plan_id}:
                 raise SagaConflict(
@@ -865,29 +1101,56 @@ class ReleaseSagaStore:
             saga.release_plan_id = release_plan_id
             saga.revision += 1
             saga.updated_at = utc_now()
-            record = {
+            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,
             }
-            saga.events.append(record)
-            _atomic_json(self._path(candidate_id), saga.to_dict())
-            self._append_event(candidate_id, record)
+            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, event: Mapping[str, Any]
-    ) -> None:
+        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(
-                    _sanitize(dict(event)),
+                    record,
                     sort_keys=True,
                     separators=(",", ":"),
                 )
@@ -895,6 +1158,117 @@ class ReleaseSagaStore:
             )
             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:
@@ -980,12 +1354,18 @@ class CapabilityController:
         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,
@@ -994,26 +1374,52 @@ class CapabilityController:
         operation: str,
         **kwargs: Any,
     ) -> dict[str, Any]:
-        reservation = self.ledger.reserve(
-            approval, self.capability, operation=operation
+        claim = self.ledger.claim(
+            approval,
+            self.capability,
+            operation=operation,
+            executor_id=self.executor_id,
+            lease_seconds=self.lease_seconds,
+            side_effect_probe=self.recovery_probe,
         )
-        if reservation["status"] == "IDEMPOTENT":
+        if claim["status"] == "IDEMPOTENT":
             return {
                 "capability": self.capability,
                 "status": "IDEMPOTENT",
             }
-        result = dict(self.adapter(**kwargs))
-        self.validate_result(approval, result)
-        self.ledger.complete(
-            approval,
-            self.capability,
-            operation=operation,
-            evidence={
-                "result_sha256": hashlib.sha256(
-                    _canonical_json(_sanitize(result))
-                ).hexdigest()
-            },
-        )
+        if claim["status"] == "BUSY":
+            return {
+                "capability": self.capability,
+                "claim_id": claim.get("claim_id"),
+                "status": "BUSY",
+            }
+        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),

tests/test_silent_release_production_contract.py

@@ -0,0 +1,367 @@
+from __future__ import annotations
+
+import dataclasses
+import json
+import multiprocessing
+import os
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
+
+from silent_release_hardening import (  # noqa: E402
+    CapabilityAuthorizationError,
+    CapabilityController,
+    CapabilityLedger,
+    ClaimProbeResult,
+    JournalIntegrityError,
+    ReleaseSagaStore,
+    SagaState,
+    SignedReleasePlanApproval,
+)
+
+
+COMMIT = "1" * 40
+TREE = "2" * 40
+ARTIFACT = "3" * 64
+MANIFEST = "4" * 64
+PLAN = "5" * 64
+POLICY = "6" * 64
+EVIDENCE = "7" * 40
+OWNER_RECORD = "8" * 40
+
+
+def approval() -> SignedReleasePlanApproval:
+    return SignedReleasePlanApproval(
+        release_plan_id="release-plan-contract-001",
+        candidate_id="rc-contract-001",
+        source_commit=COMMIT,
+        tree_oid=TREE,
+        artifact_sha256=ARTIFACT,
+        manifest_sha256=MANIFEST,
+        deployment_plan_sha256=PLAN,
+        policy_sha256=POLICY,
+        approval_plan_id="owner-plan-contract-001",
+        evidence_closure_oid=EVIDENCE,
+        owner_approval_record_oid=OWNER_RECORD,
+        capabilities=(
+            "runtime_switch",
+            "tag_publish",
+            "baseline_registry_finalize",
+        ),
+        expires_at="2099-01-01T00:00:00Z",
+        nonce="nonce-contract-000000001",
+        key_id="owner-key-contract-1",
+        signature="test-signature",
+    )
+
+
+def _claim_worker(root: str, start, queue) -> None:
+    marker = Path(root) / "adapter-calls.log"
+
+    def adapter():
+        with marker.open("a", encoding="utf-8") as output:
+            output.write(f"{os.getpid()}\n")
+            output.flush()
+            os.fsync(output.fileno())
+        time.sleep(0.05)
+        return {"verified": True}
+
+    controller = CapabilityController(
+        capability="runtime_switch",
+        ledger=CapabilityLedger(Path(root) / "ledger"),
+        adapter=adapter,
+        executor_id=f"worker-{os.getpid()}",
+        lease_seconds=10,
+    )
+    start.wait()
+    result = controller.execute(
+        approval(), operation="runtime-switch-process-race"
+    )
+    queue.put(result["status"])
+
+
+def test_journal_is_hash_chained_and_snapshot_tracks_head(tmp_path):
+    store = ReleaseSagaStore(tmp_path / "sagas")
+    saga = store.create("rc-contract-001", "release-plan-contract-001")
+    saga = store.compare_and_set(
+        saga.candidate_id,
+        expected_revision=saga.revision,
+        expected_states={SagaState.PREPARING},
+        target=SagaState.VERIFYING,
+        event="verification_started",
+    )
+    records = store.verify_journal(saga.candidate_id)
+    assert [record["sequence"] for record in records] == [1, 2]
+    assert records[0]["previous_hash"] == "0" * 64
+    assert records[1]["previous_hash"] == records[0]["event_hash"]
+    assert saga.journal_sequence == 2
+    assert saga.journal_head_hash == records[-1]["event_hash"]
+
+
+def test_journal_single_line_tamper_enters_manual_review(tmp_path):
+    store = ReleaseSagaStore(tmp_path / "sagas")
+    saga = store.create("rc-contract-001", "release-plan-contract-001")
+    journal = tmp_path / "sagas" / "rc-contract-001.events.jsonl"
+    record = json.loads(journal.read_text(encoding="utf-8").splitlines()[0])
+    record["payload"]["state"] = SagaState.COMPLETED.value
+    journal.write_text(json.dumps(record) + "\n", encoding="utf-8")
+    recovered = store.load(saga.candidate_id)
+    assert recovered.state == SagaState.MANUAL_OWNER_REVIEW_REQUIRED
+    assert recovered.journal_integrity is False
+
+
+def test_journal_truncation_is_not_silently_accepted(tmp_path):
+    store = ReleaseSagaStore(tmp_path / "sagas")
+    saga = store.create("rc-contract-001", "release-plan-contract-001")
+    store.compare_and_set(
+        saga.candidate_id,
+        expected_revision=0,
+        expected_states={SagaState.PREPARING},
+        target=SagaState.VERIFYING,
+        event="verification_started",
+    )
+    journal = tmp_path / "sagas" / "rc-contract-001.events.jsonl"
+    journal.write_text(
+        journal.read_text(encoding="utf-8").splitlines()[0] + "\n",
+        encoding="utf-8",
+    )
+    recovered = store.load(saga.candidate_id)
+    assert recovered.state == SagaState.MANUAL_OWNER_REVIEW_REQUIRED
+    assert recovered.journal_integrity is False
+
+
+def test_journal_reordering_enters_manual_review(tmp_path):
+    store = ReleaseSagaStore(tmp_path / "sagas")
+    saga = store.create("rc-contract-001", "release-plan-contract-001")
+    saga = store.compare_and_set(
+        saga.candidate_id,
+        expected_revision=0,
+        expected_states={SagaState.PREPARING},
+        target=SagaState.VERIFYING,
+        event="verification_started",
+    )
+    store.compare_and_set(
+        saga.candidate_id,
+        expected_revision=1,
+        expected_states={SagaState.VERIFYING},
+        target=SagaState.READY_FOR_APPROVAL,
+        event="verification_passed",
+    )
+    journal = tmp_path / "sagas" / "rc-contract-001.events.jsonl"
+    lines = journal.read_text(encoding="utf-8").splitlines()
+    journal.write_text(
+        "\n".join([lines[0], lines[2], lines[1]]) + "\n",
+        encoding="utf-8",
+    )
+    assert (
+        store.load(saga.candidate_id).state
+        == SagaState.MANUAL_OWNER_REVIEW_REQUIRED
+    )
+
+
+def test_journal_missing_middle_line_enters_manual_review(tmp_path):
+    store = ReleaseSagaStore(tmp_path / "sagas")
+    saga = store.create("rc-contract-001", "release-plan-contract-001")
+    saga = store.compare_and_set(
+        saga.candidate_id,
+        expected_revision=0,
+        expected_states={SagaState.PREPARING},
+        target=SagaState.VERIFYING,
+        event="verification_started",
+    )
+    store.compare_and_set(
+        saga.candidate_id,
+        expected_revision=1,
+        expected_states={SagaState.VERIFYING},
+        target=SagaState.READY_FOR_APPROVAL,
+        event="verification_passed",
+    )
+    journal = tmp_path / "sagas" / "rc-contract-001.events.jsonl"
+    lines = journal.read_text(encoding="utf-8").splitlines()
+    journal.write_text(
+        "\n".join([lines[0], lines[2]]) + "\n",
+        encoding="utf-8",
+    )
+    assert (
+        store.load(saga.candidate_id).state
+        == SagaState.MANUAL_OWNER_REVIEW_REQUIRED
+    )
+
+
+def test_crash_after_journal_before_snapshot_recovers_from_journal(tmp_path):
+    crashed = {"done": False}
+
+    def hook(event):
+        if event["payload"]["event"] == "verification_started":
+            crashed["done"] = True
+            raise RuntimeError("simulated crash after journal fsync")
+
+    root = tmp_path / "sagas"
+    store = ReleaseSagaStore(root, after_journal_hook=hook)
+    saga = store.create("rc-contract-001", "release-plan-contract-001")
+    with pytest.raises(RuntimeError, match="simulated crash"):
+        store.compare_and_set(
+            saga.candidate_id,
+            expected_revision=0,
+            expected_states={SagaState.PREPARING},
+            target=SagaState.VERIFYING,
+            event="verification_started",
+        )
+    assert crashed["done"] is True
+    recovered = ReleaseSagaStore(root).load(saga.candidate_id)
+    assert recovered.state == SagaState.VERIFYING
+    assert recovered.revision == 1
+    assert recovered.journal_sequence == 2
+
+
+def test_capability_claim_is_exclusive_and_busy(tmp_path):
+    ledger = CapabilityLedger(tmp_path / "ledger")
+    first = ledger.claim(
+        approval(),
+        "runtime_switch",
+        operation="runtime-switch-001",
+        executor_id="worker-a",
+        lease_seconds=60,
+    )
+    second = ledger.claim(
+        approval(),
+        "runtime_switch",
+        operation="runtime-switch-001",
+        executor_id="worker-b",
+        lease_seconds=60,
+    )
+    assert first["status"] == "CLAIMED"
+    assert second["status"] == "BUSY"
+    assert second["claim_id"] == first["claim_id"]
+
+
+def test_capability_complete_requires_matching_claim_id(tmp_path):
+    ledger = CapabilityLedger(tmp_path / "ledger")
+    claim = ledger.claim(
+        approval(),
+        "runtime_switch",
+        operation="runtime-switch-001",
+        executor_id="worker-a",
+        lease_seconds=60,
+    )
+    with pytest.raises(CapabilityAuthorizationError):
+        ledger.complete(
+            approval(),
+            "runtime_switch",
+            operation="runtime-switch-001",
+            claim_id="wrong-claim",
+            evidence={"verified": True},
+        )
+    result = ledger.complete(
+        approval(),
+        "runtime_switch",
+        operation="runtime-switch-001",
+        claim_id=claim["claim_id"],
+        evidence={"verified": True},
+    )
+    assert result["status"] == "COMPLETED"
+
+
+def test_expired_claim_requires_side_effect_probe_before_takeover(
+    tmp_path, monkeypatch
+):
+    ledger = CapabilityLedger(tmp_path / "ledger")
+    first = ledger.claim(
+        approval(),
+        "tag_publish",
+        operation="tag-V2R1C3",
+        executor_id="worker-a",
+        lease_seconds=1,
+        now=100.0,
+    )
+    blocked = ledger.claim(
+        approval(),
+        "tag_publish",
+        operation="tag-V2R1C3",
+        executor_id="worker-b",
+        lease_seconds=10,
+        now=102.0,
+    )
+    assert blocked["status"] == "PROBE_REQUIRED"
+    takeover = ledger.claim(
+        approval(),
+        "tag_publish",
+        operation="tag-V2R1C3",
+        executor_id="worker-b",
+        lease_seconds=10,
+        now=102.0,
+        side_effect_probe=lambda: ClaimProbeResult.NOT_APPLIED,
+    )
+    assert takeover["status"] == "CLAIMED"
+    assert takeover["claim_id"] != first["claim_id"]
+    assert takeover["attempt"] == 2
+
+
+def test_two_processes_execute_adapter_only_once(tmp_path):
+    context = multiprocessing.get_context("fork")
+    start = context.Event()
+    queue = context.Queue()
+    workers = [
+        context.Process(
+            target=_claim_worker,
+            args=(str(tmp_path), start, queue),
+        )
+        for _ in range(2)
+    ]
+    for worker in workers:
+        worker.start()
+    start.set()
+    for worker in workers:
+        worker.join(timeout=10)
+        assert worker.exitcode == 0
+    statuses = sorted(queue.get(timeout=2) for _ in workers)
+    assert statuses in (
+        ["BUSY", "EXECUTED"],
+        ["EXECUTED", "IDEMPOTENT"],
+    )
+    calls = (tmp_path / "adapter-calls.log").read_text(
+        encoding="utf-8"
+    ).splitlines()
+    assert len(calls) == 1
+
+
+@pytest.mark.parametrize(
+    "probe,expected",
+    [
+        (ClaimProbeResult.APPLIED_VERIFIED, "IDEMPOTENT"),
+        (
+            ClaimProbeResult.INDETERMINATE,
+            "MANUAL_OWNER_REVIEW_REQUIRED",
+        ),
+        (
+            ClaimProbeResult.CONFLICT,
+            "MANUAL_OWNER_REVIEW_REQUIRED",
+        ),
+    ],
+)
+def test_expired_claim_probe_is_fail_closed(tmp_path, probe, expected):
+    ledger = CapabilityLedger(tmp_path / "ledger")
+    ledger.claim(
+        approval(),
+        "tag_publish",
+        operation="tag-V2R1C3",
+        executor_id="worker-a",
+        lease_seconds=1,
+        now=100.0,
+    )
+    result = ledger.claim(
+        approval(),
+        "tag_publish",
+        operation="tag-V2R1C3",
+        executor_id="worker-b",
+        lease_seconds=10,
+        now=102.0,
+        side_effect_probe=lambda: probe,
+    )
+    assert result["status"] == expected