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/archive_encryption.py

@@ -0,0 +1,281 @@
+from __future__ import annotations
+
+import copy
+from dataclasses import dataclass
+from datetime import UTC, datetime
+import hashlib
+import json
+from typing import Any, Mapping
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal, has_scope
+from govoplan_core.core.policy import (
+    CampaignArchiveEncryptionDecision,
+    CampaignArchiveEncryptionRequest,
+    campaign_archive_encryption_policy,
+)
+from govoplan_campaign.backend.db.models import Campaign
+from govoplan_campaign.backend.runtime import get_registry
+
+
+LEGACY_ZIPCRYPTO_SCOPE = "campaigns:archive:use_legacy_zipcrypto"
+LEGACY_ZIPCRYPTO_LABEL = "Legacy ZipCrypto — Windows-compatible, weak encryption"
+
+
+class CampaignArchiveEncryptionError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class EffectiveArchiveEncryptionPolicy:
+    available: bool
+    allowed_password_encryption_methods: frozenset[str]
+    allowed_password_delivery_channels: frozenset[str]
+    policy_hash: str
+    source_path: tuple[Mapping[str, Any], ...]
+    reason: str
+    diagnostics: tuple[Mapping[str, Any], ...] = ()
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "available": self.available,
+            "allowed_password_encryption_methods": sorted(
+                self.allowed_password_encryption_methods
+            ),
+            "allowed_password_delivery_channels": sorted(
+                self.allowed_password_delivery_channels
+            ),
+            "policy_hash": self.policy_hash,
+            "source_path": [dict(item) for item in self.source_path],
+            "reason": self.reason,
+            "diagnostics": [dict(item) for item in self.diagnostics],
+            "legacy_label": LEGACY_ZIPCRYPTO_LABEL,
+        }
+
+
+def effective_archive_encryption_policy(
+    session: Session,
+    campaign: Campaign,
+) -> EffectiveArchiveEncryptionPolicy:
+    provider = campaign_archive_encryption_policy(get_registry())
+    if provider is None:
+        payload = {
+            "available": False,
+            "allowed_password_encryption_methods": ["aes"],
+            "allowed_password_delivery_channels": [
+                "in_person",
+                "letter",
+                "phone",
+                "separate_mail",
+                "sms",
+            ],
+            "source_path": [
+                {
+                    "scope_type": "system",
+                    "scope_id": None,
+                    "path": "system",
+                    "label": "Secure local fallback",
+                    "applied_fields": ["allowed_password_encryption_methods"],
+                    "policy": {
+                        "allowed_password_encryption_methods": ["aes"],
+                        "policy_provider": "unavailable",
+                    },
+                }
+            ],
+        }
+        return EffectiveArchiveEncryptionPolicy(
+            available=False,
+            allowed_password_encryption_methods=frozenset({"aes"}),
+            allowed_password_delivery_channels=frozenset(
+                {"separate_mail", "sms", "letter", "phone", "in_person"}
+            ),
+            policy_hash=_hash(payload),
+            source_path=tuple(payload["source_path"]),
+            reason=(
+                "Policy is unavailable. AES remains available through the secure "
+                "local baseline; legacy ZipCrypto fails closed."
+            ),
+        )
+    owner_type: str | None = None
+    owner_id: str | None = None
+    if campaign.owner_group_id:
+        owner_type, owner_id = "group", campaign.owner_group_id
+    elif campaign.owner_user_id:
+        owner_type, owner_id = "user", campaign.owner_user_id
+    decision: CampaignArchiveEncryptionDecision = (
+        provider.resolve_campaign_archive_encryption(
+            session,
+            request=CampaignArchiveEncryptionRequest(
+                tenant_id=campaign.tenant_id,
+                campaign_id=campaign.id,
+                owner_type=owner_type,  # type: ignore[arg-type]
+                owner_id=owner_id,
+            ),
+        )
+    )
+    return EffectiveArchiveEncryptionPolicy(
+        available=True,
+        allowed_password_encryption_methods=frozenset(
+            decision.allowed_password_encryption_methods
+        ),
+        allowed_password_delivery_channels=frozenset(
+            decision.allowed_password_delivery_channels
+        ),
+        policy_hash=decision.policy_hash,
+        source_path=tuple(step.to_dict() for step in decision.source_path),
+        reason=decision.reason or "Effective archive-encryption policy resolved.",
+        diagnostics=decision.diagnostics,
+    )
+
+
+def assert_archive_encryption_allowed(
+    session: Session,
+    campaign: Campaign,
+    raw_json: Mapping[str, Any],
+    *,
+    principal: ApiPrincipal | None = None,
+) -> EffectiveArchiveEncryptionPolicy:
+    policy = effective_archive_encryption_policy(session, campaign)
+    for archive in _archive_configs(raw_json):
+        method = str(archive.get("method") or "aes")
+        if method not in policy.allowed_password_encryption_methods:
+            raise CampaignArchiveEncryptionError(
+                f"{_method_label(method)} is blocked. {policy.reason}"
+            )
+        if method == "zip_standard":
+            if not policy.available:
+                raise CampaignArchiveEncryptionError(
+                    "Legacy ZipCrypto cannot be used while Policy is unavailable."
+                )
+            if principal is not None and not has_scope(principal, LEGACY_ZIPCRYPTO_SCOPE):
+                raise CampaignArchiveEncryptionError(
+                    f"Missing scope: {LEGACY_ZIPCRYPTO_SCOPE}"
+                )
+            if not archive.get("legacy_zipcrypto_acknowledged"):
+                raise CampaignArchiveEncryptionError(
+                    f"{LEGACY_ZIPCRYPTO_LABEL} requires explicit acknowledgement."
+                )
+            if len(str(archive.get("legacy_zipcrypto_reason") or "").strip()) < 10:
+                raise CampaignArchiveEncryptionError(
+                    f"{LEGACY_ZIPCRYPTO_LABEL} requires a reason of at least 10 characters."
+                )
+            if not archive.get("legacy_zipcrypto_acknowledged_by") or not archive.get(
+                "legacy_zipcrypto_acknowledged_at"
+            ):
+                raise CampaignArchiveEncryptionError(
+                    "Legacy ZipCrypto acknowledgement has no server-recorded actor or time. Save the campaign again."
+                )
+        if archive.get("password_enabled"):
+            # Existing campaign revisions predate the explicit field. Their
+            # model default is the separate-mail channel; apply the same
+            # normalization before policy enforcement so saved revisions do
+            # not become unusable merely because the field was omitted.
+            channel = str(
+                archive.get("password_delivery_channel") or "separate_mail"
+            )
+            if channel not in policy.allowed_password_delivery_channels:
+                raise CampaignArchiveEncryptionError(
+                    f"Password-delivery channel {channel!r} is blocked by the effective policy."
+                )
+    return policy
+
+
+def stamp_legacy_zipcrypto_acknowledgements(
+    session: Session,
+    campaign: Campaign,
+    current_raw_json: Mapping[str, Any],
+    candidate_raw_json: dict[str, Any] | None,
+    *,
+    principal: ApiPrincipal,
+) -> tuple[dict[str, Any] | None, list[dict[str, Any]]]:
+    if candidate_raw_json is None:
+        return None, []
+    candidate = copy.deepcopy(candidate_raw_json)
+    current_by_id = {
+        str(item.get("id") or index): item
+        for index, item in enumerate(_archive_configs(current_raw_json))
+    }
+    acknowledgements: list[dict[str, Any]] = []
+    policy = effective_archive_encryption_policy(session, campaign)
+    archives = _archive_configs(candidate)
+    for index, archive in enumerate(archives):
+        if str(archive.get("method") or "aes") != "zip_standard":
+            archive.pop("legacy_zipcrypto_acknowledged_by", None)
+            archive.pop("legacy_zipcrypto_acknowledged_at", None)
+            continue
+        if not has_scope(principal, LEGACY_ZIPCRYPTO_SCOPE):
+            raise CampaignArchiveEncryptionError(
+                f"Missing scope: {LEGACY_ZIPCRYPTO_SCOPE}"
+            )
+        if not policy.available or "zip_standard" not in policy.allowed_password_encryption_methods:
+            raise CampaignArchiveEncryptionError(
+                f"{LEGACY_ZIPCRYPTO_LABEL} is blocked. {policy.reason}"
+            )
+        reason = str(archive.get("legacy_zipcrypto_reason") or "").strip()
+        if not archive.get("legacy_zipcrypto_acknowledged") or len(reason) < 10:
+            raise CampaignArchiveEncryptionError(
+                f"{LEGACY_ZIPCRYPTO_LABEL} requires acknowledgement and a reason of at least 10 characters."
+            )
+        key = str(archive.get("id") or index)
+        previous = current_by_id.get(key, {})
+        unchanged = (
+            previous.get("method") == "zip_standard"
+            and previous.get("legacy_zipcrypto_acknowledged") is True
+            and str(previous.get("legacy_zipcrypto_reason") or "").strip() == reason
+            and previous.get("legacy_zipcrypto_acknowledged_by")
+            and previous.get("legacy_zipcrypto_acknowledged_at")
+        )
+        if unchanged:
+            archive["legacy_zipcrypto_acknowledged_by"] = previous[
+                "legacy_zipcrypto_acknowledged_by"
+            ]
+            archive["legacy_zipcrypto_acknowledged_at"] = previous[
+                "legacy_zipcrypto_acknowledged_at"
+            ]
+        else:
+            archive["legacy_zipcrypto_acknowledged_by"] = principal.user.id
+            archive["legacy_zipcrypto_acknowledged_at"] = datetime.now(UTC).isoformat()
+            acknowledgements.append(
+                {
+                    "archive_id": key,
+                    "reason": reason,
+                    "policy_hash": policy.policy_hash,
+                }
+            )
+    return candidate, acknowledgements
+
+
+def has_password_archives(raw_json: Mapping[str, Any]) -> bool:
+    return any(bool(item.get("password_enabled")) for item in _archive_configs(raw_json))
+
+
+def _archive_configs(raw_json: Mapping[str, Any]) -> list[dict[str, Any]]:
+    attachments = raw_json.get("attachments")
+    zip_config = attachments.get("zip") if isinstance(attachments, Mapping) else None
+    archives = zip_config.get("archives") if isinstance(zip_config, Mapping) else None
+    if isinstance(archives, list):
+        return [item for item in archives if isinstance(item, dict)]
+    return []
+
+
+def _method_label(method: str) -> str:
+    return LEGACY_ZIPCRYPTO_LABEL if method == "zip_standard" else method.upper()
+
+
+def _hash(value: object) -> str:
+    return hashlib.sha256(
+        json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode()
+    ).hexdigest()
+
+
+__all__ = [
+    "CampaignArchiveEncryptionError",
+    "EffectiveArchiveEncryptionPolicy",
+    "LEGACY_ZIPCRYPTO_LABEL",
+    "LEGACY_ZIPCRYPTO_SCOPE",
+    "assert_archive_encryption_allowed",
+    "effective_archive_encryption_policy",
+    "has_password_archives",
+    "stamp_legacy_zipcrypto_acknowledgements",
+]

src/govoplan_campaign/backend/campaign/models.py

@@ -76,6 +76,14 @@ class ZipPasswordScope(StrEnum):
     GLOBAL = "global"
 
 
+class ZipPasswordDeliveryChannel(StrEnum):
+    SEPARATE_MAIL = "separate_mail"
+    SMS = "sms"
+    LETTER = "letter"
+    PHONE = "phone"
+    IN_PERSON = "in_person"
+
+
 class ZipPasswordMode(StrEnum):
     NONE = "none"
     DIRECT = "direct"
@@ -349,6 +357,13 @@ class ZipArchiveConfig(StrictModel):
     password_field: str | None = None
     password_scope: ZipPasswordScope = ZipPasswordScope.LOCAL
     method: ZipMethod = ZipMethod.AES
+    password_delivery_channel: ZipPasswordDeliveryChannel = (
+        ZipPasswordDeliveryChannel.SEPARATE_MAIL
+    )
+    legacy_zipcrypto_acknowledged: bool = False
+    legacy_zipcrypto_reason: str | None = Field(default=None, max_length=1000)
+    legacy_zipcrypto_acknowledged_by: str | None = Field(default=None, max_length=255)
+    legacy_zipcrypto_acknowledged_at: str | None = Field(default=None, max_length=80)
 
     # Compatibility fields for campaigns created by the first single-archive
     # implementation. New WebUI campaigns use password_enabled/field/scope.
@@ -376,6 +391,20 @@ class ZipArchiveConfig(StrictModel):
             normalized["password_scope"] = ZipPasswordScope.LOCAL.value
         return normalized
 
+    @model_validator(mode="after")
+    def validate_legacy_zipcrypto_acknowledgement(self) -> "ZipArchiveConfig":
+        if self.method != ZipMethod.ZIP_STANDARD:
+            return self
+        if not self.legacy_zipcrypto_acknowledged:
+            raise ValueError(
+                "Legacy ZipCrypto requires explicit acknowledgement of its weak encryption"
+            )
+        if len((self.legacy_zipcrypto_reason or "").strip()) < 10:
+            raise ValueError(
+                "Legacy ZipCrypto requires an acknowledgement reason of at least 10 characters"
+            )
+        return self
+
 
 class ZipCollectionConfig(StrictModel):
     enabled: bool = False

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

@@ -159,6 +159,12 @@ PERMISSIONS = (
         "Build exact messages and attachment evidence.",
         "Campaigns",
     ),
+    _permission(
+        "campaigns:archive:use_legacy_zipcrypto",
+        "Use legacy ZipCrypto",
+        "Explicitly select weak Windows-compatible ZipCrypto when the effective policy permits it.",
+        "Campaign governance",
+    ),
     _permission(
         "campaigns:campaign:review",
         "Complete campaign review",
@@ -907,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"),
@@ -1025,6 +1031,33 @@ manifest = ModuleManifest(
                 ],
             },
         ),
+        DocumentationTopic(
+            id="campaigns.archive-encryption-governance",
+            title="Use governed password-protected ZIP attachments",
+            summary="Use AES by default and select weak Windows-compatible ZipCrypto only with explicit policy, permission, acknowledgement, and evidence.",
+            body=(
+                "Campaign resolves archive encryption through Policy across system, tenant, owner user or group, and campaign scopes. Password-protected archives use AES unless the complete inherited policy permits Legacy ZipCrypto — Windows-compatible, weak encryption and the actor has campaigns:archive:use_legacy_zipcrypto. A legacy selection requires a reasoned acknowledgement. Passwords are never included in Campaign evidence or the campaign message and must be conveyed through the separately selected, policy-allowed channel. Each build freezes the archive and member hashes, implementation version, policy hash and source path, acknowledgement actor, reason and time, and build identity. A more restrictive later policy blocks queueing and sending until the campaign is rebuilt; Campaign never falls back from AES to ZipCrypto after an error. Temporary plaintext and archive material is confined to the bounded build directory and removed after success or failure."
+            ),
+            documentation_types=("user", "admin"),
+            audience=("campaign_manager", "campaign_reviewer", "policy_admin"),
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("campaigns",),
+                    any_scopes=(
+                        "campaigns:campaign:update",
+                        "campaigns:campaign:review",
+                        "admin:policies:read",
+                    ),
+                ),
+            ),
+            related_modules=("policy", "audit", "access"),
+            metadata={
+                "kind": "workflow",
+                "route": "/campaigns/{campaign_id}/files",
+                "screen": "Campaign attachments",
+                "help_contexts": ["campaign.archive-encryption"],
+            },
+        ),
         DocumentationTopic(
             id="campaigns.workflow.complete-review",
             title="Inspect built messages and complete review",

src/govoplan_campaign/backend/messages/builder.py

@@ -4,7 +4,7 @@ import mimetypes
 import re
 import tempfile
 import time
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from email.message import EmailMessage
 from email.utils import make_msgid, formatdate
 from pathlib import Path
@@ -40,7 +40,10 @@ from govoplan_campaign.backend.campaign.models import (
     effective_delivery_channel_policy,
 )
 from govoplan_campaign.backend.campaign.template_values import build_template_values
-from govoplan_campaign.backend.services.zip_service import create_zip_archive
+from govoplan_campaign.backend.services.zip_service import (
+    create_zip_archive,
+    zip_archive_evidence,
+)
 from govoplan_campaign.backend.template_rendering import (
     find_unresolved_placeholders as _find_unresolved_placeholders,
     render_template as _render_template,
@@ -93,6 +96,7 @@ class _MimeBuildResult:
     build_status: BuildStatus
     validation_status: MessageValidationStatus
     attachment_count: int
+    archive_evidence: list[dict[str, object]] = field(default_factory=list)
 
 
 @dataclass(slots=True)
@@ -372,8 +376,9 @@ def _attach_files(
     resolution: EntryAttachmentResolution,
     values: dict[str, Any],
     work_dir: Path,
-) -> int:
+) -> tuple[int, list[dict[str, object]]]:
     attached_count = 0
+    evidence: list[dict[str, object]] = []
     archive_members: dict[str, list[tuple[Path, str]]] = {}
     archive_attachments: dict[str, list[ResolvedAttachment]] = {}
     used_message_filenames: set[str] = set()
@@ -429,13 +434,38 @@ def _attach_files(
             password,
             archive.method.value,
         )
+        archive_record = zip_archive_evidence(
+            archive_path,
+            members,
+            password_protected=bool(password),
+            method=archive.method.value,
+        )
+        archive_record.update(
+            {
+                "archive_id": archive.id,
+                "filename": filename,
+                "password_delivery_channel": (
+                    archive.password_delivery_channel.value if password else None
+                ),
+                "legacy_acknowledgement": (
+                    {
+                        "actor_id": archive.legacy_zipcrypto_acknowledged_by,
+                        "reason": archive.legacy_zipcrypto_reason,
+                        "recorded_at": archive.legacy_zipcrypto_acknowledged_at,
+                    }
+                    if archive.method.value == "zip_standard"
+                    else None
+                ),
+            }
+        )
+        evidence.append(archive_record)
         data, maintype, subtype = _attachment_bytes(archive_path)
         message.add_attachment(data, maintype=maintype, subtype=subtype, filename=filename)
         attached_count += 1
         for attachment in archive_attachments.get(archive.id, []):
             attachment.zip_filename = filename
 
-    return attached_count
+    return attached_count, evidence
 
 def _imap_initial_status(
     config: CampaignConfig,
@@ -527,6 +557,7 @@ def _message_draft(
     imap_status: ImapStatus | None = None,
     subject: str | None = None,
     attachment_count: int = 0,
+    archive_evidence: list[dict[str, object]] | None = None,
     issues: list[MessageIssue] | None = None,
     eml_path: str | None = None,
     eml_size: int | None = None,
@@ -566,6 +597,7 @@ def _message_draft(
         disposition_notification_to=_message_addresses(context.recipients["disposition_notification_to"]),
         attachment_count=attachment_count,
         attachments=_attachment_summaries(context.resolution),
+        archive_evidence=archive_evidence or [],
         issues=issues if issues is not None else context.issues,
         eml_path=eml_path,
         eml_size_bytes=eml_size,
@@ -763,7 +795,7 @@ def _build_mime_message(
         _populate_message_body(message, rendered)
         if work_dir is None:
             work_dir = output_dir or Path(tempfile.mkdtemp(prefix="govoplan-build-"))
-        attachment_count = _attach_files(
+        attachment_count, archive_evidence = _attach_files(
             message=message,
             config=config,
             entry=entry,
@@ -789,6 +821,7 @@ def _build_mime_message(
             build_status=BuildStatus.BUILT,
             validation_status=context.validation_status,
             attachment_count=attachment_count,
+            archive_evidence=archive_evidence,
         )
     except ZipBuildError as exc:
         context.issues.append(
@@ -893,6 +926,7 @@ def build_entry_message(
         validation_status=mime_result.validation_status,
         subject=rendered.subject,
         attachment_count=mime_result.attachment_count,
+        archive_evidence=mime_result.archive_evidence,
         eml_path=eml_path,
         eml_size=eml_size,
     )
@@ -1097,6 +1131,7 @@ def _build_residual_file_message(
             validation_status=mime_result.validation_status,
             subject=rendered.subject,
             attachment_count=mime_result.attachment_count,
+            archive_evidence=mime_result.archive_evidence,
             eml_path=eml_path,
             eml_size=eml_size,
         ),

src/govoplan_campaign/backend/messages/models.py

@@ -94,6 +94,7 @@ class MessageDraft(BaseModel):
 
     attachment_count: int = 0
     attachments: list[MessageAttachmentSummary] = Field(default_factory=list)
+    archive_evidence: list[dict[str, object]] = Field(default_factory=list)
     issues: list[MessageIssue] = Field(default_factory=list)
 
     eml_path: str | None = None

src/govoplan_campaign/backend/persistence/campaigns.py

@@ -40,6 +40,10 @@ from govoplan_campaign.backend.db.models import (
     JobSendStatus,
     JobValidationStatus,
 )
+from govoplan_campaign.backend.archive_encryption import (
+    CampaignArchiveEncryptionError,
+    assert_archive_encryption_allowed,
+)
 from govoplan_campaign.backend.campaign.loader import (
     load_campaign_json,
     validate_against_schema,
@@ -657,6 +661,15 @@ def validate_campaign_version(
         raise CampaignPersistenceError(
             "Campaign version is not accessible for this tenant"
         )
+    try:
+        archive_policy = assert_archive_encryption_allowed(
+            session,
+            campaign,
+            version.raw_json if isinstance(version.raw_json, dict) else {},
+            principal=principal,
+        )
+    except CampaignArchiveEncryptionError as exc:
+        raise CampaignPersistenceError(str(exc)) from exc
     _ensure_current_campaign_version(campaign, version, action="validate")
     if _version_is_user_locked(version) or version.workflow_state in {
         CampaignVersionWorkflowState.QUEUED.value,
@@ -734,6 +747,7 @@ def validate_campaign_version(
             "warning_count": report.warning_count,
             "validated_at": datetime.now(UTC).isoformat(),
             "validated_by_user_id": user_id,
+            "archive_encryption_policy": archive_policy.to_dict(),
         }
     )
     version.validation_summary = report_json
@@ -1430,6 +1444,11 @@ def _store_execution_snapshot(
         delivery=config.delivery,
         jobs=jobs,
         build_summary=build_summary,
+        archive_encryption=(
+            build_summary.get("archive_encryption")
+            if isinstance(build_summary.get("archive_encryption"), dict)
+            else None
+        ),
     )
     version.execution_snapshot = snapshot
     version.execution_snapshot_hash = snapshot_hash
@@ -1615,6 +1634,15 @@ def build_campaign_version(
         raise CampaignPersistenceError(
             "Campaign version is not accessible for this tenant"
         )
+    try:
+        archive_policy = assert_archive_encryption_allowed(
+            session,
+            campaign,
+            version.raw_json if isinstance(version.raw_json, dict) else {},
+            principal=principal,
+        )
+    except CampaignArchiveEncryptionError as exc:
+        raise CampaignPersistenceError(str(exc)) from exc
     _ensure_current_campaign_version(campaign, version, action="build")
     if version.workflow_state == CampaignVersionWorkflowState.COMPLETED.value:
         raise CampaignPersistenceError("Sent campaign versions cannot be rebuilt")
@@ -1749,6 +1777,25 @@ def build_campaign_version(
             )
         report_json = _campaign_build_report(result, files)
         report_json["built_by_user_id"] = user_id
+        archive_records = [
+            {
+                **archive,
+                "campaign_id": campaign.id,
+                "campaign_version_id": version.id,
+                "build_token": report_json["build_token"],
+                "built_at": report_json["built_at"],
+                "policy_hash": archive_policy.policy_hash,
+                "policy_source_path": [
+                    dict(item) for item in archive_policy.source_path
+                ],
+            }
+            for message in result.report.messages
+            for archive in message.archive_evidence
+        ]
+        report_json["archive_encryption"] = {
+            "policy": archive_policy.to_dict(),
+            "archives": archive_records,
+        }
         if resolved_print_outputs_by_index:
             first_output = next(iter(resolved_print_outputs_by_index.values()))
             report_json["print_output"] = {

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/route_support.py

@@ -13,6 +13,10 @@ from govoplan_campaign.backend.campaign.mail_profile_boundary import (
     CAMPAIGN_MAIL_SERVER_KEYS,
     campaign_mail_profile_id,
 )
+from govoplan_campaign.backend.archive_encryption import (
+    CampaignArchiveEncryptionError,
+    stamp_legacy_zipcrypto_acknowledgements,
+)
 from govoplan_campaign.backend.db.models import (
     Campaign,
     CampaignIssue,
@@ -388,7 +392,9 @@ def _update_campaign_version_detail_response(
     autosave: bool,
     audit_action: str,
 ) -> CampaignVersionDetailResponse:
-    _get_campaign_for_principal(session, campaign_id, principal, write=True)
+    campaign = _get_campaign_for_principal(
+        session, campaign_id, principal, write=True
+    )
     current_version = _get_version_for_tenant(session, version_id, principal.tenant_id)
     if payload.base_revision is None:
         error = MissingPreconditionError(
@@ -421,9 +427,40 @@ def _update_campaign_version_detail_response(
         ) from exc
     if _recipient_sections_changed(current_version.raw_json, payload.campaign_json):
         _require_permission(principal, "campaigns:recipient:write")
+    acknowledgements: list[dict[str, Any]] = []
+    try:
+        payload.campaign_json, acknowledgements = (
+            stamp_legacy_zipcrypto_acknowledgements(
+                session,
+                campaign,
+                current_version.raw_json
+                if isinstance(current_version.raw_json, dict)
+                else {},
+                payload.campaign_json,
+                principal=principal,
+            )
+        )
+    except CampaignArchiveEncryptionError as exc:
+        audit_from_principal(
+            session,
+            principal,
+            action="campaign.archive_encryption_denied",
+            object_type="campaign_version",
+            object_id=version_id,
+            details={"campaign_id": campaign_id, "reason": str(exc)},
+            commit=True,
+        )
+        raise HTTPException(
+            status_code=(
+                status.HTTP_403_FORBIDDEN
+                if "Missing scope:" in str(exc)
+                else status.HTTP_422_UNPROCESSABLE_CONTENT
+            ),
+            detail=str(exc),
+        ) from exc
     _require_mail_profile_use_if_needed(principal, payload.campaign_json)
     try:
-        return _campaign_version_detail_response(
+        result = _campaign_version_detail_response(
             session,
             principal,
             campaign_id,
@@ -462,9 +499,23 @@ def _update_campaign_version_detail_response(
                     }
                 ),
                 "legacy_mail_settings_migrated": payload.migrate_legacy_mail_settings,
+                "legacy_zipcrypto_acknowledgements": acknowledgements,
             },
             validation_error_status=status.HTTP_422_UNPROCESSABLE_CONTENT,
         )
+        for acknowledgement in acknowledgements:
+            audit_from_principal(
+                session,
+                principal,
+                action="campaign.legacy_zipcrypto_acknowledged",
+                object_type="campaign_version",
+                object_id=version_id,
+                details={"campaign_id": campaign_id, **acknowledgement},
+                commit=False,
+            )
+        if acknowledgements:
+            session.commit()
+        return result
     except RevisionConflictError as exc:
         session.rollback()
         audit_from_principal(

src/govoplan_campaign/backend/routes/campaigns.py

@@ -130,9 +130,22 @@ from govoplan_campaign.backend.route_support import (
     _write_current_version_snapshot_if_available,
     bounded_query_rows as _bounded_query_rows,
 )
+from govoplan_campaign.backend.archive_encryption import (
+    effective_archive_encryption_policy,
+)
 
 router = APIRouter(prefix="/campaigns", tags=["campaigns"])
 
+
+@router.get("/{campaign_id}/archive-encryption-policy")
+def campaign_archive_encryption_policy(
+    campaign_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:read")),
+):
+    campaign = _get_campaign_for_principal(session, campaign_id, principal)
+    return effective_archive_encryption_policy(session, campaign).to_dict()
+
 CAPABILITY_ADDRESSES_LOOKUP = "addresses.lookup"
 CAPABILITY_ADDRESSES_RECIPIENT_SOURCE = "addresses.recipient_source"
 

src/govoplan_campaign/backend/routes/versions.py

@@ -770,6 +770,14 @@ def validate_version(
     except HTTPException:
         raise
     except CampaignPersistenceError as exc:
+        if _is_archive_encryption_denial(exc):
+            _audit_archive_encryption_denial(
+                session, principal, version_id=version_id, error=exc
+            )
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+                detail=str(exc),
+            ) from exc
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
         ) from exc
@@ -883,6 +891,9 @@ def build_version(
                 "attachment_reuse": _attachment_reuse_audit_evidence(
                     result.get("attachment_reuse")
                 ),
+                "archive_encryption": _archive_encryption_audit_evidence(
+                    result.get("archive_encryption")
+                ),
             },
             commit=True,
         )
@@ -891,6 +902,18 @@ def build_version(
             include_diagnostics=has_scope(principal, "campaigns:diagnostic:read"),
         )
     except CampaignPersistenceError as exc:
+        if _is_archive_encryption_denial(exc):
+            _audit_archive_encryption_denial(
+                session, principal, version_id=version_id, error=exc
+            )
+            raise HTTPException(
+                status_code=(
+                    status.HTTP_403_FORBIDDEN
+                    if "Missing scope:" in str(exc)
+                    else status.HTTP_422_UNPROCESSABLE_CONTENT
+                ),
+                detail=str(exc),
+            ) from exc
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
         ) from exc
@@ -915,6 +938,38 @@ def build_version(
         ) from exc
 
 
+def _is_archive_encryption_denial(error: Exception) -> bool:
+    message = str(error).casefold()
+    return any(
+        marker in message
+        for marker in (
+            "archive-encryption",
+            "archive encryption",
+            "zipcrypto",
+            "password-delivery channel",
+        )
+    )
+
+
+def _audit_archive_encryption_denial(
+    session: Session,
+    principal: ApiPrincipal,
+    *,
+    version_id: str,
+    error: Exception,
+) -> None:
+    session.rollback()
+    audit_from_principal(
+        session,
+        principal,
+        action="campaign.archive_encryption_denied",
+        object_type="campaign_version",
+        object_id=version_id,
+        details={"reason": str(error)},
+        commit=True,
+    )
+
+
 def _residual_file_audit_evidence(value: object) -> dict[str, object]:
     if not isinstance(value, dict):
         return {}
@@ -945,6 +1000,25 @@ def _attachment_reuse_audit_evidence(value: object) -> dict[str, object]:
     }
 
 
+def _archive_encryption_audit_evidence(value: object) -> dict[str, object]:
+    if not isinstance(value, dict):
+        return {}
+    policy = value.get("policy")
+    archives = [
+        item for item in (value.get("archives") or []) if isinstance(item, dict)
+    ]
+    return {
+        "policy_hash": policy.get("policy_hash")
+        if isinstance(policy, dict)
+        else None,
+        "archive_count": len(archives),
+        "legacy_zipcrypto_count": sum(
+            1 for item in archives if item.get("method") == "zip_standard"
+        ),
+        "archive_sha256": [item.get("archive_sha256") for item in archives],
+    }
+
+
 def _review_decision_audit_evidence(
     version: CampaignVersion,
 ) -> dict[str, object]:

src/govoplan_campaign/backend/schema/campaign.schema.json

@@ -1625,6 +1625,44 @@
           ],
           "default": "aes"
         },
+        "password_delivery_channel": {
+          "type": "string",
+          "enum": [
+            "separate_mail",
+            "sms",
+            "letter",
+            "phone",
+            "in_person"
+          ],
+          "default": "separate_mail"
+        },
+        "legacy_zipcrypto_acknowledged": {
+          "type": "boolean",
+          "default": false
+        },
+        "legacy_zipcrypto_reason": {
+          "type": [
+            "string",
+            "null"
+          ],
+          "maxLength": 1000
+        },
+        "legacy_zipcrypto_acknowledged_by": {
+          "type": [
+            "string",
+            "null"
+          ],
+          "maxLength": 255,
+          "readOnly": true
+        },
+        "legacy_zipcrypto_acknowledged_at": {
+          "type": [
+            "string",
+            "null"
+          ],
+          "maxLength": 80,
+          "readOnly": true
+        },
         "password_mode": {
           "type": [
             "string",

src/govoplan_campaign/backend/sending/execution.py

@@ -9,6 +9,10 @@ from pydantic import BaseModel, ConfigDict
 from sqlalchemy.orm import Session
 
 from govoplan_campaign.backend.db.models import Campaign, CampaignJob, CampaignVersion, JobValidationStatus
+from govoplan_campaign.backend.archive_encryption import (
+    CampaignArchiveEncryptionError,
+    assert_archive_encryption_allowed,
+)
 from govoplan_campaign.backend.campaign.models import (
     DeliveryChannelPolicy,
     DeliveryConfig,
@@ -22,8 +26,8 @@ from govoplan_campaign.backend.campaign.mail_profile_boundary import (
 from govoplan_campaign.backend.integrations import MailProfileError, files_integration, mail_integration
 from govoplan_campaign.backend.path_security import CampaignPathSecurityError, assert_server_safe_campaign_paths
 
-SNAPSHOT_VERSION = "8"
-SUPPORTED_SNAPSHOT_VERSIONS = {"6", "7", SNAPSHOT_VERSION}
+SNAPSHOT_VERSION = "9"
+SUPPORTED_SNAPSHOT_VERSIONS = {"6", "7", "8", SNAPSHOT_VERSION}
 
 
 class ExecutionSnapshotError(RuntimeError):
@@ -57,6 +61,7 @@ class ExecutionSnapshot(BaseModel):
     queueable_job_count: int = 0
     job_manifest_sha256: str | None = None
     effective_policy_sha256: str | None = None
+    archive_encryption: dict[str, Any] | None = None
     smtp_transport_revision: str | None = None
     imap_transport_revision: str | None = None
     uses_mail: bool = True
@@ -263,6 +268,7 @@ def create_execution_snapshot(
     imap_credential_id: str | None = None,
     jobs: Iterable[CampaignJob] = (),
     build_summary: dict[str, Any] | None = None,
+    archive_encryption: dict[str, Any] | None = None,
 ) -> tuple[dict[str, Any], str]:
     raw_json = version.raw_json if isinstance(version.raw_json, dict) else {}
     job_list = list(jobs)
@@ -311,6 +317,7 @@ def create_execution_snapshot(
             delivery,
             snapshot_version=SNAPSHOT_VERSION,
         ),
+        archive_encryption=archive_encryption,
         smtp_transport_revision=smtp_transport_revision,
         imap_transport_revision=imap_transport_revision,
         uses_mail=uses_mail,
@@ -355,6 +362,39 @@ def _assert_snapshot_matches_persisted_inputs(
             "Revalidate and rebuild the campaign before delivery."
         )
 
+    campaign = session.get(Campaign, version.campaign_id)
+    if campaign is None:
+        raise ExecutionSnapshotError("Execution snapshot Campaign no longer exists")
+    try:
+        current_archive_policy = assert_archive_encryption_allowed(
+            session,
+            campaign,
+            raw_json,
+        )
+    except CampaignArchiveEncryptionError as exc:
+        raise ExecutionSnapshotError(str(exc)) from exc
+    archive_snapshot = snapshot.archive_encryption
+    configured_archives = (
+        ((raw_json.get("attachments") or {}).get("zip") or {}).get("archives")
+        if isinstance(raw_json.get("attachments"), dict)
+        else None
+    )
+    if configured_archives and not isinstance(archive_snapshot, dict):
+        raise ExecutionSnapshotError(
+            "Execution snapshot has no governed archive-encryption evidence; rebuild before delivery."
+        )
+    if isinstance(archive_snapshot, dict):
+        frozen_policy = archive_snapshot.get("policy")
+        frozen_hash = (
+            frozen_policy.get("policy_hash")
+            if isinstance(frozen_policy, dict)
+            else None
+        )
+        if frozen_hash != current_archive_policy.policy_hash:
+            raise ExecutionSnapshotError(
+                "The effective archive-encryption policy changed after build. Revalidate and rebuild before delivery."
+            )
+
     if effect_job is not None:
         if effect_job.campaign_version_id != version.id:
             raise ExecutionSnapshotError("Campaign job does not belong to the snapshotted version")
@@ -494,6 +534,12 @@ def ensure_execution_snapshot(
         delivery=config.delivery,
         jobs=jobs,
         build_summary=version.build_summary if isinstance(version.build_summary, dict) else {},
+        archive_encryption=(
+            version.build_summary.get("archive_encryption")
+            if isinstance(version.build_summary, dict)
+            and isinstance(version.build_summary.get("archive_encryption"), dict)
+            else None
+        ),
     )
     version.execution_snapshot = payload
     version.execution_snapshot_hash = digest

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:
-        try:
-            result = _deliver_job_with_recovery(
-                session,
-                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:  # keep sending other jobs and return per-job details
-            failed_count += 1
-            results.append({"job_id": job.id, "status": "failed", "message": str(exc)})
+    attempted_count = 0
+    paused_count = 0
+    pause_reason_code: str | None = None
+    batch_state = "ready"
+    batch_manager = _synchronous_smtp_batch_manager(
+        session,
+        jobs=jobs,
+        contexts=delivery_contexts,
+    )
+    try:
+        with batch_manager as smtp_batch:
+            # Queue state becomes durable only after local and SMTP
+            # DNS/connectivity/TLS/auth preflight succeeds.
+            session.commit()
+            for index, job in enumerate(jobs):
+                attempted_count += 1
+                try:
+                    result = _deliver_job_with_recovery(
+                        session,
+                        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

src/govoplan_campaign/backend/services/zip_service.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 import binascii
 from datetime import datetime
+import hashlib
+from importlib import metadata
 import secrets
 import stat
 import struct
@@ -49,6 +51,8 @@ def create_zip_archive(
     output_path.parent.mkdir(parents=True, exist_ok=True)
     members = _normalized_members(files)
     if password:
+        if method not in {ZIP_METHOD_AES, ZIP_METHOD_STANDARD}:
+            raise ValueError(f"Unsupported password-encryption method: {method}")
         if method == ZIP_METHOD_STANDARD:
             _create_zipcrypto_archive(output_path, members, password)
             return output_path
@@ -61,6 +65,51 @@ def create_zip_archive(
     return output_path
 
 
+def zip_archive_evidence(
+    output_path: Path,
+    members: Iterable[Path | ArchiveMember],
+    *,
+    password_protected: bool,
+    method: str,
+) -> dict[str, object]:
+    """Return password-free, content-addressed evidence for one built archive."""
+
+    normalized = _normalized_members(members)
+    archive_bytes = output_path.read_bytes()
+    if password_protected and method == ZIP_METHOD_AES:
+        try:
+            implementation_version = metadata.version("pyzipper")
+        except metadata.PackageNotFoundError:  # pragma: no cover - guarded by writer
+            implementation_version = "unknown"
+        implementation = "pyzipper"
+        archive_format = "WinZip AES"
+    elif password_protected and method == ZIP_METHOD_STANDARD:
+        implementation = "govoplan-campaign.zipcrypto"
+        implementation_version = "1"
+        archive_format = "Legacy ZipCrypto"
+    else:
+        implementation = "python.zipfile"
+        implementation_version = "stdlib"
+        archive_format = "ZIP (unencrypted)"
+    return {
+        "format": archive_format,
+        "method": method if password_protected else "none",
+        "password_protected": password_protected,
+        "implementation": implementation,
+        "implementation_version": implementation_version,
+        "archive_sha256": hashlib.sha256(archive_bytes).hexdigest(),
+        "archive_size_bytes": len(archive_bytes),
+        "members": [
+            {
+                "name": archive_name,
+                "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
+                "size_bytes": path.stat().st_size,
+            }
+            for path, archive_name in normalized
+        ],
+    }
+
+
 def create_encrypted_zip(output_path: Path, files: list[Path], password: str, method: str = ZIP_METHOD_AES) -> Path:
     """Backward-compatible wrapper for the original per-rule ZIP helper."""
 

tests/test_archive_encryption_governance.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_campaign.backend.archive_encryption import (
+    CampaignArchiveEncryptionError,
+    LEGACY_ZIPCRYPTO_SCOPE,
+    assert_archive_encryption_allowed,
+    effective_archive_encryption_policy,
+    stamp_legacy_zipcrypto_acknowledgements,
+)
+from govoplan_campaign.backend.db.models import Campaign
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.policy import (
+    CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION,
+    CampaignArchiveEncryptionDecision,
+    PolicySourceStep,
+)
+
+
+class _PolicyProvider:
+    def __init__(self, methods: set[str]) -> None:
+        self.methods = methods
+
+    def resolve_campaign_archive_encryption(self, session=None, *, request):
+        del session, request
+        return CampaignArchiveEncryptionDecision(
+            allowed_password_encryption_methods=frozenset(self.methods),
+            allowed_password_delivery_channels=frozenset(
+                {"separate_mail", "sms", "letter", "phone", "in_person"}
+            ),
+            policy_hash="f" * 64,
+            source_path=(
+                PolicySourceStep(
+                    scope_type="system",
+                    label="System archive-encryption policy",
+                    applied_fields=("allowed_password_encryption_methods",),
+                    policy={"allowed_password_encryption_methods": sorted(self.methods)},
+                ),
+            ),
+        )
+
+
+class _Registry:
+    def __init__(self, provider) -> None:
+        self.provider = provider
+
+    def has_capability(self, name: str) -> bool:
+        return name == CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION
+
+    def capability(self, name: str):
+        return self.provider if self.has_capability(name) else None
+
+
+class CampaignArchiveEncryptionGovernanceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        self.session = Session(self.engine)
+        self.campaign = Campaign(
+            id="campaign-1",
+            tenant_id="tenant-1",
+            external_id="example",
+            name="Example",
+            owner_user_id="user-1",
+        )
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_unavailable_policy_keeps_aes_and_fails_closed_for_legacy(self) -> None:
+        with patch("govoplan_campaign.backend.archive_encryption.get_registry", return_value=None):
+            policy = effective_archive_encryption_policy(self.session, self.campaign)
+            self.assertFalse(policy.available)
+            self.assertEqual(frozenset({"aes"}), policy.allowed_password_encryption_methods)
+            assert_archive_encryption_allowed(
+                self.session,
+                self.campaign,
+                _raw_archive("aes"),
+            )
+            with self.assertRaisesRegex(CampaignArchiveEncryptionError, "blocked"):
+                assert_archive_encryption_allowed(
+                    self.session,
+                    self.campaign,
+                    _raw_archive("zip_standard", stamped=True),
+                )
+
+    def test_existing_password_archive_inherits_separate_mail_channel(self) -> None:
+        raw = _raw_archive("aes")
+        raw["attachments"]["zip"]["archives"][0].pop("password_delivery_channel")
+        with patch(
+            "govoplan_campaign.backend.archive_encryption.get_registry",
+            return_value=None,
+        ):
+            decision = assert_archive_encryption_allowed(
+                self.session,
+                self.campaign,
+                raw,
+            )
+        self.assertIn(
+            "separate_mail",
+            decision.allowed_password_delivery_channels,
+        )
+
+    def test_legacy_selection_requires_permission_and_gets_server_stamp(self) -> None:
+        registry = _Registry(_PolicyProvider({"aes", "zip_standard"}))
+        candidate = _raw_archive("zip_standard")
+        candidate["attachments"]["zip"]["archives"][0].update(
+            {
+                "legacy_zipcrypto_acknowledged": True,
+                "legacy_zipcrypto_reason": "Recipient requires built-in Windows extraction",
+            }
+        )
+        with patch("govoplan_campaign.backend.archive_encryption.get_registry", return_value=registry):
+            with self.assertRaisesRegex(CampaignArchiveEncryptionError, "Missing scope"):
+                stamp_legacy_zipcrypto_acknowledgements(
+                    self.session,
+                    self.campaign,
+                    {},
+                    candidate,
+                    principal=_principal(set()),
+                )
+            stamped, evidence = stamp_legacy_zipcrypto_acknowledgements(
+                self.session,
+                self.campaign,
+                {},
+                candidate,
+                principal=_principal({LEGACY_ZIPCRYPTO_SCOPE}),
+            )
+            archive = stamped["attachments"]["zip"]["archives"][0]
+            self.assertEqual("user-1", archive["legacy_zipcrypto_acknowledged_by"])
+            self.assertTrue(archive["legacy_zipcrypto_acknowledged_at"])
+            self.assertEqual("f" * 64, evidence[0]["policy_hash"])
+            assert_archive_encryption_allowed(
+                self.session,
+                self.campaign,
+                stamped,
+                principal=_principal({LEGACY_ZIPCRYPTO_SCOPE}),
+            )
+
+
+def _raw_archive(method: str, *, stamped: bool = False) -> dict:
+    archive = {
+        "id": "archive-1",
+        "method": method,
+        "password_enabled": True,
+        "password_delivery_channel": "separate_mail",
+        "legacy_zipcrypto_acknowledged": method == "zip_standard",
+        "legacy_zipcrypto_reason": "Windows recipient compatibility required"
+        if method == "zip_standard"
+        else None,
+    }
+    if stamped:
+        archive.update(
+            {
+                "legacy_zipcrypto_acknowledged_by": "user-1",
+                "legacy_zipcrypto_acknowledged_at": "2026-08-20T10:00:00+00:00",
+            }
+        )
+    return {"attachments": {"zip": {"enabled": True, "archives": [archive]}}}
+
+
+def _principal(scopes: set[str]) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="user-1",
+            tenant_id="tenant-1",
+            scopes=frozenset(scopes),
+        ),
+        account=SimpleNamespace(id="account-1"),
+        user=SimpleNamespace(id="user-1"),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_execution_snapshot_integrity.py

@@ -64,6 +64,9 @@ class _Session:
     def query(self, _model):
         return _Query(self.jobs)
 
+    def get(self, _model, _identifier):
+        return SimpleNamespace(id="campaign-1")
+
 
 def _snapshotted_version(job: SimpleNamespace):
     version = SimpleNamespace(
@@ -87,9 +90,15 @@ def _snapshotted_version(job: SimpleNamespace):
 
 
 def _ensure(session: _Session, version) -> None:
-    with patch(
-        "govoplan_campaign.backend.sending.execution.files_integration",
-        return_value=SimpleNamespace(available=False),
+    with (
+        patch(
+            "govoplan_campaign.backend.sending.execution.files_integration",
+            return_value=SimpleNamespace(available=False),
+        ),
+        patch(
+            "govoplan_campaign.backend.sending.execution.assert_archive_encryption_allowed",
+            return_value=SimpleNamespace(policy_hash="archive-policy"),
+        ),
     ):
         ensure_execution_snapshot(session, version)  # type: ignore[arg-type]
 
@@ -125,12 +134,19 @@ def test_effect_check_verifies_only_the_claimed_job_in_constant_time() -> None:
     job = _job()
     version = _snapshotted_version(job)
     session = SimpleNamespace(
-        query=lambda *_args: pytest.fail("per-effect validation must not rescan every campaign job")
+        query=lambda *_args: pytest.fail("per-effect validation must not rescan every campaign job"),
+        get=lambda *_args: SimpleNamespace(id="campaign-1"),
     )
 
-    with patch(
-        "govoplan_campaign.backend.sending.execution.files_integration",
-        return_value=SimpleNamespace(available=False),
+    with (
+        patch(
+            "govoplan_campaign.backend.sending.execution.files_integration",
+            return_value=SimpleNamespace(available=False),
+        ),
+        patch(
+            "govoplan_campaign.backend.sending.execution.assert_archive_encryption_allowed",
+            return_value=SimpleNamespace(policy_hash="archive-policy"),
+        ),
     ):
         ensure_execution_snapshot(
             session,  # type: ignore[arg-type]

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_route_registration.py

@@ -42,7 +42,7 @@ def test_campaign_router_composes_every_workflow_operation_once() -> None:
     actual = _operation_keys(router)
 
     assert actual == expected
-    assert len(actual) == 80
+    assert len(actual) == 81
     assert not [operation for operation, count in Counter(actual).items() if count > 1]
 
 

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")

tests/test_zip_service.py

@@ -10,7 +10,10 @@ try:
 except ImportError:  # pragma: no cover
     pyzipper = None
 
-from govoplan_campaign.backend.services.zip_service import create_zip_archive
+from govoplan_campaign.backend.services.zip_service import (
+    create_zip_archive,
+    zip_archive_evidence,
+)
 
 
 class ZipServiceTests(unittest.TestCase):
@@ -28,6 +31,20 @@ class ZipServiceTests(unittest.TestCase):
                 self.assertEqual(info.compress_type, zipfile.ZIP_DEFLATED)
                 self.assertTrue(info.flag_bits & 0x1)
                 self.assertEqual(archive.read("message.txt", pwd=b"secret"), b"Hello Windows ZIP")
+                with self.assertRaises(RuntimeError):
+                    archive.read("message.txt", pwd=b"wrong-password")
+
+            evidence = zip_archive_evidence(
+                output,
+                [(source, "message.txt")],
+                password_protected=True,
+                method="zip_standard",
+            )
+            self.assertEqual("Legacy ZipCrypto", evidence["format"])
+            self.assertEqual("govoplan-campaign.zipcrypto", evidence["implementation"])
+            self.assertNotIn("password", evidence)
+            self.assertEqual(64, len(str(evidence["archive_sha256"])))
+            self.assertEqual(64, len(str(evidence["members"][0]["sha256"])))
 
     @unittest.skipIf(pyzipper is None, "pyzipper is not installed")
     def test_aes_password_zip_keeps_aes_encryption(self) -> None:
@@ -63,6 +80,18 @@ class ZipServiceTests(unittest.TestCase):
                 self.assertFalse(info.flag_bits & 0x1)
                 self.assertEqual(archive.read("message.txt"), b"Plain ZIP")
 
+    def test_unknown_password_method_fails_without_downgrade_or_output(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            source = root / "source.txt"
+            source.write_text("No downgrade", encoding="utf-8")
+            output = root / "unknown.zip"
+
+            with self.assertRaisesRegex(ValueError, "Unsupported"):
+                create_zip_archive(output, [source], "secret", "unknown")
+
+            self.assertFalse(output.exists())
+
 
 if __name__ == "__main__":
     unittest.main()

webui/src/api/campaigns.ts

@@ -38,6 +38,24 @@ export type CampaignShare = {
 export type CampaignShareTarget = {id: string;name: string;secondary?: string | null;};
 export type CampaignShareTargets = {users: CampaignShareTarget[];groups: CampaignShareTarget[];};
 
+export type CampaignArchiveEncryptionPolicy = {
+  available: boolean;
+  allowed_password_encryption_methods: Array<"aes" | "zip_standard">;
+  allowed_password_delivery_channels: Array<"separate_mail" | "sms" | "letter" | "phone" | "in_person">;
+  policy_hash: string;
+  source_path: Array<{
+    scope_type: string;
+    scope_id?: string | null;
+    path: string;
+    label: string;
+    applied_fields: string[];
+    policy: Record<string, unknown>;
+  }>;
+  reason: string;
+  diagnostics: Array<Record<string, unknown>>;
+  legacy_label: string;
+};
+
 export type CampaignUpdatePayload = {
   external_id?: string | null;
   name?: string | null;
@@ -1141,6 +1159,16 @@ export async function getCampaign(settings: ApiSettings, campaignId: string): Pr
   return apiFetch<CampaignListItem>(settings, `/api/v1/campaigns/${campaignId}`);
 }
 
+export async function getCampaignArchiveEncryptionPolicy(
+  settings: ApiSettings,
+  campaignId: string
+): Promise<CampaignArchiveEncryptionPolicy> {
+  return apiFetch<CampaignArchiveEncryptionPolicy>(
+    settings,
+    `/api/v1/campaigns/${encodeURIComponent(campaignId)}/archive-encryption-policy`
+  );
+}
+
 export async function updateCampaignMetadata(
 settings: ApiSettings,
 campaignId: string,

webui/src/features/campaigns/AttachmentsDataPage.tsx

@@ -2,7 +2,11 @@ import { MetricGrid } from "@govoplan/core-webui";
 import { useEffect, useMemo, useState } from "react";
 import { Pencil } from "lucide-react";
 import { useGuardedNavigate, usePlatformModuleInstalled, usePlatformUiCapability, type FilesFileExplorerUiCapability, type FilesFileSpace } from "@govoplan/core-webui";
-import type { ApiSettings } from "../../types";
+import type { ApiSettings, AuthInfo } from "../../types";
+import {
+  getCampaignArchiveEncryptionPolicy,
+  type CampaignArchiveEncryptionPolicy
+} from "../../api/campaigns";
 import { Button } from "@govoplan/core-webui";
 import { Card } from "@govoplan/core-webui";
 import { PageActionBar, PageLayout } from "@govoplan/core-webui";
@@ -22,14 +26,25 @@ import { updateNested } from "./utils/draftEditor";
 import { AttachmentRulesDataGrid } from "./components/AttachmentRulesOverlay";
 import TemplateExpressionEditorDialog from "./components/TemplateExpressionEditorDialog";
 import { countIndividualAttachmentRules, countIndividualAttachmentRulesForBasePath, createAttachmentBasePath, ensureAttachmentBasePaths, normalizeAttachmentBasePaths, normalizeAttachmentRules, normalizeAttachmentZipCollection, createAttachmentZipArchive, parseManagedAttachmentSource, removeIndividualAttachmentRulesForBasePath, summarizeAttachmentRules, type AttachmentBasePath, type AttachmentZipArchive, type AttachmentZipCollection } from "./utils/attachments";
-import { insertAfter, moveArrayItem, i18nMessage } from "@govoplan/core-webui";
+import { hasScope, insertAfter, moveArrayItem, i18nMessage } from "@govoplan/core-webui";
 import { getDraftFields, humanizeFieldName } from "./utils/fieldDefinitions";
 import { buildTemplatePreviewContext, recipientAddressTemplateFieldOptions } from "./utils/templatePlaceholders";
 
 type PathChooserState = {index: number;};
 type IndividualDisableState = {index: number;usageCount: number;};
 
-export default function AttachmentsDataPage({ settings, campaignId }: {settings: ApiSettings;campaignId: string;}) {
+const UNAVAILABLE_ARCHIVE_POLICY: CampaignArchiveEncryptionPolicy = {
+  available: false,
+  allowed_password_encryption_methods: ["aes"],
+  allowed_password_delivery_channels: ["separate_mail", "sms", "letter", "phone", "in_person"],
+  policy_hash: "",
+  source_path: [],
+  reason: "Archive-encryption policy is loading. Legacy ZipCrypto remains blocked.",
+  diagnostics: [],
+  legacy_label: "Legacy ZipCrypto — Windows-compatible, weak encryption"
+};
+
+export default function AttachmentsDataPage({ settings, auth, campaignId }: {settings: ApiSettings;auth: AuthInfo;campaignId: string;}) {
   const navigate = useGuardedNavigate();
   const filesModuleInstalled = usePlatformModuleInstalled("files");
   const filesFileExplorer = usePlatformUiCapability<FilesFileExplorerUiCapability>("files.fileExplorer");
@@ -41,6 +56,7 @@ export default function AttachmentsDataPage({ settings, campaignId }: {settings:
   const [fileSpaces, setFileSpaces] = useState<FilesFileSpace[]>([]);
   const [individualDisable, setIndividualDisable] = useState<IndividualDisableState | null>(null);
   const [zipNameEditorIndex, setZipNameEditorIndex] = useState<number | null>(null);
+  const [archivePolicy, setArchivePolicy] = useState<CampaignArchiveEncryptionPolicy>(UNAVAILABLE_ARCHIVE_POLICY);
   const version = data.currentVersion;
   const locked = isAuditLockedVersion(version, data.campaign?.current_version_id);
   const { draft, setDraft, displayDraft, dirty, saveState, localError, patch, markDirty, discardDraft, saveDraft } = useCampaignDraftEditor({
@@ -70,7 +86,15 @@ export default function AttachmentsDataPage({ settings, campaignId }: {settings:
     () => zipConfig.enabled ? validateZipArchiveNames(zipConfig.archives) : EMPTY_ZIP_ARCHIVE_NAME_VALIDATION,
     [zipConfig.archives, zipConfig.enabled]
   );
-  const canSave = dirty && !locked && Boolean(draft) && !zipArchiveNameValidation.message;
+  const canUseLegacyZipCrypto = hasScope(auth, "campaigns:archive:use_legacy_zipcrypto");
+  const legacyZipCryptoAllowed = archivePolicy.available && archivePolicy.allowed_password_encryption_methods.includes("zip_standard");
+  const legacyConfigurationInvalid = zipConfig.archives.some((archive) =>
+    archive.method === "zip_standard" && (
+      !legacyZipCryptoAllowed || !canUseLegacyZipCrypto ||
+      !archive.legacy_zipcrypto_acknowledged || archive.legacy_zipcrypto_reason.trim().length < 10
+    )
+  );
+  const canSave = dirty && !locked && Boolean(draft) && !zipArchiveNameValidation.message && !legacyConfigurationInvalid;
   const globalSummary = useMemo(() => summarizeAttachmentRules(globalRules), [globalRules]);
   const individualRulesCount = useMemo(() => countIndividualAttachmentRules(displayDraft.entries), [displayDraft.entries]);
   const attachmentPreviewEntry = useMemo(
@@ -94,6 +118,22 @@ export default function AttachmentsDataPage({ settings, campaignId }: {settings:
     return () => {cancelled = true;};
   }, [listManagedFileSpaces, settings.apiBaseUrl, settings.apiKey, settings.accessToken]);
 
+  useEffect(() => {
+    let cancelled = false;
+    setArchivePolicy(UNAVAILABLE_ARCHIVE_POLICY);
+    void getCampaignArchiveEncryptionPolicy(settings, campaignId)
+      .then((policy) => { if (!cancelled) setArchivePolicy(policy); })
+      .catch((cause) => {
+        if (!cancelled) {
+          setArchivePolicy({
+            ...UNAVAILABLE_ARCHIVE_POLICY,
+            reason: cause instanceof Error ? cause.message : String(cause)
+          });
+        }
+      });
+    return () => { cancelled = true; };
+  }, [campaignId, settings.accessToken, settings.apiBaseUrl, settings.apiKey]);
+
   function patchBasePaths(paths: AttachmentBasePath[]) {
     if (locked) return;
     const normalized = ensureAttachmentBasePaths(paths);
@@ -385,6 +425,11 @@ export default function AttachmentsDataPage({ settings, campaignId }: {settings:
           </Card>
 
           <Card title="i18n:govoplan-campaign.zip_attachments.6b58ed68" collapsible>
+            <DismissibleAlert tone={legacyZipCryptoAllowed ? "warning" : "info"} dismissible={false} compact>
+              <strong>{archivePolicy.legacy_label}</strong>: {archivePolicy.reason}
+              {archivePolicy.source_path.length > 0 && <> Source: {archivePolicy.source_path.map((step) => step.label).join(" → ")}.</>}
+              {!canUseLegacyZipCrypto && <> Your account does not have the dedicated legacy-encryption permission.</>}
+            </DismissibleAlert>
             <div className="attachment-zip-master-toggle">
               <ToggleSwitch
                 label="i18n:govoplan-campaign.enable_zip_attachments.6077075b"
@@ -403,6 +448,9 @@ export default function AttachmentsDataPage({ settings, campaignId }: {settings:
                   invalidNameIndexes: zipArchiveNameValidation.invalidIndexes,
                   onEditName: setZipNameEditorIndex,
                   passwordFields,
+                  legacyAllowed: legacyZipCryptoAllowed,
+                  canUseLegacy: canUseLegacyZipCrypto,
+                  allowedDeliveryChannels: archivePolicy.allowed_password_delivery_channels,
                   patchArchive: patchZipArchive,
                   setStandard: setStandardZipArchive,
                   addArchive: addZipArchive,
@@ -517,6 +565,9 @@ type ZipArchiveColumnContext = {
   invalidNameIndexes: ReadonlySet<number>;
   onEditName: (index: number) => void;
   passwordFields: ReturnType<typeof getDraftFields>;
+  legacyAllowed: boolean;
+  canUseLegacy: boolean;
+  allowedDeliveryChannels: CampaignArchiveEncryptionPolicy["allowed_password_delivery_channels"];
   patchArchive: (index: number, patch: Partial<AttachmentZipArchive>) => void;
   setStandard: (index: number) => void;
   addArchive: (afterIndex?: number) => void;
@@ -524,7 +575,7 @@ type ZipArchiveColumnContext = {
   removeArchive: (index: number) => void;
 };
 
-function zipArchiveColumns({ disabled, archives, invalidNameIndexes, onEditName, passwordFields, patchArchive, setStandard, addArchive, moveArchive, removeArchive }: ZipArchiveColumnContext): DataGridColumn<AttachmentZipArchive>[] {
+function zipArchiveColumns({ disabled, archives, invalidNameIndexes, onEditName, passwordFields, legacyAllowed, canUseLegacy, allowedDeliveryChannels, patchArchive, setStandard, addArchive, moveArchive, removeArchive }: ZipArchiveColumnContext): DataGridColumn<AttachmentZipArchive>[] {
   return [
   {
     id: "name", header: "i18n:govoplan-campaign.archive_name.6310f9e1", width: "minmax(360px, 1fr)", maxWidth: 640, resizable: true, sortable: true, filterable: true, sticky: "start",
@@ -557,18 +608,65 @@ function zipArchiveColumns({ disabled, archives, invalidNameIndexes, onEditName,
     value: (archive) => archive.password_enabled ? "protected" : "none"
   },
   {
-    id: "method", header: "ZIP mode", width: 280, sortable: true, filterable: true,
-    columnType: "from-list", list: { options: [{ value: "aes", label: "AES" }, { value: "zip_standard", label: "Win-compatible" }] },
+    id: "method", header: "Encryption", width: 360, sortable: true, filterable: true,
+    columnType: "from-list", list: { options: [{ value: "aes", label: "AES (strong, default)" }, { value: "zip_standard", label: "Legacy ZipCrypto — Windows-compatible, weak encryption" }] },
     render: (archive, index) =>
-    <ToggleSwitch
-      label="Win-compatible"
-      checked={archive.method === "zip_standard"}
+    <select
+      value={archive.method}
       disabled={disabled}
-      help="Win-compatible ZIP uses the legacy ZipCrypto format so password-protected archives can be opened with Windows Explorer. Use AES when recipients can use 7-Zip, NanaZip, WinRAR, or another AES-capable ZIP tool."
-      onChange={(checked) => patchArchive(index, { method: checked ? "zip_standard" : "aes" })} />,
+      aria-label="Archive password encryption"
+      onChange={(event) => {
+        const method = event.target.value === "zip_standard" ? "zip_standard" : "aes";
+        patchArchive(index, method === "zip_standard" ? {
+          method,
+          legacy_zipcrypto_acknowledged: false,
+          legacy_zipcrypto_reason: ""
+        } : {
+          method,
+          legacy_zipcrypto_acknowledged: false,
+          legacy_zipcrypto_reason: ""
+        });
+      }}>
+        <option value="aes">AES (strong, default)</option>
+        <option value="zip_standard" disabled={!legacyAllowed || !canUseLegacy}>Legacy ZipCrypto — Windows-compatible, weak encryption</option>
+      </select>,
 
     value: (archive) => archive.method
   },
+  {
+    id: "password_delivery_channel", header: "Password delivery", width: 230, sortable: true, filterable: true,
+    render: (archive, index) =>
+      <select
+        value={archive.password_delivery_channel}
+        disabled={disabled || !archive.password_enabled}
+        aria-label="Separate password-delivery channel"
+        onChange={(event) => patchArchive(index, { password_delivery_channel: event.target.value as AttachmentZipArchive["password_delivery_channel"] })}>
+        {(["separate_mail", "sms", "letter", "phone", "in_person"] as const).map((channel) =>
+          <option key={channel} value={channel} disabled={!allowedDeliveryChannels.includes(channel)}>{passwordDeliveryChannelLabel(channel)}</option>
+        )}
+      </select>,
+    value: (archive) => archive.password_delivery_channel
+  },
+  {
+    id: "legacy_acknowledgement", header: "Legacy acknowledgement", width: 380,
+    render: (archive, index) => archive.method === "zip_standard" ?
+      <div className="campaign-legacy-zipcrypto-acknowledgement">
+        <ToggleSwitch
+          label="I acknowledge that ZipCrypto encryption is weak"
+          checked={archive.legacy_zipcrypto_acknowledged}
+          disabled={disabled || !legacyAllowed || !canUseLegacy}
+          onChange={(checked) => patchArchive(index, { legacy_zipcrypto_acknowledged: checked })} />
+        <input
+          value={archive.legacy_zipcrypto_reason}
+          disabled={disabled || !archive.legacy_zipcrypto_acknowledged}
+          minLength={10}
+          maxLength={1000}
+          placeholder="Operational reason (at least 10 characters)"
+          aria-label="Reason for weak legacy encryption"
+          onChange={(event) => patchArchive(index, { legacy_zipcrypto_reason: event.target.value })} />
+      </div> : <span>Not required for AES</span>,
+    value: (archive) => archive.legacy_zipcrypto_reason
+  },
   {
     id: "password_field", header: "i18n:govoplan-campaign.password_field.a1fc8a1c", width: 230, sortable: true, filterable: true,
     columnType: "from-list", list: { options: [{ value: "", label: "i18n:govoplan-campaign.no_field.1fe00ed4" }, ...passwordFields.map((field) => ({ value: field.name, label: field.label || field.name }))] },
@@ -685,6 +783,16 @@ function uniqueStrings(values: string[]): string[] {
   return [...new Set(values.map((value) => value.trim()).filter(Boolean))];
 }
 
+function passwordDeliveryChannelLabel(channel: AttachmentZipArchive["password_delivery_channel"]): string {
+  return {
+    separate_mail: "Separate email (never this campaign message)",
+    sms: "SMS",
+    letter: "Letter",
+    phone: "Telephone",
+    in_person: "In person"
+  }[channel];
+}
+
 type AttachmentSourceColumnContext = {
   locked: boolean;
   basePaths: AttachmentBasePath[];

webui/src/features/campaigns/CampaignWorkspace.tsx

@@ -100,7 +100,7 @@ function CampaignWorkspaceInner({ settings, auth }: { settings: ApiSettings; aut
           <Route path="recipients" element={<RecipientDataPage settings={settings} campaignId={campaignId || ""} />} />
           <Route path="recipient-data" element={<Navigate to="../recipients" replace />} />
           <Route path="template" element={<TemplateDataPage settings={settings} campaignId={campaignId || ""} />} />
-          <Route path="files" element={<AttachmentsDataPage settings={settings} campaignId={campaignId || ""} />} />
+          <Route path="files" element={<AttachmentsDataPage settings={settings} auth={auth} campaignId={campaignId || ""} />} />
           <Route path="attachments" element={<Navigate to="../files" replace />} />
           <Route path="mail-settings" element={<MailSettingsPage settings={settings} campaignId={campaignId || ""} view="settings" />} />
           <Route path="mail-policy" element={<MailSettingsPage settings={settings} campaignId={campaignId || ""} view="policy" />} />

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`}

webui/src/features/campaigns/utils/attachments.ts

@@ -13,6 +13,11 @@ export type AttachmentZipArchive = {
   password_field: string;
   password_scope: AttachmentZipPasswordScope;
   method: "aes" | "zip_standard";
+  password_delivery_channel: "separate_mail" | "sms" | "letter" | "phone" | "in_person";
+  legacy_zipcrypto_acknowledged: boolean;
+  legacy_zipcrypto_reason: string;
+  legacy_zipcrypto_acknowledged_by?: string;
+  legacy_zipcrypto_acknowledged_at?: string;
   // Read-only compatibility values retained when normalizing older campaigns.
   password_mode?: "none" | "direct" | "field" | "template";
   password?: string;
@@ -32,7 +37,10 @@ export function createAttachmentZipArchive(name = "attachments.zip", standard =
     password_enabled: false,
     password_field: "",
     password_scope: "local",
-    method: "aes"
+    method: "aes",
+    password_delivery_channel: "separate_mail",
+    legacy_zipcrypto_acknowledged: false,
+    legacy_zipcrypto_reason: ""
   };
 }
 
@@ -52,6 +60,11 @@ export function normalizeAttachmentZipCollection(value: unknown): AttachmentZipC
           password_field: getText(archive, "password_field"),
           password_scope: getText(archive, "password_scope") === "global" ? "global" : "local",
           method: getText(archive, "method", "aes") === "zip_standard" ? "zip_standard" : "aes",
+          password_delivery_channel: normalizePasswordDeliveryChannel(getText(archive, "password_delivery_channel", "separate_mail")),
+          legacy_zipcrypto_acknowledged: getBool(archive, "legacy_zipcrypto_acknowledged"),
+          legacy_zipcrypto_reason: getText(archive, "legacy_zipcrypto_reason"),
+          ...(getText(archive, "legacy_zipcrypto_acknowledged_by") ? { legacy_zipcrypto_acknowledged_by: getText(archive, "legacy_zipcrypto_acknowledged_by") } : {}),
+          ...(getText(archive, "legacy_zipcrypto_acknowledged_at") ? { legacy_zipcrypto_acknowledged_at: getText(archive, "legacy_zipcrypto_acknowledged_at") } : {}),
           ...(legacyMode ? { password_mode: legacyMode } : {}),
           ...(getText(archive, "password") ? { password: getText(archive, "password") } : {}),
           ...(getText(archive, "password_template") ? { password_template: getText(archive, "password_template") } : {})
@@ -76,6 +89,9 @@ export function normalizeAttachmentZipCollection(value: unknown): AttachmentZipC
       password_field: getText(zip, "password_field"),
       password_scope: "local",
       method: getText(zip, "method", "aes") === "zip_standard" ? "zip_standard" : "aes",
+      password_delivery_channel: normalizePasswordDeliveryChannel(getText(zip, "password_delivery_channel", "separate_mail")),
+      legacy_zipcrypto_acknowledged: getBool(zip, "legacy_zipcrypto_acknowledged"),
+      legacy_zipcrypto_reason: getText(zip, "legacy_zipcrypto_reason"),
       ...(legacyMode ? { password_mode: legacyMode } : {}),
       ...(getText(zip, "password") ? { password: getText(zip, "password") } : {}),
       ...(getText(zip, "password_template") ? { password_template: getText(zip, "password_template") } : {})
@@ -83,6 +99,13 @@ export function normalizeAttachmentZipCollection(value: unknown): AttachmentZipC
   };
 }
 
+function normalizePasswordDeliveryChannel(value: string): AttachmentZipArchive["password_delivery_channel"] {
+  if (["sms", "letter", "phone", "in_person"].includes(value)) {
+    return value as AttachmentZipArchive["password_delivery_channel"];
+  }
+  return "separate_mail";
+}
+
 export function attachmentRuleZipSelection(rule: AttachmentRule): string {
   const zip = asRecord(rule.zip);
   const archiveId = getText(zip, "archive_id");

webui/src/styles/campaign-workspace.css

@@ -2025,6 +2025,16 @@
   padding-bottom: 7px;
 }
 
+.campaign-legacy-zipcrypto-acknowledgement {
+  display: grid;
+  gap: var(--space-2);
+  min-width: 20rem;
+}
+
+.campaign-legacy-zipcrypto-acknowledgement input {
+  width: 100%;
+}
+
 .attachment-zip-name-button {
   width: 100%;
   min-height: 36px;