feat: add durable mail delivery outbox

src/govoplan_mail/backend/capabilities.py

@@ -511,6 +511,32 @@ class MailCampaignCapability:
     append_campaign_message_to_sent = staticmethod(append_campaign_message_to_sent)
     wait_for_rate_limit = staticmethod(wait_for_rate_limit)
 
+    @staticmethod
+    def submit_delivery_command(session: Session, **kwargs: Any) -> dict[str, object]:
+        from govoplan_mail.backend.delivery_outbox import submit_delivery_command
+
+        return submit_delivery_command(session, **kwargs)
+
+    @staticmethod
+    def delivery_command_summary(
+        session: Session,
+        *,
+        tenant_id: str,
+        command_id: str,
+    ) -> dict[str, object]:
+        from govoplan_mail.backend.delivery_outbox import (
+            delivery_command_summary,
+            get_delivery_command,
+        )
+
+        return delivery_command_summary(
+            get_delivery_command(
+                session,
+                tenant_id=tenant_id,
+                command_id=command_id,
+            )
+        )
+
     @staticmethod
     def mock_mailbox():
         from govoplan_mail.backend.dev import mock_mailbox
@@ -521,3 +547,10 @@ class MailCampaignCapability:
 def campaign_capability(context: ModuleContext) -> MailCampaignCapability:
     configure_runtime(settings=context.settings)
     return MailCampaignCapability()
+
+
+def delivery_outbox_capability(context: ModuleContext):
+    from govoplan_mail.backend.delivery_outbox import MailDeliveryOutboxCapability
+
+    configure_runtime(settings=context.settings)
+    return MailDeliveryOutboxCapability()

src/govoplan_mail/backend/db/models.py

@@ -152,3 +152,137 @@ class MailMailboxMessageIndex(Base, TimestampMixin):
     body_preview: Mapped[str | None] = mapped_column(Text, nullable=True)
     attachment_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
     indexed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True)
+
+
+class MailDeliveryCommand(Base, TimestampMixin):
+    __tablename__ = "mail_delivery_commands"
+    __table_args__ = (
+        UniqueConstraint(
+            "tenant_id",
+            "command_type",
+            "idempotency_key",
+            name="uq_mail_delivery_commands_idempotency",
+        ),
+        Index(
+            "ix_mail_delivery_commands_dispatch",
+            "status",
+            "next_attempt_at",
+            "created_at",
+        ),
+        Index(
+            "ix_mail_delivery_commands_source",
+            "tenant_id",
+            "source_module",
+            "source_resource_type",
+            "source_resource_id",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    command_type: Mapped[str] = mapped_column(String(60), nullable=False, index=True)
+    source_module: Mapped[str] = mapped_column(String(80), nullable=False, index=True)
+    source_resource_type: Mapped[str] = mapped_column(String(80), nullable=False)
+    source_resource_id: Mapped[str | None] = mapped_column(String(120), nullable=True)
+    source_version_id: Mapped[str | None] = mapped_column(String(120), nullable=True)
+    idempotency_key: Mapped[str] = mapped_column(String(200), nullable=False)
+    canonical_request_hash: Mapped[str] = mapped_column(String(64), nullable=False)
+    profile_id: Mapped[str] = mapped_column(
+        ForeignKey("mail_server_profiles.id", ondelete="RESTRICT"),
+        nullable=False,
+        index=True,
+    )
+    smtp_server_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
+    smtp_credential_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
+    expected_smtp_transport_revision: Mapped[str] = mapped_column(String(120), nullable=False)
+    envelope_from_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True)
+    envelope_recipients_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True)
+    from_header_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True)
+    message_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True)
+    message_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
+    message_size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    recipient_count: Mapped[int] = mapped_column(Integer, nullable=False)
+    status: Mapped[str] = mapped_column(String(40), nullable=False, default="pending", index=True)
+    attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    next_attempt_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+        index=True,
+    )
+    claimed_by: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    claimed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    effect_started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    accepted_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    refused_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    refusal_summary: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default=dict)
+    refusal_details_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True)
+    failure_code: Mapped[str | None] = mapped_column(String(80), nullable=True)
+    failure_summary: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    created_by_user_id: Mapped[str | None] = mapped_column(
+        ForeignKey("access_users.id", ondelete="SET NULL"),
+        nullable=True,
+        index=True,
+    )
+    supersedes_command_id: Mapped[str | None] = mapped_column(
+        ForeignKey("mail_delivery_commands.id", ondelete="SET NULL"),
+        nullable=True,
+        index=True,
+    )
+    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True)
+    payload_purged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+
+class MailDeliveryAttempt(Base, TimestampMixin):
+    __tablename__ = "mail_delivery_attempts"
+    __table_args__ = (
+        UniqueConstraint(
+            "command_id",
+            "attempt_number",
+            name="uq_mail_delivery_attempts_number",
+        ),
+        Index("ix_mail_delivery_attempts_command_started", "command_id", "started_at"),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    command_id: Mapped[str] = mapped_column(
+        ForeignKey("mail_delivery_commands.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    attempt_number: Mapped[int] = mapped_column(Integer, nullable=False)
+    worker_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    status: Mapped[str] = mapped_column(String(40), nullable=False, index=True)
+    started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    effect_started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    accepted_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    refused_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    outcome_code: Mapped[str | None] = mapped_column(String(80), nullable=True)
+    diagnostic_summary: Mapped[str | None] = mapped_column(String(500), nullable=True)
+
+
+class MailDeliveryReconciliation(Base, TimestampMixin):
+    __tablename__ = "mail_delivery_reconciliations"
+    __table_args__ = (
+        Index(
+            "ix_mail_delivery_reconciliations_command_created",
+            "command_id",
+            "created_at",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    command_id: Mapped[str] = mapped_column(
+        ForeignKey("mail_delivery_commands.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    decision: Mapped[str] = mapped_column(String(40), nullable=False)
+    evidence_reference: Mapped[str] = mapped_column(String(500), nullable=False)
+    note_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_by_user_id: Mapped[str | None] = mapped_column(
+        ForeignKey("access_users.id", ondelete="SET NULL"),
+        nullable=True,
+        index=True,
+    )

src/govoplan_mail/backend/delivery_outbox.py

@@ -0,0 +1,880 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+from collections import Counter
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from sqlalchemy import and_, or_, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from govoplan_core.audit.logging import audit_event
+from govoplan_core.security.secrets import decrypt_secret, encrypt_secret
+from govoplan_mail.backend.capabilities import send_campaign_email_bytes
+from govoplan_mail.backend.db.models import (
+    MailDeliveryAttempt,
+    MailDeliveryCommand,
+    MailDeliveryReconciliation,
+)
+from govoplan_mail.backend.mail_profiles import MailProfileError
+from govoplan_mail.backend.sending.smtp import SmtpConfigurationError, SmtpSendError
+
+
+DISPATCHABLE_STATUSES = frozenset({"pending", "temporary_failure", "reconciled_not_accepted"})
+TERMINAL_STATUSES = frozenset(
+    {
+        "accepted",
+        "partially_refused",
+        "permanent_failure",
+        "outcome_unknown",
+        "reconciled_accepted",
+        "cancelled",
+    }
+)
+NON_RETRYABLE_STATUSES = TERMINAL_STATUSES | frozenset({"claimed", "in_progress"})
+DEFAULT_PAYLOAD_RETENTION_DAYS = 30
+STALE_CLAIM_AFTER = timedelta(minutes=10)
+
+
+class MailDeliveryError(RuntimeError):
+    pass
+
+
+class MailDeliveryIdempotencyConflict(MailDeliveryError):
+    pass
+
+
+class MailDeliveryNotFound(MailDeliveryError):
+    pass
+
+
+class MailDeliveryStateError(MailDeliveryError):
+    pass
+
+
+def utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _bounded_text(value: object | None, *, limit: int = 500) -> str | None:
+    if value is None:
+        return None
+    candidate = " ".join(str(value).split())
+    return candidate[:limit] or None
+
+
+def _canonical_hash(payload: dict[str, object]) -> str:
+    encoded = json.dumps(
+        payload,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=True,
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _encrypt_json(value: object) -> str:
+    encrypted = encrypt_secret(
+        json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
+    )
+    if not encrypted:
+        raise MailDeliveryError("Mail delivery evidence could not be encrypted")
+    return encrypted
+
+
+def _decrypt_json(value: str | None) -> Any:
+    plaintext = decrypt_secret(value)
+    if plaintext is None:
+        return None
+    return json.loads(plaintext)
+
+
+def _message_bytes(command: MailDeliveryCommand) -> bytes:
+    encoded = decrypt_secret(command.message_encrypted)
+    if encoded is None:
+        raise MailDeliveryStateError("Mail delivery payload is no longer available")
+    try:
+        message = base64.b64decode(encoded.encode("ascii"), validate=True)
+    except (ValueError, TypeError) as exc:
+        raise MailDeliveryStateError("Mail delivery payload is invalid") from exc
+    if hashlib.sha256(message).hexdigest() != command.message_sha256:
+        raise MailDeliveryStateError("Mail delivery payload integrity check failed")
+    return message
+
+
+def _delivery_payload(
+    *,
+    command_type: str,
+    source_module: str,
+    source_resource_type: str,
+    source_resource_id: str | None,
+    source_version_id: str | None,
+    profile_id: str,
+    smtp_server_id: str | None,
+    smtp_credential_id: str | None,
+    expected_smtp_transport_revision: str,
+    envelope_from: str,
+    envelope_recipients: list[str],
+    from_header: str | None,
+    message_sha256: str,
+) -> dict[str, object]:
+    return {
+        "command_type": command_type,
+        "source_module": source_module,
+        "source_resource_type": source_resource_type,
+        "source_resource_id": source_resource_id,
+        "source_version_id": source_version_id,
+        "profile_id": profile_id,
+        "smtp_server_id": smtp_server_id,
+        "smtp_credential_id": smtp_credential_id,
+        "expected_smtp_transport_revision": expected_smtp_transport_revision,
+        "envelope_from": envelope_from,
+        "envelope_recipients": envelope_recipients,
+        "from_header": from_header,
+        "message_sha256": message_sha256,
+    }
+
+
+def submit_delivery_command(
+    session: Session,
+    *,
+    tenant_id: str,
+    command_type: str,
+    source_module: str,
+    source_resource_type: str,
+    source_resource_id: str | None,
+    source_version_id: str | None,
+    idempotency_key: str,
+    profile_id: str,
+    message_bytes: bytes,
+    envelope_from: str,
+    envelope_recipients: list[str],
+    from_header: str | None,
+    expected_smtp_transport_revision: str,
+    smtp_server_id: str | None = None,
+    smtp_credential_id: str | None = None,
+    created_by_user_id: str | None = None,
+    retention_days: int = DEFAULT_PAYLOAD_RETENTION_DAYS,
+    supersedes_command_id: str | None = None,
+) -> dict[str, object]:
+    clean_key = idempotency_key.strip()
+    clean_recipients = [str(value).strip() for value in envelope_recipients if str(value).strip()]
+    if not clean_key or len(clean_key) > 200:
+        raise MailDeliveryError("A bounded idempotency key is required")
+    if not clean_recipients:
+        raise MailDeliveryError("At least one envelope recipient is required")
+    if retention_days < 1:
+        raise MailDeliveryError("Mail payload retention must be at least one day")
+    digest = hashlib.sha256(message_bytes).hexdigest()
+    request_hash = _canonical_hash(
+        _delivery_payload(
+            command_type=command_type,
+            source_module=source_module,
+            source_resource_type=source_resource_type,
+            source_resource_id=source_resource_id,
+            source_version_id=source_version_id,
+            profile_id=profile_id,
+            smtp_server_id=smtp_server_id,
+            smtp_credential_id=smtp_credential_id,
+            expected_smtp_transport_revision=expected_smtp_transport_revision,
+            envelope_from=envelope_from,
+            envelope_recipients=clean_recipients,
+            from_header=from_header,
+            message_sha256=digest,
+        )
+    )
+    existing = session.scalar(
+        select(MailDeliveryCommand).where(
+            MailDeliveryCommand.tenant_id == tenant_id,
+            MailDeliveryCommand.command_type == command_type,
+            MailDeliveryCommand.idempotency_key == clean_key,
+        )
+    )
+    if existing is not None:
+        if existing.canonical_request_hash != request_hash:
+            raise MailDeliveryIdempotencyConflict(
+                "The idempotency key is already bound to a different mail command"
+            )
+        return delivery_command_summary(existing, duplicate=True)
+
+    now = utcnow()
+    command = MailDeliveryCommand(
+        tenant_id=tenant_id,
+        command_type=command_type,
+        source_module=source_module,
+        source_resource_type=source_resource_type,
+        source_resource_id=source_resource_id,
+        source_version_id=source_version_id,
+        idempotency_key=clean_key,
+        canonical_request_hash=request_hash,
+        profile_id=profile_id,
+        smtp_server_id=smtp_server_id,
+        smtp_credential_id=smtp_credential_id,
+        expected_smtp_transport_revision=expected_smtp_transport_revision,
+        envelope_from_encrypted=encrypt_secret(envelope_from),
+        envelope_recipients_encrypted=_encrypt_json(clean_recipients),
+        from_header_encrypted=encrypt_secret(from_header),
+        message_encrypted=encrypt_secret(base64.b64encode(message_bytes).decode("ascii")),
+        message_sha256=digest,
+        message_size_bytes=len(message_bytes),
+        recipient_count=len(clean_recipients),
+        status="pending",
+        next_attempt_at=now,
+        created_by_user_id=created_by_user_id,
+        supersedes_command_id=supersedes_command_id,
+        expires_at=now + timedelta(days=retention_days),
+    )
+    try:
+        with session.begin_nested():
+            session.add(command)
+            session.flush()
+    except IntegrityError:
+        existing = session.scalar(
+            select(MailDeliveryCommand).where(
+                MailDeliveryCommand.tenant_id == tenant_id,
+                MailDeliveryCommand.command_type == command_type,
+                MailDeliveryCommand.idempotency_key == clean_key,
+            )
+        )
+        if existing is None or existing.canonical_request_hash != request_hash:
+            raise MailDeliveryIdempotencyConflict(
+                "The idempotency key is already bound to a different mail command"
+            ) from None
+        return delivery_command_summary(existing, duplicate=True)
+
+    audit_event(
+        session,
+        tenant_id=tenant_id,
+        user_id=created_by_user_id,
+        action="mail.delivery_requested",
+        object_type="mail_delivery_command",
+        object_id=command.id,
+        details={
+            "command_type": command_type,
+            "source_module": source_module,
+            "source_resource_type": source_resource_type,
+            "source_resource_id": source_resource_id,
+            "recipient_count": len(clean_recipients),
+            "message_sha256": digest,
+            "supersedes_command_id": supersedes_command_id,
+        },
+    )
+    return delivery_command_summary(command)
+
+
+def get_delivery_command(
+    session: Session,
+    *,
+    tenant_id: str,
+    command_id: str,
+) -> MailDeliveryCommand:
+    command = session.get(MailDeliveryCommand, command_id)
+    if command is None or command.tenant_id != tenant_id:
+        raise MailDeliveryNotFound("Mail delivery command not found")
+    return command
+
+
+def delivery_command_summary(
+    command: MailDeliveryCommand,
+    *,
+    duplicate: bool = False,
+) -> dict[str, object]:
+    return {
+        "id": command.id,
+        "tenant_id": command.tenant_id,
+        "command_type": command.command_type,
+        "source_module": command.source_module,
+        "source_resource_type": command.source_resource_type,
+        "source_resource_id": command.source_resource_id,
+        "source_version_id": command.source_version_id,
+        "status": command.status,
+        "recipient_count": command.recipient_count,
+        "accepted_count": command.accepted_count,
+        "refused_count": command.refused_count,
+        "refusal_summary": dict(command.refusal_summary or {}),
+        "attempt_count": command.attempt_count,
+        "failure_code": command.failure_code,
+        "failure_summary": command.failure_summary,
+        "message_sha256": command.message_sha256,
+        "message_size_bytes": command.message_size_bytes,
+        "created_at": command.created_at,
+        "completed_at": command.completed_at,
+        "expires_at": command.expires_at,
+        "payload_purged_at": command.payload_purged_at,
+        "safe_to_retry": command.status in DISPATCHABLE_STATUSES,
+        "outcome_known": command.status
+        not in {"claimed", "in_progress", "outcome_unknown"},
+        "duplicate": duplicate,
+    }
+
+
+def delivery_command_diagnostics(
+    session: Session,
+    *,
+    tenant_id: str,
+    command_id: str,
+) -> dict[str, object]:
+    command = get_delivery_command(session, tenant_id=tenant_id, command_id=command_id)
+    attempts = session.scalars(
+        select(MailDeliveryAttempt)
+        .where(MailDeliveryAttempt.command_id == command.id)
+        .order_by(MailDeliveryAttempt.attempt_number)
+    ).all()
+    reconciliations = session.scalars(
+        select(MailDeliveryReconciliation)
+        .where(MailDeliveryReconciliation.command_id == command.id)
+        .order_by(MailDeliveryReconciliation.created_at)
+    ).all()
+    refusals = _decrypt_json(command.refusal_details_encrypted) or {}
+    return {
+        **delivery_command_summary(command),
+        "refused_recipients": refusals,
+        "attempts": [
+            {
+                "id": attempt.id,
+                "attempt_number": attempt.attempt_number,
+                "worker_id": attempt.worker_id,
+                "status": attempt.status,
+                "started_at": attempt.started_at,
+                "effect_started_at": attempt.effect_started_at,
+                "completed_at": attempt.completed_at,
+                "accepted_count": attempt.accepted_count,
+                "refused_count": attempt.refused_count,
+                "outcome_code": attempt.outcome_code,
+                "diagnostic_summary": attempt.diagnostic_summary,
+            }
+            for attempt in attempts
+        ],
+        "reconciliations": [
+            {
+                "id": item.id,
+                "decision": item.decision,
+                "evidence_reference": item.evidence_reference,
+                "note": decrypt_secret(item.note_encrypted),
+                "created_by_user_id": item.created_by_user_id,
+                "created_at": item.created_at,
+            }
+            for item in reconciliations
+        ],
+    }
+
+
+def _current_attempt(
+    session: Session,
+    command: MailDeliveryCommand,
+) -> MailDeliveryAttempt | None:
+    return session.scalar(
+        select(MailDeliveryAttempt)
+        .where(
+            MailDeliveryAttempt.command_id == command.id,
+            MailDeliveryAttempt.attempt_number == command.attempt_count,
+        )
+        .limit(1)
+    )
+
+
+def _recover_stale_commands(
+    session: Session,
+    *,
+    now: datetime,
+    tenant_id: str | None,
+) -> tuple[int, int]:
+    cutoff = now - STALE_CLAIM_AFTER
+    clauses = [
+        MailDeliveryCommand.status.in_(("claimed", "in_progress")),
+        MailDeliveryCommand.claimed_at <= cutoff,
+    ]
+    if tenant_id:
+        clauses.append(MailDeliveryCommand.tenant_id == tenant_id)
+    commands = session.scalars(
+        select(MailDeliveryCommand).where(*clauses).order_by(MailDeliveryCommand.claimed_at)
+    ).all()
+    recovered = 0
+    unknown = 0
+    for command in commands:
+        attempt = _current_attempt(session, command)
+        effect_started = command.effect_started_at or (
+            attempt.effect_started_at if attempt else None
+        )
+        if effect_started is None:
+            command.status = "pending"
+            command.next_attempt_at = now
+            command.claimed_at = None
+            command.claimed_by = None
+            if attempt is not None:
+                attempt.status = "claim_abandoned"
+                attempt.completed_at = now
+                attempt.outcome_code = "worker_lost_before_effect"
+            recovered += 1
+            continue
+        command.status = "outcome_unknown"
+        command.completed_at = now
+        command.next_attempt_at = None
+        command.failure_code = "worker_lost_after_effect_start"
+        command.failure_summary = (
+            "Delivery outcome is unknown because the worker stopped after transmission began."
+        )
+        if attempt is not None:
+            attempt.status = "outcome_unknown"
+            attempt.completed_at = now
+            attempt.outcome_code = command.failure_code
+            attempt.diagnostic_summary = command.failure_summary
+        unknown += 1
+    if commands:
+        session.commit()
+    return recovered, unknown
+
+
+def _claim_command(
+    session: Session,
+    *,
+    command_id: str,
+    now: datetime,
+    worker_id: str | None,
+) -> tuple[MailDeliveryCommand, MailDeliveryAttempt] | None:
+    command = session.scalar(
+        select(MailDeliveryCommand)
+        .where(
+            MailDeliveryCommand.id == command_id,
+            MailDeliveryCommand.status.in_(DISPATCHABLE_STATUSES),
+            or_(
+                MailDeliveryCommand.next_attempt_at.is_(None),
+                MailDeliveryCommand.next_attempt_at <= now,
+            ),
+            MailDeliveryCommand.payload_purged_at.is_(None),
+        )
+        .with_for_update(skip_locked=True)
+    )
+    if command is None:
+        session.rollback()
+        return None
+    command.attempt_count += 1
+    command.status = "claimed"
+    command.claimed_by = _bounded_text(worker_id, limit=255)
+    command.claimed_at = now
+    command.effect_started_at = None
+    command.next_attempt_at = None
+    command.failure_code = None
+    command.failure_summary = None
+    attempt = MailDeliveryAttempt(
+        command_id=command.id,
+        attempt_number=command.attempt_count,
+        worker_id=command.claimed_by,
+        status="claimed",
+        started_at=now,
+    )
+    session.add(attempt)
+    session.commit()
+    return command, attempt
+
+
+def _mark_effect_started(
+    session: Session,
+    command: MailDeliveryCommand,
+    attempt: MailDeliveryAttempt,
+) -> None:
+    now = utcnow()
+    command.status = "in_progress"
+    command.effect_started_at = now
+    attempt.status = "in_progress"
+    attempt.effect_started_at = now
+    session.commit()
+
+
+def _refusal_summary(refusals: dict[str, dict[str, int | str]]) -> dict[str, int]:
+    classifications = Counter(
+        str(item.get("classification") or "unknown")
+        for item in refusals.values()
+    )
+    return dict(sorted(classifications.items()))
+
+
+def _record_outcome(
+    session: Session,
+    *,
+    command: MailDeliveryCommand,
+    attempt: MailDeliveryAttempt,
+    status: str,
+    accepted_count: int = 0,
+    refusals: dict[str, dict[str, int | str]] | None = None,
+    failure_code: str | None = None,
+    failure_summary: str | None = None,
+) -> None:
+    now = utcnow()
+    refusal_map = refusals or {}
+    command.status = status
+    command.accepted_count = accepted_count
+    command.refused_count = len(refusal_map)
+    command.refusal_summary = _refusal_summary(refusal_map)
+    command.refusal_details_encrypted = (
+        _encrypt_json(refusal_map) if refusal_map else None
+    )
+    command.failure_code = failure_code
+    command.failure_summary = _bounded_text(failure_summary)
+    command.claimed_by = None
+    command.claimed_at = None
+    command.next_attempt_at = (
+        now + timedelta(minutes=min(60, 2 ** min(command.attempt_count, 5)))
+        if status == "temporary_failure"
+        else None
+    )
+    if status != "temporary_failure":
+        command.completed_at = now
+    attempt.status = status
+    attempt.completed_at = now
+    attempt.accepted_count = accepted_count
+    attempt.refused_count = len(refusal_map)
+    attempt.outcome_code = failure_code or status
+    attempt.diagnostic_summary = _bounded_text(failure_summary)
+    session.commit()
+
+    audit_event(
+        session,
+        tenant_id=command.tenant_id,
+        user_id=command.created_by_user_id,
+        action="mail.delivery_completed",
+        object_type="mail_delivery_command",
+        object_id=command.id,
+        details={
+            "status": status,
+            "attempt_number": attempt.attempt_number,
+            "accepted_count": accepted_count,
+            "refused_count": len(refusal_map),
+            "refusal_summary": command.refusal_summary,
+            "failure_code": failure_code,
+            "source_module": command.source_module,
+            "source_resource_id": command.source_resource_id,
+        },
+    )
+    session.commit()
+
+
+def _process_claimed(
+    session: Session,
+    command: MailDeliveryCommand,
+    attempt: MailDeliveryAttempt,
+) -> str:
+    try:
+        message = _message_bytes(command)
+        envelope_from = decrypt_secret(command.envelope_from_encrypted)
+        recipients = _decrypt_json(command.envelope_recipients_encrypted)
+        from_header = decrypt_secret(command.from_header_encrypted)
+        if not envelope_from or not isinstance(recipients, list) or not recipients:
+            raise MailDeliveryStateError("Mail delivery envelope is unavailable")
+    except MailDeliveryStateError as exc:
+        _record_outcome(
+            session,
+            command=command,
+            attempt=attempt,
+            status="permanent_failure",
+            failure_code="payload_unavailable",
+            failure_summary=str(exc),
+        )
+        return "permanent_failure"
+
+    _mark_effect_started(session, command, attempt)
+    try:
+        result = send_campaign_email_bytes(
+            session,
+            tenant_id=command.tenant_id,
+            campaign_id=str(command.source_resource_id or ""),
+            profile_id=command.profile_id,
+            message_bytes=message,
+            envelope_from=envelope_from,
+            envelope_recipients=[str(item) for item in recipients],
+            from_header=from_header,
+            expected_smtp_transport_revision=command.expected_smtp_transport_revision,
+            smtp_server_id=command.smtp_server_id,
+            smtp_credential_id=command.smtp_credential_id,
+        )
+    except SmtpSendError as exc:
+        if exc.outcome_unknown:
+            status = "outcome_unknown"
+            code = "smtp_outcome_unknown"
+        elif exc.temporary:
+            status = "temporary_failure"
+            code = "smtp_temporary_failure"
+        else:
+            status = "permanent_failure"
+            code = "smtp_permanent_failure"
+        _record_outcome(
+            session,
+            command=command,
+            attempt=attempt,
+            status=status,
+            failure_code=code,
+            failure_summary=str(exc),
+        )
+        return status
+    except (MailProfileError, SmtpConfigurationError) as exc:
+        _record_outcome(
+            session,
+            command=command,
+            attempt=attempt,
+            status="permanent_failure",
+            failure_code="authorization_or_configuration_changed",
+            failure_summary=str(exc),
+        )
+        return "permanent_failure"
+    except Exception:
+        _record_outcome(
+            session,
+            command=command,
+            attempt=attempt,
+            status="outcome_unknown",
+            failure_code="unexpected_error_after_effect_start",
+            failure_summary="Mail delivery outcome is unknown after transmission began.",
+        )
+        return "outcome_unknown"
+
+    refusals = dict(result.refused_recipients)
+    accepted_count = result.accepted_count
+    if not refusals:
+        status = "accepted"
+    elif accepted_count > 0:
+        status = "partially_refused"
+    elif all(
+        item.get("classification") == "temporary" for item in refusals.values()
+    ):
+        status = "temporary_failure"
+    elif any(
+        item.get("classification") == "unknown" for item in refusals.values()
+    ):
+        status = "outcome_unknown"
+    else:
+        status = "permanent_failure"
+    _record_outcome(
+        session,
+        command=command,
+        attempt=attempt,
+        status=status,
+        accepted_count=accepted_count,
+        refusals=refusals,
+        failure_code=None if status == "accepted" else f"smtp_{status}",
+        failure_summary=None if status == "accepted" else "One or more recipients were refused.",
+    )
+    return status
+
+
+def dispatch_due(
+    session: Session,
+    *,
+    tenant_id: str | None = None,
+    limit: int = 25,
+    worker_id: str | None = None,
+) -> dict[str, object]:
+    bounded_limit = max(1, min(int(limit), 100))
+    now = utcnow()
+    recovered, recovered_unknown = _recover_stale_commands(
+        session,
+        now=now,
+        tenant_id=tenant_id,
+    )
+    clauses = [
+        MailDeliveryCommand.status.in_(DISPATCHABLE_STATUSES),
+        MailDeliveryCommand.payload_purged_at.is_(None),
+        or_(
+            MailDeliveryCommand.next_attempt_at.is_(None),
+            MailDeliveryCommand.next_attempt_at <= now,
+        ),
+    ]
+    if tenant_id:
+        clauses.append(MailDeliveryCommand.tenant_id == tenant_id)
+    command_ids = list(
+        session.scalars(
+            select(MailDeliveryCommand.id)
+            .where(*clauses)
+            .order_by(MailDeliveryCommand.next_attempt_at, MailDeliveryCommand.created_at)
+            .limit(bounded_limit)
+        ).all()
+    )
+    counters: Counter[str] = Counter()
+    processed_ids: list[str] = []
+    for command_id in command_ids:
+        claimed = _claim_command(
+            session,
+            command_id=command_id,
+            now=utcnow(),
+            worker_id=worker_id,
+        )
+        if claimed is None:
+            continue
+        command, attempt = claimed
+        outcome = _process_claimed(session, command, attempt)
+        counters[outcome] += 1
+        processed_ids.append(command.id)
+    return {
+        "selected": len(processed_ids),
+        "accepted": counters["accepted"],
+        "partially_refused": counters["partially_refused"],
+        "retrying": counters["temporary_failure"],
+        "failed": counters["permanent_failure"],
+        "outcome_unknown": counters["outcome_unknown"] + recovered_unknown,
+        "recovered_before_effect": recovered,
+        "command_ids": processed_ids,
+    }
+
+
+def reconcile_delivery_command(
+    session: Session,
+    *,
+    tenant_id: str,
+    command_id: str,
+    decision: str,
+    evidence_reference: str,
+    note: str | None,
+    user_id: str,
+) -> dict[str, object]:
+    command = get_delivery_command(session, tenant_id=tenant_id, command_id=command_id)
+    if command.status not in {"outcome_unknown", "in_progress"}:
+        raise MailDeliveryStateError(
+            "Only a delivery with an unknown outcome can be reconciled"
+        )
+    clean_decision = decision.strip().casefold()
+    if clean_decision not in {"accepted", "not_accepted"}:
+        raise MailDeliveryStateError(
+            "Reconciliation decision must be accepted or not_accepted"
+        )
+    clean_evidence = _bounded_text(evidence_reference)
+    if not clean_evidence:
+        raise MailDeliveryStateError("An evidence reference is required")
+    item = MailDeliveryReconciliation(
+        command_id=command.id,
+        decision=clean_decision,
+        evidence_reference=clean_evidence,
+        note_encrypted=encrypt_secret(note),
+        created_by_user_id=user_id,
+    )
+    session.add(item)
+    command.status = (
+        "reconciled_accepted"
+        if clean_decision == "accepted"
+        else "reconciled_not_accepted"
+    )
+    command.completed_at = utcnow() if clean_decision == "accepted" else None
+    command.next_attempt_at = None
+    command.failure_code = f"reconciled_{clean_decision}"
+    command.failure_summary = "Delivery outcome was reconciled from external evidence."
+    audit_event(
+        session,
+        tenant_id=tenant_id,
+        user_id=user_id,
+        action="mail.delivery_reconciled",
+        object_type="mail_delivery_command",
+        object_id=command.id,
+        details={
+            "decision": clean_decision,
+            "evidence_reference": clean_evidence,
+        },
+    )
+    session.commit()
+    return delivery_command_summary(command)
+
+
+def resend_delivery_command(
+    session: Session,
+    *,
+    tenant_id: str,
+    command_id: str,
+    idempotency_key: str,
+    user_id: str,
+) -> dict[str, object]:
+    command = get_delivery_command(session, tenant_id=tenant_id, command_id=command_id)
+    if command.status not in {
+        "outcome_unknown",
+        "reconciled_not_accepted",
+        "permanent_failure",
+        "partially_refused",
+    }:
+        raise MailDeliveryStateError(
+            "A deliberate resend is only available after a terminal or reconciled failure"
+        )
+    if command.payload_purged_at is not None:
+        raise MailDeliveryStateError("The retained delivery payload is no longer available")
+    message = _message_bytes(command)
+    recipients = _decrypt_json(command.envelope_recipients_encrypted)
+    envelope_from = decrypt_secret(command.envelope_from_encrypted)
+    if not envelope_from or not isinstance(recipients, list):
+        raise MailDeliveryStateError("The retained delivery envelope is unavailable")
+    result = submit_delivery_command(
+        session,
+        tenant_id=tenant_id,
+        command_type=command.command_type,
+        source_module=command.source_module,
+        source_resource_type=command.source_resource_type,
+        source_resource_id=command.source_resource_id,
+        source_version_id=command.source_version_id,
+        idempotency_key=idempotency_key,
+        profile_id=command.profile_id,
+        message_bytes=message,
+        envelope_from=envelope_from,
+        envelope_recipients=[str(item) for item in recipients],
+        from_header=decrypt_secret(command.from_header_encrypted),
+        expected_smtp_transport_revision=command.expected_smtp_transport_revision,
+        smtp_server_id=command.smtp_server_id,
+        smtp_credential_id=command.smtp_credential_id,
+        created_by_user_id=user_id,
+        retention_days=max(
+            1,
+            (command.expires_at - utcnow()).days,
+        ),
+        supersedes_command_id=command.id,
+    )
+    audit_event(
+        session,
+        tenant_id=tenant_id,
+        user_id=user_id,
+        action="mail.delivery_resend_requested",
+        object_type="mail_delivery_command",
+        object_id=str(result["id"]),
+        details={"supersedes_command_id": command.id},
+    )
+    session.commit()
+    return result
+
+
+def purge_expired(
+    session: Session,
+    *,
+    limit: int = 250,
+) -> dict[str, object]:
+    now = utcnow()
+    commands = session.scalars(
+        select(MailDeliveryCommand)
+        .where(
+            MailDeliveryCommand.expires_at <= now,
+            MailDeliveryCommand.payload_purged_at.is_(None),
+        )
+        .order_by(MailDeliveryCommand.expires_at)
+        .limit(max(1, min(int(limit), 1000)))
+    ).all()
+    for command in commands:
+        command.envelope_from_encrypted = None
+        command.envelope_recipients_encrypted = None
+        command.from_header_encrypted = None
+        command.message_encrypted = None
+        command.refusal_details_encrypted = None
+        command.payload_purged_at = now
+    reconciliations = session.scalars(
+        select(MailDeliveryReconciliation)
+        .join(
+            MailDeliveryCommand,
+            MailDeliveryCommand.id == MailDeliveryReconciliation.command_id,
+        )
+        .where(
+            MailDeliveryCommand.payload_purged_at == now,
+            MailDeliveryReconciliation.note_encrypted.is_not(None),
+        )
+    ).all()
+    for reconciliation in reconciliations:
+        reconciliation.note_encrypted = None
+    session.commit()
+    return {"purged": len(commands)}
+
+
+class MailDeliveryOutboxCapability:
+    dispatch_due = staticmethod(dispatch_due)
+    purge_expired = staticmethod(purge_expired)

src/govoplan_mail/backend/documentation.py

@@ -4,7 +4,13 @@ from typing import Any
 
 from sqlalchemy.orm import Session
 
-from govoplan_core.core.modules import DocumentationCondition, DocumentationContext, DocumentationLink, DocumentationTopic
+from govoplan_core.core.modules import (
+    DocumentationCondition,
+    DocumentationConfigurationDecision,
+    DocumentationContext,
+    DocumentationLink,
+    DocumentationTopic,
+)
 from govoplan_mail.backend.mail_profiles import (
     EffectiveMailProfilePolicy,
     MailProfileError,
@@ -35,6 +41,60 @@ def documentation_topics(context: DocumentationContext) -> tuple[DocumentationTo
     return tuple(topics)
 
 
+def documentation_configuration_states(
+    context: DocumentationContext,
+    keys: tuple[str, ...],
+) -> dict[str, DocumentationConfigurationDecision]:
+    if "mail_profile_policy" not in keys:
+        return {}
+    principal = context.principal
+    tenant_id = str(getattr(principal, "tenant_id", "") or "")
+    session = context.session
+    if not tenant_id or not isinstance(session, Session):
+        return {
+            "mail_profile_policy": DocumentationConfigurationDecision(
+                key="mail_profile_policy",
+                state="unavailable",
+                reason="The effective tenant policy cannot be evaluated in this context.",
+            )
+        }
+    try:
+        policy = effective_mail_profile_policy_for_scope(
+            session,
+            tenant_id=tenant_id,
+            scope_type="tenant",
+        )
+    except MailProfileError:
+        return {
+            "mail_profile_policy": DocumentationConfigurationDecision(
+                key="mail_profile_policy",
+                state="unavailable",
+                reason="The effective tenant policy could not be evaluated.",
+            )
+        }
+    tenant_sources = [
+        source
+        for source in policy.source_policies
+        if source.get("scope_type") == "tenant"
+    ]
+    explicitly_configured = any(
+        bool(source.get("applied_fields"))
+        for source in tenant_sources
+    )
+    return {
+        "mail_profile_policy": DocumentationConfigurationDecision(
+            key="mail_profile_policy",
+            state="enabled" if explicitly_configured else "inherited",
+            source="tenant" if explicitly_configured else "system default",
+            reason=(
+                "A tenant policy is configured."
+                if explicitly_configured
+                else "The tenant uses the inherited system policy."
+            ),
+        )
+    }
+
+
 def _tenant_mail_policy_topic(context: DocumentationContext) -> DocumentationTopic | None:
     principal = context.principal
     tenant_id = str(getattr(principal, "tenant_id", "") or "")

src/govoplan_mail/backend/manifest.py

@@ -6,9 +6,11 @@ from pathlib import Path
 from sqlalchemy import inspect
 
 from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
+from govoplan_core.core.mail import CAPABILITY_MAIL_DELIVERY_OUTBOX
 from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard
 from govoplan_core.core.modules import (
     DocumentationCondition,
+    DocumentationConfigurationProviderRegistration,
     DocumentationLink,
     DocumentationTopic,
     FrontendModule,
@@ -24,7 +26,10 @@ from govoplan_core.core.modules import (
 )
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
-from govoplan_mail.backend.documentation import documentation_topics
+from govoplan_mail.backend.documentation import (
+    documentation_configuration_states,
+    documentation_topics,
+)
 from govoplan_mail.backend.db import models as mail_models  # noqa: F401 - populate Mail ORM metadata
 
 
@@ -35,6 +40,9 @@ _mail_table_retirement_provider = drop_table_retirement_provider(
     mail_models.MailProfilePolicy,
     mail_models.MailMailboxFolderIndex,
     mail_models.MailMailboxMessageIndex,
+    mail_models.MailDeliveryReconciliation,
+    mail_models.MailDeliveryAttempt,
+    mail_models.MailDeliveryCommand,
     label="Mail",
 )
 
@@ -90,6 +98,16 @@ PERMISSIONS = (
         "Create, edit, and deactivate only the current account's user-scoped mail profiles within effective policy.",
     ),
     _permission("mail:secret:manage", "Manage mail secrets", "Create or replace stored SMTP/IMAP credentials."),
+    _permission(
+        "mail:delivery:diagnostic",
+        "Inspect mail delivery diagnostics",
+        "Inspect bounded recipient-level refusal and attempt evidence for durable Mail commands.",
+    ),
+    _permission(
+        "mail:delivery:reconcile",
+        "Reconcile mail delivery outcomes",
+        "Reconcile unknown delivery outcomes and explicitly authorize deliberate resend commands.",
+    ),
     _permission(
         "mail:secret:manage_own",
         "Manage own mail secrets",
@@ -109,6 +127,8 @@ ROLE_TEMPLATES = (
             "mail:mailbox:read",
             "mail:profile:write",
             "mail:secret:manage",
+            "mail:delivery:diagnostic",
+            "mail:delivery:reconcile",
         ),
     ),
     RoleTemplate(
@@ -158,6 +178,8 @@ manifest = ModuleManifest(
     optional_dependencies=("campaigns", "addresses"),
     provides_interfaces=(
         ModuleInterfaceProvider(name="mail.campaign_delivery", version="0.2.0"),
+        ModuleInterfaceProvider(name="mail.delivery_commands", version="0.1.0"),
+        ModuleInterfaceProvider(name="mail.delivery_outbox", version="0.1.0"),
     ),
     requires_interfaces=(
         ModuleInterfaceRequirement(
@@ -219,11 +241,18 @@ manifest = ModuleManifest(
             mail_models.MailProfilePolicy,
             mail_models.MailMailboxFolderIndex,
             mail_models.MailMailboxMessageIndex,
+            mail_models.MailDeliveryReconciliation,
+            mail_models.MailDeliveryAttempt,
+            mail_models.MailDeliveryCommand,
             label="Mail",
         ),
     ),
     capability_factories={
         "mail.campaign_delivery": lambda context: __import__("govoplan_mail.backend.capabilities", fromlist=["campaign_capability"]).campaign_capability(context),
+        CAPABILITY_MAIL_DELIVERY_OUTBOX: lambda context: __import__(
+            "govoplan_mail.backend.capabilities",
+            fromlist=["delivery_outbox_capability"],
+        ).delivery_outbox_capability(context),
     },
     documentation=(
         DocumentationTopic(
@@ -427,7 +456,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 durable recipient jobs, retry, and reconciliation. Live report emailing remains disabled until the planned Mail-owned idempotent outbox and attempt ledger are implemented.",
+            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.",
             layer="available",
             documentation_types=("admin", "user"),
             audience=("integrator", "campaign_manager", "campaign_sender", "release_reviewer"),
@@ -460,6 +489,12 @@ manifest = ModuleManifest(
         ),
     ),
     documentation_providers=(documentation_topics,),
+    documentation_configuration_providers=(
+        DocumentationConfigurationProviderRegistration(
+            keys=("mail_profile_policy",),
+            resolve=documentation_configuration_states,
+        ),
+    ),
 )
 
 

src/govoplan_mail/backend/migrations/dev_versions/82a3b4c5d6e7_mail_delivery_outbox.py

@@ -0,0 +1,22 @@
+"""add durable mail delivery outbox
+
+Revision ID: 82a3b4c5d6e7
+Revises: 7192a3bcdef0
+Create Date: 2026-07-30 00:00:00.000000
+"""
+from __future__ import annotations
+
+from importlib import import_module
+
+
+delivery_outbox = import_module(
+    "govoplan_mail.backend.migrations.versions.82a3b4c5d6e7_mail_delivery_outbox"
+)
+
+
+revision = delivery_outbox.revision
+down_revision = delivery_outbox.down_revision
+branch_labels = delivery_outbox.branch_labels
+depends_on = delivery_outbox.depends_on
+upgrade = delivery_outbox.upgrade
+downgrade = delivery_outbox.downgrade

src/govoplan_mail/backend/migrations/versions/82a3b4c5d6e7_mail_delivery_outbox.py

@@ -0,0 +1,243 @@
+"""add durable mail delivery outbox
+
+Revision ID: 82a3b4c5d6e7
+Revises: 7192a3bcdef0
+Create Date: 2026-07-30 00:00:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "82a3b4c5d6e7"
+down_revision = "7192a3bcdef0"
+branch_labels = None
+depends_on = "c91f0a72be34"
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    tables = set(inspector.get_table_names())
+    if "mail_delivery_commands" not in tables:
+        op.create_table(
+            "mail_delivery_commands",
+            sa.Column("id", sa.String(length=36), nullable=False),
+            sa.Column("tenant_id", sa.String(length=36), nullable=False),
+            sa.Column("command_type", sa.String(length=60), nullable=False),
+            sa.Column("source_module", sa.String(length=80), nullable=False),
+            sa.Column("source_resource_type", sa.String(length=80), nullable=False),
+            sa.Column("source_resource_id", sa.String(length=120), nullable=True),
+            sa.Column("source_version_id", sa.String(length=120), nullable=True),
+            sa.Column("idempotency_key", sa.String(length=200), nullable=False),
+            sa.Column("canonical_request_hash", sa.String(length=64), nullable=False),
+            sa.Column("profile_id", sa.String(length=36), nullable=False),
+            sa.Column("smtp_server_id", sa.String(length=36), nullable=True),
+            sa.Column("smtp_credential_id", sa.String(length=36), nullable=True),
+            sa.Column(
+                "expected_smtp_transport_revision",
+                sa.String(length=120),
+                nullable=False,
+            ),
+            sa.Column("envelope_from_encrypted", sa.Text(), nullable=True),
+            sa.Column("envelope_recipients_encrypted", sa.Text(), nullable=True),
+            sa.Column("from_header_encrypted", sa.Text(), nullable=True),
+            sa.Column("message_encrypted", sa.Text(), nullable=True),
+            sa.Column("message_sha256", sa.String(length=64), nullable=False),
+            sa.Column("message_size_bytes", sa.BigInteger(), nullable=False),
+            sa.Column("recipient_count", sa.Integer(), nullable=False),
+            sa.Column("status", sa.String(length=40), nullable=False),
+            sa.Column("attempt_count", sa.Integer(), nullable=False),
+            sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("claimed_by", sa.String(length=255), nullable=True),
+            sa.Column("claimed_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("effect_started_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("accepted_count", sa.Integer(), nullable=False),
+            sa.Column("refused_count", sa.Integer(), nullable=False),
+            sa.Column("refusal_summary", sa.JSON(), nullable=False),
+            sa.Column("refusal_details_encrypted", sa.Text(), nullable=True),
+            sa.Column("failure_code", sa.String(length=80), nullable=True),
+            sa.Column("failure_summary", sa.String(length=500), nullable=True),
+            sa.Column("created_by_user_id", sa.String(length=36), nullable=True),
+            sa.Column("supersedes_command_id", sa.String(length=36), nullable=True),
+            sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("payload_purged_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+            sa.ForeignKeyConstraint(
+                ["profile_id"],
+                ["mail_server_profiles.id"],
+                name=op.f(
+                    "fk_mail_delivery_commands_profile_id_mail_server_profiles"
+                ),
+                ondelete="RESTRICT",
+            ),
+            sa.ForeignKeyConstraint(
+                ["created_by_user_id"],
+                ["access_users.id"],
+                name=op.f(
+                    "fk_mail_delivery_commands_created_by_user_id_access_users"
+                ),
+                ondelete="SET NULL",
+            ),
+            sa.ForeignKeyConstraint(
+                ["supersedes_command_id"],
+                ["mail_delivery_commands.id"],
+                name=op.f(
+                    "fk_mail_delivery_commands_supersedes_command_id_mail_delivery_commands"
+                ),
+                ondelete="SET NULL",
+            ),
+            sa.PrimaryKeyConstraint(
+                "id",
+                name=op.f("pk_mail_delivery_commands"),
+            ),
+            sa.UniqueConstraint(
+                "tenant_id",
+                "command_type",
+                "idempotency_key",
+                name="uq_mail_delivery_commands_idempotency",
+            ),
+        )
+        op.create_index(
+            "ix_mail_delivery_commands_dispatch",
+            "mail_delivery_commands",
+            ["status", "next_attempt_at", "created_at"],
+            unique=False,
+        )
+        op.create_index(
+            "ix_mail_delivery_commands_source",
+            "mail_delivery_commands",
+            [
+                "tenant_id",
+                "source_module",
+                "source_resource_type",
+                "source_resource_id",
+            ],
+            unique=False,
+        )
+        for column in (
+            "tenant_id",
+            "command_type",
+            "source_module",
+            "profile_id",
+            "status",
+            "next_attempt_at",
+            "created_by_user_id",
+            "supersedes_command_id",
+            "expires_at",
+        ):
+            op.create_index(
+                op.f(f"ix_mail_delivery_commands_{column}"),
+                "mail_delivery_commands",
+                [column],
+                unique=False,
+            )
+
+    inspector = sa.inspect(op.get_bind())
+    if "mail_delivery_attempts" not in inspector.get_table_names():
+        op.create_table(
+            "mail_delivery_attempts",
+            sa.Column("id", sa.String(length=36), nullable=False),
+            sa.Column("command_id", sa.String(length=36), nullable=False),
+            sa.Column("attempt_number", sa.Integer(), nullable=False),
+            sa.Column("worker_id", sa.String(length=255), nullable=True),
+            sa.Column("status", sa.String(length=40), nullable=False),
+            sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("effect_started_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("accepted_count", sa.Integer(), nullable=False),
+            sa.Column("refused_count", sa.Integer(), nullable=False),
+            sa.Column("outcome_code", sa.String(length=80), nullable=True),
+            sa.Column("diagnostic_summary", sa.String(length=500), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+            sa.ForeignKeyConstraint(
+                ["command_id"],
+                ["mail_delivery_commands.id"],
+                name=op.f(
+                    "fk_mail_delivery_attempts_command_id_mail_delivery_commands"
+                ),
+                ondelete="CASCADE",
+            ),
+            sa.PrimaryKeyConstraint(
+                "id",
+                name=op.f("pk_mail_delivery_attempts"),
+            ),
+            sa.UniqueConstraint(
+                "command_id",
+                "attempt_number",
+                name="uq_mail_delivery_attempts_number",
+            ),
+        )
+        op.create_index(
+            "ix_mail_delivery_attempts_command_started",
+            "mail_delivery_attempts",
+            ["command_id", "started_at"],
+            unique=False,
+        )
+        for column in ("command_id", "status"):
+            op.create_index(
+                op.f(f"ix_mail_delivery_attempts_{column}"),
+                "mail_delivery_attempts",
+                [column],
+                unique=False,
+            )
+
+    inspector = sa.inspect(op.get_bind())
+    if "mail_delivery_reconciliations" not in inspector.get_table_names():
+        op.create_table(
+            "mail_delivery_reconciliations",
+            sa.Column("id", sa.String(length=36), nullable=False),
+            sa.Column("command_id", sa.String(length=36), nullable=False),
+            sa.Column("decision", sa.String(length=40), nullable=False),
+            sa.Column("evidence_reference", sa.String(length=500), nullable=False),
+            sa.Column("note_encrypted", sa.Text(), nullable=True),
+            sa.Column("created_by_user_id", sa.String(length=36), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+            sa.ForeignKeyConstraint(
+                ["command_id"],
+                ["mail_delivery_commands.id"],
+                name=op.f(
+                    "fk_mail_delivery_reconciliations_command_id_mail_delivery_commands"
+                ),
+                ondelete="CASCADE",
+            ),
+            sa.ForeignKeyConstraint(
+                ["created_by_user_id"],
+                ["access_users.id"],
+                name=op.f(
+                    "fk_mail_delivery_reconciliations_created_by_user_id_access_users"
+                ),
+                ondelete="SET NULL",
+            ),
+            sa.PrimaryKeyConstraint(
+                "id",
+                name=op.f("pk_mail_delivery_reconciliations"),
+            ),
+        )
+        op.create_index(
+            "ix_mail_delivery_reconciliations_command_created",
+            "mail_delivery_reconciliations",
+            ["command_id", "created_at"],
+            unique=False,
+        )
+        for column in ("command_id", "created_by_user_id"):
+            op.create_index(
+                op.f(f"ix_mail_delivery_reconciliations_{column}"),
+                "mail_delivery_reconciliations",
+                [column],
+                unique=False,
+            )
+
+
+def downgrade() -> None:
+    tables = set(sa.inspect(op.get_bind()).get_table_names())
+    if "mail_delivery_reconciliations" in tables:
+        op.drop_table("mail_delivery_reconciliations")
+    if "mail_delivery_attempts" in tables:
+        op.drop_table("mail_delivery_attempts")
+    if "mail_delivery_commands" in tables:
+        op.drop_table("mail_delivery_commands")

src/govoplan_mail/backend/router.py

@@ -20,6 +20,9 @@ from govoplan_mail.backend.schemas import (
     MailCredentialUpdateRequest,
     MailImapFolderListResponse,
     MailImapFolderResponse,
+    MailDeliveryCommandResponse,
+    MailDeliveryReconcileRequest,
+    MailDeliveryResendRequest,
     MailMailboxBootstrapResponse,
     MailImapTestRequest,
     MailMailboxAttachmentResponse,
@@ -79,6 +82,14 @@ 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.delivery_outbox import (
+    MailDeliveryError,
+    delivery_command_diagnostics,
+    delivery_command_summary,
+    get_delivery_command,
+    reconcile_delivery_command,
+    resend_delivery_command,
+)
 from govoplan_mail.backend.server_hierarchy import (
     MailHierarchyContext,
     MailServerHierarchyError,
@@ -120,6 +131,114 @@ DEFAULT_MAILBOX_MESSAGE_LIMIT = 50
 CAPABILITY_ADDRESSES_LOOKUP = "addresses.lookup"
 
 
+@router.get(
+    "/delivery-commands/{command_id}",
+    response_model=MailDeliveryCommandResponse,
+)
+def get_mail_delivery_command(
+    command_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+):
+    """Return the business-safe status of one tenant Mail command."""
+
+    _require_scope(principal, "mail:profile:read")
+    try:
+        command = get_delivery_command(
+            session,
+            tenant_id=principal.tenant_id,
+            command_id=command_id,
+        )
+    except MailDeliveryError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    return MailDeliveryCommandResponse(result=delivery_command_summary(command))
+
+
+@router.get(
+    "/delivery-commands/{command_id}/diagnostics",
+    response_model=MailDeliveryCommandResponse,
+)
+def get_mail_delivery_command_diagnostics(
+    command_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+):
+    """Return bounded recipient and attempt evidence to diagnostic actors."""
+
+    _require_scope(principal, "mail:delivery:diagnostic")
+    try:
+        result = delivery_command_diagnostics(
+            session,
+            tenant_id=principal.tenant_id,
+            command_id=command_id,
+        )
+    except MailDeliveryError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    return MailDeliveryCommandResponse(result=result)
+
+
+@router.post(
+    "/delivery-commands/{command_id}/reconcile",
+    response_model=MailDeliveryCommandResponse,
+)
+def reconcile_mail_delivery_command(
+    command_id: str,
+    payload: MailDeliveryReconcileRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+):
+    _require_scope(principal, "mail:delivery:reconcile")
+    try:
+        result = reconcile_delivery_command(
+            session,
+            tenant_id=principal.tenant_id,
+            command_id=command_id,
+            decision=payload.decision,
+            evidence_reference=payload.evidence_reference,
+            note=payload.note,
+            user_id=principal.user.id,
+        )
+    except MailDeliveryError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(exc),
+        ) from exc
+    return MailDeliveryCommandResponse(result=result)
+
+
+@router.post(
+    "/delivery-commands/{command_id}/resend",
+    response_model=MailDeliveryCommandResponse,
+)
+def resend_mail_delivery_command(
+    command_id: str,
+    payload: MailDeliveryResendRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+):
+    _require_scope(principal, "mail:delivery:reconcile")
+    try:
+        result = resend_delivery_command(
+            session,
+            tenant_id=principal.tenant_id,
+            command_id=command_id,
+            idempotency_key=payload.idempotency_key,
+            user_id=principal.user.id,
+        )
+    except MailDeliveryError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(exc),
+        ) from exc
+    return MailDeliveryCommandResponse(result=result)
+
+
 def _capability_payload(value: object) -> dict[str, Any]:
     if dataclasses.is_dataclass(value):
         return dataclasses.asdict(value)

src/govoplan_mail/backend/schemas.py

@@ -436,3 +436,21 @@ class MailMailboxMessageResponse(BaseModel):
     port: int | None = None
     security: str | None = None
     message: MailMailboxMessageDetailResponse
+
+
+class MailDeliveryCommandResponse(BaseModel):
+    result: dict[str, Any]
+
+
+class MailDeliveryReconcileRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    decision: Literal["accepted", "not_accepted"]
+    evidence_reference: str = Field(min_length=1, max_length=500)
+    note: str | None = Field(default=None, max_length=4000)
+
+
+class MailDeliveryResendRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    idempotency_key: str = Field(min_length=1, max_length=200)