Fence governed connector acquisitions

README.md

@@ -28,6 +28,16 @@ connector implementations or stores connector credentials.
 Database, REST/HTTP, directory, managed-file, and warehouse providers can
 implement the same origin contract without changing Datasources or Dataflow.
 
+Governed sanctions and feed snapshot acquisitions use Core recovery operations.
+The source revision/cursor, redacted dry-run decision, canonical request digest,
+and distributed lease are durable before network I/O. Immutable snapshot rows
+and the terminal recovery checkpoint commit atomically, and an
+`Idempotency-Key` replays the committed result without contacting the provider.
+Current acquisition transports are read-only. The exported external-mutation
+recovery contract requires stable idempotency, provider verification, and
+operator reconciliation, but no connector currently claims a production write
+or delete path.
+
 Development:
 
 ```bash

docs/CONNECTOR_SOURCE_LIFECYCLE.md

@@ -138,6 +138,26 @@ Every connector that writes to an external system needs a reconciliation story:
 - explicit requested, approved, dispatched, possibly-executed, confirmed, and
   reconciled/corrected effect states
 
+## Durable recovery operations
+
+Connectors declares two recovery classes. A read-only acquisition into an
+immutable snapshot is `atomic`: the source revision or conditional cursor,
+redacted dry-run decision, canonical request digest, and distributed
+tenant/provider lease are durable before the fetch. The acquired domain rows
+and terminal Core recovery checkpoint commit in one PostgreSQL transaction. A
+caller-supplied `Idempotency-Key` replays that committed result without a second
+provider request. A failed or stale transaction has no remote mutation and may
+be repeated only as a new deliberate acquisition.
+
+An external create, update, publish, or delete is `forward_recovery`. It must
+start through the connector mutation recovery contract with a stable
+idempotency key, SHA-256 request digest, source revision/cursor, and dry-run
+evidence. Definitive rejection is terminal. A timeout or lost acknowledgement
+after dispatch is `outcome_unknown` and blocks replay until the owning connector
+verifies provider state. The contract and conformance tests exist; no current
+connector advertises a production external mutation, so write/delete adoption
+remains explicitly planned rather than implied.
+
 ## Provider Declaration
 
 An executable connector type should publish machine-readable metadata for:

src/govoplan_connectors/backend/manifest.py

@@ -88,6 +88,11 @@ ARCHITECTURE = ModuleArchitectureDeclaration(
             reference="tests/test_sanctions_sources.py",
             summary="Exercises source acquisition health, checksums, retries, and immutable evidence.",
         ),
+        ModuleMaturityEvidence(
+            kind="recovery",
+            reference="tests/test_recovery.py",
+            summary="Proves atomic snapshot commits, idempotent replay, distributed fences, tamper rejection, and unknown external-effect handling.",
+        ),
         ModuleMaturityEvidence(
             kind="documentation",
             reference="docs/CONNECTOR_SOURCE_LIFECYCLE.md",
@@ -149,8 +154,13 @@ EXTERNAL_PROVIDERS = (
             freshness="Snapshot acquisition time and source timestamp are exposed.",
             health="Import validation and source-read failures are explicit.",
             max_read_items=1000,
+            idempotency="Feed imports accept a caller request key and replay the same committed immutable source without refetching.",
+            retry="Read-only acquisition may be retried only as a new deliberate request after a failed atomic operation.",
+            outcome_unknown="Provider reads do not mutate remote state; an uncertain database commit is resolved by the atomic recovery transaction.",
+            outcome_unknown_supported=False,
             evidence="Rows, schema, fingerprint, source metadata, and acquisition provenance remain linked.",
             correction="Import a replacement snapshot; retain the prior snapshot as evidence.",
+            rollback="Snapshot rows and the terminal recovery checkpoint commit or roll back together.",
             reconciliation="Compare source and snapshot fingerprints before selecting a new current state.",
             outage="Existing snapshots remain available and visibly stale; no live-source claim is made.",
             classifications=("internal", "confidential", "restricted"),
@@ -187,12 +197,18 @@ EXTERNAL_PROVIDERS = (
         ),
         behavior=ProviderBehaviorDeclaration(
             revision_tokens="Provider source version, ETag, Last-Modified, and SHA-256 digest are retained when available.",
-            concurrency="Refreshes use conditional source requests and create immutable snapshots.",
+            concurrency="Refreshes use conditional source requests, a distributed per-tenant/provider fence, and immutable snapshots.",
             freshness="Latest successful acquisition, source timestamp, and stale health are reported.",
             health="Transport, parsing, source-change, and malformed-source states are explicit.",
             max_read_items=5000,
+            idempotency="A caller request key identifies one acquisition run and replays its committed result without contacting the source again.",
+            retry="Bounded HTTP retries are safe because acquisition is read-only; failed runs require a new deliberate request key.",
+            timeout_seconds=30,
+            outcome_unknown="The external operation is read-only; snapshot rows and recovery evidence commit atomically.",
+            outcome_unknown_supported=False,
             evidence="Raw source bytes, checksum, acquisition run, parser result, and normalized entry count are linked.",
             correction="A corrected source creates a new immutable snapshot and acquisition run.",
+            rollback="A failed database transaction leaves no snapshot and the stale atomic fence resolves as failed.",
             reconciliation="Compare source version and digest, then preserve both prior and corrected evidence.",
             outage="The latest accepted snapshot stays usable with stale/unavailable source health.",
             classifications=("public", "internal"),
@@ -494,7 +510,11 @@ manifest = ModuleManifest(
                 "the official United Nations Security Council consolidated "
                 "XML source. Each fetch records conditional transport "
                 "evidence, bounded retries, health state, source metadata, "
-                "raw evidence, and a SHA-256 checksum. Risk Compliance owns "
+                "raw evidence, and a SHA-256 checksum. Refreshes acquire a "
+                "distributed recovery fence before provider I/O; the immutable "
+                "snapshot and terminal recovery checkpoint then commit in one "
+                "transaction. A repeated request key returns the same result. "
+                "Risk Compliance owns "
                 "normalization, matching, legal review, and dispositions."
             ),
             layer="available",

src/govoplan_connectors/backend/recovery.py

@@ -0,0 +1,381 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+from typing import Any
+from uuid import NAMESPACE_URL, uuid4, uuid5
+
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.core.recovery import (
+    RecoveryGuaranteeError,
+    RecoveryMode,
+    RecoveryPlan,
+    RecoveryStatus,
+)
+from govoplan_core.core.recovery_runtime import (
+    DurableRecoveryOperation,
+    RecoveryOperationBusy,
+    RecoveryOperationStateConflict,
+    begin_durable_recovery_operation,
+)
+from govoplan_core.core.runtime_coordination import process_runtime_identity
+
+
+class ConnectorRecoveryError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class ConnectorRecoveryDeclaration:
+    operation_type: str
+    mode: RecoveryMode
+    provider_mutation: bool
+    idempotency: str
+    verification: tuple[str, ...]
+    recovery: tuple[str, ...]
+    implemented: bool
+
+
+CONNECTOR_RECOVERY_OPERATIONS = (
+    ConnectorRecoveryDeclaration(
+        operation_type="read-snapshot",
+        mode=RecoveryMode.ATOMIC,
+        provider_mutation=False,
+        idempotency=(
+            "Caller-supplied request keys replay a committed immutable snapshot; "
+            "otherwise each deliberate acquisition receives a generated key."
+        ),
+        verification=(
+            "provider revision or conditional cursor is recorded before fetch",
+            "domain snapshot and terminal recovery checkpoint commit together",
+            "stored bytes and provider evidence are checksum verified",
+        ),
+        recovery=(
+            "a stale running transaction is failed after its database transaction rolls back",
+            "a new deliberate acquisition may then use a new request key",
+        ),
+        implemented=True,
+    ),
+    ConnectorRecoveryDeclaration(
+        operation_type="external-mutation",
+        mode=RecoveryMode.FORWARD_RECOVERY,
+        provider_mutation=True,
+        idempotency="A stable caller key and canonical request digest are mandatory.",
+        verification=(
+            "record the remote revision and bounded provider result",
+            "verify the provider state before reporting success",
+        ),
+        recovery=(
+            "unknown outcomes remain unresolved until provider-backed reconciliation",
+            "never retry the same remote effect solely to reconstruct local state",
+        ),
+        implemented=False,
+    ),
+)
+
+
+def connector_session_factory(session: Session) -> sessionmaker[Session]:
+    bind = session.get_bind()
+    if bind is None:
+        raise ConnectorRecoveryError("Connector recovery requires a bound database session")
+    return sessionmaker(bind=bind, expire_on_commit=False)
+
+
+def _digest(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def _clean_key(value: str | None) -> str:
+    clean = str(value or "").strip()
+    if clean and len(clean) > 500:
+        raise ConnectorRecoveryError("Connector idempotency keys are limited to 500 characters")
+    return clean or str(uuid4())
+
+
+def _stable_resource_id(
+    *,
+    tenant_id: str,
+    provider_id: str,
+    operation_type: str,
+    request_key: str,
+) -> str:
+    return str(
+        uuid5(
+            NAMESPACE_URL,
+            f"govoplan:{tenant_id}:{provider_id}:{operation_type}:{request_key}",
+        )
+    )
+
+
+@dataclass(slots=True)
+class ConnectorReadSnapshotRecovery:
+    operation: DurableRecoveryOperation | None
+    operation_id: str
+    request_key: str
+    resource_id: str
+    replayed: bool
+
+    def commit_success(self, session: Session, *, evidence: dict[str, Any]) -> None:
+        if self.operation is None:
+            raise ConnectorRecoveryError("A replayed connector read cannot be committed again")
+        try:
+            self.operation.commit_atomic_success(session, evidence=evidence)
+        except Exception as exc:
+            raise ConnectorRecoveryError(
+                "The connector snapshot and recovery evidence did not commit atomically"
+            ) from exc
+
+    def commit_failure(
+        self,
+        session: Session,
+        *,
+        summary: str,
+        evidence: dict[str, Any],
+    ) -> None:
+        if self.operation is None:
+            raise ConnectorRecoveryError("A replayed connector read cannot be failed again")
+        try:
+            self.operation.commit_atomic_failure(
+                session,
+                summary=summary,
+                evidence=evidence,
+            )
+        except Exception as exc:
+            raise ConnectorRecoveryError(
+                "The connector failure evidence did not commit atomically"
+            ) from exc
+
+    def fail_without_projection(
+        self,
+        *,
+        summary: str,
+        code: str,
+    ) -> None:
+        if self.operation is None:
+            return
+        self.operation.fail(
+            summary=summary,
+            evidence={
+                "verified": True,
+                "checks": {
+                    "provider_mutation": False,
+                    "projection_committed": False,
+                    "failure_code": code,
+                },
+            },
+        )
+
+
+def begin_connector_read_snapshot(
+    session: Session,
+    *,
+    tenant_id: str,
+    provider_id: str,
+    idempotency_key: str | None,
+    source_revision: str | None,
+    cursor: str | None,
+    dry_run_evidence: dict[str, Any],
+    request_metadata: dict[str, Any] | None = None,
+    resource_type: str = "connector_sync_run",
+) -> ConnectorReadSnapshotRecovery:
+    request_key = _clean_key(idempotency_key)
+    resource_id = _stable_resource_id(
+        tenant_id=tenant_id,
+        provider_id=provider_id,
+        operation_type="read-snapshot",
+        request_key=request_key,
+    )
+    request = {
+        "tenant_id": tenant_id,
+        "provider_id": provider_id,
+        "dry_run": dry_run_evidence,
+        "request_key_sha256": _digest(request_key),
+        **dict(request_metadata or {}),
+    }
+    try:
+        started = begin_durable_recovery_operation(
+            connector_session_factory(session),
+            identity=process_runtime_identity(),
+            module_id="connectors",
+            operation_type="read-snapshot",
+            idempotency_key=f"connector-read:{_digest(f'{tenant_id}:{provider_id}:{request_key}')}",
+            request=request,
+            recovery_plan=RecoveryPlan(
+                mode=RecoveryMode.ATOMIC,
+                preconditions=(
+                    "the actor is authorized for the connector source",
+                    "the provider request is read-only",
+                    "the source revision, cursor, and dry-run decision are durable",
+                ),
+                verification_steps=(
+                    "validate the bounded provider response and source revision",
+                    "commit the immutable snapshot and terminal checkpoint atomically",
+                    "compare the stored content digest with the acquired bytes",
+                ),
+            ),
+            precondition_evidence={
+                "provider_id": provider_id,
+                "source_revision": source_revision,
+                "cursor_sha256": _digest(cursor) if cursor else None,
+                "dry_run": dry_run_evidence,
+                "provider_mutation": False,
+            },
+            lease_resource_key=f"connectors:read:{tenant_id}:{_digest(provider_id)[:40]}",
+            lease_ttl_seconds=15 * 60,
+            resource_type=resource_type,
+            resource_id=resource_id,
+            metadata={
+                "resources": ["postgresql", "external-provider"],
+                "provider_mutation": False,
+                "recovery_declaration": "read-snapshot",
+            },
+        )
+    except RecoveryOperationBusy as exc:
+        raise ConnectorRecoveryError(
+            "Another runtime is already acquiring this connector source"
+        ) from exc
+    except RecoveryOperationStateConflict as exc:
+        raise ConnectorRecoveryError(
+            "This connector request is active or unresolved; reconcile it before retrying"
+        ) from exc
+    except (RecoveryGuaranteeError, RuntimeError) as exc:
+        raise ConnectorRecoveryError(
+            "The connector recovery ledger is unavailable; the provider was not contacted"
+        ) from exc
+    return ConnectorReadSnapshotRecovery(
+        operation=started.operation,
+        operation_id=started.operation_id,
+        request_key=request_key,
+        resource_id=resource_id,
+        replayed=started.replayed,
+    )
+
+
+@dataclass(slots=True)
+class ConnectorExternalMutationRecovery:
+    operation: DurableRecoveryOperation | None
+    operation_id: str
+    replayed: bool
+
+    def succeed(self, *, provider_evidence: dict[str, Any]) -> None:
+        if self.operation is not None:
+            self.operation.succeed(evidence=provider_evidence)
+
+    def reject(self, *, summary: str, provider_code: str) -> None:
+        if self.operation is not None:
+            self.operation.reject(
+                summary=summary,
+                evidence={
+                    "verified": True,
+                    "checks": {"provider_rejection": provider_code},
+                },
+            )
+
+    def outcome_unknown(self, *, summary: str, provider_code: str) -> None:
+        if self.operation is not None:
+            self.operation.unresolved(
+                status=RecoveryStatus.OUTCOME_UNKNOWN,
+                summary=summary,
+                evidence={"effect_started": True, "provider_code": provider_code},
+                failure_summary="Inspect provider state before any retry",
+            )
+
+
+def begin_connector_external_mutation(
+    session: Session,
+    *,
+    tenant_id: str,
+    provider_id: str,
+    idempotency_key: str,
+    request_sha256: str,
+    source_revision: str | None,
+    cursor: str | None,
+    dry_run_evidence: dict[str, Any],
+    resource_type: str,
+    resource_id: str,
+) -> ConnectorExternalMutationRecovery:
+    if not str(idempotency_key or "").strip():
+        raise ConnectorRecoveryError("External connector mutations require an idempotency key")
+    request_key = _clean_key(idempotency_key)
+    if len(request_sha256) != 64 or any(
+        character not in "0123456789abcdefABCDEF" for character in request_sha256
+    ):
+        raise ConnectorRecoveryError("External connector mutations require a SHA-256 request digest")
+    try:
+        started = begin_durable_recovery_operation(
+            connector_session_factory(session),
+            identity=process_runtime_identity(),
+            module_id="connectors",
+            operation_type="external-mutation",
+            idempotency_key=f"connector-write:{_digest(f'{tenant_id}:{provider_id}:{request_key}')}",
+            request={
+                "tenant_id": tenant_id,
+                "provider_id": provider_id,
+                "request_sha256": request_sha256,
+                "source_revision": source_revision,
+                "cursor_sha256": _digest(cursor) if cursor else None,
+                "dry_run": dry_run_evidence,
+            },
+            recovery_plan=RecoveryPlan(
+                mode=RecoveryMode.FORWARD_RECOVERY,
+                preconditions=(
+                    "the actor and effective connector policy authorize the mutation",
+                    "a stable idempotency key and canonical request digest are present",
+                    "the dry-run and source revision evidence are durable",
+                ),
+                forward_recovery_steps=(
+                    "inspect provider state without repeating the mutation",
+                    "record whether the provider accepted the requested revision",
+                    "retry only under a new deliberate key when absence is proven",
+                ),
+                verification_steps=(
+                    "compare provider identity and revision with the canonical request",
+                    "verify the consuming domain state independently",
+                ),
+            ),
+            precondition_evidence={
+                "request_sha256": request_sha256,
+                "source_revision": source_revision,
+                "cursor_sha256": _digest(cursor) if cursor else None,
+                "dry_run": dry_run_evidence,
+                "provider_mutation": True,
+            },
+            lease_resource_key=(
+                f"connectors:write:{tenant_id}:{_digest(provider_id)[:24]}:"
+                f"{_digest(resource_id)[:24]}"
+            ),
+            lease_ttl_seconds=15 * 60,
+            resource_type=resource_type,
+            resource_id=resource_id,
+            metadata={
+                "resources": ["postgresql", "queue", "external-provider"],
+                "provider_mutation": True,
+                "recovery_declaration": "external-mutation",
+            },
+        )
+    except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc:
+        raise ConnectorRecoveryError(
+            "This external connector effect is active or unresolved"
+        ) from exc
+    except (RecoveryGuaranteeError, RuntimeError) as exc:
+        raise ConnectorRecoveryError(
+            "The connector recovery ledger is unavailable; no external mutation started"
+        ) from exc
+    return ConnectorExternalMutationRecovery(
+        operation=started.operation,
+        operation_id=started.operation_id,
+        replayed=started.replayed,
+    )
+
+
+__all__ = [
+    "CONNECTOR_RECOVERY_OPERATIONS",
+    "ConnectorExternalMutationRecovery",
+    "ConnectorReadSnapshotRecovery",
+    "ConnectorRecoveryDeclaration",
+    "ConnectorRecoveryError",
+    "begin_connector_external_mutation",
+    "begin_connector_read_snapshot",
+    "connector_session_factory",
+]

src/govoplan_connectors/backend/router.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 from dataclasses import asdict
+import hashlib
+from typing import Annotated
 
-from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response, status
 from sqlalchemy.orm import Session
 
 from govoplan_core.audit.logging import audit_event
@@ -42,6 +44,10 @@ from govoplan_connectors.backend.schemas import (
     TabularSourceResponse,
 )
 from govoplan_connectors.backend.feeds import ConnectorFeedProvider, feed_rows
+from govoplan_connectors.backend.recovery import (
+    ConnectorRecoveryError,
+    begin_connector_read_snapshot,
+)
 from govoplan_connectors.backend.sanctions_sources import (
     SANCTIONS_READ_SCOPE,
     SANCTIONS_REFRESH_SCOPE,
@@ -108,6 +114,18 @@ def _feed_http_error(exc: FeedCapabilityError) -> HTTPException:
     )
 
 
+def _recovery_http_error(exc: ConnectorRecoveryError) -> HTTPException:
+    detail = str(exc)
+    return HTTPException(
+        status_code=(
+            status.HTTP_409_CONFLICT
+            if "already" in detail.casefold() or "active" in detail.casefold()
+            else status.HTTP_503_SERVICE_UNAVAILABLE
+        ),
+        detail=detail,
+    )
+
+
 @router.post("/feeds/preview", response_model=FeedDocumentResponse)
 def api_preview_feed(
     payload: FeedAcquireRequest,
@@ -133,8 +151,45 @@ def api_import_feed_snapshot(
     payload: FeedImportRequest,
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
+    idempotency_key: Annotated[
+        str | None,
+        Header(alias="Idempotency-Key", max_length=500),
+    ] = None,
 ) -> TabularSourceResponse:
     _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
+    try:
+        recovery = begin_connector_read_snapshot(
+            session,
+            tenant_id=principal.tenant_id,
+            provider_id="connectors.feed_snapshot",
+            idempotency_key=idempotency_key,
+            source_revision=None,
+            cursor=None,
+            dry_run_evidence={
+                "performed": False,
+                "reason": "read-only acquisition into an immutable snapshot",
+            },
+            request_metadata={
+                "source_url_sha256": hashlib.sha256(
+                    payload.url.encode("utf-8")
+                ).hexdigest(),
+                "source_name": payload.source_name,
+                "max_entries": payload.max_entries,
+            },
+            resource_type="connector_tabular_source",
+        )
+    except ConnectorRecoveryError as exc:
+        raise _recovery_http_error(exc) from exc
+    if recovery.replayed:
+        try:
+            source = provider.get_source(
+                session,
+                principal,
+                source_ref=f"snapshot:{recovery.resource_id}",
+            )
+        except TabularSourceError as exc:
+            raise _http_error(exc) from exc
+        return _source_response(source)
     try:
         document = feed_transport.fetch(
             payload.url,
@@ -170,8 +225,14 @@ def api_import_feed_snapshot(
                     },
                 },
             ),
+            source_id=recovery.resource_id,
         )
     except (FeedCapabilityError, TabularSourceError) as exc:
+        session.rollback()
+        recovery.fail_without_projection(
+            summary="The read-only feed import failed before a snapshot committed",
+            code=exc.__class__.__name__,
+        )
         if isinstance(exc, FeedCapabilityError):
             raise _feed_http_error(exc) from exc
         raise _http_error(exc) from exc
@@ -190,7 +251,22 @@ def api_import_feed_snapshot(
             "row_count": source.row_count,
         },
     )
-    session.commit()
+    try:
+        recovery.commit_success(
+            session,
+            evidence={
+                "verified": True,
+                "checks": {
+                    "snapshot_ref": source.ref,
+                    "snapshot_fingerprint": source.fingerprint,
+                    "feed_sha256": document.sha256,
+                    "row_count": source.row_count,
+                    "provider_mutation": False,
+                },
+            },
+        )
+    except ConnectorRecoveryError as exc:
+        raise _recovery_http_error(exc) from exc
     return _source_response(source)
 
 
@@ -390,6 +466,10 @@ def api_refresh_sanctions_source(
     provider_id: str,
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
+    idempotency_key: Annotated[
+        str | None,
+        Header(alias="Idempotency-Key", max_length=500),
+    ] = None,
 ) -> SanctionsRefreshResponse:
     _require_any_scope(
         principal,
@@ -401,7 +481,10 @@ def api_refresh_sanctions_source(
             session,
             principal,
             provider_id=provider_id,
+            idempotency_key=idempotency_key,
         )
+    except ConnectorRecoveryError as exc:
+        raise _recovery_http_error(exc) from exc
     except SanctionsSourceError as exc:
         raise _sanctions_http_error(exc) from exc
     audit_event(

src/govoplan_connectors/backend/sanctions_sources.py

@@ -26,6 +26,11 @@ from govoplan_connectors.backend.db.models import (
     ConnectorSanctionsAcquisitionRun,
     ConnectorSanctionsSnapshot,
 )
+from govoplan_connectors.backend.recovery import (
+    ConnectorRecoveryError,
+    begin_connector_read_snapshot,
+    connector_session_factory,
+)
 from govoplan_core.auth import ApiPrincipal, has_scope
 from govoplan_core.core.sanctions import (
     SanctionsSnapshotPayload,
@@ -320,8 +325,9 @@ class SqlSanctionsSnapshotProvider:
         principal: object,
         *,
         provider_id: str,
+        idempotency_key: str | None = None,
     ) -> SanctionsAcquisitionResult:
-        db, api_principal = _context(
+        caller_db, api_principal = _context(
             session,
             principal,
             SANCTIONS_REFRESH_SCOPE,
@@ -331,23 +337,101 @@ class SqlSanctionsSnapshotProvider:
             raise SanctionsSourceNotFoundError(
                 "Sanctions source provider is not available."
             )
-        now = utcnow()
+        factory = connector_session_factory(caller_db)
-        latest = db.scalar(
+        with factory() as preflight_db:
-            select(ConnectorSanctionsSnapshot)
+            latest = _latest_snapshot(
-            .where(
+                preflight_db,
-                ConnectorSanctionsSnapshot.tenant_id
+                tenant_id=api_principal.tenant_id,
-                == api_principal.tenant_id,
+                provider_id=provider_id,
-                ConnectorSanctionsSnapshot.provider_id
-                == provider_id,
-            )
-            .order_by(
-                ConnectorSanctionsSnapshot.acquired_at.desc(),
-                ConnectorSanctionsSnapshot.id.desc(),
-            )
-            .limit(1)
             )
         conditional_headers = _conditional_headers(latest)
+        source_revision = latest.source_version if latest is not None else None
+        cursor = "\n".join(
+            f"{key.casefold()}:{value}"
+            for key, value in sorted(conditional_headers.items())
+        ) or None
+        recovery = begin_connector_read_snapshot(
+            caller_db,
+            tenant_id=api_principal.tenant_id,
+            provider_id=provider_id,
+            idempotency_key=idempotency_key,
+            source_revision=source_revision,
+            cursor=cursor,
+            dry_run_evidence={
+                "performed": False,
+                "reason": "read-only immutable source acquisition",
+            },
+            request_metadata={
+                "source_id": definition.source_id,
+                "subject_data_transmitted": False,
+            },
+        )
+        if recovery.replayed:
+            with factory() as replay_db:
+                replayed = _acquisition_result(
+                    replay_db,
+                    run_id=recovery.resource_id,
+                    tenant_id=api_principal.tenant_id,
+                )
+            if replayed is None:
+                raise ConnectorRecoveryError(
+                    "The completed connector acquisition has no matching durable result"
+                )
+            return replayed
+
+        with factory() as work_db:
+            result = self._refresh_source_transaction(
+                work_db,
+                api_principal,
+                definition=definition,
+                provider_id=provider_id,
+                run_id=recovery.resource_id,
+                latest=latest,
+                conditional_headers=conditional_headers,
+                recovery_operation_id=recovery.operation_id,
+                request_key_sha256=hashlib.sha256(
+                    recovery.request_key.encode("utf-8")
+                ).hexdigest(),
+            )
+            evidence = _acquisition_recovery_evidence(
+                work_db,
+                result=result,
+                tenant_id=api_principal.tenant_id,
+            )
+            try:
+                if result.status in {"succeeded", "not_modified"}:
+                    recovery.commit_success(work_db, evidence=evidence)
+                else:
+                    recovery.commit_failure(
+                        work_db,
+                        summary=(
+                            "The read-only connector acquisition failed without "
+                            "mutating the provider"
+                        ),
+                        evidence=evidence,
+                    )
+            except Exception as exc:
+                raise ConnectorRecoveryError(
+                    "Connector acquisition state could not be committed with its recovery evidence"
+                ) from exc
+        return result
+
+    def _refresh_source_transaction(
+        self,
+        db: Session,
+        api_principal: ApiPrincipal,
+        *,
+        definition: SanctionsSourceDefinition,
+        provider_id: str,
+        run_id: str,
+        latest: ConnectorSanctionsSnapshot | None,
+        conditional_headers: Mapping[str, str],
+        recovery_operation_id: str,
+        request_key_sha256: str,
+    ) -> SanctionsAcquisitionResult:
+        now = utcnow()
         run = ConnectorSanctionsAcquisitionRun(
+            id=run_id,
             tenant_id=api_principal.tenant_id,
             provider_id=provider_id,
             source_id=definition.source_id,
@@ -363,6 +447,8 @@ class SqlSanctionsSnapshotProvider:
                     conditional_headers
                 ),
                 "subject_data_transmitted": False,
+                "recovery_operation_id": recovery_operation_id,
+                "request_key_sha256": request_key_sha256,
             },
             response_evidence={},
             started_at=now,
@@ -727,6 +813,95 @@ def _conditional_headers(
     return result
 
 
+def _latest_snapshot(
+    session: Session,
+    *,
+    tenant_id: str,
+    provider_id: str,
+) -> ConnectorSanctionsSnapshot | None:
+    return session.scalar(
+        select(ConnectorSanctionsSnapshot)
+        .where(
+            ConnectorSanctionsSnapshot.tenant_id == tenant_id,
+            ConnectorSanctionsSnapshot.provider_id == provider_id,
+        )
+        .order_by(
+            ConnectorSanctionsSnapshot.acquired_at.desc(),
+            ConnectorSanctionsSnapshot.id.desc(),
+        )
+        .limit(1)
+    )
+
+
+def _acquisition_result(
+    session: Session,
+    *,
+    run_id: str,
+    tenant_id: str,
+) -> SanctionsAcquisitionResult | None:
+    run = session.scalar(
+        select(ConnectorSanctionsAcquisitionRun).where(
+            ConnectorSanctionsAcquisitionRun.id == run_id,
+            ConnectorSanctionsAcquisitionRun.tenant_id == tenant_id,
+        )
+    )
+    if run is None:
+        return None
+    snapshot = (
+        session.get(ConnectorSanctionsSnapshot, run.snapshot_id)
+        if run.snapshot_id
+        else None
+    )
+    return SanctionsAcquisitionResult(
+        run_id=run.id,
+        provider_id=run.provider_id,
+        status=run.status,
+        snapshot=_snapshot_dto(snapshot) if snapshot is not None else None,
+        error=run.error,
+    )
+
+
+def _acquisition_recovery_evidence(
+    session: Session,
+    *,
+    result: SanctionsAcquisitionResult,
+    tenant_id: str,
+) -> dict[str, object]:
+    run = session.get(ConnectorSanctionsAcquisitionRun, result.run_id)
+    snapshot = (
+        session.get(ConnectorSanctionsSnapshot, run.snapshot_id)
+        if run is not None and run.snapshot_id
+        else None
+    )
+    checksum_matches = bool(
+        snapshot is None
+        or hashlib.sha256(snapshot.raw_content).hexdigest() == snapshot.sha256
+    )
+    successful = result.status in {"succeeded", "not_modified"}
+    verified = bool(
+        run is not None
+        and run.tenant_id == tenant_id
+        and run.status == result.status
+        and run.finished_at is not None
+        and checksum_matches
+        and (not successful or snapshot is not None)
+    )
+    return {
+        "verified": verified,
+        "checks": {
+            "run_present": run is not None,
+            "run_status": result.status,
+            "run_finished": bool(run is not None and run.finished_at is not None),
+            "snapshot_present": snapshot is not None,
+            "snapshot_checksum_matches": checksum_matches,
+            "provider_mutation": False,
+        },
+        "run_id": result.run_id,
+        "snapshot_sha256": snapshot.sha256 if snapshot is not None else None,
+        "source_version": snapshot.source_version if snapshot is not None else None,
+    }
+
+
 def _snapshot_record(
     session: Session,
     *,

src/govoplan_connectors/backend/tabular_sources.py

@@ -130,6 +130,7 @@ class SqlTabularSourceProvider:
         principal: object,
         *,
         snapshot: TabularSnapshotInput,
+        source_id: str | None = None,
     ) -> TabularSource:
         db, api_principal = _context(session, principal, WRITE_SCOPE)
         name = snapshot.name.strip()
@@ -178,6 +179,8 @@ class SqlTabularSourceProvider:
             created_by=actor_id,
             updated_by=actor_id,
         )
+        if source_id:
+            item.id = source_id
         db.add(item)
         db.flush()
         return _source_dto(item)
@@ -296,6 +299,7 @@ def _context(
     if required_scope == READ_SCOPE:
         accepted_scopes.update(
             {
+                WRITE_SCOPE,
                 "datasources:catalogue:read",
                 "datasources:source:write",
                 "datasources:source:admin",

tests/test_recovery.py

@@ -0,0 +1,263 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+import unittest
+from unittest.mock import patch
+
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session
+
+from govoplan_connectors.backend.db.models import ConnectorTabularSource
+from govoplan_connectors.backend.feeds import ConnectorFeedProvider
+from govoplan_connectors.backend.recovery import (
+    CONNECTOR_RECOVERY_OPERATIONS,
+    ConnectorRecoveryError,
+    begin_connector_external_mutation,
+    begin_connector_read_snapshot,
+)
+from govoplan_connectors.backend.router import api_import_feed_snapshot
+from govoplan_connectors.backend.schemas import FeedImportRequest
+from govoplan_connectors.backend.tabular_sources import WRITE_SCOPE
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.recovery import (
+    RecoveryCheckpoint,
+    RecoveryOperation,
+    RecoveryStatus,
+)
+from govoplan_core.core.recovery_runtime import (
+    RecoveryOperationStateConflict,
+    claim_durable_recovery_operation,
+)
+from govoplan_core.core.tabular_sources import TabularSnapshotInput
+from govoplan_core.core.runtime_coordination import (
+    DistributedLease,
+    RuntimeIdentity,
+    bind_process_runtime_identity,
+)
+from govoplan_core.db.base import Base
+from govoplan_connectors.backend.tabular_sources import SqlTabularSourceProvider
+
+
+RSS = b"""<?xml version="1.0"?>
+<rss version="2.0"><channel><title>Updates</title>
+<link>https://example.test/</link><description>Updates</description>
+<item><guid>1</guid><title>One</title></item></channel></rss>"""
+
+
+def _identity(node: str, incarnation: str) -> RuntimeIdentity:
+    return RuntimeIdentity(
+        installation_id="connector-recovery-tests",
+        node_id=node,
+        incarnation=incarnation,
+        role="worker",
+        software_version="test",
+        composition_hash="a" * 64,
+    )
+
+
+def _principal() -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            tenant_id="tenant-1",
+            scopes=frozenset({WRITE_SCOPE}),
+        ),
+        account=object(),
+        user=object(),
+    )
+
+
+class ConnectorRecoveryTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        Base.metadata.create_all(
+            self.engine,
+            tables=(
+                DistributedLease.__table__,
+                RecoveryOperation.__table__,
+                RecoveryCheckpoint.__table__,
+                ConnectorTabularSource.__table__,
+            ),
+        )
+        self.session = Session(self.engine, expire_on_commit=False)
+        bind_process_runtime_identity(_identity("node-1", "incarnation-1"))
+
+    def tearDown(self) -> None:
+        bind_process_runtime_identity(None)
+        self.session.close()
+        self.engine.dispose()
+
+    def test_feed_snapshot_and_recovery_checkpoint_commit_atomically_and_replay(self) -> None:
+        document = ConnectorFeedProvider().parse(
+            RSS,
+            source_url="https://example.test/feed.xml",
+        )
+        payload = FeedImportRequest(
+            url="https://example.test/feed.xml",
+            name="Updates",
+            source_name="updates",
+        )
+        with (
+            patch(
+                "govoplan_connectors.backend.router.feed_transport.fetch",
+                return_value=document,
+            ) as fetch,
+            patch("govoplan_connectors.backend.router.audit_event"),
+        ):
+            first = api_import_feed_snapshot(
+                payload,
+                session=self.session,
+                principal=_principal(),
+                idempotency_key="feed-import-1",
+            )
+            replay = api_import_feed_snapshot(
+                payload,
+                session=self.session,
+                principal=_principal(),
+                idempotency_key="feed-import-1",
+            )
+
+        self.assertEqual(first.ref, replay.ref)
+        fetch.assert_called_once()
+        operation = self.session.scalar(select(RecoveryOperation))
+        assert operation is not None
+        self.assertEqual(RecoveryStatus.SUCCEEDED.value, operation.status)
+        self.assertEqual(
+            first.ref.removeprefix("snapshot:"),
+            operation.resource_id,
+        )
+
+    def test_recovery_metadata_distinguishes_reads_from_external_mutations(self) -> None:
+        declarations = {
+            item.operation_type: item for item in CONNECTOR_RECOVERY_OPERATIONS
+        }
+
+        self.assertFalse(declarations["read-snapshot"].provider_mutation)
+        self.assertTrue(declarations["read-snapshot"].implemented)
+        self.assertTrue(declarations["external-mutation"].provider_mutation)
+        self.assertFalse(declarations["external-mutation"].implemented)
+
+    def test_stale_atomic_connector_fence_fails_without_claiming_an_effect(self) -> None:
+        recovery = begin_connector_read_snapshot(
+            self.session,
+            tenant_id="tenant-1",
+            provider_id="provider-1",
+            idempotency_key="read-1",
+            source_revision="revision-1",
+            cursor="cursor-1",
+            dry_run_evidence={"performed": True, "approved": True},
+        )
+        lease = self.session.scalar(select(DistributedLease))
+        assert lease is not None
+        lease.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1)
+        self.session.commit()
+        bind_process_runtime_identity(_identity("node-2", "incarnation-2"))
+
+        with self.assertRaises(RecoveryOperationStateConflict):
+            claim_durable_recovery_operation(
+                recovery.operation.session_factory,
+                identity=_identity("node-2", "incarnation-2"),
+                operation_id=recovery.operation_id,
+            )
+
+        operation = self.session.get(RecoveryOperation, recovery.operation_id)
+        self.session.refresh(operation)
+        self.assertEqual(RecoveryStatus.FAILED.value, operation.status)
+
+    def test_external_mutation_unknown_outcome_blocks_blind_retry(self) -> None:
+        kwargs = {
+            "tenant_id": "tenant-1",
+            "provider_id": "provider-1",
+            "idempotency_key": "publish-1",
+            "request_sha256": "b" * 64,
+            "source_revision": "revision-1",
+            "cursor": None,
+            "dry_run_evidence": {"performed": True, "approved": True},
+            "resource_type": "external_record",
+            "resource_id": "record-1",
+        }
+        recovery = begin_connector_external_mutation(self.session, **kwargs)
+        recovery.outcome_unknown(
+            summary="The provider connection closed after dispatch",
+            provider_code="connection_closed",
+        )
+
+        with self.assertRaises(ConnectorRecoveryError):
+            begin_connector_external_mutation(self.session, **kwargs)
+        operation = self.session.get(RecoveryOperation, recovery.operation_id)
+        self.session.refresh(operation)
+        self.assertEqual(RecoveryStatus.OUTCOME_UNKNOWN.value, operation.status)
+
+    def test_tampered_chain_rolls_back_the_atomic_connector_projection(self) -> None:
+        recovery = begin_connector_read_snapshot(
+            self.session,
+            tenant_id="tenant-1",
+            provider_id="provider-1",
+            idempotency_key="tampered-read",
+            source_revision=None,
+            cursor=None,
+            dry_run_evidence={"performed": False, "reason": "read-only"},
+        )
+        checkpoint = self.session.scalar(
+            select(RecoveryCheckpoint)
+            .where(RecoveryCheckpoint.operation_id == recovery.operation_id)
+            .order_by(RecoveryCheckpoint.sequence)
+            .limit(1)
+        )
+        assert checkpoint is not None
+        checkpoint.summary = "tampered"
+        self.session.commit()
+        source = SqlTabularSourceProvider().create_snapshot(
+            self.session,
+            _principal(),
+            snapshot=TabularSnapshotInput(
+                name="Tampered",
+                source_name="tampered",
+                rows=({"id": 1},),
+            ),
+            source_id=recovery.resource_id,
+        )
+
+        with self.assertRaises(ConnectorRecoveryError):
+            recovery.commit_success(
+                self.session,
+                evidence={
+                    "verified": True,
+                    "checks": {"snapshot_ref": source.ref},
+                },
+            )
+
+        self.assertIsNone(
+            self.session.get(ConnectorTabularSource, recovery.resource_id)
+        )
+        operation = self.session.get(RecoveryOperation, recovery.operation_id)
+        self.session.refresh(operation)
+        self.assertEqual(RecoveryStatus.RUNNING.value, operation.status)
+
+    def test_definitive_external_rejection_is_terminal(self) -> None:
+        recovery = begin_connector_external_mutation(
+            self.session,
+            tenant_id="tenant-1",
+            provider_id="provider-1",
+            idempotency_key="publish-rejected",
+            request_sha256="c" * 64,
+            source_revision="revision-1",
+            cursor=None,
+            dry_run_evidence={"performed": True, "approved": True},
+            resource_type="external_record",
+            resource_id="record-2",
+        )
+        recovery.reject(
+            summary="The provider rejected the requested revision",
+            provider_code="revision_conflict",
+        )
+
+        operation = self.session.get(RecoveryOperation, recovery.operation_id)
+        self.session.refresh(operation)
+        self.assertEqual(RecoveryStatus.REJECTED.value, operation.status)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_sanctions_sources.py

@@ -27,6 +27,16 @@ from govoplan_connectors.backend.sanctions_sources import (
 )
 from govoplan_core.auth import ApiPrincipal
 from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.recovery import (
+    RecoveryCheckpoint,
+    RecoveryOperation,
+    RecoveryStatus,
+)
+from govoplan_core.core.runtime_coordination import (
+    DistributedLease,
+    RuntimeIdentity,
+    bind_process_runtime_identity,
+)
 from govoplan_core.db.base import Base, utcnow
 
 
@@ -89,13 +99,27 @@ class SanctionsSourcesTests(unittest.TestCase):
         Base.metadata.create_all(
             self.engine,
             tables=(
+                DistributedLease.__table__,
+                RecoveryOperation.__table__,
+                RecoveryCheckpoint.__table__,
                 ConnectorSanctionsAcquisitionRun.__table__,
                 ConnectorSanctionsSnapshot.__table__,
             ),
         )
         self.session = Session(self.engine)
+        bind_process_runtime_identity(
+            RuntimeIdentity(
+                installation_id="connectors-tests",
+                node_id="connectors-test-node",
+                incarnation="connectors-test-incarnation",
+                role="worker",
+                software_version="test",
+                composition_hash="a" * 64,
+            )
+        )
 
     def tearDown(self) -> None:
+        bind_process_runtime_identity(None)
         self.session.close()
         self.engine.dispose()
 
@@ -234,6 +258,7 @@ class SanctionsSourcesTests(unittest.TestCase):
             first.snapshot.ref.removeprefix("sanctions-snapshot:"),
         )
         record.acquired_at = utcnow() - timedelta(days=3)
+        self.session.commit()
 
         result = provider.refresh_source(
             self.session,
@@ -244,6 +269,48 @@ class SanctionsSourcesTests(unittest.TestCase):
         self.assertEqual("stale", result.status)
         self.assertEqual(first.snapshot.ref, result.snapshot.ref)
 
+    def test_idempotent_refresh_replays_the_committed_acquisition(self) -> None:
+        transport = _Transport((response(),))
+        provider = SqlSanctionsSnapshotProvider(transport)
+
+        first = provider.refresh_source(
+            self.session,
+            principal(),
+            provider_id=UNSC_PROVIDER_ID,
+            idempotency_key="scheduled-refresh-1",
+        )
+        replay = provider.refresh_source(
+            self.session,
+            principal(),
+            provider_id=UNSC_PROVIDER_ID,
+            idempotency_key="scheduled-refresh-1",
+        )
+
+        self.assertEqual(first.run_id, replay.run_id)
+        self.assertEqual(first.snapshot.ref, replay.snapshot.ref)
+        self.assertEqual([], transport.responses)
+        operation = self.session.query(RecoveryOperation).one()
+        self.assertEqual(RecoveryStatus.SUCCEEDED.value, operation.status)
+
+    def test_provider_failure_commits_failed_run_and_terminal_recovery(self) -> None:
+        provider = SqlSanctionsSnapshotProvider(
+            _Transport((SanctionsSourceError("offline"),))
+        )
+
+        result = provider.refresh_source(
+            self.session,
+            principal(),
+            provider_id=UNSC_PROVIDER_ID,
+        )
+
+        self.assertEqual("unavailable", result.status)
+        operation = self.session.query(RecoveryOperation).one()
+        self.assertEqual(RecoveryStatus.FAILED.value, operation.status)
+        self.assertEqual(
+            result.run_id,
+            self.session.query(ConnectorSanctionsAcquisitionRun).one().id,
+        )
+
     def test_snapshot_access_is_tenant_and_scope_isolated(self) -> None:
         provider = SqlSanctionsSnapshotProvider()
         created = provider.refresh_source(