Fence Mail provider and mailbox recovery effects

README.md

@@ -45,6 +45,18 @@ SMTP effects decrypt only SMTP credentials; Sent-folder effects decrypt only
 IMAP credentials. A connection loss after an effect starts is surfaced as an
 unknown outcome. Campaign does not automatically retry an unknown IMAP append,
 preventing silent duplicate Sent copies while an operator inspects the mailbox.
+Every current outbox, Campaign SMTP, and Campaign Sent-folder attempt also
+starts a Mail-owned Core recovery operation under a stable per-attempt effect
+identifier before contacting the provider. Evidence contains only message,
+address, and folder digests plus bounded outcome counts. A completed effect is
+never replayed to repair caller state; unknown outcomes require explicit
+provider-backed reconciliation.
+
+Read-only mailbox folder/message indexing and bounce/calendar-reply scans use
+distributed recovery fences. Cache rows or source cursors commit before an
+independent verification closes the operation. A failed read rolls back and is
+safe to repeat because these paths never move, delete, flag, or otherwise
+mutate provider messages.
 
 The existing SMTP/IMAP credential-inheritance policy remains part of the Mail
 policy model for compatibility. Campaign delivery requires effective

docs/MAIL_HANDBOOK.md

@@ -280,6 +280,20 @@ implemented.
 
 ### SMTP/IMAP incidents
 
+Current SMTP delivery and Sent-folder APPEND paths start a Mail-owned Core
+recovery operation before the network effect. Outbox attempts use the durable
+command and attempt number; Campaign jobs and single-message actions pass their
+own stable attempt identifiers through the versioned capability. A matching
+completed identifier is never sent or appended again merely to reconstruct
+caller state. Provider acceptance, definitive rejection, and outcome-unknown
+states are recorded independently of the consuming transaction.
+
+Mailbox folder/message indexing and configured bounce/calendar-reply scans are
+read-only provider operations. They acquire distributed per-profile/folder or
+per-source fences, commit bounded projection state, and verify that state in an
+independent session. A failed read is rolled back and can be repeated; it is not
+treated as an unknown provider mutation.
+
 1. Stop new consumer work if duplicate effects or credential compromise are
    possible.
 2. Preserve safe Mail, consumer-job, worker, and provider evidence.
@@ -291,6 +305,10 @@ implemented.
    merely to recreate a Sent copy.
 6. Record the incident/reconciliation reference in the consuming domain's audit
    trail without copying raw provider secrets or message content unnecessarily.
+7. For an outbox `outcome_unknown`, reconcile the Mail command with provider
+   evidence. Confirmed acceptance closes the provider operation as succeeded;
+   confirmed absence records verified recovery and permits only a new,
+   deliberate attempt identifier.
 
 ### Delivery-status and calendar-reply sources
 
@@ -408,9 +426,6 @@ Before claiming a Mail composition is production-ready:
 
 ## Explicitly planned, not yet claimed
 
-- Durable, idempotent Campaign report delivery with Mail-owned attempts,
-  unknown-outcome reconciliation, and partial-refusal evidence
-  ([`govoplan-mail#17`](https://git.add-ideas.de/GovOPlaN/govoplan-mail/issues/17)).
 - Canonical audit events for profile tests and the remaining profile/policy
   administration lifecycle, plus an operator-visible Redis-throttling
   degradation signal.
@@ -422,5 +437,7 @@ Before claiming a Mail composition is production-ready:
   is stable.
 - POP3 except for a future explicit legacy download/import requirement.
 - A full mail client with compose/reply/move/delete/read-state mutation.
+- Recovery-ledger adoption for future provider-side move, delete, and flag
+  mutations; no such production path exists in the current read-only mailbox.
 - Proof that process-local throttling coordinates multiple workers when Redis
   is unavailable; it deliberately does not.

src/govoplan_mail/backend/bounce_processing.py

@@ -38,6 +38,10 @@ from govoplan_mail.backend.server_hierarchy import (
     resolve_mail_transport,
 )
 from govoplan_mail.backend.runtime import get_registry
+from govoplan_mail.backend.recovery import (
+    MailRecoveryError,
+    begin_bounce_scan_recovery,
+)
 
 
 class MailBounceError(RuntimeError):
@@ -465,49 +469,66 @@ class SqlMailBounceProcessingProvider(MailBounceProcessingProvider):
             raise MailBounceError(
                 "Bounce-source IMAP settings changed; review and save the source before scanning."
             )
-        page = list_imap_uids_since(
+        try:
-            imap_config=resolved.config,
+            recovery = begin_bounce_scan_recovery(
-            folder=source.folder,
-            highest_uid=source.highest_processed_uid,
-            expected_uidvalidity=source.uidvalidity,
-            limit=limit,
-        )
-        found = 0
-        calendar_replies = 0
-        highest = 0 if page.cursor_reset else source.highest_processed_uid
-        for uid in page.uids:
-            raw = get_imap_raw_message(
-                imap_config=resolved.config,
-                folder=source.folder,
-                uid=uid,
-            )
-            calendar_replies += _reconcile_calendar_replies(
-                session,
                 tenant_id=source.tenant_id,
                 profile_id=source.profile_id,
+                source_id=source.id,
                 folder=source.folder,
-                uid=uid,
-                raw_message=raw.raw,
             )
-            found += len(
+        except MailRecoveryError as exc:
-                self.process_raw_message(
+            raise MailBounceError(str(exc)) from exc
+        try:
+            page = list_imap_uids_since(
+                imap_config=resolved.config,
+                folder=source.folder,
+                highest_uid=source.highest_processed_uid,
+                expected_uidvalidity=source.uidvalidity,
+                limit=limit,
+            )
+            found = 0
+            calendar_replies = 0
+            highest = 0 if page.cursor_reset else source.highest_processed_uid
+            for uid in page.uids:
+                raw = get_imap_raw_message(
+                    imap_config=resolved.config,
+                    folder=source.folder,
+                    uid=uid,
+                )
+                calendar_replies += _reconcile_calendar_replies(
                     session,
                     tenant_id=source.tenant_id,
                     profile_id=source.profile_id,
                     folder=source.folder,
                     uid=uid,
                     raw_message=raw.raw,
-                    reconcile_calendar=False,
                 )
-            )
+                found += len(
-            highest = max(highest, int(uid))
+                    self.process_raw_message(
-        now = utcnow()
+                        session,
-        source.uidvalidity = page.uidvalidity
+                        tenant_id=source.tenant_id,
-        source.highest_processed_uid = highest
+                        profile_id=source.profile_id,
-        source.last_scanned_at = now
+                        folder=source.folder,
-        source.last_success_at = now
+                        uid=uid,
-        source.last_error = None
+                        raw_message=raw.raw,
-        session.flush()
+                        reconcile_calendar=False,
+                    )
+                )
+                highest = max(highest, int(uid))
+            now = utcnow()
+            source.uidvalidity = page.uidvalidity
+            source.highest_processed_uid = highest
+            source.last_scanned_at = now
+            source.last_success_at = now
+            source.last_error = None
+            session.flush()
+            session.commit()
+        except Exception as exc:
+            session.rollback()
+            if not recovery.operation.closed:
+                recovery.reject(code=exc.__class__.__name__)
+            raise
+        recovery.complete(highest_uid=highest, uidvalidity=page.uidvalidity)
         return len(page.uids), found, calendar_replies
 
 

src/govoplan_mail/backend/capabilities.py

@@ -30,6 +30,10 @@ from govoplan_mail.backend.server_hierarchy import (
     select_mail_transport,
 )
 from govoplan_mail.backend.runtime import configure_runtime
+from govoplan_mail.backend.recovery import (
+    MailRecoveryError,
+    begin_provider_effect_recovery,
+)
 from govoplan_mail.backend.sending.imap import (
     ImapAppendError,
     ImapConfigurationError,
@@ -288,6 +292,9 @@ def send_campaign_email_bytes(
     expected_smtp_transport_revision: str,
     smtp_server_id: str | None = None,
     smtp_credential_id: str | None = None,
+    recovery_effect_id: str | None = None,
+    recovery_resource_type: str | None = None,
+    recovery_resource_id: str | None = None,
 ) -> CampaignSmtpDeliveryResult:
     selection = _selection_payload(
         profile_id=profile_id,
@@ -362,6 +369,25 @@ def send_campaign_email_bytes(
         )
     except MailProfileError:
         raise MailProfileError("Mail delivery is blocked by the effective Mail policy.") from None
+    try:
+        recovery = begin_provider_effect_recovery(
+            kind="smtp-delivery",
+            effect_id=recovery_effect_id,
+            tenant_id=tenant_id,
+            profile_id=profile_id,
+            message_bytes=message_bytes,
+            expected_transport_revision=expected_smtp_transport_revision,
+            recipient_count=len(envelope_recipients),
+            resource_type=recovery_resource_type,
+            resource_id=recovery_resource_id,
+        )
+    except MailRecoveryError as exc:
+        raise SmtpConfigurationError(str(exc)) from None
+    if recovery is not None and recovery.replayed:
+        raise SmtpSendError(
+            "The matching SMTP effect already succeeded; reconcile caller state without resending.",
+            outcome_unknown=True,
+        )
     try:
         result = send_email_bytes(
             message_bytes,
@@ -370,18 +396,43 @@ def send_campaign_email_bytes(
             envelope_recipients=envelope_recipients,
         )
     except SmtpSendError as exc:
-        raise _sanitized_smtp_error(exc) from None
+        sanitized = _sanitized_smtp_error(exc)
-    except SmtpConfigurationError:
+        if recovery is not None:
+            if sanitized.outcome_unknown:
+                recovery.unknown(code="smtp_outcome_unknown", summary=str(sanitized))
+            else:
+                recovery.reject(code="smtp_rejected", summary=str(sanitized))
+        raise sanitized from None
+    except SmtpConfigurationError as exc:
+        if recovery is not None:
+            recovery.reject(code="smtp_configuration", summary=str(exc))
         raise SmtpConfigurationError("The selected Mail profile's SMTP configuration is unusable.") from None
     except Exception:
+        if recovery is not None:
+            recovery.unknown(
+                code="unexpected_provider_error",
+                summary="SMTP outcome is unknown after an unexpected provider failure",
+            )
         raise SmtpSendError(
             "Mail delivery outcome is unknown after the provider effect started.",
             outcome_unknown=True,
         ) from None
-    return CampaignSmtpDeliveryResult(
+    sanitized_result = CampaignSmtpDeliveryResult(
         envelope_recipients=list(result.envelope_recipients),
         refused_recipients=_sanitized_refusals(result.refused_recipients),
     )
+    if recovery is not None:
+        try:
+            recovery.succeed_smtp(
+                accepted_count=sanitized_result.accepted_count,
+                refused_recipients=sanitized_result.refused_recipients,
+            )
+        except Exception:
+            raise SmtpSendError(
+                "SMTP returned an outcome, but durable recovery evidence could not be finalized.",
+                outcome_unknown=True,
+            ) from None
+    return sanitized_result
 
 
 def append_campaign_message_to_sent(
@@ -398,6 +449,9 @@ def append_campaign_message_to_sent(
     smtp_credential_id: str | None = None,
     imap_server_id: str | None = None,
     imap_credential_id: str | None = None,
+    recovery_effect_id: str | None = None,
+    recovery_resource_type: str | None = None,
+    recovery_resource_id: str | None = None,
 ) -> CampaignImapAppendResult:
     selection = _selection_payload(
         profile_id=profile_id,
@@ -487,17 +541,57 @@ def append_campaign_message_to_sent(
         )
     except MailProfileError:
         raise MailProfileError("Appending to Sent is blocked by the effective Mail policy.") from None
+    try:
+        recovery = begin_provider_effect_recovery(
+            kind="imap-append",
+            effect_id=recovery_effect_id,
+            tenant_id=tenant_id,
+            profile_id=profile_id,
+            message_bytes=message_bytes,
+            expected_transport_revision=expected_imap_transport_revision,
+            folder=folder,
+            resource_type=recovery_resource_type,
+            resource_id=recovery_resource_id,
+        )
+    except MailRecoveryError as exc:
+        raise ImapConfigurationError(str(exc)) from None
+    if recovery is not None and recovery.replayed:
+        raise ImapAppendError(
+            "The matching IMAP append already succeeded; reconcile caller state without appending again.",
+            outcome_unknown=True,
+        )
     try:
         result = append_message_to_sent(message_bytes, imap_config=imap, folder=folder)
     except ImapAppendError as exc:
-        raise _sanitized_imap_error(exc) from None
+        sanitized = _sanitized_imap_error(exc)
-    except ImapConfigurationError:
+        if recovery is not None:
+            if sanitized.outcome_unknown:
+                recovery.unknown(code="imap_outcome_unknown", summary=str(sanitized))
+            else:
+                recovery.reject(code="imap_rejected", summary=str(sanitized))
+        raise sanitized from None
+    except ImapConfigurationError as exc:
+        if recovery is not None:
+            recovery.reject(code="imap_configuration", summary=str(exc))
         raise ImapConfigurationError("The selected Mail profile's IMAP configuration is unusable.") from None
     except Exception:
+        if recovery is not None:
+            recovery.unknown(
+                code="unexpected_provider_error",
+                summary="IMAP APPEND outcome is unknown after an unexpected provider failure",
+            )
         raise ImapAppendError(
             "The Sent-folder append outcome is unknown; inspect the mailbox before retrying.",
             outcome_unknown=True,
         ) from None
+    if recovery is not None:
+        try:
+            recovery.succeed_imap(folder=result.folder)
+        except Exception:
+            raise ImapAppendError(
+                "IMAP APPEND returned success, but durable recovery evidence could not be finalized.",
+                outcome_unknown=True,
+            ) from None
     return CampaignImapAppendResult(folder=result.folder)
 
 

src/govoplan_mail/backend/delivery_outbox.py

@@ -22,6 +22,10 @@ from govoplan_mail.backend.db.models import (
     MailDeliveryReconciliation,
 )
 from govoplan_mail.backend.mail_profiles import MailProfileError
+from govoplan_mail.backend.recovery import (
+    MailRecoveryError,
+    reconcile_outbox_provider_effect,
+)
 from govoplan_mail.backend.sending.smtp import SmtpConfigurationError, SmtpSendError
 
 
@@ -606,6 +610,11 @@ def _process_claimed(
             expected_smtp_transport_revision=command.expected_smtp_transport_revision,
             smtp_server_id=command.smtp_server_id,
             smtp_credential_id=command.smtp_credential_id,
+            recovery_effect_id=(
+                f"outbox:{command.id}:smtp-attempt:{attempt.attempt_number}"
+            ),
+            recovery_resource_type="mail_delivery_command",
+            recovery_resource_id=command.id,
         )
     except SmtpSendError as exc:
         if exc.outcome_unknown:
@@ -758,6 +767,15 @@ def reconcile_delivery_command(
     clean_evidence = _bounded_text(evidence_reference)
     if not clean_evidence:
         raise MailDeliveryStateError("An evidence reference is required")
+    try:
+        reconcile_outbox_provider_effect(
+            command_id=command.id,
+            effect_occurred=clean_decision == "accepted",
+            evidence_reference=clean_evidence,
+            user_id=user_id,
+        )
+    except MailRecoveryError as exc:
+        raise MailDeliveryStateError(str(exc)) from exc
     item = MailDeliveryReconciliation(
         command_id=command.id,
         decision=clean_decision,

src/govoplan_mail/backend/mailbox_index.py

@@ -86,6 +86,20 @@ def cache_mailbox_folders(
         .filter(MailMailboxFolderIndex.tenant_id == tenant_id, MailMailboxFolderIndex.profile_id == profile_id)
         .all()
     }
+    expected_names = {folder.name for folder in result.folders}
+    removed_names = set(existing) - expected_names
+    if removed_names:
+        (
+            session.query(MailMailboxMessageIndex)
+            .filter(
+                MailMailboxMessageIndex.tenant_id == tenant_id,
+                MailMailboxMessageIndex.profile_id == profile_id,
+                MailMailboxMessageIndex.folder.in_(removed_names),
+            )
+            .delete(synchronize_session=False)
+        )
+        for name in removed_names:
+            session.delete(existing[name])
     for folder in result.folders:
         row = existing.get(folder.name)
         if row is None:

src/govoplan_mail/backend/manifest.py

@@ -639,7 +639,7 @@ manifest = ModuleManifest(
             id="mail.reference.campaign-delivery-contract",
             title="Integrate Campaign through the Mail delivery contract",
             summary="Campaign freezes a Mail profile reference and opaque revision; Mail re-authorizes, revision-checks, resolves credentials, and performs the effect in one call.",
-            body="The mail.campaign_delivery 0.2 contract never returns decrypted credentials or resolved SMTP/IMAP configuration. Mail compares the expected random transport revision before decrypting protocol-specific credentials and returns only bounded sanitized outcomes. Campaign owns ordinary recipient jobs; report messages use Mail's encrypted idempotent delivery-command and attempt ledger. Effect-start evidence prevents blind redelivery, unknown outcomes require explicit reconciliation, and raw recipient refusals require Mail diagnostic authority.",
+            body="The mail.campaign_delivery 0.2 contract never returns decrypted credentials or resolved SMTP/IMAP configuration. Mail compares the expected random transport revision before decrypting protocol-specific credentials and returns only bounded sanitized outcomes. Campaign owns ordinary recipient jobs; report messages use Mail's encrypted idempotent delivery-command and attempt ledger. Every current SMTP and Sent-folder attempt passes a stable effect identifier into a Mail-owned Core recovery operation before provider contact. Effect-start evidence prevents blind redelivery, unknown outcomes require explicit reconciliation, and raw recipient refusals require Mail diagnostic authority.",
             layer="available",
             documentation_types=("admin", "user"),
             audience=("integrator", "campaign_manager", "campaign_sender", "release_reviewer"),
@@ -697,7 +697,7 @@ manifest = ModuleManifest(
         maturity="vertical_slice",
         documentation_ref="docs/MAIL_HANDBOOK.md",
         test_ref="tests/test_delivery_outbox.py",
-        known_limits=("Provider recovery drills and a complete webmail profile are not reference-ready.",),
+        known_limits=("A complete webmail profile and recovery adoption for future provider-side mailbox mutations are not reference-ready.",),
         supported_authority_modes=(
             "external_authoritative",
             "external_mirror",

src/govoplan_mail/backend/recovery.py

@@ -0,0 +1,679 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+from threading import Lock
+from uuid import uuid4
+
+from sqlalchemy import select
+
+from govoplan_core.core.recovery import (
+    RecoveryGuaranteeError,
+    RecoveryMode,
+    RecoveryOperation,
+    RecoveryPlan,
+    RecoveryStatus,
+)
+from govoplan_core.core.recovery_runtime import (
+    DurableRecoveryOperation,
+    RecoveryOperationBusy,
+    RecoveryOperationStateConflict,
+    begin_durable_recovery_operation,
+    claim_durable_recovery_operation,
+)
+from govoplan_core.core.runtime_coordination import process_runtime_identity
+from govoplan_core.db.session import get_database
+from govoplan_mail.backend.db.models import (
+    MailBounceSource,
+    MailMailboxFolderIndex,
+    MailMailboxMessageIndex,
+)
+from govoplan_mail.backend.sending.imap import (
+    ImapFolderListResult,
+    ImapMailboxMessageListResult,
+)
+
+
+class MailRecoveryError(RuntimeError):
+    pass
+
+
+class MailboxRefreshBusy(MailRecoveryError):
+    pass
+
+
+_local_refresh_lock = Lock()
+_local_refreshes: set[str] = set()
+
+
+def _claim_local_refresh(key: str) -> bool:
+    with _local_refresh_lock:
+        if key in _local_refreshes:
+            return False
+        _local_refreshes.add(key)
+        return True
+
+
+def _release_local_refresh(key: str) -> None:
+    with _local_refresh_lock:
+        _local_refreshes.discard(key)
+
+
+def _digest(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def _bounded_idempotency(kind: str, effect_id: str) -> str:
+    return f"mail-{kind}:{_digest(effect_id)}"
+
+
+@dataclass(slots=True)
+class ProviderEffectRecovery:
+    operation: DurableRecoveryOperation | None
+    operation_id: str
+    replayed: bool
+    kind: str
+
+    def reject(self, *, code: str, summary: str) -> None:
+        if self.operation is None:
+            return
+        self.operation.reject(
+            summary=summary,
+            evidence={
+                "verified": True,
+                "checks": {"provider_rejection": code},
+                "effect_kind": self.kind,
+            },
+        )
+
+    def unknown(self, *, code: str, summary: str) -> None:
+        if self.operation is None:
+            return
+        self.operation.unresolved(
+            status=RecoveryStatus.OUTCOME_UNKNOWN,
+            summary=summary,
+            evidence={"effect_started": True, "failure_code": code},
+            failure_summary=(
+                "Inspect the provider and reconcile the outcome before any retry"
+            ),
+        )
+
+    def succeed_smtp(
+        self,
+        *,
+        accepted_count: int,
+        refused_recipients: dict[str, dict[str, int | str]],
+    ) -> None:
+        if self.operation is None:
+            return
+        refused = [
+            {
+                "recipient_sha256": _digest(address.casefold()),
+                "classification": str(item.get("classification") or "unknown"),
+                "status_code": int(item.get("status_code") or 0),
+            }
+            for address, item in sorted(refused_recipients.items())
+        ]
+        evidence = {
+            "verified": True,
+            "checks": {
+                "provider_returned": True,
+                "accepted_count": accepted_count,
+                "refused_count": len(refused),
+            },
+            "accepted_count": accepted_count,
+            "refused": refused,
+        }
+        if accepted_count > 0:
+            self.operation.succeed(evidence=evidence)
+        else:
+            self.operation.reject(
+                summary="SMTP definitively accepted no recipients",
+                evidence=evidence,
+            )
+
+    def succeed_imap(self, *, folder: str) -> None:
+        if self.operation is None:
+            return
+        self.operation.succeed(
+            evidence={
+                "verified": True,
+                "checks": {"provider_append_returned": True},
+                "folder_sha256": _digest(folder),
+            }
+        )
+
+
+def begin_provider_effect_recovery(
+    *,
+    kind: str,
+    effect_id: str | None,
+    tenant_id: str,
+    profile_id: str,
+    message_bytes: bytes,
+    expected_transport_revision: str | None,
+    recipient_count: int | None = None,
+    folder: str | None = None,
+    resource_type: str | None = None,
+    resource_id: str | None = None,
+) -> ProviderEffectRecovery | None:
+    """Fence a Mail-owned provider mutation before any network effect.
+
+    ``effect_id`` is optional only for compatibility with callers predating the
+    recovery contract. Current Mail and Campaign paths always supply one.
+    """
+
+    clean_effect_id = str(effect_id or "").strip()
+    if not clean_effect_id:
+        return None
+    if kind not in {"smtp-delivery", "imap-append"}:
+        raise ValueError("Unsupported Mail provider recovery kind")
+    message_sha256 = hashlib.sha256(message_bytes).hexdigest()
+    request = {
+        "tenant_id": tenant_id,
+        "profile_id": profile_id,
+        "message_sha256": message_sha256,
+        "message_size_bytes": len(message_bytes),
+        "transport_revision": expected_transport_revision,
+        "recipient_count": recipient_count,
+        "folder_sha256": _digest(folder) if folder else None,
+        "effect_id_sha256": _digest(clean_effect_id),
+    }
+    try:
+        started = begin_durable_recovery_operation(
+            get_database().SessionLocal,
+            identity=process_runtime_identity(),
+            module_id="mail",
+            operation_type=kind,
+            idempotency_key=_bounded_idempotency(kind, clean_effect_id),
+            request=request,
+            recovery_plan=RecoveryPlan(
+                mode=RecoveryMode.FORWARD_RECOVERY,
+                preconditions=(
+                    "Mail authorized the profile and effective transport policy",
+                    "the caller supplied a stable effect identifier",
+                    "the request records message and address digests, never content or credentials",
+                ),
+                forward_recovery_steps=(
+                    "inspect provider evidence without repeating the effect",
+                    "record whether the provider accepted the effect",
+                    "retry only under a new deliberate attempt identifier when absence is proven",
+                ),
+                verification_steps=(
+                    "compare the provider outcome with the message digest and effect identifier",
+                    "verify the caller's durable attempt state independently",
+                ),
+            ),
+            precondition_evidence=request,
+            lease_resource_key=f"mail:{kind}:{tenant_id}:{_digest(clean_effect_id)[:40]}",
+            lease_ttl_seconds=15 * 60,
+            resource_type=resource_type,
+            resource_id=resource_id,
+            metadata={"resources": ["postgresql", kind.split("-", 1)[0]]},
+        )
+    except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc:
+        raise MailRecoveryError(
+            "This Mail provider effect already has an active or unresolved recovery record"
+        ) from exc
+    except (RecoveryGuaranteeError, RuntimeError) as exc:
+        raise MailRecoveryError(
+            "The Mail recovery ledger is unavailable; no provider effect was started"
+        ) from exc
+    return ProviderEffectRecovery(
+        operation=started.operation,
+        operation_id=started.operation_id,
+        replayed=started.replayed,
+        kind=kind,
+    )
+
+
+def reconcile_outbox_provider_effect(
+    *,
+    command_id: str,
+    effect_occurred: bool,
+    evidence_reference: str,
+    user_id: str,
+) -> bool:
+    """Resolve the newest unknown SMTP effect for a durable outbox command."""
+
+    factory = get_database().SessionLocal
+    with factory() as evidence_session:
+        operation = evidence_session.scalar(
+            select(RecoveryOperation)
+            .where(
+                RecoveryOperation.module_id == "mail",
+                RecoveryOperation.operation_type == "smtp-delivery",
+                RecoveryOperation.resource_type == "mail_delivery_command",
+                RecoveryOperation.resource_id == command_id,
+            )
+            .order_by(RecoveryOperation.created_at.desc(), RecoveryOperation.id.desc())
+            .limit(1)
+        )
+        if operation is None:
+            return False
+        if effect_occurred and operation.status == RecoveryStatus.SUCCEEDED.value:
+            return True
+        if not effect_occurred and operation.status == RecoveryStatus.RECOVERED.value:
+            return True
+        if operation.status != RecoveryStatus.OUTCOME_UNKNOWN.value:
+            raise MailRecoveryError(
+                f"The Mail recovery record is already {operation.status}"
+            )
+        operation_id = operation.id
+    try:
+        recovery = claim_durable_recovery_operation(
+            factory,
+            identity=process_runtime_identity(),
+            operation_id=operation_id,
+            lease_ttl_seconds=15 * 60,
+        )
+        recovery.resolve_unknown(
+            effect_occurred=effect_occurred,
+            summary="An operator reconciled the SMTP provider outcome",
+            evidence={
+                "verified": True,
+                "checks": {
+                    "provider_evidence_sha256": _digest(evidence_reference),
+                    "reconciled_by_user_id": user_id,
+                },
+                "effect_occurred": effect_occurred,
+            },
+        )
+    except (RecoveryGuaranteeError, RecoveryOperationBusy) as exc:
+        raise MailRecoveryError(
+            "The Mail provider recovery record could not be reconciled"
+        ) from exc
+    return True
+
+
+@dataclass(slots=True)
+class MailboxRefreshRecovery:
+    operation: DurableRecoveryOperation
+    tenant_id: str
+    profile_id: str
+    folder: str
+    local_key: str
+
+    def complete_folders(self, result: ImapFolderListResult) -> None:
+        expected = {item.name for item in result.folders}
+        with get_database().SessionLocal() as session:
+            rows = session.scalars(
+                select(MailMailboxFolderIndex).where(
+                    MailMailboxFolderIndex.tenant_id == self.tenant_id,
+                    MailMailboxFolderIndex.profile_id == self.profile_id,
+                )
+            ).all()
+        present = {row.folder for row in rows}
+        verified = present == expected
+        evidence = {
+            "verified": verified,
+            "checks": {
+                "expected_folder_count": len(expected),
+                "indexed_folder_count": len(present),
+                "expected_folders_sha256": _digest("\n".join(sorted(expected))),
+                "indexed_folders_sha256": _digest("\n".join(sorted(present))),
+            },
+        }
+        try:
+            if verified:
+                self.operation.succeed(evidence=evidence)
+            else:
+                self.operation.unresolved(
+                    status=RecoveryStatus.RECOVERY_REQUIRED,
+                    summary="The IMAP folder read committed an incomplete mailbox index",
+                    evidence=evidence,
+                    failure_summary="Repeat the read-only index refresh under a new fence",
+                )
+        finally:
+            _release_local_refresh(self.local_key)
+
+    def complete_messages(self, result: ImapMailboxMessageListResult) -> None:
+        expected = {item.uid for item in result.messages}
+        with get_database().SessionLocal() as session:
+            rows = session.scalars(
+                select(MailMailboxMessageIndex).where(
+                    MailMailboxMessageIndex.tenant_id == self.tenant_id,
+                    MailMailboxMessageIndex.profile_id == self.profile_id,
+                    MailMailboxMessageIndex.folder == result.folder,
+                    MailMailboxMessageIndex.uid.in_(expected or {""}),
+                )
+            ).all()
+            folder = session.scalar(
+                select(MailMailboxFolderIndex).where(
+                    MailMailboxFolderIndex.tenant_id == self.tenant_id,
+                    MailMailboxFolderIndex.profile_id == self.profile_id,
+                    MailMailboxFolderIndex.folder == result.folder,
+                )
+            )
+        present = {row.uid for row in rows}
+        verified = (
+            present == expected
+            and folder is not None
+            and folder.uidvalidity == result.uidvalidity
+            and folder.message_count == result.total_count
+        )
+        evidence = {
+            "verified": verified,
+            "checks": {
+                "expected_message_count": len(expected),
+                "indexed_message_count": len(present),
+                "uids_sha256": _digest("\n".join(sorted(expected))),
+                "uidvalidity_matches": bool(
+                    folder is not None and folder.uidvalidity == result.uidvalidity
+                ),
+                "total_count_matches": bool(
+                    folder is not None and folder.message_count == result.total_count
+                ),
+            },
+        }
+        try:
+            if verified:
+                self.operation.succeed(evidence=evidence)
+            else:
+                self.operation.unresolved(
+                    status=RecoveryStatus.RECOVERY_REQUIRED,
+                    summary="The IMAP message read committed an incomplete mailbox index",
+                    evidence=evidence,
+                    failure_summary="Repeat the read-only index refresh under a new fence",
+                )
+        finally:
+            _release_local_refresh(self.local_key)
+
+    def complete_bootstrap(
+        self,
+        folders: ImapFolderListResult,
+        messages: ImapMailboxMessageListResult,
+    ) -> None:
+        expected_folders = {item.name for item in folders.folders}
+        expected_uids = {item.uid for item in messages.messages}
+        with get_database().SessionLocal() as session:
+            folder_rows = session.scalars(
+                select(MailMailboxFolderIndex).where(
+                    MailMailboxFolderIndex.tenant_id == self.tenant_id,
+                    MailMailboxFolderIndex.profile_id == self.profile_id,
+                )
+            ).all()
+            message_rows = session.scalars(
+                select(MailMailboxMessageIndex).where(
+                    MailMailboxMessageIndex.tenant_id == self.tenant_id,
+                    MailMailboxMessageIndex.profile_id == self.profile_id,
+                    MailMailboxMessageIndex.folder == messages.folder,
+                    MailMailboxMessageIndex.uid.in_(expected_uids or {""}),
+                )
+            ).all()
+            selected_folder = session.scalar(
+                select(MailMailboxFolderIndex).where(
+                    MailMailboxFolderIndex.tenant_id == self.tenant_id,
+                    MailMailboxFolderIndex.profile_id == self.profile_id,
+                    MailMailboxFolderIndex.folder == messages.folder,
+                )
+            )
+        present_folders = {row.folder for row in folder_rows}
+        present_uids = {row.uid for row in message_rows}
+        verified = (
+            present_folders == expected_folders
+            and present_uids == expected_uids
+            and selected_folder is not None
+            and selected_folder.uidvalidity == messages.uidvalidity
+            and selected_folder.message_count == messages.total_count
+        )
+        evidence = {
+            "verified": verified,
+            "checks": {
+                "expected_folder_count": len(expected_folders),
+                "indexed_folder_count": len(present_folders),
+                "expected_message_count": len(expected_uids),
+                "indexed_message_count": len(present_uids),
+                "folder_set_matches": present_folders == expected_folders,
+                "message_window_matches": present_uids == expected_uids,
+                "uidvalidity_matches": bool(
+                    selected_folder is not None
+                    and selected_folder.uidvalidity == messages.uidvalidity
+                ),
+                "total_count_matches": bool(
+                    selected_folder is not None
+                    and selected_folder.message_count == messages.total_count
+                ),
+            },
+        }
+        try:
+            if verified:
+                self.operation.succeed(evidence=evidence)
+            else:
+                self.operation.unresolved(
+                    status=RecoveryStatus.RECOVERY_REQUIRED,
+                    summary="The IMAP bootstrap committed an incomplete mailbox index",
+                    evidence=evidence,
+                    failure_summary="Repeat the read-only index refresh under a new fence",
+                )
+        finally:
+            _release_local_refresh(self.local_key)
+
+    def reject(self, *, summary: str, code: str) -> None:
+        try:
+            self.operation.reject(
+                summary=summary,
+                evidence={
+                    "verified": True,
+                    "checks": {
+                        "provider_mutation": False,
+                        "business_transaction_rolled_back": True,
+                        "failure_code": code,
+                    },
+                },
+            )
+        finally:
+            _release_local_refresh(self.local_key)
+
+
+@dataclass(slots=True)
+class BounceScanRecovery:
+    operation: DurableRecoveryOperation
+    source_id: str
+    local_key: str
+
+    def complete(self, *, highest_uid: int, uidvalidity: str | None) -> None:
+        with get_database().SessionLocal() as session:
+            source = session.get(MailBounceSource, self.source_id)
+        verified = bool(
+            source is not None
+            and source.highest_processed_uid == highest_uid
+            and source.uidvalidity == uidvalidity
+            and source.last_success_at is not None
+        )
+        evidence = {
+            "verified": verified,
+            "checks": {
+                "source_present": source is not None,
+                "cursor_matches": bool(
+                    source is not None
+                    and source.highest_processed_uid == highest_uid
+                ),
+                "uidvalidity_matches": bool(
+                    source is not None and source.uidvalidity == uidvalidity
+                ),
+                "success_recorded": bool(
+                    source is not None and source.last_success_at is not None
+                ),
+            },
+        }
+        try:
+            if verified:
+                self.operation.succeed(evidence=evidence)
+            else:
+                self.operation.unresolved(
+                    status=RecoveryStatus.RECOVERY_REQUIRED,
+                    summary="The read-only bounce scan did not persist its verified cursor",
+                    evidence=evidence,
+                    failure_summary="Repeat the source scan under a new distributed fence",
+                )
+        finally:
+            _release_local_refresh(self.local_key)
+
+    def reject(self, *, code: str) -> None:
+        try:
+            self.operation.reject(
+                summary="The read-only bounce scan failed before its cursor committed",
+                evidence={
+                    "verified": True,
+                    "checks": {
+                        "provider_mutation": False,
+                        "business_transaction_rolled_back": True,
+                        "failure_code": code,
+                    },
+                },
+            )
+        finally:
+            _release_local_refresh(self.local_key)
+
+
+def begin_mailbox_refresh_recovery(
+    *,
+    tenant_id: str,
+    profile_id: str,
+    folder: str,
+    purpose: str,
+) -> MailboxRefreshRecovery:
+    folder_sha256 = _digest(folder)
+    refresh_id = str(uuid4())
+    local_key = f"mailbox:{tenant_id}:{profile_id}:{folder_sha256}"
+    if not _claim_local_refresh(local_key):
+        raise MailboxRefreshBusy(
+            "Another request in this runtime is already refreshing this mailbox index"
+        )
+    try:
+        started = begin_durable_recovery_operation(
+            get_database().SessionLocal,
+            identity=process_runtime_identity(),
+            module_id="mail",
+            operation_type="mailbox-index-refresh",
+            idempotency_key=f"mailbox-refresh:{refresh_id}",
+            request={
+                "tenant_id": tenant_id,
+                "profile_id": profile_id,
+                "folder_sha256": folder_sha256,
+                "purpose": purpose,
+            },
+            recovery_plan=RecoveryPlan(
+                mode=RecoveryMode.ATOMIC,
+                preconditions=(
+                    "the actor is authorized to read the selected Mail profile",
+                    "the IMAP operation is read-only",
+                ),
+                verification_steps=(
+                    "reload the bounded mailbox index through an independent session",
+                    "compare provider UID and folder metadata with committed cache rows",
+                ),
+            ),
+            precondition_evidence={
+                "profile_id": profile_id,
+                "folder_sha256": folder_sha256,
+                "provider_mutation": False,
+            },
+            lease_resource_key=f"mail:mailbox-refresh:{tenant_id}:{profile_id}:{folder_sha256[:24]}",
+            lease_ttl_seconds=5 * 60,
+            resource_type="mail_profile",
+            resource_id=profile_id,
+            metadata={"resources": ["postgresql", "imap"], "purpose": purpose},
+        )
+    except RecoveryOperationBusy as exc:
+        _release_local_refresh(local_key)
+        raise MailboxRefreshBusy(
+            "Another runtime is already refreshing this mailbox index"
+        ) from exc
+    except (RecoveryOperationStateConflict, RecoveryGuaranteeError, RuntimeError) as exc:
+        _release_local_refresh(local_key)
+        raise MailRecoveryError(
+            "The mailbox recovery fence is unavailable; IMAP was not read"
+        ) from exc
+    if started.operation is None:
+        _release_local_refresh(local_key)
+        raise MailRecoveryError("A mailbox refresh cannot replay a completed read")
+    return MailboxRefreshRecovery(
+        operation=started.operation,
+        tenant_id=tenant_id,
+        profile_id=profile_id,
+        folder=folder,
+        local_key=local_key,
+    )
+
+
+def begin_bounce_scan_recovery(
+    *,
+    tenant_id: str,
+    profile_id: str,
+    source_id: str,
+    folder: str,
+) -> BounceScanRecovery:
+    folder_sha256 = _digest(folder)
+    local_key = f"bounce:{tenant_id}:{source_id}"
+    if not _claim_local_refresh(local_key):
+        raise MailboxRefreshBusy(
+            "Another request in this runtime is already scanning this bounce source"
+        )
+    try:
+        started = begin_durable_recovery_operation(
+            get_database().SessionLocal,
+            identity=process_runtime_identity(),
+            module_id="mail",
+            operation_type="bounce-source-scan",
+            idempotency_key=f"bounce-scan:{uuid4()}",
+            request={
+                "tenant_id": tenant_id,
+                "profile_id": profile_id,
+                "source_id": source_id,
+                "folder_sha256": folder_sha256,
+            },
+            recovery_plan=RecoveryPlan(
+                mode=RecoveryMode.ATOMIC,
+                preconditions=(
+                    "the configured source transport revision still matches",
+                    "the IMAP scan does not alter provider mailbox state",
+                ),
+                verification_steps=(
+                    "reload the bounce source cursor independently",
+                    "verify UIDVALIDITY, highest UID, and success timestamp",
+                ),
+            ),
+            precondition_evidence={
+                "source_id": source_id,
+                "folder_sha256": folder_sha256,
+                "provider_mutation": False,
+            },
+            lease_resource_key=f"mail:bounce-scan:{tenant_id}:{source_id}",
+            lease_ttl_seconds=15 * 60,
+            resource_type="mail_bounce_source",
+            resource_id=source_id,
+            metadata={"resources": ["postgresql", "imap"]},
+        )
+    except RecoveryOperationBusy as exc:
+        _release_local_refresh(local_key)
+        raise MailboxRefreshBusy(
+            "Another runtime is already scanning this bounce source"
+        ) from exc
+    except (RecoveryOperationStateConflict, RecoveryGuaranteeError, RuntimeError) as exc:
+        _release_local_refresh(local_key)
+        raise MailRecoveryError(
+            "The bounce-source recovery fence is unavailable; IMAP was not read"
+        ) from exc
+    if started.operation is None:
+        _release_local_refresh(local_key)
+        raise MailRecoveryError("A bounce-source scan cannot replay a completed read")
+    return BounceScanRecovery(
+        operation=started.operation,
+        source_id=source_id,
+        local_key=local_key,
+    )
+
+
+__all__ = [
+    "MailRecoveryError",
+    "BounceScanRecovery",
+    "MailboxRefreshBusy",
+    "MailboxRefreshRecovery",
+    "ProviderEffectRecovery",
+    "begin_mailbox_refresh_recovery",
+    "begin_bounce_scan_recovery",
+    "begin_provider_effect_recovery",
+    "reconcile_outbox_provider_effect",
+]

src/govoplan_mail/backend/router.py

@@ -89,6 +89,11 @@ from govoplan_mail.backend.mail_profiles import (
 )
 from govoplan_mail.backend.config import ImapConfig, SmtpConfig
 from govoplan_mail.backend.runtime import get_registry
+from govoplan_mail.backend.recovery import (
+    MailRecoveryError,
+    MailboxRefreshBusy,
+    begin_mailbox_refresh_recovery,
+)
 from govoplan_mail.backend.delivery_outbox import (
     MailDeliveryError,
     delivery_command_diagnostics,
@@ -2639,10 +2644,30 @@ def list_profile_mailbox_folders(
                     detected_sent_folder=None,
                 )
                 return _mailbox_folder_response(result, from_cache=True, refreshing=False, indexed_at=cached.indexed_at)
-        result = list_imap_folders(imap_config=imap, include_status=include_status)
+        recovery = begin_mailbox_refresh_recovery(
-        cache_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result)
+            tenant_id=principal.tenant_id,
-        session.commit()
+            profile_id=profile_id,
+            folder="*",
+            purpose="folders",
+        )
+        try:
+            result = list_imap_folders(imap_config=imap, include_status=include_status)
+            cache_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result)
+            session.commit()
+        except Exception as exc:
+            session.rollback()
+            if not recovery.operation.closed:
+                recovery.reject(
+                    summary="The read-only IMAP folder refresh did not commit",
+                    code=exc.__class__.__name__,
+                )
+            raise
+        recovery.complete_folders(result)
         return _mailbox_folder_response(result)
+    except MailboxRefreshBusy as exc:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
+    except MailRecoveryError as exc:
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
     except MailProfileError as exc:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
     except ImapConfigurationError as exc:
@@ -2724,16 +2749,32 @@ def bootstrap_profile_mailbox(
                         ),
                     )
 
-        result = load_imap_mailbox_bootstrap(
+        recovery = begin_mailbox_refresh_recovery(
-            imap_config=imap,
+            tenant_id=principal.tenant_id,
+            profile_id=profile_id,
             folder=folder,
-            limit=limit,
+            purpose="bootstrap",
-            offset=offset,
-            include_folder_status=include_status,
         )
-        cache_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result.folders)
+        try:
-        cache_mailbox_messages(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result.messages)
+            result = load_imap_mailbox_bootstrap(
-        session.commit()
+                imap_config=imap,
+                folder=folder,
+                limit=limit,
+                offset=offset,
+                include_folder_status=include_status,
+            )
+            cache_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result.folders)
+            cache_mailbox_messages(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result.messages)
+            session.commit()
+        except Exception as exc:
+            session.rollback()
+            if not recovery.operation.closed:
+                recovery.reject(
+                    summary="The read-only IMAP bootstrap did not commit",
+                    code=exc.__class__.__name__,
+                )
+            raise
+        recovery.complete_bootstrap(result.folders, result.messages)
         next_cursor, cursor_stable = _next_mailbox_cursor(
             tenant_id=principal.tenant_id,
             profile_id=profile_id,
@@ -2764,6 +2805,10 @@ def bootstrap_profile_mailbox(
                 messages=result.messages.messages,
             ),
         )
+    except MailboxRefreshBusy as exc:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
+    except MailRecoveryError as exc:
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
     except MailProfileError as exc:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
     except ImapConfigurationError as exc:
@@ -2830,16 +2875,32 @@ def list_profile_mailbox_messages(
                     refreshing=False,
                     indexed_at=cached.indexed_at,
                 )
-        result = list_imap_messages(
+        recovery = begin_mailbox_refresh_recovery(
-            imap_config=imap,
+            tenant_id=principal.tenant_id,
+            profile_id=profile_id,
             folder=folder,
-            limit=effective_limit,
+            purpose="messages",
-            offset=effective_offset,
-            after_uid=after_uid,
-            expected_uidvalidity=expected_uidvalidity,
         )
-        cache_mailbox_messages(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result)
+        try:
-        session.commit()
+            result = list_imap_messages(
+                imap_config=imap,
+                folder=folder,
+                limit=effective_limit,
+                offset=effective_offset,
+                after_uid=after_uid,
+                expected_uidvalidity=expected_uidvalidity,
+            )
+            cache_mailbox_messages(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result)
+            session.commit()
+        except Exception as exc:
+            session.rollback()
+            if not recovery.operation.closed:
+                recovery.reject(
+                    summary="The read-only IMAP message refresh did not commit",
+                    code=exc.__class__.__name__,
+                )
+            raise
+        recovery.complete_messages(result)
         full = bool(cursor_values is not None and result.cursor_reset)
         next_cursor, cursor_stable = _next_mailbox_cursor(
             tenant_id=principal.tenant_id,
@@ -2866,6 +2927,10 @@ def list_profile_mailbox_messages(
             full=full or not cursor_stable,
             messages=result.messages,
         )
+    except MailboxRefreshBusy as exc:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
+    except MailRecoveryError as exc:
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
     except MailProfileError as exc:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
     except ImapConfigurationError as exc:

src/govoplan_mail/backend/sending/imap.py

@@ -1203,7 +1203,7 @@ def list_imap_uids_since(
         effective_highest = 0 if cursor_reset else highest_uid
         numeric = sorted(
             int(uid)
-            for uid in _search_all_uids(client)
+            for uid in _search_message_uids(client)
             if uid.isdigit() and int(uid) > effective_highest
         )
         return ImapUidListResult(

tests/test_imap_parser.py

@@ -19,6 +19,7 @@ from govoplan_mail.backend.sending.imap import (
     _sequence_set,
     append_message_to_sent,
     list_imap_messages,
+    list_imap_uids_since,
 )
 
 
@@ -207,6 +208,39 @@ class ImapMessagePaginationTests(unittest.TestCase):
 
 
 class ImapMailboxCommandTests(unittest.TestCase):
+    def test_watcher_lists_new_uids_oldest_first(self):
+        class Client:
+            def uid(self, command, _charset, criterion):
+                self.search = (command, criterion)
+                return "OK", [b"5 9 7"]
+
+            def logout(self):
+                return "BYE", [b"logged out"]
+
+        client = Client()
+        config = ImapConfig(
+            host="imap.example.org",
+            username="user",
+            password="secret",
+        )
+        with (
+            patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client),
+            patch(
+                "govoplan_mail.backend.sending.imap._select_readonly",
+                return_value=(3, "42"),
+            ),
+        ):
+            result = list_imap_uids_since(
+                imap_config=config,
+                folder="INBOX",
+                highest_uid=6,
+                expected_uidvalidity="42",
+                limit=2,
+            )
+
+        self.assertEqual(["7", "9"], result.uids)
+        self.assertEqual(("search", "ALL"), client.search)
+
     def test_select_quotes_mailbox_name_with_spaces(self):
         class Client:
             untagged_responses = {"EXISTS": [b"0"], "UIDVALIDITY": [b"1"]}

tests/test_profile_actor_authorization.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import unittest
 from types import SimpleNamespace
-from unittest.mock import ANY, patch
+from unittest.mock import ANY, Mock, patch
 
 from fastapi import HTTPException
 from sqlalchemy.dialects import postgresql
@@ -791,11 +791,20 @@ class ProfileActorAuthorizationTests(unittest.TestCase):
             ],
             detected_sent_folder=None,
         )
+        recovery = SimpleNamespace(
+            operation=SimpleNamespace(closed=False),
+            reject=Mock(),
+            complete_folders=Mock(),
+        )
         with (
             patch("govoplan_mail.backend.router._imap_config_for_principal", return_value=imap),
             patch("govoplan_mail.backend.router.cached_mailbox_folders", return_value=stale),
             patch("govoplan_mail.backend.router.list_imap_folders", return_value=provider_result) as provider,
             patch("govoplan_mail.backend.router.cache_mailbox_folders") as cache,
+            patch(
+                "govoplan_mail.backend.router.begin_mailbox_refresh_recovery",
+                return_value=recovery,
+            ),
         ):
             response = router.list_profile_mailbox_folders(
                 "profile-1",
@@ -807,6 +816,7 @@ class ProfileActorAuthorizationTests(unittest.TestCase):
 
         provider.assert_called_once_with(imap_config=imap, include_status=False)
         cache.assert_called_once()
+        recovery.complete_folders.assert_called_once_with(provider_result)
         self.assertFalse(response.from_cache)
         self.assertFalse(response.refreshing)
         self.assertEqual(session.commits, 1)

tests/test_recovery.py

@@ -0,0 +1,312 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+from sqlalchemy import Column, String, Table, create_engine, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.core.recovery import (
+    RecoveryCheckpoint,
+    RecoveryGuaranteeError,
+    RecoveryOperation,
+    RecoveryStatus,
+)
+from govoplan_core.core.runtime_coordination import (
+    DistributedLease,
+    RuntimeIdentity,
+    bind_process_runtime_identity,
+)
+from govoplan_core.db.base import Base
+from govoplan_core.db.session import configure_database, reset_database
+from govoplan_mail.backend.db.models import (
+    MailMailboxFolderIndex,
+    MailMailboxMessageIndex,
+    MailServerProfile,
+)
+from govoplan_mail.backend.mailbox_index import cache_mailbox_folders
+from govoplan_mail.backend.recovery import (
+    MailRecoveryError,
+    MailboxRefreshBusy,
+    begin_mailbox_refresh_recovery,
+    begin_provider_effect_recovery,
+    reconcile_outbox_provider_effect,
+)
+from govoplan_mail.backend.sending.imap import (
+    ImapFolderListResult,
+    ImapMailboxInfo,
+)
+
+
+class MailRecoveryTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tempdir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tempdir.cleanup)
+        database_path = Path(self.tempdir.name) / "mail-recovery.sqlite3"
+        self.engine = create_engine(f"sqlite:///{database_path}")
+        access_users = Base.metadata.tables.get("access_users")
+        if access_users is None:
+            access_users = Table(
+                "access_users",
+                Base.metadata,
+                Column("id", String(36), primary_key=True),
+            )
+        Base.metadata.create_all(
+            self.engine,
+            tables=[
+                access_users,
+                DistributedLease.__table__,
+                RecoveryOperation.__table__,
+                RecoveryCheckpoint.__table__,
+                MailServerProfile.__table__,
+                MailMailboxFolderIndex.__table__,
+                MailMailboxMessageIndex.__table__,
+            ],
+        )
+        configure_database(
+            f"sqlite:///{database_path}",
+            engine=self.engine,
+            dispose_previous=True,
+        )
+        bind_process_runtime_identity(
+            RuntimeIdentity(
+                installation_id="mail-recovery-tests",
+                node_id="mail-test-node",
+                incarnation="mail-test-incarnation",
+                role="worker",
+                software_version="test",
+                composition_hash="a" * 64,
+            )
+        )
+        self.SessionLocal = sessionmaker(
+            bind=self.engine,
+            class_=Session,
+            expire_on_commit=False,
+        )
+        with self.SessionLocal() as session:
+            session.add(
+                MailServerProfile(
+                    id="profile-1",
+                    tenant_id="tenant-1",
+                    scope_type="tenant",
+                    scope_id="tenant-1",
+                    name="Recovery profile",
+                    slug="recovery-profile",
+                    smtp_config={},
+                )
+            )
+            session.commit()
+        self.addCleanup(self._cleanup_runtime)
+
+    def _cleanup_runtime(self) -> None:
+        bind_process_runtime_identity(None)
+        reset_database()
+        self.engine.dispose()
+
+    def test_smtp_effect_is_durable_before_provider_and_redacted_on_success(self) -> None:
+        recovery = begin_provider_effect_recovery(
+            kind="smtp-delivery",
+            effect_id="outbox:command-1:smtp-attempt:1",
+            tenant_id="tenant-1",
+            profile_id="profile-1",
+            message_bytes=b"Subject: Recovery\r\n\r\nBody",
+            expected_transport_revision="revision-1",
+            recipient_count=1,
+            resource_type="mail_delivery_command",
+            resource_id="command-1",
+        )
+        assert recovery is not None and recovery.operation is not None
+        with self.SessionLocal() as session:
+            operation = session.get(RecoveryOperation, recovery.operation_id)
+            assert operation is not None
+            self.assertEqual(RecoveryStatus.RUNNING.value, operation.status)
+
+        recovery.succeed_smtp(
+            accepted_count=1,
+            refused_recipients={},
+        )
+
+        with self.SessionLocal() as session:
+            operation = session.get(RecoveryOperation, recovery.operation_id)
+            assert operation is not None
+            self.assertEqual(RecoveryStatus.SUCCEEDED.value, operation.status)
+            evidence = json.dumps(
+                [
+                    item.evidence
+                    for item in session.scalars(
+                        select(RecoveryCheckpoint).where(
+                            RecoveryCheckpoint.operation_id == operation.id
+                        )
+                    )
+                ]
+            )
+            self.assertNotIn("recipient@example.test", evidence)
+            self.assertNotIn("Subject: Recovery", evidence)
+
+    def test_unknown_smtp_outcome_blocks_replay_until_reconciled(self) -> None:
+        kwargs = {
+            "kind": "smtp-delivery",
+            "effect_id": "outbox:command-2:smtp-attempt:1",
+            "tenant_id": "tenant-1",
+            "profile_id": "profile-1",
+            "message_bytes": b"Subject: Unknown\r\n\r\nBody",
+            "expected_transport_revision": "revision-1",
+            "recipient_count": 1,
+            "resource_type": "mail_delivery_command",
+            "resource_id": "command-2",
+        }
+        recovery = begin_provider_effect_recovery(**kwargs)
+        assert recovery is not None
+        recovery.unknown(
+            code="socket_closed_after_data",
+            summary="SMTP outcome is unknown",
+        )
+
+        with self.assertRaises(MailRecoveryError):
+            begin_provider_effect_recovery(**kwargs)
+
+        self.assertTrue(
+            reconcile_outbox_provider_effect(
+                command_id="command-2",
+                effect_occurred=False,
+                evidence_reference="provider-case-42",
+                user_id="operator-1",
+            )
+        )
+        with self.SessionLocal() as session:
+            operation = session.get(RecoveryOperation, recovery.operation_id)
+            assert operation is not None
+            self.assertEqual(RecoveryStatus.RECOVERED.value, operation.status)
+
+    def test_mailbox_refresh_verifies_the_committed_index(self) -> None:
+        result = ImapFolderListResult(
+            host="imap.example.test",
+            port=993,
+            security="tls",
+            folders=[
+                ImapMailboxInfo(
+                    name="INBOX",
+                    flags=["\\HasNoChildren"],
+                    message_count=4,
+                    unseen_count=1,
+                )
+            ],
+        )
+        recovery = begin_mailbox_refresh_recovery(
+            tenant_id="tenant-1",
+            profile_id="profile-1",
+            folder="*",
+            purpose="folders",
+        )
+        with self.SessionLocal() as session:
+            session.add(
+                MailMailboxFolderIndex(
+                    tenant_id="tenant-1",
+                    profile_id="profile-1",
+                    folder="Removed",
+                    flags=[],
+                )
+            )
+            session.commit()
+            cache_mailbox_folders(
+                session,
+                tenant_id="tenant-1",
+                profile_id="profile-1",
+                result=result,
+            )
+            session.commit()
+        recovery.complete_folders(result)
+
+        with self.SessionLocal() as session:
+            self.assertEqual(
+                ["INBOX"],
+                list(
+                    session.scalars(
+                        select(MailMailboxFolderIndex.folder).order_by(
+                            MailMailboxFolderIndex.folder
+                        )
+                    )
+                ),
+            )
+            operation = session.get(
+                RecoveryOperation,
+                recovery.operation.operation_id,
+            )
+            assert operation is not None
+            self.assertEqual(RecoveryStatus.SUCCEEDED.value, operation.status)
+
+    def test_missing_runtime_identity_fails_before_a_provider_effect_can_start(self) -> None:
+        bind_process_runtime_identity(None)
+        with self.assertRaises(MailRecoveryError):
+            begin_provider_effect_recovery(
+                kind="imap-append",
+                effect_id="campaign-job:job-1:imap-attempt:1",
+                tenant_id="tenant-1",
+                profile_id="profile-1",
+                message_bytes=b"message",
+                expected_transport_revision="revision-1",
+                folder="Sent",
+            )
+
+    def test_tampered_provider_evidence_cannot_be_marked_successful(self) -> None:
+        recovery = begin_provider_effect_recovery(
+            kind="smtp-delivery",
+            effect_id="outbox:command-3:smtp-attempt:1",
+            tenant_id="tenant-1",
+            profile_id="profile-1",
+            message_bytes=b"message",
+            expected_transport_revision="revision-1",
+            recipient_count=1,
+            resource_type="mail_delivery_command",
+            resource_id="command-3",
+        )
+        assert recovery is not None
+        with self.SessionLocal() as session:
+            checkpoint = 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"
+            session.commit()
+
+        with self.assertRaises(RecoveryGuaranteeError):
+            recovery.succeed_smtp(accepted_count=1, refused_recipients={})
+        with self.SessionLocal() as session:
+            operation = session.get(RecoveryOperation, recovery.operation_id)
+            assert operation is not None
+            self.assertNotEqual(RecoveryStatus.SUCCEEDED.value, operation.status)
+
+    def test_mailbox_refresh_has_a_cross_runtime_fence(self) -> None:
+        recovery = begin_mailbox_refresh_recovery(
+            tenant_id="tenant-1",
+            profile_id="profile-1",
+            folder="INBOX",
+            purpose="messages",
+        )
+        bind_process_runtime_identity(
+            RuntimeIdentity(
+                installation_id="mail-recovery-tests",
+                node_id="mail-test-node-2",
+                incarnation="mail-test-incarnation-2",
+                role="worker",
+                software_version="test",
+                composition_hash="a" * 64,
+            )
+        )
+        with self.assertRaises(MailboxRefreshBusy):
+            begin_mailbox_refresh_recovery(
+                tenant_id="tenant-1",
+                profile_id="profile-1",
+                folder="INBOX",
+                purpose="messages",
+            )
+        recovery.reject(summary="Test refresh stopped", code="test")
+
+
+if __name__ == "__main__":
+    unittest.main()