feat: add reusable SMTP batch sessions

README.md

@@ -41,6 +41,18 @@ and requires explicit evidence-backed reconciliation before any deliberate
 resend. Business readers receive only counts and sanitized state; recipient
 refusal details require `mail:delivery:diagnostic`.
 
+Synchronous Campaign batches now preflight DNS, egress, connectivity, TLS, and
+authentication before the first message, then reuse the authorized SMTP
+connection for the bounded batch. A health check precedes reuse; a stale
+connection is reopened before the next message, while a connection loss after
+DATA starts remains outcome-unknown and is never replayed automatically.
+Systemic authentication, sender, and connectivity failures pause remaining
+Campaign jobs instead of producing one failure per recipient. Deployment
+operators can disable reuse or bound connection lifetime and reconnects with
+`GOVOPLAN_SMTP_BATCH_REUSE`, `GOVOPLAN_SMTP_BATCH_MAX_MESSAGES`,
+`GOVOPLAN_SMTP_BATCH_RECONNECT_ATTEMPTS`, and
+`GOVOPLAN_SMTP_BATCH_HEALTH_CHECK`.
+
 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,

src/govoplan_mail/backend/capabilities.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+from contextlib import contextmanager
+from contextvars import ContextVar
 from dataclasses import dataclass
 from email.message import EmailMessage
 from email.utils import formatdate, make_msgid
-from typing import Any
+from typing import Any, Iterator
 
 from sqlalchemy.orm import Session
 
@@ -40,13 +42,27 @@ from govoplan_mail.backend.sending.imap import (
     append_message_to_sent,
 )
 from govoplan_mail.backend.sending.rate_limit import wait_for_rate_limit
-from govoplan_mail.backend.sending.smtp import SmtpConfigurationError, SmtpSendError, send_email_bytes
+from govoplan_mail.backend.sending.smtp import (
+    SmtpBatchSession,
+    SmtpConfigurationError,
+    SmtpSendError,
+    send_email_bytes,
+)
+
+
+_ACTIVE_SMTP_BATCH: ContextVar[SmtpBatchSession | None] = ContextVar(
+    "govoplan_mail_active_smtp_batch",
+    default=None,
+)
 
 
 @dataclass(frozen=True, slots=True)
 class CampaignSmtpDeliveryResult:
     envelope_recipients: list[str]
     refused_recipients: dict[str, dict[str, int | str]]
+    connection_sequence: int = 1
+    session_reused: bool = False
+    reconnect_count: int = 0
 
     @property
     def accepted_count(self) -> int:
@@ -58,6 +74,23 @@ class CampaignImapAppendResult:
     folder: str
 
 
+@dataclass(frozen=True, slots=True)
+class CampaignSmtpBatchState:
+    session: SmtpBatchSession
+
+    @property
+    def status(self) -> str:
+        return "ready"
+
+    @property
+    def connection_count(self) -> int:
+        return self.session.connection_count
+
+    @property
+    def reconnect_count(self) -> int:
+        return self.session.reconnect_count
+
+
 def _sanitized_refusals(
     refused_recipients: dict[str, tuple[int, bytes | str]],
 ) -> dict[str, dict[str, int | str]]:
@@ -95,6 +128,9 @@ def _sanitized_smtp_error(exc: SmtpSendError) -> SmtpSendError:
         message,
         temporary=exc.temporary,
         outcome_unknown=exc.outcome_unknown,
+        systemic=exc.systemic,
+        reason_code=exc.reason_code,
+        phase=exc.phase,
     )
 
 
@@ -279,6 +315,106 @@ def campaign_profile_delivery_summary(
     }
 
 
+@contextmanager
+def campaign_smtp_batch(
+    session: Session,
+    *,
+    tenant_id: str,
+    campaign_id: str,
+    profile_id: str,
+    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,
+) -> Iterator[CampaignSmtpBatchState]:
+    """Preflight and retain one authorized SMTP connection for a batch."""
+
+    selection = _selection_payload(
+        profile_id=profile_id,
+        smtp_server_id=smtp_server_id,
+        smtp_credential_id=smtp_credential_id,
+    )
+    try:
+        profile = _authorized_campaign_profile(
+            session,
+            tenant_id=tenant_id,
+            campaign_id=campaign_id,
+            profile_id=profile_id,
+            selection=selection,
+        )
+    except MailProfileError:
+        raise
+    except Exception:
+        raise SmtpConfigurationError("The selected Mail profile's SMTP configuration is unusable.") from None
+    if _supports_hierarchy(session):
+        context = _campaign_hierarchy_context(
+            session,
+            tenant_id=tenant_id,
+            campaign_id=campaign_id,
+        )
+        try:
+            selected_smtp = select_mail_transport(
+                session,
+                profile=profile,
+                protocol="smtp",
+                context=context,
+                server_id=smtp_server_id,
+                credential_id=smtp_credential_id,
+            )
+        except MailServerHierarchyError as exc:
+            raise MailProfileError(str(exc)) from exc
+        current_revision = selected_smtp.transport_revision
+    else:
+        context = None
+        current_revision = campaign_profile_transport_revisions(profile)["smtp"]
+    if current_revision != expected_smtp_transport_revision:
+        raise MailProfileError(
+            "The selected Mail profile's SMTP settings changed after this campaign was built. "
+            "Revalidate and rebuild the campaign before delivery."
+        )
+    try:
+        smtp = (
+            resolve_mail_transport(
+                session,
+                profile=profile,
+                protocol="smtp",
+                context=context,
+                server_id=smtp_server_id,
+                credential_id=smtp_credential_id,
+            ).config
+            if context is not None
+            else smtp_config_from_profile(profile)
+        )
+    except MailProfileError:
+        raise
+    except Exception:
+        raise SmtpConfigurationError("The selected Mail profile's SMTP configuration is unusable.") from None
+    try:
+        assert_mail_policy_allows_send(
+            session,
+            tenant_id=tenant_id,
+            campaign_id=campaign_id,
+            smtp=smtp,
+            imap=None,
+            envelope_sender=envelope_from,
+            from_header=from_header,
+            recipients=envelope_recipients,
+        )
+    except MailProfileError:
+        raise MailProfileError("Mail delivery is blocked by the effective Mail policy.") from None
+
+    smtp_session = SmtpBatchSession(smtp)
+    smtp_session.preflight()
+    token = _ACTIVE_SMTP_BATCH.set(smtp_session)
+    try:
+        yield CampaignSmtpBatchState(session=smtp_session)
+    finally:
+        _ACTIVE_SMTP_BATCH.reset(token)
+        smtp_session.close()
+
+
 def send_campaign_email_bytes(
     session: Session,
     *,
@@ -394,6 +530,7 @@ def send_campaign_email_bytes(
             smtp_config=smtp,
             envelope_from=envelope_from,
             envelope_recipients=envelope_recipients,
+            batch_session=_ACTIVE_SMTP_BATCH.get(),
         )
     except SmtpSendError as exc:
         sanitized = _sanitized_smtp_error(exc)
@@ -420,6 +557,9 @@ def send_campaign_email_bytes(
     sanitized_result = CampaignSmtpDeliveryResult(
         envelope_recipients=list(result.envelope_recipients),
         refused_recipients=_sanitized_refusals(result.refused_recipients),
+        connection_sequence=getattr(result, "connection_sequence", 0),
+        session_reused=getattr(result, "session_reused", False),
+        reconnect_count=getattr(result, "reconnect_count", 0),
     )
     if recovery is not None:
         try:
@@ -604,6 +744,7 @@ class MailCampaignCapability:
     assert_campaign_mail_policy_allows_json = staticmethod(assert_campaign_mail_policy_allows_json)
     mail_profile_id_from_campaign_json = staticmethod(mail_profile_id_from_campaign_json)
     campaign_profile_delivery_summary = staticmethod(campaign_profile_delivery_summary)
+    campaign_smtp_batch = staticmethod(campaign_smtp_batch)
     send_campaign_email_bytes = staticmethod(send_campaign_email_bytes)
     append_campaign_message_to_sent = staticmethod(append_campaign_message_to_sent)
     wait_for_rate_limit = staticmethod(wait_for_rate_limit)

src/govoplan_mail/backend/manifest.py

@@ -929,7 +929,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. 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. Mail outbox dispatch and retention scans are partitioned by tenant entitlement, so disabling Mail leaves accepted commands and evidence untouched for operator resolution.",
+            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. Synchronous batches authorize the complete recipient set and preflight DNS, egress, connectivity, TLS, and authentication before the first effect. Mail reuses the bounded connection when deployment policy permits, health-checks it before reuse, and reconnects before the next message when a stale connection is detected. A connection loss after DATA begins remains outcome-unknown and is never replayed. Systemic authentication, sender, or connectivity failures carry stable reason codes so Campaign pauses remaining queued work and shows connection, reconnect, failure, and pause progress. 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. Mail outbox dispatch and retention scans are partitioned by tenant entitlement, so disabling Mail leaves accepted commands and evidence untouched for operator resolution.",
             layer="available",
             documentation_types=("admin", "user"),
             audience=("integrator", "campaign_manager", "campaign_sender", "release_reviewer"),
@@ -952,7 +952,7 @@ manifest = ModuleManifest(
                 "route": "/campaigns/{campaign_id}/mail-settings",
                 "screen": "Campaign Mail settings",
                 "section": "Mail-owned profile and transport boundary",
-                "verification": "Prove stale revisions fail before credential decryption, SMTP never decrypts IMAP credentials, IMAP never decrypts SMTP credentials, provider details are sanitized, and the interface/version gate passes.",
+                "verification": "Prove stale revisions fail before credential decryption, batch preflight fails before DATA, two messages reuse one healthy connection, a stale connection reconnects before the next message, post-DATA disconnect is never replayed, systemic failures pause remaining jobs, provider details are sanitized, and the interface/version gate passes.",
                 "related_topic_ids": [
                     "mail.profile-ownership-and-consumers",
                     "campaigns.mail-profile-user-journey",

src/govoplan_mail/backend/sending/smtp.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import copy
 import logging
+import os
 import smtplib
 import ssl
 from dataclasses import dataclass
@@ -64,10 +65,22 @@ class SmtpSendError(RuntimeError):
     started, so automatic retry is intentionally forbidden.
     """
 
-    def __init__(self, message: str, *, temporary: bool = False, outcome_unknown: bool = False):
+    def __init__(
+        self,
+        message: str,
+        *,
+        temporary: bool = False,
+        outcome_unknown: bool = False,
+        systemic: bool = False,
+        reason_code: str | None = None,
+        phase: str = "send",
+    ):
         super().__init__(message)
         self.temporary = temporary
         self.outcome_unknown = outcome_unknown
+        self.systemic = systemic
+        self.reason_code = reason_code
+        self.phase = phase
 
 
 @dataclass(frozen=True, slots=True)
@@ -86,12 +99,315 @@ class SmtpSendResult:
     envelope_from: str
     envelope_recipients: list[str]
     refused_recipients: dict[str, tuple[int, bytes | str]]
+    connection_sequence: int = 1
+    session_reused: bool = False
+    reconnect_count: int = 0
 
     @property
     def accepted_count(self) -> int:
         return len(self.envelope_recipients) - len(self.refused_recipients)
 
 
+@dataclass(frozen=True, slots=True)
+class SmtpBatchPolicy:
+    reuse_connections: bool = True
+    max_messages_per_connection: int = 100
+    reconnect_attempts: int = 1
+    health_check_before_reuse: bool = True
+
+    @classmethod
+    def from_environment(cls) -> "SmtpBatchPolicy":
+        return cls(
+            reuse_connections=_environment_bool("GOVOPLAN_SMTP_BATCH_REUSE", True),
+            max_messages_per_connection=_environment_int(
+                "GOVOPLAN_SMTP_BATCH_MAX_MESSAGES",
+                default=100,
+                minimum=1,
+                maximum=10_000,
+            ),
+            reconnect_attempts=_environment_int(
+                "GOVOPLAN_SMTP_BATCH_RECONNECT_ATTEMPTS",
+                default=1,
+                minimum=0,
+                maximum=5,
+            ),
+            health_check_before_reuse=_environment_bool(
+                "GOVOPLAN_SMTP_BATCH_HEALTH_CHECK",
+                True,
+            ),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class SmtpBatchPreflightResult:
+    ready: bool
+    authenticated: bool
+    connection_sequence: int
+    reconnect_count: int
+
+
+class SmtpBatchSession:
+    """Bounded reusable SMTP connection for one already-authorized batch."""
+
+    def __init__(
+        self,
+        smtp_config: SmtpConfig,
+        *,
+        policy: SmtpBatchPolicy | None = None,
+    ) -> None:
+        self.smtp_config = smtp_config
+        self.policy = policy or SmtpBatchPolicy.from_environment()
+        self._smtp: smtplib.SMTP | None = None
+        self._connection_sequence = 0
+        self._reconnect_count = 0
+        self._messages_on_connection = 0
+        self._closed = False
+
+    @property
+    def connection_count(self) -> int:
+        return self._connection_sequence
+
+    @property
+    def reconnect_count(self) -> int:
+        return self._reconnect_count
+
+    def preflight(self) -> SmtpBatchPreflightResult:
+        """Validate DNS/egress/connectivity/TLS/auth before a provider effect."""
+
+        if self._closed:
+            raise SmtpSendError(
+                "SMTP batch session is closed.",
+                systemic=True,
+                reason_code="batch_session_closed",
+                phase="preflight",
+            )
+        _require_smtp_config(self.smtp_config)
+        if is_mock_smtp_host(self.smtp_config.host):
+            if self._connection_sequence == 0:
+                self._connection_sequence = 1
+            return self._preflight_result()
+        if self._smtp is None:
+            self._connect_with_retries()
+        return self._preflight_result()
+
+    def send(
+        self,
+        message: EmailMessage | bytes,
+        *,
+        envelope_from: str,
+        envelope_recipients: list[str],
+    ) -> SmtpSendResult:
+        host, port, recipients = _prepare_smtp_send(
+            smtp_config=self.smtp_config,
+            envelope_from=envelope_from,
+            envelope_recipients=envelope_recipients,
+        )
+        if is_mock_smtp_host(self.smtp_config.host):
+            preflight = self.preflight()
+            _accepted, refused = _send_mock_smtp_payload(
+                message,
+                smtp_config=self.smtp_config,
+                envelope_from=envelope_from,
+                envelope_recipients=recipients,
+            )
+            self._messages_on_connection += 1
+            return _smtp_send_result(
+                smtp_config=self.smtp_config,
+                host=host,
+                port=port,
+                envelope_from=envelope_from,
+                envelope_recipients=recipients,
+                refused=refused,
+                connection_sequence=preflight.connection_sequence,
+                session_reused=self._messages_on_connection > 1,
+                reconnect_count=preflight.reconnect_count,
+            )
+
+        reused = self._prepare_connection_for_send()
+        smtp = self._smtp
+        if smtp is None:  # Defensive: preflight either opens or raises.
+            raise SmtpSendError(
+                "SMTP preflight did not establish a connection.",
+                temporary=True,
+                systemic=True,
+                reason_code="smtp_connectivity_unavailable",
+                phase="preflight",
+            )
+        try:
+            if isinstance(message, bytes):
+                refused = smtp.sendmail(envelope_from, recipients, message)
+            else:
+                refused = smtp.send_message(
+                    message,
+                    from_addr=envelope_from,
+                    to_addrs=recipients,
+                )
+        except smtplib.SMTPRecipientsRefused as exc:
+            raise SmtpSendError(
+                f"all SMTP recipients were refused: {_decode_refused(exc.recipients)}",
+                temporary=False,
+                reason_code="smtp_recipients_refused",
+            ) from exc
+        except smtplib.SMTPSenderRefused as exc:
+            self._discard_connection()
+            raise SmtpSendError(
+                f"SMTP sender was refused: {exc.smtp_code} {exc.smtp_error!r}",
+                temporary=400 <= int(exc.smtp_code) < 500,
+                systemic=True,
+                reason_code="smtp_sender_refused",
+            ) from exc
+        except smtplib.SMTPResponseException as exc:
+            disconnected = int(exc.smtp_code) == 421
+            if disconnected:
+                self._discard_connection()
+            raise SmtpSendError(
+                f"SMTP error: {exc.smtp_code} {exc.smtp_error!r}",
+                temporary=400 <= int(exc.smtp_code) < 500,
+                systemic=disconnected,
+                reason_code="smtp_connection_closed" if disconnected else "smtp_message_rejected",
+            ) from exc
+        except (OSError, smtplib.SMTPServerDisconnected, smtplib.SMTPException) as exc:
+            self._discard_connection()
+            raise SmtpSendError(
+                f"SMTP outcome is unknown after transmission started: {exc}",
+                outcome_unknown=True,
+                systemic=True,
+                reason_code="smtp_connection_lost_after_transmission",
+            ) from exc
+
+        self._messages_on_connection += 1
+        result = _smtp_send_result(
+            smtp_config=self.smtp_config,
+            host=host,
+            port=port,
+            envelope_from=envelope_from,
+            envelope_recipients=recipients,
+            refused=refused,
+            connection_sequence=self._connection_sequence,
+            session_reused=reused,
+            reconnect_count=self._reconnect_count,
+        )
+        if not self.policy.reuse_connections:
+            self._discard_connection()
+        return result
+
+    def close(self) -> None:
+        self._closed = True
+        self._discard_connection()
+
+    def __enter__(self) -> "SmtpBatchSession":
+        self.preflight()
+        return self
+
+    def __exit__(self, _exc_type, _exc, _traceback) -> None:
+        self.close()
+
+    def _preflight_result(self) -> SmtpBatchPreflightResult:
+        return SmtpBatchPreflightResult(
+            ready=True,
+            authenticated=bool(self.smtp_config.username and self.smtp_config.password),
+            connection_sequence=self._connection_sequence,
+            reconnect_count=self._reconnect_count,
+        )
+
+    def _prepare_connection_for_send(self) -> bool:
+        reused = self._smtp is not None and self._messages_on_connection > 0
+        if self._smtp is not None and self._messages_on_connection >= self.policy.max_messages_per_connection:
+            self._discard_connection()
+            reused = False
+        elif reused and self.policy.health_check_before_reuse:
+            try:
+                code, _message = self._smtp.noop()
+                if int(code) >= 400:
+                    raise smtplib.SMTPServerDisconnected(f"SMTP NOOP returned {code}")
+            except (OSError, smtplib.SMTPException):
+                self._discard_connection()
+                reused = False
+        self.preflight()
+        return reused and self._smtp is not None
+
+    def _connect_with_retries(self) -> None:
+        last_error: BaseException | None = None
+        for attempt in range(self.policy.reconnect_attempts + 1):
+            try:
+                smtp = _open_smtp(self.smtp_config)
+            except smtplib.SMTPAuthenticationError as exc:
+                raise SmtpSendError(
+                    "SMTP authentication failed during batch preflight.",
+                    systemic=True,
+                    reason_code="smtp_authentication_failed",
+                    phase="preflight",
+                ) from exc
+            except SmtpConfigurationError:
+                raise
+            except smtplib.SMTPResponseException as exc:
+                temporary = 400 <= int(exc.smtp_code) < 500
+                last_error = exc
+                if not temporary or attempt >= self.policy.reconnect_attempts:
+                    raise SmtpSendError(
+                        "SMTP server rejected batch preflight.",
+                        temporary=temporary,
+                        systemic=True,
+                        reason_code="smtp_preflight_rejected",
+                        phase="preflight",
+                    ) from exc
+                continue
+            except (OSError, smtplib.SMTPException) as exc:
+                last_error = exc
+                if attempt >= self.policy.reconnect_attempts:
+                    raise SmtpSendError(
+                        "SMTP connectivity is unavailable during batch preflight.",
+                        temporary=True,
+                        systemic=True,
+                        reason_code="smtp_connectivity_unavailable",
+                        phase="preflight",
+                    ) from exc
+                continue
+            self._smtp = smtp
+            if attempt > 0 or self._connection_sequence > 0:
+                self._reconnect_count += 1
+            self._connection_sequence += 1
+            self._messages_on_connection = 0
+            return
+        raise SmtpSendError(
+            f"SMTP batch preflight failed: {last_error}",
+            temporary=True,
+            systemic=True,
+            reason_code="smtp_connectivity_unavailable",
+            phase="preflight",
+        )
+
+    def _discard_connection(self) -> None:
+        smtp, self._smtp = self._smtp, None
+        self._messages_on_connection = 0
+        if smtp is None:
+            return
+        try:
+            smtp.quit()
+        except Exception as quit_exc:
+            _log_smtp_cleanup_failure("closing batch connection", quit_exc)
+            try:
+                smtp.close()
+            except Exception as close_exc:
+                _log_smtp_cleanup_failure("closing batch socket", close_exc)
+
+
+def _environment_bool(name: str, default: bool) -> bool:
+    value = os.getenv(name)
+    if value is None:
+        return default
+    return value.strip().casefold() in {"1", "true", "yes", "on"}
+
+
+def _environment_int(name: str, *, default: int, minimum: int, maximum: int) -> int:
+    value = os.getenv(name)
+    try:
+        parsed = int(value) if value is not None else default
+    except ValueError:
+        parsed = default
+    return max(minimum, min(maximum, parsed))
+
+
 def _log_smtp_cleanup_failure(action: str, exc: BaseException) -> None:
     logger.debug("SMTP cleanup failed while %s: %s", action, exc, exc_info=True)
 
@@ -324,15 +640,27 @@ def _send_network_smtp_payload(
         raise SmtpSendError(
             f"SMTP authentication failed: {exc.smtp_code} {exc.smtp_error!r}",
             temporary=False,
+            systemic=True,
+            reason_code="smtp_authentication_failed",
+            phase="preflight",
         ) from exc
     except smtplib.SMTPResponseException as exc:
         raise SmtpSendError(
             f"SMTP connection error: {exc.smtp_code} {exc.smtp_error!r}",
             temporary=400 <= int(exc.smtp_code) < 500,
+            systemic=True,
+            reason_code="smtp_preflight_rejected",
+            phase="preflight",
         ) from exc
     except (OSError, smtplib.SMTPException) as exc:
         # No message transmission has begun yet; a later explicit retry is safe.
-        raise SmtpSendError(f"SMTP connection failed: {exc}", temporary=True) from exc
+        raise SmtpSendError(
+            f"SMTP connection failed: {exc}",
+            temporary=True,
+            systemic=True,
+            reason_code="smtp_connectivity_unavailable",
+            phase="preflight",
+        ) from exc
 
     try:
         if isinstance(message, bytes):
@@ -352,6 +680,8 @@ def _send_network_smtp_payload(
         raise SmtpSendError(
             f"SMTP sender was refused: {exc.smtp_code} {exc.smtp_error!r}",
             temporary=400 <= int(exc.smtp_code) < 500,
+            systemic=True,
+            reason_code="smtp_sender_refused",
         ) from exc
     except smtplib.SMTPResponseException as exc:
         # An explicit SMTP response means the server rejected the transaction;
@@ -359,6 +689,8 @@ def _send_network_smtp_payload(
         raise SmtpSendError(
             f"SMTP error: {exc.smtp_code} {exc.smtp_error!r}",
             temporary=400 <= int(exc.smtp_code) < 500,
+            systemic=int(exc.smtp_code) == 421,
+            reason_code="smtp_connection_closed" if int(exc.smtp_code) == 421 else "smtp_message_rejected",
         ) from exc
     except (OSError, smtplib.SMTPServerDisconnected, smtplib.SMTPException) as exc:
         # A connection loss after DATA began can happen after the server accepted
@@ -366,6 +698,8 @@ def _send_network_smtp_payload(
         raise SmtpSendError(
             f"SMTP outcome is unknown after transmission started: {exc}",
             outcome_unknown=True,
+            systemic=True,
+            reason_code="smtp_connection_lost_after_transmission",
         ) from exc
     finally:
         try:
@@ -387,6 +721,9 @@ def _smtp_send_result(
     envelope_from: str,
     envelope_recipients: list[str],
     refused: dict[str, tuple[int, bytes]],
+    connection_sequence: int = 1,
+    session_reused: bool = False,
+    reconnect_count: int = 0,
 ) -> SmtpSendResult:
     return SmtpSendResult(
         host=host,
@@ -395,6 +732,9 @@ def _smtp_send_result(
         envelope_from=envelope_from,
         envelope_recipients=list(envelope_recipients),
         refused_recipients=_decode_refused(refused),
+        connection_sequence=connection_sequence,
+        session_reused=session_reused,
+        reconnect_count=reconnect_count,
     )
 
 
@@ -404,15 +744,19 @@ def send_email_bytes(
     smtp_config: SmtpConfig,
     envelope_from: str,
     envelope_recipients: list[str],
+    batch_session: SmtpBatchSession | None = None,
 ) -> SmtpSendResult:
     """Send exact RFC 5322 bytes through SMTP without reserializing the message."""
 
-    return _send_smtp_payload(
+    if batch_session is not None:
+        if batch_session.smtp_config != smtp_config:
+            raise SmtpConfigurationError("SMTP batch session does not match the resolved transport.")
+        return batch_session.send(
             message_bytes,
-        smtp_config=smtp_config,
             envelope_from=envelope_from,
             envelope_recipients=envelope_recipients,
         )
+    return _send_smtp_payload(message_bytes, smtp_config=smtp_config, envelope_from=envelope_from, envelope_recipients=envelope_recipients)
 
 
 def send_email_message(

tests/test_smtp_helpers.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import smtplib
 import unittest
 from unittest.mock import patch
 
@@ -7,6 +8,9 @@ from govoplan_core.security.outbound_http import OutboundHttpBlocked
 from govoplan_mail.backend.config import SmtpConfig
 from govoplan_mail.backend.sending.smtp import (
     SmtpConfigurationError,
+    SmtpBatchPolicy,
+    SmtpBatchSession,
+    SmtpSendError,
     _open_smtp,
     _prepare_smtp_send,
     _smtp_send_result,
@@ -70,6 +74,109 @@ class SmtpSendHelperTests(unittest.TestCase):
         self.assertEqual(result.accepted_count, 1)
         self.assertEqual(result.refused_recipients["blocked@example.org"], (550, "blocked"))
 
+    def test_batch_preflight_reuses_one_authenticated_connection(self):
+        config = SmtpConfig(host="smtp.example.org", port=587, security="starttls")
+        smtp = _FakeSmtp()
+        with patch("govoplan_mail.backend.sending.smtp._open_smtp", return_value=smtp) as opener:
+            with SmtpBatchSession(config) as batch:
+                first = batch.send(b"first", envelope_from="sender@example.org", envelope_recipients=["one@example.org"])
+                second = batch.send(b"second", envelope_from="sender@example.org", envelope_recipients=["two@example.org"])
+
+        opener.assert_called_once_with(config)
+        self.assertFalse(first.session_reused)
+        self.assertTrue(second.session_reused)
+        self.assertEqual(1, second.connection_sequence)
+        self.assertEqual([b"first", b"second"], smtp.messages)
+        self.assertTrue(smtp.quit_called)
+
+    def test_batch_reconnects_before_next_message_when_health_check_fails(self):
+        config = SmtpConfig(host="smtp.example.org", port=587, security="starttls")
+        first_smtp = _FakeSmtp(noop_error_on_call=1)
+        second_smtp = _FakeSmtp()
+        policy = SmtpBatchPolicy(reconnect_attempts=1)
+        with patch(
+            "govoplan_mail.backend.sending.smtp._open_smtp",
+            side_effect=[first_smtp, second_smtp],
+        ) as opener:
+            with SmtpBatchSession(config, policy=policy) as batch:
+                batch.send(b"first", envelope_from="sender@example.org", envelope_recipients=["one@example.org"])
+                result = batch.send(b"second", envelope_from="sender@example.org", envelope_recipients=["two@example.org"])
+
+        self.assertEqual(2, opener.call_count)
+        self.assertEqual(2, result.connection_sequence)
+        self.assertEqual(1, result.reconnect_count)
+        self.assertEqual([b"first"], first_smtp.messages)
+        self.assertEqual([b"second"], second_smtp.messages)
+
+    def test_preflight_retries_a_transient_connection_failure(self):
+        config = SmtpConfig(host="smtp.example.org", port=587, security="starttls")
+        smtp = _FakeSmtp()
+        with patch(
+            "govoplan_mail.backend.sending.smtp._open_smtp",
+            side_effect=[OSError("temporary DNS failure"), smtp],
+        ) as opener:
+            with SmtpBatchSession(config, policy=SmtpBatchPolicy(reconnect_attempts=1)) as batch:
+                result = batch.send(b"message", envelope_from="sender@example.org", envelope_recipients=["one@example.org"])
+
+        self.assertEqual(2, opener.call_count)
+        self.assertEqual(1, result.reconnect_count)
+
+    def test_connection_loss_after_send_starts_is_unknown_and_never_replayed(self):
+        config = SmtpConfig(host="smtp.example.org", port=587, security="starttls")
+        smtp = _FakeSmtp(send_error=smtplib.SMTPServerDisconnected("lost"))
+        with patch("govoplan_mail.backend.sending.smtp._open_smtp", return_value=smtp), self.assertRaises(SmtpSendError) as raised:
+            with SmtpBatchSession(config) as batch:
+                batch.send(b"one", envelope_from="sender@example.org", envelope_recipients=["one@example.org"])
+
+        self.assertTrue(raised.exception.outcome_unknown)
+        self.assertTrue(raised.exception.systemic)
+        self.assertEqual("smtp_connection_lost_after_transmission", raised.exception.reason_code)
+        self.assertEqual(1, smtp.send_calls)
+
+    def test_authentication_preflight_is_systemic_and_blocks_batch(self):
+        config = SmtpConfig(host="smtp.example.org", port=587, security="starttls")
+        error = smtplib.SMTPAuthenticationError(535, b"bad credentials")
+        with patch("govoplan_mail.backend.sending.smtp._open_smtp", side_effect=error), self.assertRaises(SmtpSendError) as raised:
+            SmtpBatchSession(config).preflight()
+
+        self.assertTrue(raised.exception.systemic)
+        self.assertFalse(raised.exception.temporary)
+        self.assertEqual("preflight", raised.exception.phase)
+        self.assertEqual("smtp_authentication_failed", raised.exception.reason_code)
+
+
+class _FakeSmtp:
+    def __init__(self, *, noop_error_on_call: int | None = None, send_error: BaseException | None = None):
+        self.noop_error_on_call = noop_error_on_call
+        self.send_error = send_error
+        self.noop_calls = 0
+        self.send_calls = 0
+        self.messages: list[bytes] = []
+        self.quit_called = False
+
+    def noop(self):
+        self.noop_calls += 1
+        if self.noop_error_on_call == self.noop_calls:
+            raise smtplib.SMTPServerDisconnected("stale")
+        return 250, b"ok"
+
+    def sendmail(self, _sender, _recipients, message):
+        self.send_calls += 1
+        if self.send_error is not None:
+            raise self.send_error
+        self.messages.append(message)
+        return {}
+
+    def send_message(self, message, **_kwargs):
+        return self.sendmail(None, None, message.as_bytes())
+
+    def quit(self):
+        self.quit_called = True
+        return 221, b"bye"
+
+    def close(self):
+        return None
+
 
 if __name__ == "__main__":
     unittest.main()