Fence Campaign delivery effects in recovery ledger

docs/CAMPAIGN_DELIVERY_RUNBOOK.md

@@ -65,6 +65,17 @@ Before the first live send for a sender domain or mail-server profile:
 
 ## Outcome Handling
 
+Each real Campaign job delivery is represented by a Core recovery-ledger
+operation before the worker claims the job or invokes Mail, Postbox, or print.
+Synchronous batches use the same boundary after their batch-wide preflight.
+Explicit test/resend actions and post-acceptance IMAP appends use separate
+action-fenced operations. Operations store only opaque IDs, digests, channel
+policy, and bounded status evidence. A verified acceptance becomes
+`succeeded`, a definitive pre-effect or provider rejection becomes `rejected`,
+and uncertain or stranded effects remain `outcome_unknown` or
+`recovery_required` in Ops. Campaign jobs, message actions, and channel-attempt
+records remain the business source of truth.
+
 - `smtp_accepted`: Do not retry. If IMAP append is enabled and pending, run or
   enqueue the append action.
 - `failed_temporary`: Retry explicitly after checking the error and retry count.

src/govoplan_campaign/backend/manifest.py

@@ -1052,7 +1052,7 @@ manifest = ModuleManifest(
             id="campaigns.reference.shared-build-artifacts",
             title="Operate Campaign build artifacts across workers",
             summary="Generated messages use shared object storage and are verified before delivery.",
-            body="Campaign stores generated EML under opaque shared object keys and records expected size, SHA-256 digest, and Message-ID in each job. A worker may run on another node and verifies that evidence before delivery. Before object or Files-owned output effects, a fenced Core recovery operation records the canonical build request, validated-version evidence, and reserved object prefix. Object-only failures prove compensation; managed Files output and uncertain cleanup remain explicit forward-recovery or recovery-required work in Ops. Retention retains metadata when deletion fails. Never copy or edit runtime object keys as business data.",
+            body="Campaign stores generated EML under opaque shared object keys and records expected size, SHA-256 digest, and Message-ID in each job. A worker may run on another node and verifies that evidence before delivery. Before object or Files-owned output effects, a fenced Core recovery operation records the canonical build request, validated-version evidence, and reserved object prefix. Before a real Mail, Postbox, or print effect, a separate job-fenced operation records immutable message and recipient digests and later verifies the authoritative Campaign/channel attempt state. Definitive rejection is distinct from accepted, outcome-unknown, and recovery-required work in Ops. Object-only failures prove compensation; managed Files output and uncertain cleanup remain explicit forward-recovery work. Retention retains metadata when deletion fails. Never copy or edit runtime object keys as business data.",
             layer="evidence",
             documentation_types=("admin",),
             audience=("campaign_operator", "platform_operator", "release_reviewer"),

src/govoplan_campaign/backend/sending/jobs.py

@@ -15,6 +15,13 @@ from sqlalchemy import func
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
+from govoplan_core.core.recovery import RecoveryMode, RecoveryPlan, RecoveryStatus
+from govoplan_core.core.recovery_runtime import (
+    DurableRecoveryOperation,
+    RecoveryOperationBusy,
+    RecoveryOperationStateConflict,
+    begin_durable_recovery_operation,
+)
 from govoplan_core.core.notifications import (
     NotificationDispatchRequest,
     notification_dispatch_provider,
@@ -23,7 +30,9 @@ from govoplan_core.core.object_storage import (
     StorageBackendError,
     configured_storage_backend,
 )
+from govoplan_core.core.runtime_coordination import process_runtime_identity
 from govoplan_core.audit.logging import audit_event
+from govoplan_core.db.session import get_database
 from govoplan_core.security.redaction import redact_secret_values
 from govoplan_core.settings import settings as core_settings
 from govoplan_campaign.backend.db.models import (
@@ -1074,15 +1083,9 @@ def send_campaign_now(
     skipped_after_queue = 0
     for job in jobs:
         try:
-            claimed = _claimed_campaign_job_for_delivery(session, job)
-            if isinstance(claimed, SendJobResult):
-                result = claimed
-            else:
-                claimed_job, claim_token = claimed
-                result = _send_claimed_campaign_job(
+            result = _deliver_job_with_recovery(
                 session,
-                    job=claimed_job,
-                    claim_token=claim_token,
+                job=job,
                 context=delivery_contexts[job.id],
                 use_rate_limit=use_rate_limit,
                 enqueue_imap_task=enqueue_imap_task,
@@ -2096,6 +2099,134 @@ def _single_message_action_response(
     }
 
 
+def _begin_single_action_delivery_recovery(
+    *,
+    action: CampaignMessageAction,
+    job: CampaignJob,
+    delivery_context: _SendJobDeliveryContext,
+):
+    return begin_durable_recovery_operation(
+        get_database().SessionLocal,
+        identity=process_runtime_identity(),
+        module_id="campaigns",
+        operation_type="single-message-external-delivery",
+        idempotency_key=f"campaign-message-action:{action.id}",
+        request={
+            "tenant_id": action.tenant_id,
+            "campaign_id": action.campaign_id,
+            "version_id": action.campaign_version_id,
+            "job_id": action.job_id,
+            "action_id": action.id,
+            "action_kind": action.kind,
+            "message_sha256": action.message_sha256,
+            "recipient_manifest_sha256": action.recipient_manifest_sha256,
+            "smtp_transport_revision_sha256": hashlib.sha256(
+                str(delivery_context.snapshot.smtp_transport_revision).encode("utf-8")
+            ).hexdigest(),
+        },
+        recovery_plan=RecoveryPlan(
+            mode=RecoveryMode.FORWARD_RECOVERY,
+            preconditions=(
+                "the immutable single-message action is durably recorded",
+                "the message and SMTP transport revision passed preflight",
+            ),
+            forward_recovery_steps=(
+                "inspect the Campaign message action and SMTP provider evidence",
+                "reconcile acceptance, rejection, or an unknown outcome",
+            ),
+            verification_steps=(
+                "reload the message action through an independent session",
+                "verify its terminal outcome and attempt evidence",
+            ),
+        ),
+        precondition_evidence={
+            "action_id": action.id,
+            "action_status": action.status,
+            "action_kind": action.kind,
+            "job_id": job.id,
+            "message_sha256": action.message_sha256,
+            "recipient_manifest_sha256": action.recipient_manifest_sha256,
+        },
+        lease_resource_key=f"campaign:message-action:{action.tenant_id}:{action.id}",
+        lease_ttl_seconds=15 * 60,
+        resource_type="campaign_message_action",
+        resource_id=action.id,
+        metadata={"resources": ["postgresql", "smtp"]},
+    )
+
+
+def _single_action_recovery_evidence(
+    action_id: str,
+) -> tuple[str, dict[str, Any]]:
+    with get_database().SessionLocal() as evidence_session:
+        action = evidence_session.get(CampaignMessageAction, action_id)
+        if action is None:
+            return "recovery_required", {
+                "verified": False,
+                "action_present": False,
+            }
+        latest = (
+            evidence_session.query(CampaignMessageActionAttempt)
+            .filter(CampaignMessageActionAttempt.action_id == action.id)
+            .order_by(CampaignMessageActionAttempt.attempt_number.desc())
+            .first()
+        )
+        evidence = {
+            "verified": True,
+            "checks": {
+                "action_state_reloaded": True,
+                "action_attempt_compared": True,
+            },
+            "action_present": True,
+            "action_id": action.id,
+            "action_kind": action.kind,
+            "action_status": action.status,
+            "attempt_status": latest.status if latest else None,
+            "accepted_count": action.accepted_count,
+            "refused_count": action.refused_count,
+        }
+        if action.status in {"accepted", "accepted_with_refusals"}:
+            return "succeeded", evidence
+        if action.status == "outcome_unknown":
+            return "outcome_unknown", evidence
+        if action.status in {
+            "failed_temporary",
+            "failed_permanent",
+            "initiation_failed",
+        }:
+            return "failed", evidence
+        return "recovery_required", evidence
+
+
+def _finish_single_action_delivery_recovery(
+    operation: DurableRecoveryOperation,
+    *,
+    action_id: str,
+) -> None:
+    outcome, evidence = _single_action_recovery_evidence(action_id)
+    if outcome == "succeeded":
+        operation.succeed(evidence=evidence)
+    elif outcome == "failed":
+        operation.reject(
+            summary="The single-message SMTP effect was definitively rejected",
+            evidence=evidence,
+        )
+    elif outcome == "outcome_unknown":
+        operation.unresolved(
+            status=RecoveryStatus.OUTCOME_UNKNOWN,
+            summary="The single-message SMTP outcome requires reconciliation",
+            evidence=evidence,
+            failure_summary="The message may have been accepted by SMTP",
+        )
+    else:
+        operation.unresolved(
+            status=RecoveryStatus.RECOVERY_REQUIRED,
+            summary="The single-message action did not reach a verified terminal state",
+            evidence=evidence,
+            failure_summary="The message action requires forward recovery",
+        )
+
+
 def _send_single_message_direct(
     session: Session,
     *,
@@ -2127,6 +2258,53 @@ def _send_single_message_direct(
         messages_per_minute=delivery_context.snapshot.delivery.rate_limit.messages_per_minute,
         enabled=use_rate_limit,
     )
+    try:
+        recovery_start = _begin_single_action_delivery_recovery(
+            action=action,
+            job=job,
+            delivery_context=delivery_context,
+        )
+    except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc:
+        _finish_single_message_action(
+            session,
+            action=action,
+            status="initiation_failed",
+            error_type=exc.__class__.__name__,
+            error_message=str(exc),
+            final_send_status=job.send_status,
+        )
+        raise QueueingError(str(exc)) from exc
+    if recovery_start.replayed or recovery_start.operation is None:
+        session.refresh(action)
+        return _single_message_action_response(action, duplicate=True)
+    operation = recovery_start.operation
+    try:
+        result = _perform_single_message_direct_effect(
+            session,
+            campaign=campaign,
+            version=version,
+            job=job,
+            action=action,
+            delivery_context=delivery_context,
+            enqueue_imap_task=enqueue_imap_task,
+        )
+    except Exception:
+        _finish_single_action_delivery_recovery(operation, action_id=action.id)
+        raise
+    _finish_single_action_delivery_recovery(operation, action_id=action.id)
+    return result
+
+
+def _perform_single_message_direct_effect(
+    session: Session,
+    *,
+    campaign: Campaign,
+    version: CampaignVersion,
+    job: CampaignJob,
+    action: CampaignMessageAction,
+    delivery_context: _SendJobDeliveryContext,
+    enqueue_imap_task: bool,
+) -> dict[str, Any]:
     attempt = _start_single_message_action_attempt(session, action)
     try:
         result = mail_integration().send_campaign_email_bytes(
@@ -2726,7 +2904,12 @@ def _recipients_from_job(job: CampaignJob) -> list[str]:
     return list(dict.fromkeys(recipients))
 
 
-def _claim_job_for_sending(session: Session, job: CampaignJob) -> str | None:
+def _claim_job_for_sending(
+    session: Session,
+    job: CampaignJob,
+    *,
+    claim_token: str | None = None,
+) -> str | None:
     """Atomically claim a queued job and return the claim token.
 
     A duplicate task can observe CLAIMED/SENDING but cannot acquire a second
@@ -2735,7 +2918,7 @@ def _claim_job_for_sending(session: Session, job: CampaignJob) -> str | None:
     could race a slow worker.
     """
 
-    claim_token = str(uuid4())
+    effective_claim_token = claim_token or str(uuid4())
     changed = (
         session.query(CampaignJob)
         .filter(
@@ -2748,7 +2931,7 @@ def _claim_job_for_sending(session: Session, job: CampaignJob) -> str | None:
                 CampaignJob.queue_status: JobQueueStatus.SENDING.value,
                 CampaignJob.send_status: JobSendStatus.CLAIMED.value,
                 CampaignJob.claimed_at: _utcnow(),
-                CampaignJob.claim_token: claim_token,
+                CampaignJob.claim_token: effective_claim_token,
                 CampaignJob.last_error: None,
             },
             synchronize_session=False,
@@ -2756,7 +2939,7 @@ def _claim_job_for_sending(session: Session, job: CampaignJob) -> str | None:
     )
     session.commit()
     session.expire_all()
-    return claim_token if changed == 1 else None
+    return effective_claim_token if changed == 1 else None
 
 
 def _record_attempt_start(
@@ -2946,6 +3129,184 @@ def _campaign_delivery_status_pair(
     return None
 
 
+def _canonical_delivery_sha256(value: object) -> str:
+    return hashlib.sha256(
+        json.dumps(
+            value,
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=True,
+            default=str,
+        ).encode("utf-8")
+    ).hexdigest()
+
+
+def _begin_job_delivery_recovery(
+    *,
+    job: CampaignJob,
+    context: _SendJobDeliveryContext,
+    claim_token: str,
+):
+    channel_policy = DeliveryChannelPolicy(job.delivery_channel_policy)
+    resources = ["postgresql"]
+    if channel_policy.uses_mail:
+        resources.append("smtp")
+    if channel_policy.uses_postbox:
+        resources.append("postbox")
+    if channel_policy.uses_print:
+        resources.append("print-provider")
+    claim_sha256 = hashlib.sha256(claim_token.encode("utf-8")).hexdigest()
+    recipient_manifest_sha256 = _canonical_delivery_sha256(
+        _recipients_from_job(job)
+    )
+    return begin_durable_recovery_operation(
+        get_database().SessionLocal,
+        identity=process_runtime_identity(),
+        module_id="campaigns",
+        operation_type="external-channel-delivery",
+        idempotency_key=f"campaign-delivery:{job.id}:{claim_sha256[:32]}",
+        request={
+            "tenant_id": job.tenant_id,
+            "campaign_id": job.campaign_id,
+            "version_id": job.campaign_version_id,
+            "job_id": job.id,
+            "channel_policy": channel_policy.value,
+            "message_sha256": job.eml_sha256,
+            "recipient_manifest_sha256": recipient_manifest_sha256,
+            "execution_snapshot_sha256": context.version.execution_snapshot_hash,
+        },
+        recovery_plan=RecoveryPlan(
+            mode=RecoveryMode.FORWARD_RECOVERY,
+            preconditions=(
+                "the immutable built job is explicitly queued",
+                "message and transport revisions passed preflight",
+            ),
+            forward_recovery_steps=(
+                "inspect provider and module attempt evidence",
+                "reconcile accepted, rejected, or outcome-unknown channel state",
+            ),
+            verification_steps=(
+                "reload the Campaign job through an independent session",
+                "compare final channel and attempt states",
+            ),
+        ),
+        precondition_evidence={
+            "job_id": job.id,
+            "queue_status": job.queue_status,
+            "send_status": job.send_status,
+            "build_status": job.build_status,
+            "validation_status": job.validation_status,
+            "message_sha256": job.eml_sha256,
+            "recipient_manifest_sha256": recipient_manifest_sha256,
+            "claim_sha256": claim_sha256,
+        },
+        lease_resource_key=f"campaign:delivery:{job.tenant_id}:{job.id}",
+        lease_ttl_seconds=30 * 60,
+        resource_type="campaign_job",
+        resource_id=job.id,
+        metadata={
+            "resources": resources,
+            "channel_policy": channel_policy.value,
+        },
+    )
+
+
+def _job_delivery_recovery_evidence(job_id: str) -> tuple[str, dict[str, Any]]:
+    with get_database().SessionLocal() as evidence_session:
+        job = evidence_session.get(CampaignJob, job_id)
+        if job is None:
+            return "recovery_required", {
+                "verified": False,
+                "job_present": False,
+            }
+        latest_smtp = (
+            evidence_session.query(SendAttempt)
+            .filter(SendAttempt.job_id == job.id)
+            .order_by(SendAttempt.attempt_number.desc())
+            .first()
+        )
+        latest_postbox = (
+            evidence_session.query(PostboxDeliveryAttempt)
+            .filter(PostboxDeliveryAttempt.job_id == job.id)
+            .order_by(PostboxDeliveryAttempt.attempt_number.desc())
+            .first()
+        )
+        latest_print = (
+            evidence_session.query(PrintOutputAttempt)
+            .filter(PrintOutputAttempt.job_id == job.id)
+            .order_by(PrintOutputAttempt.attempt_number.desc())
+            .first()
+        )
+        evidence = {
+            "verified": True,
+            "checks": {
+                "job_state_reloaded": True,
+                "channel_attempts_compared": True,
+            },
+            "job_present": True,
+            "job_id": job.id,
+            "queue_status": job.queue_status,
+            "send_status": job.send_status,
+            "postbox_status": job.postbox_status,
+            "print_status": job.print_status,
+            "attempt_count": job.attempt_count,
+            "postbox_attempt_count": job.postbox_attempt_count,
+            "print_attempt_count": job.print_attempt_count,
+            "latest_attempts": {
+                "smtp": latest_smtp.status if latest_smtp else None,
+                "postbox": latest_postbox.status if latest_postbox else None,
+                "print": latest_print.status if latest_print else None,
+            },
+        }
+        if (
+            job.send_status == JobSendStatus.OUTCOME_UNKNOWN.value
+            or job.postbox_status == JobPostboxStatus.OUTCOME_UNKNOWN.value
+        ):
+            return "outcome_unknown", evidence
+        if (
+            job.send_status
+            in {JobSendStatus.CLAIMED.value, JobSendStatus.SENDING.value}
+            or job.postbox_status == JobPostboxStatus.DELIVERING.value
+            or job.print_status == JobPrintStatus.ACCEPTING.value
+        ):
+            return "recovery_required", evidence
+        if job.send_status in {
+            *FULLY_ACCEPTED_STATUSES,
+            JobSendStatus.PARTIALLY_ACCEPTED.value,
+        }:
+            return "succeeded", evidence
+        return "failed", evidence
+
+
+def _finish_job_delivery_recovery(
+    operation: DurableRecoveryOperation,
+    *,
+    job_id: str,
+) -> None:
+    outcome, evidence = _job_delivery_recovery_evidence(job_id)
+    if outcome == "succeeded":
+        operation.succeed(evidence=evidence)
+    elif outcome == "failed":
+        operation.reject(
+            summary="Campaign delivery completed with a verified provider rejection",
+            evidence=evidence,
+        )
+    elif outcome == "outcome_unknown":
+        operation.unresolved(
+            status=RecoveryStatus.OUTCOME_UNKNOWN,
+            summary="Campaign delivery provider outcome requires reconciliation",
+            evidence=evidence,
+            failure_summary="A Campaign delivery effect may have been accepted",
+        )
+    else:
+        operation.unresolved(
+            status=RecoveryStatus.RECOVERY_REQUIRED,
+            summary="Campaign delivery did not reach a verified terminal state",
+            evidence=evidence,
+            failure_summary="Campaign delivery state requires forward recovery",
+        )
+
+
 def send_campaign_job(
     session: Session,
     *,
@@ -2986,11 +3347,68 @@ def send_campaign_job(
             message=f"Would deliver via {'; '.join(descriptions)}",
         )
 
-    claimed = _claimed_campaign_job_for_delivery(session, job)
+    return _deliver_job_with_recovery(
+        session,
+        job=job,
+        context=context,
+        use_rate_limit=use_rate_limit,
+        enqueue_imap_task=enqueue_imap_task,
+    )
+
+
+def _deliver_job_with_recovery(
+    session: Session,
+    *,
+    job: CampaignJob,
+    context: _SendJobDeliveryContext,
+    use_rate_limit: bool,
+    enqueue_imap_task: bool,
+) -> SendJobResult:
+    claim_token = str(uuid4())
+    try:
+        recovery_start = _begin_job_delivery_recovery(
+            job=job,
+            context=context,
+            claim_token=claim_token,
+        )
+    except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc:
+        return SendJobResult(
+            job_id=job.id,
+            status="recovery_blocked",
+            attempt_number=job.attempt_count,
+            message=str(exc),
+        )
+    if recovery_start.replayed or recovery_start.operation is None:
+        return SendJobResult(
+            job_id=job.id,
+            status="already_completed",
+            attempt_number=job.attempt_count,
+            message="The durable delivery operation already completed.",
+        )
+    recovery_operation = recovery_start.operation
+    claimed = _claimed_campaign_job_for_delivery(
+        session,
+        job,
+        claim_token=claim_token,
+    )
     if isinstance(claimed, SendJobResult):
+        recovery_operation.reject(
+            summary="Campaign job could not be claimed before any provider effect",
+            evidence={
+                "verified": True,
+                "checks": {
+                    "provider_effect_started": False,
+                    "claim_rejected": True,
+                },
+                "provider_effect_started": False,
+                "job_id": job.id,
+                "claim_result": claimed.status,
+            },
+        )
         return claimed
     claimed_job, claim_token = claimed
-    return _send_claimed_campaign_job(
+    try:
+        result = _send_claimed_campaign_job(
             session,
             job=claimed_job,
             claim_token=claim_token,
@@ -2998,6 +3416,11 @@ def send_campaign_job(
             use_rate_limit=use_rate_limit,
             enqueue_imap_task=enqueue_imap_task,
         )
+    except Exception:
+        _finish_job_delivery_recovery(recovery_operation, job_id=job.id)
+        raise
+    _finish_job_delivery_recovery(recovery_operation, job_id=job.id)
+    return result
 
 
 def _preflight_send_campaign_job(
@@ -3100,15 +3523,21 @@ def _send_job_delivery_context(
 def _claimed_campaign_job_for_delivery(
     session: Session,
     job: CampaignJob,
+    *,
+    claim_token: str | None = None,
 ) -> tuple[CampaignJob, str] | SendJobResult:
-    claim_token = _claim_job_for_sending(session, job)
-    if claim_token is None:
+    effective_claim = _claim_job_for_sending(
+        session,
+        job,
+        claim_token=claim_token,
+    )
+    if effective_claim is None:
         return _not_claimed_send_job_result(session, job)
 
     job = session.get(CampaignJob, job.id)
     if job is None:
         raise SendJobError("Claimed campaign job disappeared before send.")
-    return job, claim_token
+    return job, effective_claim
 
 
 def _not_claimed_send_job_result(session: Session, job: CampaignJob) -> SendJobResult:
@@ -3878,7 +4307,12 @@ def _imap_attempt_count(session: Session, job_id: str) -> int:
     )
 
 
-def _claim_job_for_imap_append(session: Session, job: CampaignJob) -> str | None:
+def _claim_job_for_imap_append(
+    session: Session,
+    job: CampaignJob,
+    *,
+    claim_token: str | None = None,
+) -> str | None:
     """Atomically grant one worker permission to invoke the IMAP provider."""
 
     if not _mail_was_accepted(
@@ -3891,7 +4325,7 @@ def _claim_job_for_imap_append(session: Session, job: CampaignJob) -> str | None
         ),
     ):
         return None
-    claim_token = str(uuid4())
+    effective_claim_token = claim_token or str(uuid4())
     changed = (
         session.query(CampaignJob)
         .filter(
@@ -3905,7 +4339,7 @@ def _claim_job_for_imap_append(session: Session, job: CampaignJob) -> str | None
             {
                 CampaignJob.imap_status: JobImapStatus.APPENDING.value,
                 CampaignJob.imap_claimed_at: _utcnow(),
-                CampaignJob.imap_claim_token: claim_token,
+                CampaignJob.imap_claim_token: effective_claim_token,
                 CampaignJob.last_error: None,
             },
             synchronize_session=False,
@@ -3913,7 +4347,7 @@ def _claim_job_for_imap_append(session: Session, job: CampaignJob) -> str | None
     )
     session.commit()
     session.expire_all()
-    return claim_token if changed == 1 else None
+    return effective_claim_token if changed == 1 else None
 
 
 def _record_imap_attempt_start(
@@ -4269,10 +4703,17 @@ def _imap_append_dry_run(
 
 
 def _claim_imap_append(
-    session: Session, job: CampaignJob
+    session: Session,
+    job: CampaignJob,
+    *,
+    claim_token: str | None = None,
 ) -> _ClaimedImapAppend | AppendSentResult:
-    claim_token = _claim_job_for_imap_append(session, job)
-    if claim_token is None:
+    effective_claim = _claim_job_for_imap_append(
+        session,
+        job,
+        claim_token=claim_token,
+    )
+    if effective_claim is None:
         current = session.get(CampaignJob, job.id)
         if current is None:
             raise SendJobError(f"Job disappeared while claiming IMAP append: {job.id}")
@@ -4290,8 +4731,8 @@ def _claim_imap_append(
         raise SendJobError("Claimed campaign job disappeared before IMAP append")
     return _ClaimedImapAppend(
         job=claimed_job,
-        attempt=_record_imap_attempt_start(session, claimed_job, claim_token),
-        claim_token=claim_token,
+        attempt=_record_imap_attempt_start(session, claimed_job, effective_claim),
+        claim_token=effective_claim,
     )
 
 
@@ -4392,6 +4833,127 @@ def _perform_imap_append(
         )
 
 
+def _begin_imap_append_recovery(
+    *,
+    job: CampaignJob,
+    context: _ImapAppendContext,
+    claim_token: str,
+):
+    claim_sha256 = hashlib.sha256(claim_token.encode("utf-8")).hexdigest()
+    return begin_durable_recovery_operation(
+        get_database().SessionLocal,
+        identity=process_runtime_identity(),
+        module_id="campaigns",
+        operation_type="imap-sent-append",
+        idempotency_key=f"campaign-imap:{job.id}:{claim_sha256[:32]}",
+        request={
+            "tenant_id": job.tenant_id,
+            "campaign_id": job.campaign_id,
+            "version_id": job.campaign_version_id,
+            "job_id": job.id,
+            "message_sha256": job.eml_sha256,
+            "folder_sha256": hashlib.sha256(
+                context.folder.encode("utf-8")
+            ).hexdigest(),
+            "imap_transport_revision_sha256": hashlib.sha256(
+                context.snapshot.imap_transport_revision.encode("utf-8")
+            ).hexdigest(),
+        },
+        recovery_plan=RecoveryPlan(
+            mode=RecoveryMode.FORWARD_RECOVERY,
+            preconditions=(
+                "SMTP acceptance is durable",
+                "the exact EML and IMAP transport revision passed preflight",
+            ),
+            forward_recovery_steps=(
+                "inspect the target mailbox and append attempt",
+                "reconcile as appended or not appended before retry",
+            ),
+            verification_steps=(
+                "reload the Campaign IMAP state through an independent session",
+                "compare the latest append attempt",
+            ),
+        ),
+        precondition_evidence={
+            "job_id": job.id,
+            "send_status": job.send_status,
+            "imap_status": job.imap_status,
+            "message_sha256": job.eml_sha256,
+            "claim_sha256": claim_sha256,
+        },
+        lease_resource_key=f"campaign:imap:{job.tenant_id}:{job.id}",
+        lease_ttl_seconds=15 * 60,
+        resource_type="campaign_job",
+        resource_id=job.id,
+        metadata={"resources": ["postgresql", "imap"]},
+    )
+
+
+def _imap_recovery_evidence(job_id: str) -> tuple[str, dict[str, Any]]:
+    with get_database().SessionLocal() as evidence_session:
+        job = evidence_session.get(CampaignJob, job_id)
+        if job is None:
+            return "recovery_required", {
+                "verified": False,
+                "job_present": False,
+            }
+        latest = (
+            evidence_session.query(ImapAppendAttempt)
+            .filter(ImapAppendAttempt.job_id == job.id)
+            .order_by(ImapAppendAttempt.attempt_number.desc())
+            .first()
+        )
+        evidence = {
+            "verified": True,
+            "checks": {
+                "job_state_reloaded": True,
+                "append_attempt_compared": True,
+            },
+            "job_present": True,
+            "job_id": job.id,
+            "send_status": job.send_status,
+            "imap_status": job.imap_status,
+            "attempt_count": _imap_attempt_count(evidence_session, job.id),
+            "latest_attempt_status": latest.status if latest else None,
+        }
+        if job.imap_status == JobImapStatus.APPENDED.value:
+            return "succeeded", evidence
+        if job.imap_status == JobImapStatus.FAILED.value:
+            return "failed", evidence
+        if job.imap_status == JobImapStatus.OUTCOME_UNKNOWN.value:
+            return "outcome_unknown", evidence
+        return "recovery_required", evidence
+
+
+def _finish_imap_append_recovery(
+    operation: DurableRecoveryOperation,
+    *,
+    job_id: str,
+) -> None:
+    outcome, evidence = _imap_recovery_evidence(job_id)
+    if outcome == "succeeded":
+        operation.succeed(evidence=evidence)
+    elif outcome == "failed":
+        operation.reject(
+            summary="The IMAP provider definitively rejected the append",
+            evidence=evidence,
+        )
+    elif outcome == "outcome_unknown":
+        operation.unresolved(
+            status=RecoveryStatus.OUTCOME_UNKNOWN,
+            summary="The IMAP append outcome requires mailbox reconciliation",
+            evidence=evidence,
+            failure_summary="The exact message may already exist in the Sent folder",
+        )
+    else:
+        operation.unresolved(
+            status=RecoveryStatus.RECOVERY_REQUIRED,
+            summary="The IMAP append did not reach a verified terminal state",
+            evidence=evidence,
+            failure_summary="The IMAP append requires forward recovery",
+        )
+
+
 def append_sent_for_job(
     session: Session, *, job_id: str, dry_run: bool = False
 ) -> AppendSentResult:
@@ -4408,10 +4970,53 @@ def append_sent_for_job(
         return prepared
     if dry_run:
         return _imap_append_dry_run(session, job, prepared)
-    claimed = _claim_imap_append(session, job)
+    claim_token = str(uuid4())
+    try:
+        recovery_start = _begin_imap_append_recovery(
+            job=job,
+            context=prepared,
+            claim_token=claim_token,
+        )
+    except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc:
+        return AppendSentResult(
+            job_id=job.id,
+            status="recovery_blocked",
+            attempt_number=_imap_attempt_count(session, job.id),
+            message=str(exc),
+        )
+    if recovery_start.replayed or recovery_start.operation is None:
+        return AppendSentResult(
+            job_id=job.id,
+            status="already_completed",
+            attempt_number=_imap_attempt_count(session, job.id),
+        )
+    recovery_operation = recovery_start.operation
+    claimed = _claim_imap_append(
+        session,
+        job,
+        claim_token=claim_token,
+    )
     if isinstance(claimed, AppendSentResult):
+        recovery_operation.reject(
+            summary="The IMAP append claim was rejected before any provider effect",
+            evidence={
+                "verified": True,
+                "checks": {
+                    "provider_effect_started": False,
+                    "claim_rejected": True,
+                },
+                "provider_effect_started": False,
+                "claim_result": claimed.status,
+            },
+        )
         return claimed
-    return _perform_imap_append(session, claimed, prepared)
+    try:
+        result = _perform_imap_append(session, claimed, prepared)
+    except Exception:
+        _finish_imap_append_recovery(recovery_operation, job_id=job.id)
+        raise
+    _finish_imap_append_recovery(recovery_operation, job_id=job.id)
+    return result
 
 
 def enqueue_pending_imap_appends(

tests/conftest.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import pytest
+
+from govoplan_core.core import runtime_coordination
+from govoplan_core.core.runtime_coordination import (
+    RuntimeIdentity,
+    bind_process_runtime_identity,
+)
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _bind_test_runtime_identity():
+    previous = runtime_coordination._process_runtime_identity
+    bind_process_runtime_identity(
+        RuntimeIdentity(
+            installation_id="campaign-tests",
+            node_id="campaign-test-process",
+            incarnation="campaign-test-incarnation",
+            role="test",
+            software_version="test",
+            composition_hash="a" * 64,
+        )
+    )
+    try:
+        yield
+    finally:
+        bind_process_runtime_identity(previous)

tests/test_campaign_delivery_recovery.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+from unittest.mock import Mock
+
+import pytest
+
+from govoplan_core.core.recovery import RecoveryStatus
+from govoplan_campaign.backend.sending import jobs
+
+
+@pytest.mark.parametrize(
+    ("outcome", "expected_method", "expected_status"),
+    [
+        ("succeeded", "succeed", None),
+        ("failed", "reject", None),
+        ("outcome_unknown", "unresolved", RecoveryStatus.OUTCOME_UNKNOWN),
+        ("recovery_required", "unresolved", RecoveryStatus.RECOVERY_REQUIRED),
+    ],
+)
+def test_delivery_recovery_maps_verified_job_state(
+    monkeypatch,
+    outcome: str,
+    expected_method: str,
+    expected_status: RecoveryStatus | None,
+) -> None:
+    evidence = {
+        "verified": outcome != "recovery_required",
+        "checks": {"job_state_reloaded": True},
+        "send_status": outcome,
+    }
+    monkeypatch.setattr(
+        jobs,
+        "_job_delivery_recovery_evidence",
+        lambda _job_id: (outcome, evidence),
+    )
+    operation = Mock()
+
+    jobs._finish_job_delivery_recovery(operation, job_id="job-1")
+
+    method = getattr(operation, expected_method)
+    method.assert_called_once()
+    if expected_status is not None:
+        assert method.call_args.kwargs["status"] == expected_status
+    for other in {"succeed", "reject", "unresolved"} - {expected_method}:
+        getattr(operation, other).assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("outcome", "expected_method", "expected_status"),
+    [
+        ("succeeded", "succeed", None),
+        ("failed", "reject", None),
+        ("outcome_unknown", "unresolved", RecoveryStatus.OUTCOME_UNKNOWN),
+        ("recovery_required", "unresolved", RecoveryStatus.RECOVERY_REQUIRED),
+    ],
+)
+def test_imap_recovery_maps_verified_append_state(
+    monkeypatch,
+    outcome: str,
+    expected_method: str,
+    expected_status: RecoveryStatus | None,
+) -> None:
+    evidence = {
+        "verified": outcome != "recovery_required",
+        "checks": {"job_state_reloaded": True},
+        "imap_status": outcome,
+    }
+    monkeypatch.setattr(
+        jobs,
+        "_imap_recovery_evidence",
+        lambda _job_id: (outcome, evidence),
+    )
+    operation = Mock()
+
+    jobs._finish_imap_append_recovery(operation, job_id="job-1")
+
+    method = getattr(operation, expected_method)
+    method.assert_called_once()
+    if expected_status is not None:
+        assert method.call_args.kwargs["status"] == expected_status
+    for other in {"succeed", "reject", "unresolved"} - {expected_method}:
+        getattr(operation, other).assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("outcome", "expected_method", "expected_status"),
+    [
+        ("succeeded", "succeed", None),
+        ("failed", "reject", None),
+        ("outcome_unknown", "unresolved", RecoveryStatus.OUTCOME_UNKNOWN),
+        ("recovery_required", "unresolved", RecoveryStatus.RECOVERY_REQUIRED),
+    ],
+)
+def test_single_action_recovery_maps_verified_action_state(
+    monkeypatch,
+    outcome: str,
+    expected_method: str,
+    expected_status: RecoveryStatus | None,
+) -> None:
+    evidence = {
+        "verified": outcome != "recovery_required",
+        "checks": {"action_state_reloaded": True},
+        "action_status": outcome,
+    }
+    monkeypatch.setattr(
+        jobs,
+        "_single_action_recovery_evidence",
+        lambda _action_id: (outcome, evidence),
+    )
+    operation = Mock()
+
+    jobs._finish_single_action_delivery_recovery(operation, action_id="action-1")
+
+    method = getattr(operation, expected_method)
+    method.assert_called_once()
+    if expected_status is not None:
+        assert method.call_args.kwargs["status"] == expected_status
+    for other in {"succeed", "reject", "unresolved"} - {expected_method}:
+        getattr(operation, other).assert_not_called()

tests/test_imap_append_idempotency.py

@@ -159,6 +159,11 @@ def test_post_provider_persistence_failure_freezes_imap_retry() -> None:
         patch("govoplan_campaign.backend.sending.jobs._load_eml_bytes_for_job", return_value=b"message"),
         patch("govoplan_campaign.backend.sending.jobs._claim_job_for_imap_append", return_value="claim-1"),
         patch("govoplan_campaign.backend.sending.jobs._record_imap_attempt_start", return_value=attempt),
+        patch(
+            "govoplan_campaign.backend.sending.jobs._begin_imap_append_recovery",
+            return_value=SimpleNamespace(replayed=False, operation=MagicMock()),
+        ),
+        patch("govoplan_campaign.backend.sending.jobs._finish_imap_append_recovery"),
         patch("govoplan_campaign.backend.sending.jobs.mail_integration", return_value=Mail()),
         patch(
             "govoplan_campaign.backend.sending.jobs._record_imap_append_success",

tests/test_single_message_actions.py

@@ -170,6 +170,13 @@ class CampaignSingleMessageActionTests(unittest.TestCase):
                 "govoplan_campaign.backend.sending.jobs.mail_integration",
                 return_value=mail,
             ),
+            patch(
+                "govoplan_campaign.backend.sending.jobs._begin_single_action_delivery_recovery",
+                return_value=SimpleNamespace(replayed=False, operation=Mock()),
+            ),
+            patch(
+                "govoplan_campaign.backend.sending.jobs._finish_single_action_delivery_recovery"
+            ),
             patch(
                 "govoplan_campaign.backend.sending.jobs.audit_event"
             ),
@@ -269,6 +276,13 @@ class CampaignSingleMessageActionTests(unittest.TestCase):
                     "govoplan_campaign.backend.sending.jobs.mail_integration",
                     return_value=mail,
                 ),
+                patch(
+                    "govoplan_campaign.backend.sending.jobs._begin_single_action_delivery_recovery",
+                    return_value=SimpleNamespace(replayed=False, operation=Mock()),
+                ),
+                patch(
+                    "govoplan_campaign.backend.sending.jobs._finish_single_action_delivery_recovery"
+                ),
                 patch(
                     "govoplan_campaign.backend.sending.jobs.files_integration"
                 ) as files,