Implement durable module recovery operations

docs/DURABLE_RECOVERY_OPERATIONS.md

@@ -0,0 +1,25 @@
+# Durable Recovery Operations
+
+Modules must use `begin_durable_recovery_operation` for work whose effects can
+outlive the caller's SQLAlchemy transaction. The helper commits the canonical
+request hash, recovery plan, precondition evidence, running state, and lease
+fence before the caller mutates object storage, a queue, a filesystem, or an
+external provider.
+
+Each later checkpoint is written through an independent database session. A
+business-transaction rollback therefore cannot erase evidence of an earlier
+effect. Successful completion requires concrete verification checks and a valid
+hash chain. Compensation likewise records recovery-required, recovering, and
+verified-recovered checkpoints rather than reporting an ordinary failure.
+
+If a runtime disappears, another runtime may claim the operation only after the
+lease expires. The takeover records both fences. A stale compensatable operation
+becomes recovery-required; a stale forward-only or irreversible external effect
+becomes outcome-unknown; a database-only atomic operation is recorded failed
+because its transaction rolled back. Takeover never re-executes the original
+request automatically.
+
+Evidence and metadata may contain opaque references, digests, counts, and
+provider result codes. They must never contain credentials or resolved secrets.
+Ops is the platform surface for unresolved operation status; owning modules must
+provide the reconciliation action and business-level explanation.

src/govoplan_core/core/recovery_runtime.py

@@ -0,0 +1,446 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.recovery import (
+    RecoveryGuaranteeError,
+    RecoveryMode,
+    RecoveryOperation,
+    RecoveryPlan,
+    RecoveryStatus,
+    plan_recovery_operation,
+    prepare_recovery_operation,
+    record_recovery_checkpoint,
+    start_recovery_operation,
+    transition_recovery_operation,
+    verify_recovery_evidence_chain,
+)
+from govoplan_core.core.runtime_coordination import (
+    LeaseClaim,
+    RuntimeIdentity,
+    acquire_lease,
+    release_lease,
+    renew_lease,
+)
+
+
+SessionFactory = Callable[[], Session]
+
+
+class RecoveryOperationBusy(RecoveryGuaranteeError):
+    pass
+
+
+class RecoveryOperationStateConflict(RecoveryGuaranteeError):
+    def __init__(self, operation_id: str, status: str) -> None:
+        self.operation_id = operation_id
+        self.status = status
+        super().__init__(
+            f"Recovery operation {operation_id} is already {status}; "
+            "reconcile it before starting another effect"
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class DurableRecoveryStart:
+    operation_id: str
+    status: str
+    replayed: bool
+    operation: DurableRecoveryOperation | None
+
+
+@dataclass(slots=True)
+class DurableRecoveryOperation:
+    """Append checkpoints in independent, committed transactions.
+
+    The caller's business transaction may roll back without erasing evidence
+    that an object, queue, filesystem, or provider effect already occurred.
+    """
+
+    session_factory: SessionFactory
+    operation_id: str
+    lease_claim: LeaseClaim
+    lease_ttl_seconds: int
+    closed: bool = False
+
+    def checkpoint(
+        self,
+        *,
+        kind: str,
+        summary: str,
+        evidence: dict[str, Any],
+    ) -> None:
+        with self.session_factory() as session:
+            operation, claim = self._locked_and_renewed(session)
+            record_recovery_checkpoint(
+                session,
+                operation,
+                kind=kind,
+                summary=summary,
+                evidence=evidence,
+                lease_claim=claim,
+            )
+            self._verify_chain(session)
+            session.commit()
+
+    def succeed(self, *, evidence: dict[str, Any]) -> None:
+        with self.session_factory() as session:
+            operation, claim = self._locked_and_renewed(session)
+            transition_recovery_operation(
+                session,
+                operation,
+                status=RecoveryStatus.SUCCEEDED,
+                kind="verified-success",
+                summary="Operation effects and authoritative state were verified",
+                evidence=evidence,
+                lease_claim=claim,
+            )
+            self._verify_chain(session)
+            release_lease(session, claim)
+            session.commit()
+        self.closed = True
+
+    def compensate(
+        self,
+        *,
+        failure_summary: str,
+        failure_evidence: dict[str, Any],
+        recovery_evidence: dict[str, Any],
+    ) -> None:
+        with self.session_factory() as session:
+            operation, claim = self._locked_and_renewed(session)
+            if operation.status == RecoveryStatus.RUNNING.value:
+                operation = transition_recovery_operation(
+                    session,
+                    operation,
+                    status=RecoveryStatus.RECOVERY_REQUIRED,
+                    kind="compensation-required",
+                    summary="The started operation requires explicit compensation",
+                    evidence=failure_evidence,
+                    failure_summary=failure_summary,
+                    lease_claim=claim,
+                )
+            if operation.status == RecoveryStatus.RECOVERY_REQUIRED.value:
+                operation = transition_recovery_operation(
+                    session,
+                    operation,
+                    status=RecoveryStatus.RECOVERING,
+                    kind="compensation-started",
+                    summary="Compensation started",
+                    evidence={"failure_summary": failure_summary},
+                    lease_claim=claim,
+                )
+            transition_recovery_operation(
+                session,
+                operation,
+                status=RecoveryStatus.RECOVERED,
+                kind="compensation-verified",
+                summary="Compensation restored the declared invariant",
+                evidence=recovery_evidence,
+                lease_claim=claim,
+            )
+            self._verify_chain(session)
+            release_lease(session, claim)
+            session.commit()
+        self.closed = True
+
+    def unresolved(
+        self,
+        *,
+        status: RecoveryStatus,
+        summary: str,
+        evidence: dict[str, Any],
+        failure_summary: str,
+    ) -> None:
+        if status not in {
+            RecoveryStatus.OUTCOME_UNKNOWN,
+            RecoveryStatus.RECOVERY_REQUIRED,
+        }:
+            raise ValueError("Unresolved operations require an unresolved status")
+        with self.session_factory() as session:
+            operation, claim = self._locked_and_renewed(session)
+            transition_recovery_operation(
+                session,
+                operation,
+                status=status,
+                kind="unresolved-effect",
+                summary=summary,
+                evidence=evidence,
+                failure_summary=failure_summary,
+                lease_claim=claim,
+            )
+            self._verify_chain(session)
+            release_lease(session, claim)
+            session.commit()
+        self.closed = True
+
+    def release_unresolved(self) -> None:
+        """Release authority after a process-local exception.
+
+        This does not alter the operation state. A later recovery claim treats a
+        stale `running` operation according to its declared recovery mode.
+        """
+
+        if self.closed:
+            return
+        with self.session_factory() as session:
+            operation, claim = self._locked_and_renewed(session)
+            record_recovery_checkpoint(
+                session,
+                operation,
+                kind="authority-released",
+                summary="Execution authority was released without a terminal claim",
+                evidence={"status": operation.status},
+                lease_claim=claim,
+            )
+            self._verify_chain(session)
+            release_lease(session, claim)
+            session.commit()
+        self.closed = True
+
+    def _locked_and_renewed(
+        self,
+        session: Session,
+    ) -> tuple[RecoveryOperation, LeaseClaim]:
+        if self.closed:
+            raise RecoveryGuaranteeError("Recovery operation handle is closed")
+        claim = renew_lease(
+            session,
+            self.lease_claim,
+            ttl_seconds=self.lease_ttl_seconds,
+        )
+        operation = session.execute(
+            select(RecoveryOperation)
+            .where(RecoveryOperation.id == self.operation_id)
+            .with_for_update()
+        ).scalar_one()
+        self.lease_claim = claim
+        return operation, claim
+
+    def _verify_chain(self, session: Session) -> None:
+        if not verify_recovery_evidence_chain(session, self.operation_id):
+            raise RecoveryGuaranteeError(
+                "Recovery checkpoint chain verification failed"
+            )
+
+
+def begin_durable_recovery_operation(
+    session_factory: SessionFactory,
+    *,
+    identity: RuntimeIdentity,
+    module_id: str,
+    operation_type: str,
+    idempotency_key: str,
+    request: dict[str, Any],
+    recovery_plan: RecoveryPlan,
+    precondition_evidence: dict[str, Any],
+    lease_resource_key: str,
+    lease_ttl_seconds: int = 300,
+    resource_type: str | None = None,
+    resource_id: str | None = None,
+    metadata: dict[str, Any] | None = None,
+) -> DurableRecoveryStart:
+    if lease_ttl_seconds < 1:
+        raise ValueError("Recovery lease TTL must be at least one second")
+    with session_factory() as session:
+        claim = acquire_lease(
+            session,
+            installation_id=identity.installation_id,
+            resource_key=lease_resource_key,
+            holder_node_id=identity.node_id,
+            holder_incarnation=identity.incarnation,
+            ttl_seconds=lease_ttl_seconds,
+            metadata={
+                "module_id": module_id,
+                "operation_type": operation_type,
+            },
+        )
+        if claim is None:
+            raise RecoveryOperationBusy(
+                f"Another runtime owns the recovery fence for {lease_resource_key}"
+            )
+        existing = session.execute(
+            select(RecoveryOperation).where(
+                RecoveryOperation.installation_id == identity.installation_id,
+                RecoveryOperation.module_id == module_id,
+                RecoveryOperation.idempotency_key == idempotency_key,
+            )
+        ).scalar_one_or_none()
+        operation = plan_recovery_operation(
+            session,
+            installation_id=identity.installation_id,
+            module_id=module_id,
+            operation_type=operation_type,
+            idempotency_key=idempotency_key,
+            request=request,
+            recovery_plan=recovery_plan,
+            resource_type=resource_type,
+            resource_id=resource_id,
+            lease_claim=claim,
+            metadata=metadata,
+        )
+        if existing is not None:
+            if operation.status == RecoveryStatus.SUCCEEDED.value:
+                release_lease(session, claim)
+                session.commit()
+                return DurableRecoveryStart(
+                    operation_id=operation.id,
+                    status=operation.status,
+                    replayed=True,
+                    operation=None,
+                )
+            session.rollback()
+            raise RecoveryOperationStateConflict(operation.id, operation.status)
+        prepare_recovery_operation(
+            session,
+            operation,
+            evidence=precondition_evidence,
+            lease_claim=claim,
+        )
+        start_recovery_operation(
+            session,
+            operation,
+            evidence={"lease_resource_key": lease_resource_key},
+            lease_claim=claim,
+        )
+        if not verify_recovery_evidence_chain(session, operation.id):
+            raise RecoveryGuaranteeError(
+                "Recovery checkpoint chain verification failed before side effects"
+            )
+        session.commit()
+        return DurableRecoveryStart(
+            operation_id=operation.id,
+            status=operation.status,
+            replayed=False,
+            operation=DurableRecoveryOperation(
+                session_factory=session_factory,
+                operation_id=operation.id,
+                lease_claim=claim,
+                lease_ttl_seconds=lease_ttl_seconds,
+            ),
+        )
+
+
+def claim_durable_recovery_operation(
+    session_factory: SessionFactory,
+    *,
+    identity: RuntimeIdentity,
+    operation_id: str,
+    lease_ttl_seconds: int = 300,
+) -> DurableRecoveryOperation:
+    with session_factory() as session:
+        candidate = session.get(RecoveryOperation, operation_id)
+        if candidate is None:
+            raise RecoveryGuaranteeError("Recovery operation was not found")
+        if candidate.status in {
+            RecoveryStatus.SUCCEEDED.value,
+            RecoveryStatus.FAILED.value,
+            RecoveryStatus.RECOVERED.value,
+            RecoveryStatus.MANUAL_INTERVENTION.value,
+        }:
+            raise RecoveryOperationStateConflict(candidate.id, candidate.status)
+        if not candidate.lease_resource_key:
+            raise RecoveryGuaranteeError(
+                "Recovery takeover requires an operation-bound lease resource"
+            )
+        claim = acquire_lease(
+            session,
+            installation_id=identity.installation_id,
+            resource_key=candidate.lease_resource_key,
+            holder_node_id=identity.node_id,
+            holder_incarnation=identity.incarnation,
+            ttl_seconds=lease_ttl_seconds,
+            metadata={"recovery_operation_id": candidate.id},
+        )
+        if claim is None:
+            raise RecoveryOperationBusy(
+                f"Another runtime owns recovery operation {candidate.id}"
+            )
+        operation = session.execute(
+            select(RecoveryOperation)
+            .where(RecoveryOperation.id == operation_id)
+            .with_for_update()
+        ).scalar_one()
+        previous_fence = {
+            "holder_node_id": operation.holder_node_id,
+            "holder_incarnation": operation.holder_incarnation,
+            "fence_number": operation.fencing_token,
+        }
+        operation.holder_node_id = claim.holder_node_id
+        operation.holder_incarnation = claim.holder_incarnation
+        operation.fencing_token = claim.fencing_token
+        session.add(operation)
+        record_recovery_checkpoint(
+            session,
+            operation,
+            kind="fence-takeover",
+            summary="A new runtime claimed explicit recovery authority",
+            evidence={
+                "previous_fence": previous_fence,
+                "new_fence_number": claim.fencing_token,
+            },
+            lease_claim=claim,
+        )
+        if operation.status == RecoveryStatus.RUNNING.value:
+            mode = RecoveryMode(operation.mode)
+            if mode == RecoveryMode.ATOMIC:
+                transition_recovery_operation(
+                    session,
+                    operation,
+                    status=RecoveryStatus.FAILED,
+                    kind="stale-atomic-operation",
+                    summary="The stale database-only transaction rolled back",
+                    evidence={"previous_fence": previous_fence},
+                    failure_summary="Execution authority expired before commit",
+                    lease_claim=claim,
+                )
+            elif mode in {RecoveryMode.COMPENSATION, RecoveryMode.SNAPSHOT_RESTORE}:
+                transition_recovery_operation(
+                    session,
+                    operation,
+                    status=RecoveryStatus.RECOVERY_REQUIRED,
+                    kind="stale-effect-requires-recovery",
+                    summary="Execution authority expired after effects may have started",
+                    evidence={"previous_fence": previous_fence},
+                    failure_summary="Execution authority expired during a non-atomic operation",
+                    lease_claim=claim,
+                )
+            else:
+                transition_recovery_operation(
+                    session,
+                    operation,
+                    status=RecoveryStatus.OUTCOME_UNKNOWN,
+                    kind="stale-effect-outcome-unknown",
+                    summary="Execution authority expired after an external effect may have started",
+                    evidence={"previous_fence": previous_fence},
+                    failure_summary="External effect outcome requires reconciliation",
+                    lease_claim=claim,
+                )
+        if not verify_recovery_evidence_chain(session, operation.id):
+            raise RecoveryGuaranteeError("Recovery checkpoint chain verification failed")
+        if operation.status == RecoveryStatus.FAILED.value:
+            release_lease(session, claim)
+            session.commit()
+            raise RecoveryOperationStateConflict(operation.id, operation.status)
+        session.commit()
+        return DurableRecoveryOperation(
+            session_factory=session_factory,
+            operation_id=operation.id,
+            lease_claim=claim,
+            lease_ttl_seconds=lease_ttl_seconds,
+        )
+
+
+__all__ = [
+    "DurableRecoveryOperation",
+    "DurableRecoveryStart",
+    "RecoveryOperationBusy",
+    "RecoveryOperationStateConflict",
+    "begin_durable_recovery_operation",
+    "claim_durable_recovery_operation",
+]

src/govoplan_core/server/default_config.py

@@ -12,6 +12,7 @@ from govoplan_core.db.bootstrap import bootstrap_dev_data, create_all_tables
 from govoplan_core.db.session import get_database
 from govoplan_core.server.config import GovoplanServerConfig
 from govoplan_core.server.runtime_agent import RuntimeNodeAgent
+from govoplan_core.core.runtime_coordination import RuntimeIdentity, runtime_identity
 from govoplan_core.settings import Settings, settings
 
 
@@ -49,11 +50,19 @@ async def lifespan(app: FastAPI):
         if registry is not None
         else ()
     )
+    configured_identity = getattr(
+        app.state,
+        "govoplan_runtime_identity",
+        None,
+    )
     runtime_agent = RuntimeNodeAgent(
         settings=settings,
         software_version=app.version,
         module_ids=module_ids,
         metadata={"process": "api"},
+        identity=configured_identity
+        if isinstance(configured_identity, RuntimeIdentity)
+        else None,
     )
     await runtime_agent.start()
     app.state.govoplan_runtime_agent = runtime_agent
@@ -73,6 +82,16 @@ def register_health_details(
     active_settings = (
         config_settings if isinstance(config_settings, Settings) else settings
     )
+    module_ids = tuple(manifest.id for manifest in registry.manifests())
+    if not isinstance(
+        getattr(app.state, "govoplan_runtime_identity", None),
+        RuntimeIdentity,
+    ):
+        app.state.govoplan_runtime_identity = runtime_identity(
+            active_settings,
+            software_version=app.version,
+            module_ids=module_ids,
+        )
 
     @app.get("/health/details")
     def health_details(

src/govoplan_core/server/runtime_agent.py

@@ -17,6 +17,21 @@ from govoplan_core.db.session import get_database
 logger = logging.getLogger("govoplan.runtime")
 
 
+def application_runtime_identity(app: object) -> RuntimeIdentity:
+    """Return the registered identity used to fence request-owned effects."""
+
+    state = getattr(app, "state", None)
+    agent = getattr(state, "govoplan_runtime_agent", None)
+    identity = getattr(agent, "identity", None) or getattr(
+        state,
+        "govoplan_runtime_identity",
+        None,
+    )
+    if not isinstance(identity, RuntimeIdentity):
+        raise RuntimeError("The application runtime identity is not available")
+    return identity
+
+
 class RuntimeNodeAgent:
     """Register one process in the shared runtime directory and heartbeat it."""
 
@@ -30,9 +45,10 @@ class RuntimeNodeAgent:
         node_id: str | None = None,
         queues: tuple[str, ...] | None = None,
         metadata: dict[str, Any] | None = None,
+        identity: RuntimeIdentity | None = None,
     ) -> None:
         self.settings = settings
-        self.identity: RuntimeIdentity = runtime_identity(
+        self.identity: RuntimeIdentity = identity or runtime_identity(
             settings,
             software_version=software_version,
             module_ids=module_ids,
@@ -126,4 +142,4 @@ class RuntimeNodeAgent:
         return True
 
 
-__all__ = ["RuntimeNodeAgent"]
+__all__ = ["RuntimeNodeAgent", "application_runtime_identity"]

tests/test_api_smoke.py

@@ -44,6 +44,11 @@ from govoplan_core.db.migrations import alembic_config
 from govoplan_core.db.session import configure_database, set_database
 from govoplan_core.core.change_sequence import decode_sequence_watermark, prune_sequence_entries
 from govoplan_core.core.pagination import encode_keyset_cursor, keyset_query_fingerprint
+from govoplan_core.core.recovery import (
+    RecoveryOperation,
+    RecoveryStatus,
+    verify_recovery_evidence_chain,
+)
 from govoplan_core.tenancy.scope import create_scope_tables, scope_registry
 from govoplan_access.backend.permissions.catalog import permission_catalog as access_permission_catalog
 
@@ -101,6 +106,14 @@ class ApiSmokeTests(unittest.TestCase):
         payload = response.json()
         return {"Authorization": f"Bearer {payload['access_token']}"}, payload
 
+    def _stored_campaign_eml(self, job: object) -> bytes:
+        from govoplan_files.backend.storage.backends import get_storage_backend
+
+        self.assertIsNone(getattr(job, "eml_local_path", None))
+        storage_key = getattr(job, "eml_storage_key", None)
+        self.assertTrue(storage_key)
+        return get_storage_backend().get_bytes(str(storage_key))
+
     def _create_test_mail_profile(
         self,
         headers: dict[str, str],
@@ -3304,6 +3317,29 @@ class ApiSmokeTests(unittest.TestCase):
             json={"write_eml": False},
         )
         self.assertEqual(built.status_code, 200, built.text)
+        replayed_build = self.client.post(
+            f"/api/v1/campaigns/versions/{version_id}/build",
+            headers=headers,
+            json={"write_eml": False},
+        )
+        self.assertEqual(replayed_build.status_code, 200, replayed_build.text)
+        with SessionLocal() as session:
+            operations = (
+                session.query(RecoveryOperation)
+                .filter(
+                    RecoveryOperation.module_id == "campaigns",
+                    RecoveryOperation.resource_id == version_id,
+                )
+                .all()
+            )
+            self.assertEqual(1, len(operations))
+            self.assertEqual(
+                RecoveryStatus.SUCCEEDED.value,
+                operations[0].status,
+            )
+            self.assertTrue(
+                verify_recovery_evidence_chain(session, operations[0].id)
+            )
         self.assertEqual(built.json()["built_count"], 1)
 
         mocked = self.client.post(
@@ -3742,7 +3778,8 @@ class ApiSmokeTests(unittest.TestCase):
 
         with SessionLocal() as session:
             job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one()
-            self.assertTrue(job.eml_local_path)
+            self.assertIsNone(job.eml_local_path)
+            self.assertTrue(job.eml_storage_key)
             built_use = (
                 session.query(CampaignAttachmentUse)
                 .filter(
@@ -3979,7 +4016,9 @@ class ApiSmokeTests(unittest.TestCase):
             job = session.query(CampaignJob).filter(CampaignJob.campaign_id == campaign_id).one()
             self.assertEqual([item["email"] for item in job.resolved_recipients["from_all"]], ["local-from@example.org"])
             self.assertEqual([item["email"] for item in job.resolved_recipients["to"]], ["global-to@example.org", "local-to@example.org"])
-            message = BytesParser(policy=policy.default).parsebytes(Path(job.eml_local_path).read_bytes())
+            message = BytesParser(policy=policy.default).parsebytes(
+                self._stored_campaign_eml(job)
+            )
         self.assertIsNone(message["Sender"])
         self.assertEqual([address.addr_spec for address in message["From"].addresses], ["local-from@example.org"])
         self.assertEqual([address.addr_spec for address in message["To"].addresses], ["global-to@example.org", "local-to@example.org"])
@@ -4229,8 +4268,9 @@ class ApiSmokeTests(unittest.TestCase):
 
         with SessionLocal() as session:
             job = session.query(CampaignJob).filter(CampaignJob.campaign_id == campaign_id).one()
-            eml_path = Path(job.eml_local_path)
-            message = BytesParser(policy=policy.default).parsebytes(eml_path.read_bytes())
+            message = BytesParser(policy=policy.default).parsebytes(
+                self._stored_campaign_eml(job)
+            )
             uses = (
                 session.query(CampaignAttachmentUse)
                 .filter(
@@ -4337,8 +4377,7 @@ class ApiSmokeTests(unittest.TestCase):
 
         with SessionLocal() as session:
             job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one()
-            self.assertIsNotNone(job.eml_local_path)
-            generated_eml = Path(job.eml_local_path).read_bytes()
+            generated_eml = self._stored_campaign_eml(job)
 
         sent = self.client.post(
             f"/api/v1/campaigns/{campaign_id}/send-now",
@@ -4368,13 +4407,19 @@ class ApiSmokeTests(unittest.TestCase):
         )
 
         from govoplan_campaign.backend.db.models import CampaignJob
+        from govoplan_files.backend.storage.backends import get_storage_backend
         from govoplan_mail.backend.dev.mock_mailbox import list_records
 
         with SessionLocal() as session:
             job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one()
-            self.assertIsNotNone(job.eml_local_path)
-            eml_path = Path(job.eml_local_path)
-            eml_path.write_bytes(eml_path.read_bytes() + b"\r\nX-Tampered: true\r\n")
+            storage_key = job.eml_storage_key
+            generated_eml = self._stored_campaign_eml(job)
+        assert storage_key is not None
+        get_storage_backend().put_bytes(
+            storage_key,
+            generated_eml + b"\r\nX-Tampered: true\r\n",
+            content_type="message/rfc822",
+        )
 
         sent = self.client.post(
             f"/api/v1/campaigns/{campaign_id}/send-now",

tests/test_recovery_runtime.py

@@ -0,0 +1,182 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import sessionmaker
+import pytest
+
+from govoplan_core.core.recovery import (
+    RecoveryCheckpoint,
+    RecoveryGuaranteeError,
+    RecoveryMode,
+    RecoveryOperation,
+    RecoveryPlan,
+    RecoveryStatus,
+    verify_recovery_evidence_chain,
+)
+from govoplan_core.core.recovery_runtime import (
+    RecoveryOperationBusy,
+    RecoveryOperationStateConflict,
+    begin_durable_recovery_operation,
+    claim_durable_recovery_operation,
+)
+from govoplan_core.core.runtime_coordination import DistributedLease, RuntimeIdentity
+from govoplan_core.db.base import Base
+
+
+def _fixture():
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    Base.metadata.create_all(
+        engine,
+        tables=[
+            DistributedLease.__table__,
+            RecoveryOperation.__table__,
+            RecoveryCheckpoint.__table__,
+        ],
+    )
+    return engine, sessionmaker(bind=engine, expire_on_commit=False)
+
+
+def _identity(node: str, incarnation: str) -> RuntimeIdentity:
+    return RuntimeIdentity(
+        installation_id="installation-1",
+        node_id=node,
+        incarnation=incarnation,
+        role="worker",
+        software_version="test",
+        composition_hash="a" * 64,
+    )
+
+
+def _start(factory, identity, *, key: str = "build-1"):
+    return begin_durable_recovery_operation(
+        factory,
+        identity=identity,
+        module_id="campaigns",
+        operation_type="build-artifacts",
+        idempotency_key=key,
+        request={"version_id": "version-1", "write_eml": True},
+        recovery_plan=RecoveryPlan(
+            mode=RecoveryMode.COMPENSATION,
+            preconditions=("validated version is locked",),
+            compensation_steps=("delete build object prefix",),
+            verification_steps=("compare database and object manifests",),
+        ),
+        precondition_evidence={"validation_sha256": "b" * 64},
+        lease_resource_key="campaign:build:version-1",
+        resource_type="campaign_version",
+        resource_id="version-1",
+    )
+
+
+def test_durable_operation_commits_before_caller_effect_and_replays_success() -> None:
+    engine, factory = _fixture()
+    try:
+        started = _start(factory, _identity("worker-1", "incarnation-1"))
+        assert started.operation is not None
+        with factory() as session:
+            persisted = session.get(RecoveryOperation, started.operation_id)
+            assert persisted is not None
+            assert persisted.status == RecoveryStatus.RUNNING.value
+            assert persisted.checkpoint_count == 3
+
+        started.operation.checkpoint(
+            kind="object-prefix-reserved",
+            summary="Build object prefix reserved",
+            evidence={"prefix": "campaign-artifacts/build-1/"},
+        )
+        started.operation.succeed(
+            evidence={
+                "verified": True,
+                "checks": {"database_manifest": "matched", "object_manifest": "matched"},
+            }
+        )
+
+        replay = _start(factory, _identity("worker-2", "incarnation-2"))
+        assert replay.replayed is True
+        assert replay.operation is None
+        with factory() as session:
+            assert verify_recovery_evidence_chain(session, started.operation_id)
+    finally:
+        engine.dispose()
+
+
+def test_same_fence_cannot_start_duplicate_running_operation() -> None:
+    engine, factory = _fixture()
+    try:
+        identity = _identity("worker-1", "incarnation-1")
+        started = _start(factory, identity)
+        with pytest.raises(RecoveryOperationStateConflict, match="already running"):
+            _start(factory, identity)
+        assert started.operation is not None
+        started.operation.release_unresolved()
+    finally:
+        engine.dispose()
+
+
+def test_other_runtime_cannot_use_an_active_fence() -> None:
+    engine, factory = _fixture()
+    try:
+        started = _start(factory, _identity("worker-1", "incarnation-1"))
+        with pytest.raises(RecoveryOperationBusy):
+            _start(factory, _identity("worker-2", "incarnation-2"), key="build-2")
+        assert started.operation is not None
+        started.operation.release_unresolved()
+    finally:
+        engine.dispose()
+
+
+def test_expired_crash_fence_is_taken_over_as_recovery_required() -> None:
+    engine, factory = _fixture()
+    try:
+        started = _start(factory, _identity("worker-1", "incarnation-1"))
+        assert started.operation is not None
+        with factory() as session:
+            lease = session.execute(select(DistributedLease)).scalar_one()
+            lease.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1)
+            session.add(lease)
+            session.commit()
+
+        recovery = claim_durable_recovery_operation(
+            factory,
+            identity=_identity("worker-2", "incarnation-2"),
+            operation_id=started.operation_id,
+        )
+        with factory() as session:
+            operation = session.get(RecoveryOperation, started.operation_id)
+            assert operation is not None
+            assert operation.status == RecoveryStatus.RECOVERY_REQUIRED.value
+            assert operation.fencing_token == 2
+            assert verify_recovery_evidence_chain(session, operation.id)
+        recovery.compensate(
+            failure_summary="worker stopped during object publication",
+            failure_evidence={"object_prefix": "campaign-artifacts/build-1/"},
+            recovery_evidence={
+                "verified": True,
+                "checks": {"object_prefix_empty": True},
+            },
+        )
+    finally:
+        engine.dispose()
+
+
+def test_tampered_checkpoint_blocks_verified_success() -> None:
+    engine, factory = _fixture()
+    try:
+        started = _start(factory, _identity("worker-1", "incarnation-1"))
+        assert started.operation is not None
+        with factory() as session:
+            checkpoint = session.execute(
+                select(RecoveryCheckpoint).order_by(RecoveryCheckpoint.sequence)
+            ).scalars().first()
+            assert checkpoint is not None
+            checkpoint.summary = "tampered"
+            session.add(checkpoint)
+            session.commit()
+        with pytest.raises(RecoveryGuaranteeError, match="chain verification failed"):
+            started.operation.succeed(
+                evidence={"verified": True, "checks": {"objects": "matched"}}
+            )
+    finally:
+        engine.dispose()