feat: pause campaign batches on SMTP failure

README.md

@@ -120,6 +120,11 @@ Platform RBAC and governance rules are documented in `govoplan-core/docs/`.
 - [Campaign handbook](docs/CAMPAIGN_HANDBOOK.md) provides the adaptive user, process, governance, technical, and operations perspectives.
 - [Campaign delivery runbook](docs/CAMPAIGN_DELIVERY_RUNBOOK.md) covers queueing, local vs Celery operation, retries, reconciliation, reports, and the live SMTP/IMAP test checklist.
 - Immediate delivery is bounded to 25 exact eligible recipient jobs by default. Deployments may set `GOVOPLAN_CAMPAIGN_SYNCHRONOUS_SEND_MAX_RECIPIENTS` (0–500), and tenants may narrow that ceiling through `campaign_delivery_policy.synchronous_send_max_recipients` in tenant settings.
+- Immediate Mail delivery preflights the selected SMTP transport before the
+  first effect and reuses a healthy bounded connection through Mail. Review and
+  send reports the batch state, connection/reconnect counts, and paused count.
+  A systemic authentication, sender, or connectivity failure pauses remaining
+  jobs; correct and test the Mail profile before explicitly resuming them.
 - Report-email preview uses the selected version's stored v5 Mail-profile evidence. Live report email fails closed until [govoplan-mail#17](https://git.add-ideas.de/GovOPlaN/govoplan-mail/issues/17) provides a durable, idempotent Mail-owned outbox and transport-attempt ledger; per-job CSV is off by default and requires `campaigns:recipient:export` when requested.
 - [Campaign/Mail profile boundary](docs/MAIL_PROFILE_BOUNDARY.md) defines profile-only delivery, runtime resolution, execution evidence, and the fail-closed legacy migration path.
 - [Recipient import guide](docs/RECIPIENT_IMPORT_GUIDE.md) covers user/admin workflows, mapping profiles, validation, and import evidence.

docs/CAMPAIGN_DELIVERY_RUNBOOK.md

@@ -62,6 +62,10 @@ Before the first live send for a sender domain or mail-server profile:
 6. If a synchronous request is used, keep Review and send open: it polls the
    durable counters while the request runs. A rejection occurs before SMTP and
    directs oversized runs to workers.
+7. Review the SMTP batch line. `ready` means DNS/connectivity/TLS/auth preflight
+   succeeded. Connection and reconnect counts explain reuse. `paused` means a
+   systemic transport failure stopped the remaining jobs before their SMTP
+   effect; test/correct the Mail profile and explicitly resume the queue.
 
 ## Outcome Handling
 
@@ -88,6 +92,9 @@ unknown provider attempt merely to repair the other layer's state.
 - `failed_temporary`: Retry explicitly after checking the error and retry count.
 - `failed_permanent`: Retry only if the operator has corrected the root cause and
   intentionally includes permanent failures.
+- `paused` after a systemic SMTP failure: do not resume until the shared Mail
+  profile passes its connection test. Authentication, sender rejection, and
+  unavailable connectivity affect the batch rather than one recipient.
 - `outcome_unknown`: Do not retry directly. Check SMTP logs, mailbox evidence, or
   provider control panels, then reconcile as accepted or not sent.
 - `claimed` or `sending` that does not progress: treat as a worker interruption.

docs/CAMPAIGN_HANDBOOK.md

@@ -611,7 +611,6 @@ The following are part of the selected reference journey but are not implied by
 the current baseline:
 
 - the final audited **test / single send / single resend** semantics;
-- reusable SMTP batch sessions and their measured throughput benefit;
 - durable, idempotent Campaign report delivery through a Mail-owned outbox
   ([`govoplan-mail#17`](https://git.add-ideas.de/GovOPlaN/govoplan-mail/issues/17));
 - a fully packaged one-command Campaign reference composition with production

src/govoplan_campaign/backend/integrations.py

@@ -74,11 +74,21 @@ class SmtpConfigurationError(RuntimeError):
 
 class SmtpSendError(RuntimeError):
     def __init__(
-        self, message: str, *, temporary: bool = False, outcome_unknown: bool = False
+        self,
+        message: str,
+        *,
+        temporary: bool = False,
+        outcome_unknown: bool = False,
+        systemic: bool = False,
+        reason_code: str | None = None,
+        phase: str = "send",
     ) -> None:
         super().__init__(message)
         self.temporary = temporary
         self.outcome_unknown = outcome_unknown
+        self.systemic = systemic
+        self.reason_code = reason_code
+        self.phase = phase
 
 
 class ImapConfigurationError(RuntimeError):
@@ -298,6 +308,9 @@ class MailCampaignIntegration:
                 str(exc),
                 temporary=bool(getattr(exc, "temporary", False)),
                 outcome_unknown=bool(getattr(exc, "outcome_unknown", False)),
+                systemic=bool(getattr(exc, "systemic", False)),
+                reason_code=str(getattr(exc, "reason_code", "") or "") or None,
+                phase=str(getattr(exc, "phase", "send") or "send"),
             ) from exc
         except getattr(
             delegate, "SmtpConfigurationError", SmtpConfigurationError
@@ -306,6 +319,30 @@ class MailCampaignIntegration:
         except getattr(delegate, "MailProfileError", MailProfileError) as exc:
             raise MailProfileError(str(exc)) from exc
 
+    @contextmanager
+    def campaign_smtp_batch(self, *args: Any, **kwargs: Any) -> Iterator[Any]:
+        delegate = self._require()
+        method = getattr(delegate, "campaign_smtp_batch", None)
+        if not callable(method):
+            yield None
+            return
+        try:
+            with method(*args, **kwargs) as state:
+                yield state
+        except getattr(delegate, "SmtpSendError", SmtpSendError) as exc:
+            raise SmtpSendError(
+                str(exc),
+                temporary=bool(getattr(exc, "temporary", False)),
+                outcome_unknown=bool(getattr(exc, "outcome_unknown", False)),
+                systemic=bool(getattr(exc, "systemic", False)),
+                reason_code=str(getattr(exc, "reason_code", "") or "") or None,
+                phase=str(getattr(exc, "phase", "preflight") or "preflight"),
+            ) from exc
+        except getattr(delegate, "SmtpConfigurationError", SmtpConfigurationError) as exc:
+            raise SmtpConfigurationError(str(exc)) from exc
+        except getattr(delegate, "MailProfileError", MailProfileError) as exc:
+            raise MailProfileError(str(exc)) from exc
+
     def append_campaign_message_to_sent(self, *args: Any, **kwargs: Any) -> Any:
         delegate = self._require()
         try:

src/govoplan_campaign/backend/manifest.py

@@ -913,7 +913,7 @@ manifest = ModuleManifest(
             id="campaigns.mail-profile-operations",
             title="Operate profile-backed campaign delivery",
             summary="Workers re-authorize and resolve Mail profiles at execution time while Campaign retains only opaque Mail-owned revisions and outcomes.",
-            body="A legacy snapshot, unauthorized or inactive profile, profile-reference mismatch, or changed SMTP/IMAP transport revision stops delivery. Preserve the record, migrate or correct the profile selection, revalidate, rebuild, and only then queue again. Password-only rotation remains possible without copying secrets into Campaign. Uncertain SMTP and IMAP effects remain blocked until an evidence-backed operator reconciliation. If Campaign becomes unavailable to the tenant after a job was accepted, the worker leaves the job untouched and reports an operator action instead of sending or dropping it.",
+            body="A legacy snapshot, unauthorized or inactive profile, profile-reference mismatch, or changed SMTP/IMAP transport revision stops delivery. Synchronous Mail batches preflight DNS, connectivity, TLS, and authentication before their first effect, reuse a bounded healthy SMTP connection, and reconnect before a later message when the old connection is stale. Review and send shows batch, connection, reconnect, failure, and pause counts. A systemic authentication, sender, or connectivity failure pauses remaining queued jobs with a stable reason code; correct and test the Mail profile before explicitly resuming. A connection loss after DATA begins stays outcome-unknown and is never replayed automatically. Preserve a stopped record, migrate or correct the profile selection, revalidate, rebuild, and only then queue again. Password-only rotation remains possible without copying secrets into Campaign. Uncertain SMTP and IMAP effects remain blocked until an evidence-backed operator reconciliation. If Campaign becomes unavailable to the tenant after a job was accepted, the worker leaves the job untouched and reports an operator action instead of sending or dropping it.",
             layer="configured",
             documentation_types=("admin",),
             audience=("campaign_sender", "campaign_operator", "mail_admin"),

src/govoplan_campaign/backend/response_security.py

@@ -48,7 +48,12 @@ _SEND_NOW_RESULT_KEYS = (
     "failed_count",
     "outcome_unknown_count",
     "skipped_count",
+    "paused_count",
     "preflight_count",
+    "batch_state",
+    "batch_pause_reason_code",
+    "smtp_connection_count",
+    "smtp_reconnect_count",
     "delivery_mode",
     "dry_run",
 )

src/govoplan_campaign/backend/sending/jobs.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import hashlib
 import json
 from collections import Counter
+from contextlib import nullcontext
 from dataclasses import asdict, dataclass
 from datetime import datetime, timezone
 from email import policy
@@ -225,6 +226,11 @@ class SendCampaignNowResult:
     failed_count: int
     outcome_unknown_count: int
     skipped_count: int
+    paused_count: int = 0
+    batch_state: str = "not_started"
+    batch_pause_reason_code: str | None = None
+    smtp_connection_count: int = 0
+    smtp_reconnect_count: int = 0
     preflight_count: int = 0
     synchronous_send_policy: dict[str, Any] | None = None
     dry_run: bool = False
@@ -239,7 +245,12 @@ class SendCampaignNowResult:
             "failed_count": self.failed_count,
             "outcome_unknown_count": self.outcome_unknown_count,
             "skipped_count": self.skipped_count,
+            "paused_count": self.paused_count,
             "preflight_count": self.preflight_count,
+            "batch_state": self.batch_state,
+            "batch_pause_reason_code": self.batch_pause_reason_code,
+            "smtp_connection_count": self.smtp_connection_count,
+            "smtp_reconnect_count": self.smtp_reconnect_count,
             "delivery_mode": "synchronous",
             "synchronous_send_policy": self.synchronous_send_policy or {},
             "dry_run": self.dry_run,
@@ -1070,48 +1081,91 @@ def send_campaign_now(
         jobs=jobs,
         policy=synchronous_policy,
     )
-    # Queue state and its inbox notification become durable only after every
-    # message and the selected transport revision have passed preflight. This
-    # preserves late-ack recovery without leaving rejected work eligible for a
-    # background worker.
-    session.commit()
-
     results: list[dict[str, Any]] = []
     sent_count = 0
     failed_count = 0
     outcome_unknown_count = 0
     skipped_after_queue = 0
-    for job in jobs:
+    attempted_count = 0
-        try:
+    paused_count = 0
-            result = _deliver_job_with_recovery(
+    pause_reason_code: str | None = None
-                session,
+    batch_state = "ready"
-                job=job,
+    batch_manager = _synchronous_smtp_batch_manager(
-                context=delivery_contexts[job.id],
+        session,
-                use_rate_limit=use_rate_limit,
+        jobs=jobs,
-                enqueue_imap_task=enqueue_imap_task,
+        contexts=delivery_contexts,
-            )
+    )
-            result_dict = result.as_dict()
+    try:
-            results.append(result_dict)
+        with batch_manager as smtp_batch:
-            if result.status in DELIVERY_ACCEPTED_STATUSES | {"already_accepted"}:
+            # Queue state becomes durable only after local and SMTP
-                sent_count += 1
+            # DNS/connectivity/TLS/auth preflight succeeds.
-            elif result.status == JobSendStatus.OUTCOME_UNKNOWN.value:
+            session.commit()
-                outcome_unknown_count += 1
+            for index, job in enumerate(jobs):
-            else:
+                attempted_count += 1
-                skipped_after_queue += 1
+                try:
-        except Exception as exc:  # keep sending other jobs and return per-job details
+                    result = _deliver_job_with_recovery(
-            failed_count += 1
+                        session,
-            results.append({"job_id": job.id, "status": "failed", "message": str(exc)})
+                        job=job,
+                        context=delivery_contexts[job.id],
+                        use_rate_limit=use_rate_limit,
+                        enqueue_imap_task=enqueue_imap_task,
+                    )
+                    result_dict = result.as_dict()
+                    results.append(result_dict)
+                    if result.status in DELIVERY_ACCEPTED_STATUSES | {"already_accepted"}:
+                        sent_count += 1
+                    elif result.status == JobSendStatus.OUTCOME_UNKNOWN.value:
+                        outcome_unknown_count += 1
+                    else:
+                        skipped_after_queue += 1
+                except Exception as exc:
+                    failed_count += 1
+                    results.append({"job_id": job.id, "status": "failed", "message": str(exc)})
+                    if isinstance(exc, SmtpSendError) and exc.systemic:
+                        pause_reason_code = exc.reason_code or "smtp_systemic_failure"
+                        paused_count = _pause_jobs_after_systemic_smtp_failure(
+                            session,
+                            campaign_id=job.campaign_id,
+                            exclude_job_id=job.id,
+                            reason_code=pause_reason_code,
+                        )
+                        batch_state = "paused"
+                        for remaining in jobs[index + 1 :]:
+                            results.append(
+                                {
+                                    "job_id": remaining.id,
+                                    "status": "paused",
+                                    "message": "Batch paused after a systemic SMTP failure.",
+                                }
+                            )
+                        break
+            smtp_connection_count = int(getattr(smtp_batch, "connection_count", 0) or 0)
+            smtp_reconnect_count = int(getattr(smtp_batch, "reconnect_count", 0) or 0)
+    except (MailProfileError, SmtpConfigurationError, SmtpSendError, OSError) as exc:
+        session.rollback()
+        reason_code = str(getattr(exc, "reason_code", "") or "smtp_batch_preflight_failed")
+        raise SynchronousSendRejected(
+            "SMTP batch preflight could not validate DNS, connectivity, TLS, and authentication; no message was sent.",
+            reason=reason_code,
+            eligible_count=len(jobs),
+            policy=synchronous_policy,
+        ) from exc
 
     return SendCampaignNowResult(
         campaign_id=campaign.id,
         version_id=version.id,
-        attempted_count=len(jobs),
+        attempted_count=attempted_count,
         sent_count=sent_count,
         failed_count=failed_count,
         outcome_unknown_count=outcome_unknown_count,
         skipped_count=queue_result.skipped_count
         + queue_result.blocked_count
         + skipped_after_queue,
+        paused_count=paused_count,
+        batch_state=batch_state,
+        batch_pause_reason_code=pause_reason_code,
+        smtp_connection_count=smtp_connection_count,
+        smtp_reconnect_count=smtp_reconnect_count,
         preflight_count=len(delivery_contexts),
         synchronous_send_policy=synchronous_policy.as_dict(),
         dry_run=False,
@@ -1193,6 +1247,100 @@ def _preflight_synchronous_send_batch(
     return contexts
 
 
+def _synchronous_smtp_batch_manager(
+    session: Session,
+    *,
+    jobs: list[CampaignJob],
+    contexts: dict[str, _SendJobDeliveryContext],
+):
+    mail_items = [
+        (job, contexts[job.id])
+        for job in jobs
+        if DeliveryChannelPolicy(job.delivery_channel_policy).uses_mail
+    ]
+    if not mail_items:
+        return nullcontext(None)
+    first_job, first_context = mail_items[0]
+    envelope_froms = {str(context.envelope_from or "") for _job, context in mail_items}
+    transport_keys = {
+        (
+            context.snapshot.mail_profile_id,
+            context.snapshot.smtp_transport_revision,
+            context.snapshot.smtp_server_id,
+            context.snapshot.smtp_credential_id,
+        )
+        for _job, context in mail_items
+    }
+    if len(envelope_froms) != 1 or "" in envelope_froms or len(transport_keys) != 1:
+        raise SynchronousSendRejected(
+            "A synchronous SMTP batch requires one frozen sender and transport selection.",
+            reason="smtp_batch_transport_mismatch",
+            eligible_count=len(jobs),
+        )
+    recipients = sorted(
+        {
+            recipient
+            for _job, context in mail_items
+            for recipient in context.envelope_recipients
+        }
+    )
+    return mail_integration().campaign_smtp_batch(
+        session,
+        tenant_id=first_job.tenant_id,
+        campaign_id=first_job.campaign_id,
+        profile_id=first_context.snapshot.mail_profile_id,
+        envelope_from=str(first_context.envelope_from),
+        envelope_recipients=recipients,
+        from_header=_from_header_from_job(first_job),
+        expected_smtp_transport_revision=first_context.snapshot.smtp_transport_revision or "",
+        smtp_server_id=first_context.snapshot.smtp_server_id,
+        smtp_credential_id=first_context.snapshot.smtp_credential_id,
+    )
+
+
+def _pause_jobs_after_systemic_smtp_failure(
+    session: Session,
+    *,
+    campaign_id: str,
+    exclude_job_id: str,
+    reason_code: str,
+) -> int:
+    reason = f"SMTP batch paused after systemic failure ({reason_code[:80]})."
+    changed = (
+        session.query(CampaignJob)
+        .filter(
+            CampaignJob.campaign_id == campaign_id,
+            CampaignJob.id != exclude_job_id,
+            CampaignJob.queue_status == JobQueueStatus.QUEUED.value,
+            CampaignJob.send_status.in_(
+                [JobSendStatus.QUEUED.value, JobSendStatus.FAILED_TEMPORARY.value]
+            ),
+        )
+        .update(
+            {
+                CampaignJob.queue_status: JobQueueStatus.PAUSED.value,
+                CampaignJob.last_error: reason,
+            },
+            synchronize_session=False,
+        )
+    )
+    campaign = session.get(Campaign, campaign_id)
+    if changed and campaign is not None:
+        campaign.status = CampaignStatus.READY_TO_QUEUE.value
+        session.add(campaign)
+    audit_event(
+        session,
+        tenant_id=campaign.tenant_id if campaign is not None else None,
+        user_id=None,
+        action="campaign.smtp_batch_paused",
+        object_type="campaign",
+        object_id=campaign_id,
+        details={"reason_code": reason_code[:80], "paused_count": int(changed)},
+    )
+    session.commit()
+    return int(changed)
+
+
 def enqueue_existing_queued_jobs(
     session: Session, *, tenant_id: str, campaign_id: str
 ) -> int:
@@ -3652,6 +3800,13 @@ def _send_claimed_mail_only_job(
         outcome_unknown = _record_smtp_send_error(
             session, job=job, attempt=attempt, exc=exc
         )
+        if exc.systemic:
+            _pause_jobs_after_systemic_smtp_failure(
+                session,
+                campaign_id=job.campaign_id,
+                exclude_job_id=job.id,
+                reason_code=exc.reason_code or "smtp_systemic_failure",
+            )
         if outcome_unknown is not None:
             return outcome_unknown
         raise

tests/test_mail_profile_boundary.py

@@ -176,7 +176,7 @@ def test_new_execution_snapshot_stores_reference_and_evidence_not_transport_mate
         delivery=DeliveryConfig(),
     )
 
-    assert payload["snapshot_version"] == "8"
+    assert payload["snapshot_version"] == "9"
     assert payload["mail_profile_id"] == "profile-1"
     assert "smtp" not in payload
     assert "imap" not in payload

tests/test_send_now_response_security.py

@@ -38,6 +38,11 @@ def test_send_now_omits_provider_and_recipient_text_from_response_and_audit() ->
         failed_count=1,
         outcome_unknown_count=0,
         skipped_count=0,
+        paused_count=1,
+        batch_state="paused",
+        batch_pause_reason_code="smtp_authentication_failed",
+        smtp_connection_count=1,
+        smtp_reconnect_count=0,
         preflight_count=2,
         synchronous_send_policy={
             "max_recipient_jobs": 25,
@@ -113,7 +118,12 @@ def test_send_now_omits_provider_and_recipient_text_from_response_and_audit() ->
         "failed_count",
         "outcome_unknown_count",
         "skipped_count",
+        "paused_count",
         "preflight_count",
+        "batch_state",
+        "batch_pause_reason_code",
+        "smtp_connection_count",
+        "smtp_reconnect_count",
         "delivery_mode",
         "dry_run",
         "synchronous_send_policy",

tests/test_sending_jobs.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import unittest
 from types import SimpleNamespace
-from unittest.mock import patch
+from unittest.mock import ANY, patch
 
 from govoplan_campaign.backend.db.models import (
     JobBuildStatus,
@@ -17,6 +17,7 @@ from govoplan_campaign.backend.sending.jobs import (
     _select_campaign_jobs_for_queue,
     _send_claimed_campaign_job,
 )
+from govoplan_campaign.backend.integrations import SmtpSendError
 
 
 class FakeSession:
@@ -220,6 +221,62 @@ class CampaignQueueSelectionTests(unittest.TestCase):
         self.assertTrue(session.rolled_back)
         self.assertIn("Automatic retry is stopped", mark_unknown.call_args.kwargs["reason"])
 
+    def test_systemic_smtp_failure_pauses_remaining_campaign_jobs(self):
+        job = SimpleNamespace(
+            id="job-1",
+            tenant_id="tenant-1",
+            campaign_id="campaign-1",
+            campaign_version_id="version-1",
+            delivery_channel_policy="mail",
+            resolved_recipients={"from": {"email": "sender@example.test"}},
+        )
+        context = SimpleNamespace(
+            snapshot=SimpleNamespace(
+                mail_profile_id="profile-1",
+                smtp_server_id=None,
+                smtp_credential_id=None,
+                smtp_transport_revision="revision-1",
+                delivery=SimpleNamespace(rate_limit=SimpleNamespace(messages_per_minute=60)),
+            ),
+            message_bytes=b"message",
+            envelope_from="sender@example.test",
+            envelope_recipients=["recipient@example.test"],
+        )
+
+        class Mail:
+            def wait_for_rate_limit(self, **_kwargs):
+                return None
+
+            def send_campaign_email_bytes(self, *_args, **_kwargs):
+                raise SmtpSendError(
+                    "SMTP authentication failed.",
+                    systemic=True,
+                    reason_code="smtp_authentication_failed",
+                )
+
+        with (
+            patch("govoplan_campaign.backend.sending.jobs.mail_integration", return_value=Mail()),
+            patch("govoplan_campaign.backend.sending.jobs._record_attempt_start", return_value=SimpleNamespace(attempt_number=1)),
+            patch("govoplan_campaign.backend.sending.jobs._record_smtp_send_error", return_value=None),
+            patch("govoplan_campaign.backend.sending.jobs._pause_jobs_after_systemic_smtp_failure", return_value=4) as pause,
+            self.assertRaises(SmtpSendError),
+        ):
+            _send_claimed_campaign_job(
+                object(),  # type: ignore[arg-type]
+                job=job,  # type: ignore[arg-type]
+                claim_token="claim-1",
+                context=context,  # type: ignore[arg-type]
+                use_rate_limit=False,
+                enqueue_imap_task=False,
+            )
+
+        pause.assert_called_once_with(
+            ANY,
+            campaign_id="campaign-1",
+            exclude_job_id="job-1",
+            reason_code="smtp_authentication_failed",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_synchronous_delivery_policy.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from contextlib import contextmanager
 from types import SimpleNamespace
 from unittest.mock import Mock, patch
 
@@ -23,7 +24,9 @@ from govoplan_campaign.backend.sending.jobs import (
     QueueCampaignResult,
     SynchronousSendRejected,
     _ensure_synchronous_send_count_allowed,
+    _pause_jobs_after_systemic_smtp_failure,
     _preflight_synchronous_send_batch,
+    _synchronous_smtp_batch_manager,
     queue_campaign_jobs,
     send_campaign_now,
     synchronous_send_candidate_jobs,
@@ -364,6 +367,72 @@ def test_batch_preflight_checks_every_message_before_provider_effects() -> None:
     provider.send_campaign_email_bytes.assert_not_called()
 
 
+def test_smtp_batch_manager_preflights_combined_recipients_and_frozen_transport() -> None:
+    snapshot = SimpleNamespace(
+        uses_mail=True,
+        mail_profile_id="profile-1",
+        smtp_transport_revision="revision-1",
+        smtp_server_id="server-1",
+        smtp_credential_id="credential-1",
+    )
+    jobs = [
+        SimpleNamespace(
+            id="one",
+            tenant_id="tenant-1",
+            campaign_id="campaign-1",
+            delivery_channel_policy="mail",
+            resolved_recipients={"from": {"email": "sender@example.test"}},
+        ),
+        SimpleNamespace(
+            id="two",
+            tenant_id="tenant-1",
+            campaign_id="campaign-1",
+            delivery_channel_policy="mail",
+            resolved_recipients={"from": {"email": "sender@example.test"}},
+        ),
+    ]
+    contexts = {
+        "one": SimpleNamespace(snapshot=snapshot, envelope_from="sender@example.test", envelope_recipients=["one@example.test"]),
+        "two": SimpleNamespace(snapshot=snapshot, envelope_from="sender@example.test", envelope_recipients=["two@example.test", "one@example.test"]),
+    }
+    state = SimpleNamespace(connection_count=1, reconnect_count=0)
+
+    @contextmanager
+    def batch(_session, **kwargs):
+        batch.kwargs = kwargs
+        yield state
+
+    provider = SimpleNamespace(campaign_smtp_batch=batch)
+    with patch("govoplan_campaign.backend.sending.jobs.mail_integration", return_value=provider):
+        with _synchronous_smtp_batch_manager(object(), jobs=jobs, contexts=contexts) as opened:  # type: ignore[arg-type]
+            assert opened is state
+
+    assert batch.kwargs["envelope_recipients"] == ["one@example.test", "two@example.test"]
+    assert batch.kwargs["expected_smtp_transport_revision"] == "revision-1"
+
+
+def test_systemic_failure_pauses_only_remaining_queued_jobs() -> None:
+    session = Mock()
+    session.query.return_value.filter.return_value.update.return_value = 3
+    campaign = SimpleNamespace(id="campaign-1", tenant_id="tenant-1", status="sending")
+    session.get.return_value = campaign
+    with patch("govoplan_campaign.backend.sending.jobs.audit_event") as audit:
+        paused = _pause_jobs_after_systemic_smtp_failure(
+            session,
+            campaign_id="campaign-1",
+            exclude_job_id="failed-job",
+            reason_code="smtp_authentication_failed",
+        )
+
+    assert paused == 3
+    assert campaign.status == "ready_to_queue"
+    session.commit.assert_called_once_with()
+    assert audit.call_args.kwargs["details"] == {
+        "reason_code": "smtp_authentication_failed",
+        "paused_count": 3,
+    }
+
+
 def test_rejected_synchronous_preflight_rolls_back_staged_queue_before_audit() -> None:
     session = Mock()
     campaign = SimpleNamespace(id="campaign-1", current_version_id="version-1")

webui/src/features/campaigns/ReviewSendPage.tsx

@@ -932,10 +932,11 @@ export default function ReviewSendPage({
       const sent = result.sent_count ?? 0;
       const failed = result.failed_count ?? 0;
       const unknown = result.outcome_unknown_count ?? 0;
+      const paused = result.paused_count ?? 0;
       setMessage(
         effectiveDryRun ?
         "i18n:govoplan-campaign.dry_run_finished_no_message_was_sent.c026c6cd" :
-        `Send finished. SMTP accepted ${String(sent)} message(s), failed ${String(failed)}, outcome unknown ${String(unknown)}.`
+        `Send finished. SMTP accepted ${String(sent)} message(s), failed ${String(failed)}, outcome unknown ${String(unknown)}, paused ${String(paused)}.`
       );
       setSendConfirmOpen(false);
       await reload();
@@ -1765,6 +1766,13 @@ export default function ReviewSendPage({
             {sendResult &&
             <div className="review-flow-data-section">
                 <p className="muted small-note">i18n:govoplan-campaign.attempted.a9eb9c90 {String(sendResult.attempted_count ?? "—")}i18n:govoplan-campaign.smtp_accepted.a5d0dccc {String(sendResult.sent_count ?? "—")}i18n:govoplan-campaign.failed.fac9f871 {String(sendResult.failed_count ?? "—")}i18n:govoplan-campaign.outcome_unknown.4383023a {String(sendResult.outcome_unknown_count ?? 0)}i18n:govoplan-campaign.skipped.6b98496c {String(sendResult.skipped_count ?? "—")}.</p>
+                <p className="muted small-note">
+                  SMTP batch: {humanize(String(sendResult.batch_state ?? "not_started"))} · connections {String(sendResult.smtp_connection_count ?? 0)} · reconnects {String(sendResult.smtp_reconnect_count ?? 0)} · paused {String(sendResult.paused_count ?? 0)}.
+                </p>
+                {sendResult.batch_state === "paused" &&
+                  <DismissibleAlert tone="warning" resetKey={String(sendResult.batch_pause_reason_code ?? "smtp_systemic_failure")}>
+                    Remaining messages were paused before SMTP after a systemic transport failure ({String(sendResult.batch_pause_reason_code ?? "smtp_systemic_failure")}). Review the Mail profile, then resume the campaign queue.
+                  </DismissibleAlert>}
                 {sendResultRows.length > 0 &&
               <DataGrid
                 id={`campaign-${campaignId}-workflow-send-results`}