feat(mail): return exact messages from Quick Access

README.md

@@ -19,6 +19,15 @@ This repository owns:
 
 Core owns auth, tenants, RBAC evaluation, database/session primitives, secret helpers, CSRF/API helpers, and shell layout.
 
+Mail publishes `privacy.dsar.mail` for Core's governed data-subject-request
+workflow. It isolates matching mailbox header parties and returns bounded index,
+personal-profile, delivery, reconciliation, and bounce metadata. SMTP/IMAP
+configuration and credentials, encrypted messages and envelopes, folder/UID
+locators, worker and idempotency state, diagnostics, and opaque evidence are
+excluded. Delivery and bounce outcomes remain retained evidence; mailbox and
+profile changes require coordinated Mail and external-provider review, so the
+provider does not perform direct erasure.
+
 ## Profile and credential ownership
 
 Mail profiles are separate governed definitions. Mail owns their SMTP/IMAP
@@ -41,6 +50,18 @@ and requires explicit evidence-backed reconciliation before any deliberate
 resend. Business readers receive only counts and sanitized state; recipient
 refusal details require `mail:delivery:diagnostic`.
 
+Synchronous Campaign batches now preflight DNS, egress, connectivity, TLS, and
+authentication before the first message, then reuse the authorized SMTP
+connection for the bounded batch. A health check precedes reuse; a stale
+connection is reopened before the next message, while a connection loss after
+DATA starts remains outcome-unknown and is never replayed automatically.
+Systemic authentication, sender, and connectivity failures pause remaining
+Campaign jobs instead of producing one failure per recipient. Deployment
+operators can disable reuse or bound connection lifetime and reconnects with
+`GOVOPLAN_SMTP_BATCH_REUSE`, `GOVOPLAN_SMTP_BATCH_MAX_MESSAGES`,
+`GOVOPLAN_SMTP_BATCH_RECONNECT_ATTEMPTS`, and
+`GOVOPLAN_SMTP_BATCH_HEALTH_CHECK`.
+
 SMTP effects decrypt only SMTP credentials; Sent-folder effects decrypt only
 IMAP credentials. A connection loss after an effect starts is surfaced as an
 unknown outcome. Campaign does not automatically retry an unknown IMAP append,

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-mail"
-version = "0.1.18"
+version = "0.1.19"
 description = "GovOPlaN mail module with backend and WebUI integration."
 readme = "README.md"
 requires-python = ">=3.12"

src/govoplan_mail/backend/capabilities.py

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

src/govoplan_mail/backend/dsar_provider.py

@@ -0,0 +1,554 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from datetime import datetime, timezone
+from email.utils import getaddresses
+
+from sqlalchemy import func, or_
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_mail.backend.db.models import (
+    MailBounceObservation,
+    MailDeliveryAttempt,
+    MailDeliveryCommand,
+    MailDeliveryReconciliation,
+    MailMailboxMessageIndex,
+    MailServerProfile,
+)
+
+
+MAIL_DSAR_CAPABILITY = dsar_capability_name("mail")
+_MAX_RECORDS = 5_000
+
+
+class MailDsarProvider:
+    provider_id = "mail"
+    module_id = "mail"
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]:
+        db = _session(session)
+        email = _subject_email(subject)
+        membership_ids = _membership_ids(subject)
+        references = _mail_references(subject)
+        if email is None and not membership_ids and not references:
+            return ()
+
+        records: list[DsarRecordRef] = []
+
+        def append(record: DsarRecordRef) -> None:
+            if len(records) >= _MAX_RECORDS:
+                raise ValueError(
+                    "Mail DSAR match limit exceeded; narrow the subject selectors."
+                )
+            records.append(record)
+
+        profiles = _matching_profiles(
+            db,
+            tenant_id=tenant_id,
+            membership_ids=membership_ids,
+            profile_id=references.get("profile"),
+        )
+        for profile in profiles:
+            append(
+                _record(
+                    "mail_server_profile",
+                    profile.id,
+                    "mail_profile",
+                    profile.name,
+                    {
+                        "match_fields": _profile_matching_fields(
+                            profile, membership_ids
+                        ),
+                        "name": profile.name,
+                        "slug": profile.slug,
+                        "description": profile.description,
+                        "scope_type": profile.scope_type,
+                        "is_active": profile.is_active,
+                        "inherit_to_lower_scopes": profile.inherit_to_lower_scopes,
+                    },
+                    observed_at=profile.updated_at,
+                    source_path="/settings?section=mail-profiles",
+                )
+            )
+
+        messages = _matching_messages(
+            db,
+            tenant_id=tenant_id,
+            email=email,
+            message_id=references.get("message_index"),
+        )
+        for message in messages:
+            matching_headers = _matching_headers(message, email)
+            append(
+                _record(
+                    "mailbox_message_index",
+                    message.id,
+                    "mailbox_message",
+                    message.subject or "Mailbox message",
+                    {
+                        "match_fields": list(matching_headers),
+                        "subject": _bounded_text(message.subject),
+                        "matching_headers": matching_headers,
+                        "date": message.date,
+                        "flags": tuple(
+                            str(flag)[:100] for flag in (message.flags or ())[:32]
+                        ),
+                        "size_bytes": message.size_bytes,
+                        "body_preview": _bounded_text(message.body_preview),
+                        "attachment_count": message.attachment_count,
+                        "indexed_at": _iso(message.indexed_at),
+                    },
+                    observed_at=message.updated_at,
+                    source_path="/mail",
+                )
+            )
+
+        bounces = _matching_bounces(
+            db,
+            tenant_id=tenant_id,
+            email=email,
+            bounce_id=references.get("bounce"),
+        )
+        bounce_command_ids = {row.command_id for row in bounces if row.command_id}
+        commands = _matching_commands(
+            db,
+            tenant_id=tenant_id,
+            membership_ids=membership_ids,
+            command_id=references.get("command"),
+            related_command_ids=bounce_command_ids,
+        )
+        command_ids = {row.id for row in commands}
+        for command in commands:
+            append(
+                _record(
+                    "mail_delivery_command",
+                    command.id,
+                    "mail_delivery_evidence",
+                    f"Mail {command.command_type} command",
+                    {
+                        "match_fields": (
+                            ["created_by_user_id"]
+                            if command.created_by_user_id in membership_ids
+                            else (
+                                ["reference"]
+                                if command.id == references.get("command")
+                                else ["bounce"]
+                            )
+                        ),
+                        "command_type": command.command_type,
+                        "source_module": command.source_module,
+                        "source_resource_type": command.source_resource_type,
+                        "message_sha256": command.message_sha256,
+                        "rfc_message_id": command.rfc_message_id,
+                        "message_size_bytes": command.message_size_bytes,
+                        "recipient_count": command.recipient_count,
+                        "status": command.status,
+                        "attempt_count": command.attempt_count,
+                        "effect_started_at": _iso(command.effect_started_at),
+                        "completed_at": _iso(command.completed_at),
+                        "accepted_count": command.accepted_count,
+                        "refused_count": command.refused_count,
+                        "failure_code": command.failure_code,
+                        "payload_purged_at": _iso(command.payload_purged_at),
+                    },
+                    observed_at=command.updated_at,
+                    immutable=True,
+                    retention_reason="Mail delivery commands are immutable transport, retry, and outcome evidence; encrypted payload retention is governed separately.",
+                )
+            )
+
+        for attempt in _command_attempts(db, command_ids):
+            append(
+                _record(
+                    "mail_delivery_attempt",
+                    attempt.id,
+                    "mail_delivery_evidence",
+                    f"Mail delivery attempt {attempt.attempt_number}",
+                    {
+                        "command_id": attempt.command_id,
+                        "attempt_number": attempt.attempt_number,
+                        "status": attempt.status,
+                        "started_at": _iso(attempt.started_at),
+                        "effect_started_at": _iso(attempt.effect_started_at),
+                        "completed_at": _iso(attempt.completed_at),
+                        "accepted_count": attempt.accepted_count,
+                        "refused_count": attempt.refused_count,
+                        "outcome_code": attempt.outcome_code,
+                    },
+                    observed_at=attempt.updated_at,
+                    immutable=True,
+                    retention_reason="Per-attempt Mail outcome state is immutable delivery and recovery evidence.",
+                )
+            )
+
+        for reconciliation in _command_reconciliations(db, command_ids):
+            append(
+                _record(
+                    "mail_delivery_reconciliation",
+                    reconciliation.id,
+                    "mail_delivery_evidence",
+                    "Mail delivery reconciliation",
+                    {
+                        "command_id": reconciliation.command_id,
+                        "decision": reconciliation.decision,
+                    },
+                    observed_at=reconciliation.updated_at,
+                    immutable=True,
+                    retention_reason="Mail reconciliation decisions are immutable authorization and recovery evidence.",
+                )
+            )
+
+        for bounce in bounces:
+            append(
+                _record(
+                    "mail_bounce_observation",
+                    bounce.id,
+                    "mail_bounce_evidence",
+                    "Mail bounce observation",
+                    {
+                        "match_fields": (
+                            ["recipient"]
+                            if email and _normalized_email(bounce.recipient) == email
+                            else ["reference"]
+                        ),
+                        "command_id": bounce.command_id
+                        if bounce.command_id in command_ids
+                        else None,
+                        "recipient": email
+                        if email and _normalized_email(bounce.recipient) == email
+                        else None,
+                        "action": bounce.action,
+                        "status_code": bounce.status_code,
+                        "permanent": bounce.permanent,
+                        "observed_at": _iso(bounce.observed_at),
+                        "matched": bounce.matched,
+                    },
+                    observed_at=bounce.updated_at,
+                    immutable=True,
+                    retention_reason="Bounce observations are immutable delivery-status and suppression evidence.",
+                    source_path="/mail/bounces",
+                )
+            )
+        return tuple(records)
+
+    def plan_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        records: Sequence[DsarRecordRef],
+    ) -> Sequence[DsarErasureActionRef]:
+        del session, tenant_id, subject
+        actions = []
+        for record in records:
+            if (
+                record.provider_id != self.provider_id
+                or record.module_id != self.module_id
+            ):
+                raise ValueError("Mail DSAR received a foreign provider record.")
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=f"mail:{'retain' if record.immutable_evidence else 'review'}:{record.resource_type}:{record.resource_id}",
+                    provider_id="mail",
+                    module_id="mail",
+                    kind="retain" if record.immutable_evidence else "manual_review",
+                    resource_type=record.resource_type,
+                    resource_id=record.resource_id,
+                    title=f"{'Retain' if record.immutable_evidence else 'Review'} {record.title}",
+                    rationale=record.retention_reason
+                    or "Mailbox indexes and personal profiles must be reviewed through Mail and the authoritative external mailbox lifecycle.",
+                    executable=False,
+                )
+            )
+        return tuple(actions)
+
+    def execute_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        actions: Sequence[DsarErasureActionRef],
+        request_id: str,
+    ) -> Sequence[DsarExecutionResultRef]:
+        del session, tenant_id, subject, request_id
+        return tuple(
+            DsarExecutionResultRef(
+                action_id=action.action_id,
+                status="blocked",
+                summary="Mail erasure requires an authorized Mail/external-mailbox lifecycle action; the DSAR provider does not mutate it directly.",
+            )
+            for action in actions
+        )
+
+
+def _matching_profiles(
+    session: Session,
+    *,
+    tenant_id: str,
+    membership_ids: set[str],
+    profile_id: str | None,
+) -> list[MailServerProfile]:
+    conditions = []
+    if profile_id:
+        conditions.append(MailServerProfile.id == profile_id)
+    if membership_ids:
+        conditions.extend(
+            (
+                MailServerProfile.created_by_user_id.in_(membership_ids),
+                MailServerProfile.updated_by_user_id.in_(membership_ids),
+                (MailServerProfile.scope_type == "user")
+                & MailServerProfile.scope_id.in_(membership_ids),
+            )
+        )
+    if not conditions:
+        return []
+    return _bounded_rows(
+        session.query(MailServerProfile)
+        .filter(MailServerProfile.tenant_id == tenant_id, or_(*conditions))
+        .order_by(MailServerProfile.id)
+    )
+
+
+def _profile_matching_fields(
+    row: MailServerProfile, membership_ids: set[str]
+) -> list[str]:
+    fields = []
+    if row.scope_type == "user" and row.scope_id in membership_ids:
+        fields.append("scope_id")
+    for field in ("created_by_user_id", "updated_by_user_id"):
+        if getattr(row, field) in membership_ids:
+            fields.append(field)
+    return fields
+
+
+def _matching_messages(
+    session: Session, *, tenant_id: str, email: str | None, message_id: str | None
+) -> list[MailMailboxMessageIndex]:
+    conditions = []
+    if message_id:
+        conditions.append(MailMailboxMessageIndex.id == message_id)
+    if email:
+        pattern = f"%{_escape_like(email)}%"
+        conditions.extend(
+            func.lower(field).like(pattern, escape="\\")
+            for field in (
+                MailMailboxMessageIndex.from_header,
+                MailMailboxMessageIndex.to_header,
+                MailMailboxMessageIndex.cc_header,
+            )
+        )
+    if not conditions:
+        return []
+    candidates = _bounded_rows(
+        session.query(MailMailboxMessageIndex)
+        .filter(MailMailboxMessageIndex.tenant_id == tenant_id, or_(*conditions))
+        .order_by(MailMailboxMessageIndex.id)
+    )
+    return [
+        row
+        for row in candidates
+        if row.id == message_id or _matching_headers(row, email)
+    ]
+
+
+def _matching_headers(
+    row: MailMailboxMessageIndex, email: str | None
+) -> dict[str, list[dict[str, str | None]]]:
+    if email is None:
+        return {}
+    result = {}
+    for role, value in (
+        ("from", row.from_header),
+        ("to", row.to_header),
+        ("cc", row.cc_header),
+    ):
+        matches = [
+            {"email": address.casefold(), "name": name or None}
+            for name, address in getaddresses([value or ""])
+            if address.casefold() == email
+        ]
+        if matches:
+            result[role] = matches[:64]
+    return result
+
+
+def _matching_bounces(
+    session: Session, *, tenant_id: str, email: str | None, bounce_id: str | None
+) -> list[MailBounceObservation]:
+    conditions = []
+    if bounce_id:
+        conditions.append(MailBounceObservation.id == bounce_id)
+    if email:
+        conditions.append(func.lower(MailBounceObservation.recipient) == email)
+    if not conditions:
+        return []
+    return _bounded_rows(
+        session.query(MailBounceObservation)
+        .filter(MailBounceObservation.tenant_id == tenant_id, or_(*conditions))
+        .order_by(MailBounceObservation.id)
+    )
+
+
+def _matching_commands(
+    session: Session,
+    *,
+    tenant_id: str,
+    membership_ids: set[str],
+    command_id: str | None,
+    related_command_ids: set[str],
+) -> list[MailDeliveryCommand]:
+    conditions = []
+    if command_id:
+        conditions.append(MailDeliveryCommand.id == command_id)
+    if related_command_ids:
+        conditions.append(MailDeliveryCommand.id.in_(related_command_ids))
+    if membership_ids:
+        conditions.append(MailDeliveryCommand.created_by_user_id.in_(membership_ids))
+    if not conditions:
+        return []
+    return _bounded_rows(
+        session.query(MailDeliveryCommand)
+        .filter(MailDeliveryCommand.tenant_id == tenant_id, or_(*conditions))
+        .order_by(MailDeliveryCommand.id)
+    )
+
+
+def _command_attempts(
+    session: Session, command_ids: set[str]
+) -> list[MailDeliveryAttempt]:
+    if not command_ids:
+        return []
+    return _bounded_rows(
+        session.query(MailDeliveryAttempt)
+        .filter(MailDeliveryAttempt.command_id.in_(command_ids))
+        .order_by(MailDeliveryAttempt.id)
+    )
+
+
+def _command_reconciliations(
+    session: Session, command_ids: set[str]
+) -> list[MailDeliveryReconciliation]:
+    if not command_ids:
+        return []
+    return _bounded_rows(
+        session.query(MailDeliveryReconciliation)
+        .filter(MailDeliveryReconciliation.command_id.in_(command_ids))
+        .order_by(MailDeliveryReconciliation.id)
+    )
+
+
+def _mail_references(subject: DsarSubjectRef) -> dict[str, str]:
+    aliases = {
+        "mail.profile": "profile",
+        "mail.message_index": "message_index",
+        "mail.delivery_command": "command",
+        "mail.bounce_observation": "bounce",
+    }
+    return {
+        target: value
+        for key, target in aliases.items()
+        if (value := str(subject.external_references.get(key) or "").strip())
+    }
+
+
+def _membership_ids(subject: DsarSubjectRef) -> set[str]:
+    values = [subject.membership_id]
+    values.extend(
+        subject.external_references.get(key)
+        for key in (
+            "mail.user",
+            "mail.membership",
+            "access.membership",
+            "membership_id",
+        )
+    )
+    return {value for item in values if (value := str(item or "").strip())}
+
+
+def _subject_email(subject: DsarSubjectRef) -> str | None:
+    values = [subject.email, subject.external_references.get("mail.email")]
+    normalized = {email for value in values if (email := _normalized_email(value))}
+    return normalized.pop() if len(normalized) == 1 else None
+
+
+def _normalized_email(value: object) -> str | None:
+    if not isinstance(value, str):
+        return None
+    value = value.strip().casefold()
+    return value or None
+
+
+def _record(
+    resource_type: str,
+    resource_id: str,
+    category: str,
+    title: str,
+    data: Mapping[str, object],
+    *,
+    observed_at: datetime | None = None,
+    immutable: bool = False,
+    retention_reason: str | None = None,
+    source_path: str | None = None,
+) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="mail",
+        module_id="mail",
+        resource_type=resource_type,
+        resource_id=resource_id,
+        category=category,
+        title=title,
+        data=data,
+        observed_at=observed_at,
+        immutable_evidence=immutable,
+        retention_reason=retention_reason,
+        source_path=source_path,
+    )
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Mail DSAR provider requires a SQLAlchemy session.")
+    return value
+
+
+def _bounded_rows(query: object) -> list[object]:
+    rows = query.limit(_MAX_RECORDS + 1).all()  # type: ignore[attr-defined]
+    if len(rows) > _MAX_RECORDS:
+        raise ValueError(
+            "Mail DSAR match limit exceeded; narrow the subject selectors."
+        )
+    return rows
+
+
+def _bounded_text(value: str | None) -> str | None:
+    return value[:2_000] if value else None
+
+
+def _escape_like(value: str) -> str:
+    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+
+
+def _iso(value: datetime | None) -> str | None:
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=timezone.utc)
+    return value.isoformat()
+
+
+__all__ = ["MAIL_DSAR_CAPABILITY", "MailDsarProvider"]

src/govoplan_mail/backend/manifest.py

@@ -15,6 +15,7 @@ from govoplan_core.core.mail import (
 from govoplan_core.core.postbox import CAPABILITY_POSTBOX_DELIVERY
 from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard
 from govoplan_core.core.modules import (
+    CapabilityDocumentation,
     DocumentationCondition,
     DocumentationConfigurationProviderRegistration,
     DocumentationLink,
@@ -54,6 +55,7 @@ from govoplan_mail.backend.provider_state import (
     smtp_provider_states,
 )
 from govoplan_mail.backend.db import models as mail_models  # noqa: F401 - populate Mail ORM metadata
+from govoplan_mail.backend.dsar_provider import MAIL_DSAR_CAPABILITY
 from govoplan_mail.backend.search_source import create_mail_search_source
 
 
@@ -105,6 +107,13 @@ def _configuration_provider(context: ModuleContext) -> object:
     return SqlMailConfigurationProvider()
 
 
+def _mail_dsar_provider(context: ModuleContext) -> object:
+    del context
+    from govoplan_mail.backend.dsar_provider import MailDsarProvider
+
+    return MailDsarProvider()
+
+
 def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
     module_id, resource, action = scope.split(":", 2)
     return PermissionDefinition(
@@ -311,7 +320,7 @@ IMAP_PROVIDER = ExternalProviderDeclaration(
 manifest = ModuleManifest(
     id="mail",
     name="Mail",
-    version="0.1.18",
+    version="0.1.19",
     required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
     optional_dependencies=("campaigns", "addresses", "calendar", "postbox", "search"),
     provides_interfaces=(
@@ -321,6 +330,7 @@ manifest = ModuleManifest(
         ModuleInterfaceProvider(name="mail.notification_delivery", version="0.1.0"),
         ModuleInterfaceProvider(name="mail.bounce_processing", version="0.1.0"),
         ModuleInterfaceProvider(name=CAPABILITY_MAIL_POSTBOX_BRIDGE, version="1.0.0"),
+        ModuleInterfaceProvider(name=MAIL_DSAR_CAPABILITY, version="0.1.0"),
     ),
     requires_interfaces=(
         ModuleInterfaceRequirement(
@@ -347,6 +357,12 @@ manifest = ModuleManifest(
             version_max_exclusive="0.2.0",
             optional=True,
         ),
+        ModuleInterfaceRequirement(
+            name="addresses.contact_writer",
+            version_min="0.1.0",
+            version_max_exclusive="0.2.0",
+            optional=True,
+        ),
         ModuleInterfaceRequirement(
             name=CAPABILITY_POSTBOX_DELIVERY,
             version_min="0.1.0",
@@ -420,7 +436,9 @@ manifest = ModuleManifest(
                 full_page_path="/mail",
                 required_any=("mail:mailbox:read",),
                 order=10,
-                modes=("browse", "compose"),
+                modes=("browse", "compose", "select"),
+                returned_reference_kinds=("mail.message",),
+                help_context_id="mail.quick_access.messages",
             ),
         ),
     ),
@@ -467,8 +485,57 @@ manifest = ModuleManifest(
             "govoplan_mail.backend.postbox_bridge",
             fromlist=["create_postbox_bridge"],
         ).create_postbox_bridge(context),
+        MAIL_DSAR_CAPABILITY: _mail_dsar_provider,
+    },
+    capability_documentation={
+        MAIL_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Mail data-subject request provider",
+            summary="Finds isolated mailbox-index, personal-profile, delivery, reconciliation, and bounce metadata without exposing transport secrets.",
+            contract_version="0.1.0",
+            documentation_types=("admin",),
+            audience=("privacy_officer", "mail_admin", "records_manager"),
+        ),
     },
     documentation=(
+        DocumentationTopic(
+            id="mail.privacy.data-subject-requests",
+            title="Review Mail data in a data-subject request",
+            summary="Collect tenant-scoped Mail metadata while preserving transport evidence and external-mailbox authority.",
+            body=(
+                "Mail's DSAR provider searches the effective tenant by normalized header or bounce-recipient email, direct membership references, and namespaced Mail profile, message-index, delivery-command, or bounce references. It isolates only matching From, To, and Cc parties and returns bounded message-index content, safe personal-profile metadata, and delivery, attempt, reconciliation, and bounce outcomes. It excludes SMTP/IMAP configuration, usernames and credentials, encrypted messages and envelopes, refusal detail, mailbox folders and UIDs, endpoint and credential identifiers, idempotency and worker claims, diagnostics, error text, and opaque evidence. "
+                "Delivery attempts, reconciliation decisions, and bounce observations remain retained as immutable transport and recovery evidence. Mailbox indexes and personal profiles require manual review through Mail and the authoritative external mailbox. The provider performs no direct erasure because deleting a derived index alone would not delete its external source, while changing a profile or delivery command can affect credentials, other users, and preserved evidence."
+            ),
+            layer="configured",
+            documentation_types=("admin",),
+            audience=("privacy_officer", "mail_admin", "records_manager", "operator"),
+            order=35,
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("mail", "access"),
+                    any_scopes=("access:privacy:read", "access:privacy:manage", "access:privacy:erase"),
+                ),
+            ),
+            links=(
+                DocumentationLink(label="Data-subject requests", href="/admin?section=tenant-data-subject-requests", kind="runtime"),
+                DocumentationLink(label="Mail handbook", href="govoplan-mail/docs/MAIL_HANDBOOK.md", kind="repository"),
+            ),
+            related_modules=("access", "audit", "campaigns", "postbox"),
+            metadata={
+                "kind": "workflow",
+                "route": "/admin?section=tenant-data-subject-requests",
+                "help_contexts": ["admin.privacy.data-subject-requests"],
+                "steps": [
+                    "Run the Mail provider search and review mailbox-index, profile, delivery, reconciliation, and bounce dispositions.",
+                    "Retain immutable transport evidence with its reason.",
+                    "Coordinate approved mailbox content deletion with the authoritative external mailbox and then refresh the derived index.",
+                    "Use Mail profile lifecycle controls for approved personal-profile changes; do not edit encrypted payload or evidence rows directly.",
+                ],
+                "limitations": [
+                    "Encrypted outbound payloads cannot be searched by recipient without an independently corroborated Mail command reference.",
+                    "External mailbox deletion is outside the DSAR provider and must be coordinated with the configured provider.",
+                ],
+            },
+        ),
         DocumentationTopic(
             id="mail.postbox.bridge",
             title="Bridge selected Mail observations into Postbox",
@@ -556,6 +623,10 @@ manifest = ModuleManifest(
                 "appears inside the shared Messages drawer alongside independent Postbox and future chat contributions. "
                 "Recent entries open the exact authorized profile, folder, and message; Drafts resolves the configured or "
                 "provider-detected Drafts folder, and the full Mail page keeps the versioned Quick Access return context. "
+                "When an active Case requests a selection, choosing a recent message returns only its tenant-bound profile, "
+                "folder, UID version, label, and owner route through the shared result contract. Cases discards the label and "
+                "does not receive headers, participants, preview, body, attachments, credentials, or an access decision; Mail "
+                "reauthorizes the mailbox list now and the exact message again whenever its owner route is opened. "
                 "Compose deliberately launches the user's configured mail application because GovOPlaN Mail's mailbox is "
                 "read-only; it neither selects a GovOPlaN transport profile nor claims a GovOPlaN delivery. The shared drawer "
                 "does not merge channel state, credentials, custody, delivery semantics, or authorization."
@@ -574,6 +645,10 @@ manifest = ModuleManifest(
                         "Aktuelle Einträge öffnen das genaue berechtigte Profil, den Ordner und die Nachricht; Entwürfe verwendet "
                         "den konfigurierten oder vom Anbieter erkannten Entwurfsordner. Die vollständige Mail-Seite erhält den "
                         "versionierten Rücksprungkontext. Verfassen öffnet bewusst die konfigurierte Mail-Anwendung des Benutzers, "
+                        "Bei einem aktiven Vorgang gibt die Auswahl einer aktuellen Nachricht nur den mandantengebundenen Profil-, "
+                        "Ordner- und UID-Verweis sowie die Eigentümerroute über den gemeinsamen Ergebnisvertrag zurück. Vorgänge "
+                        "übernehmen weder Kopfzeilen, Beteiligte, Vorschau, Inhalt, Anlagen, Zugangsdaten noch eine Zugriffsentscheidung; "
+                        "Mail prüft den Zugriff beim Auflisten und beim späteren Öffnen erneut. "
                         "da das GovOPlaN-Mail-Postfach nur lesend arbeitet; dabei wird weder ein GovOPlaN-Transportprofil gewählt "
                         "noch eine GovOPlaN-Zustellung behauptet. Die gemeinsame Darstellung führt weder Kanalzustand noch "
                         "Zugangsdaten, Verwahrung, Zustelllogik oder Berechtigungen zusammen."
@@ -583,6 +658,52 @@ manifest = ModuleManifest(
             metadata={"kind": "reference", "help_contexts": ["mail.quick_access.messages"]},
             order=37,
         ),
+        DocumentationTopic(
+            id="mail.address-book-integration",
+            title="Use address-book contacts in Mail",
+            summary="Autocomplete recipients and add message participants to an explicitly writable address book when Addresses is installed.",
+            body=(
+                "Mail resolves the optional addresses.lookup and addresses.contact_writer capabilities through the platform registry. "
+                "Quick Access compose accepts manual recipients in every installation and adds visible contact suggestions when lookup is available, "
+                "then opens the user's configured mail application. A selected mailbox message offers add-contact actions only through writer decisions "
+                "returned for the current principal. Read-only connector books and policy-blocked targets remain disabled with the owning Addresses reason. "
+                "Mail never imports Addresses models, bypasses its scope checks, or changes its read-only mailbox and external-compose custody model."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("mail_user", "mail_admin", "administrator"),
+            related_modules=("addresses",),
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("mail",),
+                    required_scopes=("mail:profile:use",),
+                ),
+            ),
+            links=(
+                DocumentationLink(label="Mail", href="/mail", kind="runtime"),
+                DocumentationLink(label="Addresses", href="/addresses", kind="runtime"),
+            ),
+            translations={
+                "de": {
+                    "title": "Adressbuchkontakte in Mail verwenden",
+                    "summary": "Empfänger vervollständigen und Nachrichtenbeteiligte in einem ausdrücklich beschreibbaren Adressbuch speichern, wenn Addresses installiert ist.",
+                    "body": (
+                        "Mail löst die optionalen Capabilities addresses.lookup und addresses.contact_writer über die Plattformregistrierung auf. "
+                        "Beim Verfassen im Schnellzugriff können Empfänger immer manuell eingegeben werden; bei verfügbarer Suche kommen sichtbare Kontaktvorschläge hinzu, "
+                        "anschließend öffnet sich die konfigurierte Mail-Anwendung. Für eine ausgewählte Postfachnachricht werden Kontaktaktionen ausschließlich anhand der "
+                        "Writer-Entscheidungen für den aktuellen Principal angeboten. Schreibgeschützte Connector-Adressbücher und durch Richtlinien gesperrte Ziele bleiben "
+                        "mit dem von Addresses gelieferten Grund deaktiviert. Mail importiert keine Addresses-Modelle, umgeht keine Bereichsprüfung und ändert weder das "
+                        "nur lesende Postfach noch das Verwahrungsmodell des externen Verfassens."
+                    ),
+                }
+            },
+            metadata={
+                "kind": "workflow",
+                "route": "/mail",
+                "help_contexts": ["mail.quick_access.messages", "mail.mailbox"],
+            },
+            order=38,
+        ),
         DocumentationTopic(
             id="mail.search.mailbox-messages",
             title="Search authorized mailbox messages",
@@ -598,7 +719,7 @@ manifest = ModuleManifest(
             documentation_types=("admin", "user"),
             audience=("mail_user", "mail_admin", "administrator"),
             related_modules=("search",),
-            order=38,
+            order=39,
         ),
         DocumentationTopic(
             id="mail.profiles-and-policy",
@@ -877,7 +998,7 @@ manifest = ModuleManifest(
             id="mail.reference.campaign-delivery-contract",
             title="Integrate Campaign through the Mail delivery contract",
             summary="Campaign freezes a Mail profile reference and opaque revision; Mail re-authorizes, revision-checks, resolves credentials, and performs the effect in one call.",
-            body="The mail.campaign_delivery 0.2 contract never returns decrypted credentials or resolved SMTP/IMAP configuration. Mail compares the expected random transport revision before decrypting protocol-specific credentials and returns only bounded sanitized outcomes. Campaign owns ordinary recipient jobs; report messages use Mail's encrypted idempotent delivery-command and attempt ledger. Every current SMTP and Sent-folder attempt passes a stable effect identifier into a Mail-owned Core recovery operation before provider contact. Effect-start evidence prevents blind redelivery, unknown outcomes require explicit reconciliation, and raw recipient refusals require Mail diagnostic authority. Mail outbox dispatch and retention scans are partitioned by tenant entitlement, so disabling Mail leaves accepted commands and evidence untouched for operator resolution.",
+            body="The mail.campaign_delivery 0.2 contract never returns decrypted credentials or resolved SMTP/IMAP configuration. Mail compares the expected random transport revision before decrypting protocol-specific credentials and returns only bounded sanitized outcomes. Synchronous batches authorize the complete recipient set and preflight DNS, egress, connectivity, TLS, and authentication before the first effect. Mail reuses the bounded connection when deployment policy permits, health-checks it before reuse, and reconnects before the next message when a stale connection is detected. A connection loss after DATA begins remains outcome-unknown and is never replayed. Systemic authentication, sender, or connectivity failures carry stable reason codes so Campaign pauses remaining queued work and shows connection, reconnect, failure, and pause progress. Campaign owns ordinary recipient jobs; report messages use Mail's encrypted idempotent delivery-command and attempt ledger. Every current SMTP and Sent-folder attempt passes a stable effect identifier into a Mail-owned Core recovery operation before provider contact. Effect-start evidence prevents blind redelivery, unknown outcomes require explicit reconciliation, and raw recipient refusals require Mail diagnostic authority. Mail outbox dispatch and retention scans are partitioned by tenant entitlement, so disabling Mail leaves accepted commands and evidence untouched for operator resolution.",
             layer="available",
             documentation_types=("admin", "user"),
             audience=("integrator", "campaign_manager", "campaign_sender", "release_reviewer"),
@@ -900,7 +1021,7 @@ manifest = ModuleManifest(
                 "route": "/campaigns/{campaign_id}/mail-settings",
                 "screen": "Campaign Mail settings",
                 "section": "Mail-owned profile and transport boundary",
-                "verification": "Prove stale revisions fail before credential decryption, SMTP never decrypts IMAP credentials, IMAP never decrypts SMTP credentials, provider details are sanitized, and the interface/version gate passes.",
+                "verification": "Prove stale revisions fail before credential decryption, batch preflight fails before DATA, two messages reuse one healthy connection, a stale connection reconnects before the next message, post-DATA disconnect is never replayed, systemic failures pause remaining jobs, provider details are sanitized, and the interface/version gate passes.",
                 "related_topic_ids": [
                     "mail.profile-ownership-and-consumers",
                     "campaigns.mail-profile-user-journey",

src/govoplan_mail/backend/router.py

@@ -11,6 +11,10 @@ from sqlalchemy.orm import Session
 from govoplan_mail.backend.schemas import (
     MailAddressLookupCandidate,
     MailAddressLookupResponse,
+    MailAddressWriteTarget,
+    MailAddressWriteTargetResponse,
+    MailContactCreateRequest,
+    MailContactCreateResponse,
     MailConnectionTestResponse,
     MailBounceObservationListResponse,
     MailBounceObservationResponse,
@@ -149,6 +153,7 @@ MAIL_CREDENTIAL_RESOURCE = "mail_credential"
 MAILBOX_MESSAGES_CURSOR_SCOPE = "mail.mailbox.messages.v1"
 DEFAULT_MAILBOX_MESSAGE_LIMIT = 50
 CAPABILITY_ADDRESSES_LOOKUP = "addresses.lookup"
+CAPABILITY_ADDRESSES_CONTACT_WRITER = "addresses.contact_writer"
 bounce_provider = SqlMailBounceProcessingProvider()
 
 
@@ -416,6 +421,15 @@ def _capability_payload(value: object) -> dict[str, Any]:
         "source_ref",
         "source_revision",
         "provenance",
+        "address_book_label",
+        "operation",
+        "allowed",
+        "reason",
+        "message",
+        "scope_type",
+        "scope_id",
+        "read_only",
+        "required_scopes",
     ):
         if hasattr(value, key):
             payload[key] = getattr(value, key)
@@ -1361,6 +1375,78 @@ def lookup_mail_addresses(
     )
 
 
+@router.get("/address-write-targets", response_model=MailAddressWriteTargetResponse)
+def list_mail_address_write_targets(
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> MailAddressWriteTargetResponse:
+    _require_scope(principal, "mail:profile:use")
+    capability = _registry_capability(CAPABILITY_ADDRESSES_CONTACT_WRITER)
+    if capability is None or not hasattr(capability, "list_write_targets"):
+        return MailAddressWriteTargetResponse(available=False, targets=[])
+    targets = getattr(capability, "list_write_targets")(session, principal, operation="create_contact")
+    return MailAddressWriteTargetResponse(
+        available=True,
+        targets=[MailAddressWriteTarget.model_validate(_capability_payload(target)) for target in targets],
+    )
+
+
+@router.post(
+    "/address-contacts",
+    response_model=MailContactCreateResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def create_mail_address_contact(
+    payload: MailContactCreateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> MailContactCreateResponse:
+    _require_scope(principal, "mail:profile:use")
+    capability = _registry_capability(CAPABILITY_ADDRESSES_CONTACT_WRITER)
+    if capability is None or not all(
+        hasattr(capability, method)
+        for method in ("can_write_to_address_book", "create_contact")
+    ):
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Address-book contact writing is not available.",
+        )
+    decision = getattr(capability, "can_write_to_address_book")(
+        session,
+        principal,
+        address_book_id=payload.address_book_id,
+        operation="create_contact",
+    )
+    decision_payload = _capability_payload(decision)
+    if decision_payload.get("allowed") is not True:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=str(decision_payload.get("message") or "The selected address book is not writable."),
+        )
+    try:
+        result = getattr(capability, "create_contact")(
+            session,
+            principal,
+            address_book_id=payload.address_book_id,
+            payload={
+                "display_name": payload.display_name or payload.email,
+                "emails": [{"label": "Mail", "email": payload.email, "is_primary": True}],
+            },
+            provenance={
+                "consumer_module": MAIL_MODULE_ID,
+                "consumer_workflow": "mailbox_add_contact",
+            },
+        )
+        session.commit()
+    except ValueError as exc:
+        session.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=str(exc),
+        ) from exc
+    return MailContactCreateResponse.model_validate(_capability_payload(result))
+
+
 @router.get("/settings/delta", response_model=MailSettingsDeltaResponse)
 def mail_settings_delta(
     scope_type: str = Query(default="tenant"),

src/govoplan_mail/backend/schemas.py

@@ -344,6 +344,43 @@ class MailAddressLookupResponse(BaseModel):
     candidates: list[MailAddressLookupCandidate] = Field(default_factory=list)
 
 
+class MailAddressWriteTarget(BaseModel):
+    address_book_id: str
+    address_book_label: str | None = None
+    operation: str = "create_contact"
+    allowed: bool = False
+    reason: str
+    message: str
+    scope_type: str | None = None
+    scope_id: str | None = None
+    source_kind: str | None = None
+    read_only: bool = False
+    required_scopes: list[str] = Field(default_factory=list)
+    provenance: dict[str, Any] = Field(default_factory=dict)
+
+
+class MailAddressWriteTargetResponse(BaseModel):
+    available: bool = False
+    targets: list[MailAddressWriteTarget] = Field(default_factory=list)
+
+
+class MailContactCreateRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    address_book_id: str = Field(min_length=1, max_length=36)
+    display_name: str | None = Field(default=None, max_length=255)
+    email: str = Field(min_length=3, max_length=320, pattern=r"^[^\s@]+@[^\s@]+\.[^\s@]+$")
+
+
+class MailContactCreateResponse(BaseModel):
+    contact_id: str
+    address_book_id: str
+    display_name: str
+    email: str | None = None
+    source_kind: str = "local"
+    provenance: dict[str, Any] = Field(default_factory=dict)
+
+
 class MailConnectionTestResponse(BaseModel):
     ok: bool
     protocol: Literal["smtp", "imap"]

src/govoplan_mail/backend/sending/smtp.py

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

tests/test_address_integration.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from fastapi import HTTPException
+from pydantic import ValidationError
+
+from govoplan_mail.backend.router import (
+    create_mail_address_contact,
+    list_mail_address_write_targets,
+    lookup_mail_addresses,
+)
+from govoplan_mail.backend.schemas import MailContactCreateRequest
+
+
+class _Session:
+    def __init__(self) -> None:
+        self.commits = 0
+        self.rollbacks = 0
+
+    def commit(self) -> None:
+        self.commits += 1
+
+    def rollback(self) -> None:
+        self.rollbacks += 1
+
+
+class _Writer:
+    def __init__(self, *, allowed: bool = True, read_only: bool = False) -> None:
+        self.allowed = allowed
+        self.read_only = read_only
+        self.created_payload = None
+        self.created_provenance = None
+
+    def list_write_targets(self, _session, _principal, *, operation):
+        return (
+            SimpleNamespace(
+                address_book_id="book-1",
+                address_book_label="Personal contacts",
+                operation=operation,
+                allowed=self.allowed,
+                reason="allowed" if self.allowed else "read_only_source",
+                message="Contact can be added." if self.allowed else "This source is read-only.",
+                scope_type="user",
+                scope_id="user-1",
+                source_kind="local" if self.allowed else "ldap",
+                read_only=self.read_only,
+                required_scopes=("addresses:contacts:write",),
+                provenance={"policy": "addresses"},
+            ),
+        )
+
+    def can_write_to_address_book(self, _session, _principal, *, address_book_id, operation):
+        return self.list_write_targets(_session, _principal, operation=operation)[0]
+
+    def create_contact(self, _session, _principal, *, address_book_id, payload, provenance):
+        self.created_payload = payload
+        self.created_provenance = provenance
+        return SimpleNamespace(
+            contact_id="contact-1",
+            address_book_id=address_book_id,
+            display_name=payload["display_name"],
+            email=payload["emails"][0]["email"],
+            source_kind="local",
+            provenance=provenance,
+        )
+
+
+def _principal():
+    return SimpleNamespace(has=lambda scope: scope == "mail:profile:use")
+
+
+class MailAddressIntegrationTests(unittest.TestCase):
+    def test_optional_capabilities_fail_open_for_mail(self) -> None:
+        with patch("govoplan_mail.backend.router._registry_capability", return_value=None):
+            lookup = lookup_mail_addresses(query="ada", limit=25, session=_Session(), principal=_principal())
+            targets = list_mail_address_write_targets(session=_Session(), principal=_principal())
+
+        self.assertFalse(lookup.available)
+        self.assertEqual(lookup.candidates, [])
+        self.assertFalse(targets.available)
+        self.assertEqual(targets.targets, [])
+
+    def test_write_target_preserves_read_only_decision(self) -> None:
+        writer = _Writer(allowed=False, read_only=True)
+        with patch("govoplan_mail.backend.router._registry_capability", return_value=writer):
+            response = list_mail_address_write_targets(session=_Session(), principal=_principal())
+
+        self.assertTrue(response.available)
+        self.assertFalse(response.targets[0].allowed)
+        self.assertTrue(response.targets[0].read_only)
+        self.assertEqual(response.targets[0].reason, "read_only_source")
+
+    def test_blocked_target_cannot_be_bypassed_by_create(self) -> None:
+        session = _Session()
+        writer = _Writer(allowed=False, read_only=True)
+        with (
+            patch("govoplan_mail.backend.router._registry_capability", return_value=writer),
+            self.assertRaises(HTTPException) as raised,
+        ):
+            create_mail_address_contact(
+                MailContactCreateRequest(
+                    address_book_id="book-1",
+                    display_name="Ada Lovelace",
+                    email="ada@example.test",
+                ),
+                session=session,
+                principal=_principal(),
+            )
+
+        self.assertEqual(raised.exception.status_code, 422)
+        self.assertEqual(session.commits, 0)
+
+    def test_allowed_create_uses_writer_and_records_consumer_provenance(self) -> None:
+        session = _Session()
+        writer = _Writer()
+        with patch("govoplan_mail.backend.router._registry_capability", return_value=writer):
+            result = create_mail_address_contact(
+                MailContactCreateRequest(
+                    address_book_id="book-1",
+                    display_name="Ada Lovelace",
+                    email="ada@example.test",
+                ),
+                session=session,
+                principal=_principal(),
+            )
+
+        self.assertEqual(result.contact_id, "contact-1")
+        self.assertEqual(writer.created_payload["emails"][0]["email"], "ada@example.test")
+        self.assertEqual(writer.created_provenance["consumer_module"], "mail")
+        self.assertEqual(session.commits, 1)
+
+    def test_proxy_rejects_invalid_email_before_calling_writer(self) -> None:
+        with self.assertRaises(ValidationError):
+            MailContactCreateRequest(address_book_id="book-1", email="not-an-email")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_dsar_provider.py

@@ -0,0 +1,374 @@
+from __future__ import annotations
+
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_access.backend.db.models import Account, Group, User
+from govoplan_core.core.change_sequence import ChangeSequenceEntry
+from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    DataSubjectRequest,
+    create_data_subject_request,
+    plan_data_subject_erasure,
+    search_data_subject_request,
+)
+from govoplan_mail.backend.db.models import (
+    MailBounceObservation,
+    MailDeliveryAttempt,
+    MailDeliveryCommand,
+    MailDeliveryReconciliation,
+    MailMailboxMessageIndex,
+    MailServerProfile,
+)
+from govoplan_mail.backend.dsar_provider import MAIL_DSAR_CAPABILITY, MailDsarProvider
+from govoplan_mail.backend.manifest import manifest
+
+
+class _Registry:
+    def __init__(self, provider, active=True):
+        self.provider = provider
+        self.active = active
+
+    def capability_names(self):
+        return (MAIL_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        assert name == MAIL_DSAR_CAPABILITY
+        return "mail"
+
+    def tenant_entitlement_resolver(self):
+        active = self.active
+
+        class Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type(
+                    "State", (), {"effective_modules": ("mail",) if active else ()}
+                )()
+
+        return Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del session, kwargs
+        assert name == MAIL_DSAR_CAPABILITY
+        return self.provider
+
+
+class MailDsarProviderTests(unittest.TestCase):
+    def setUp(self):
+        self.engine = create_engine("sqlite:///:memory:", future=True)
+        Base.metadata.create_all(
+            self.engine,
+            tables=[
+                Account.__table__,
+                User.__table__,
+                Group.__table__,
+                ChangeSequenceEntry.__table__,
+                DataSubjectRequest.__table__,
+                MailServerProfile.__table__,
+                MailMailboxMessageIndex.__table__,
+                MailDeliveryCommand.__table__,
+                MailDeliveryAttempt.__table__,
+                MailDeliveryReconciliation.__table__,
+                MailBounceObservation.__table__,
+            ],
+        )
+        self.session = sessionmaker(bind=self.engine, future=True)()
+        now = datetime.now(timezone.utc)
+        account = Account(
+            id="account-1",
+            email="subject@example.test",
+            normalized_email="subject@example.test",
+            display_name="Subject",
+        )
+        user = User(
+            id="membership-1",
+            tenant_id="tenant-1",
+            account_id=account.id,
+            email="subject@example.test",
+            display_name="Subject",
+        )
+        profile = MailServerProfile(
+            id="profile-subject",
+            tenant_id="tenant-1",
+            scope_type="user",
+            scope_id=user.id,
+            name="Personal mail",
+            slug="personal",
+            smtp_config={"host": "smtp-secret-do-not-export"},
+            smtp_username="smtp-user-do-not-export",
+            smtp_password_encrypted="smtp-cipher-do-not-export",
+            imap_config={"host": "imap-secret-do-not-export"},
+            imap_username="imap-user-do-not-export",
+            imap_password_encrypted="imap-cipher-do-not-export",
+            created_by_user_id=user.id,
+        )
+        message = MailMailboxMessageIndex(
+            id="message-subject",
+            tenant_id="tenant-1",
+            profile_id=profile.id,
+            folder="INBOX-secret-do-not-export",
+            uid="uid-secret-do-not-export",
+            uid_int=1,
+            sort_position=1,
+            subject="Subject notice",
+            from_header="Office <office@example.test>",
+            to_header="Subject Person <Subject@Example.Test>",
+            cc_header="Unrelated Person <other@example.test>",
+            date="2026-08-20",
+            message_id="message-locator-do-not-export",
+            flags=["\\Seen"],
+            size_bytes=42,
+            body_preview="Message preview for the subject",
+            attachment_count=1,
+            indexed_at=now,
+        )
+        unrelated = MailMailboxMessageIndex(
+            id="message-other",
+            tenant_id="tenant-1",
+            profile_id=profile.id,
+            folder="INBOX",
+            uid="2",
+            uid_int=2,
+            sort_position=2,
+            subject="Unrelated message do not export",
+            from_header="other@example.test",
+            to_header="someone@example.test",
+            indexed_at=now,
+        )
+        tenant_two_profile = MailServerProfile(
+            id="profile-tenant-2",
+            tenant_id="tenant-2",
+            name="Tenant two",
+            slug="tenant-two",
+            smtp_config={},
+        )
+        tenant_two = MailMailboxMessageIndex(
+            id="message-tenant-2",
+            tenant_id="tenant-2",
+            profile_id=tenant_two_profile.id,
+            folder="INBOX",
+            uid="1",
+            uid_int=1,
+            sort_position=1,
+            subject="Tenant two message do not export",
+            to_header="subject@example.test",
+            indexed_at=now,
+        )
+        command = MailDeliveryCommand(
+            id="command-subject",
+            tenant_id="tenant-1",
+            command_type="send",
+            source_module="notifications",
+            source_resource_type="notification",
+            idempotency_key="idempotency-do-not-export",
+            canonical_request_hash="a" * 64,
+            profile_id=profile.id,
+            expected_smtp_transport_revision="revision-secret",
+            envelope_recipients_encrypted="recipient-cipher-do-not-export",
+            message_encrypted="message-cipher-do-not-export",
+            message_sha256="b" * 64,
+            rfc_message_id="rfc-message-id",
+            message_size_bytes=100,
+            recipient_count=1,
+            status="succeeded",
+            attempt_count=1,
+            accepted_count=1,
+            created_by_user_id=user.id,
+            completed_at=now,
+            expires_at=now + timedelta(days=30),
+        )
+        attempt = MailDeliveryAttempt(
+            id="attempt-subject",
+            command_id=command.id,
+            attempt_number=1,
+            worker_id="worker-do-not-export",
+            status="succeeded",
+            started_at=now,
+            completed_at=now,
+            accepted_count=1,
+            diagnostic_summary="diagnostic-do-not-export",
+        )
+        reconciliation = MailDeliveryReconciliation(
+            id="reconciliation-subject",
+            command_id=command.id,
+            decision="confirmed_sent",
+            evidence_reference="private-reference-do-not-export",
+            note_encrypted="private-note-do-not-export",
+            created_by_user_id=user.id,
+        )
+        bounce = MailBounceObservation(
+            id="bounce-subject",
+            tenant_id="tenant-1",
+            profile_id=profile.id,
+            folder="bounce-folder-do-not-export",
+            uid="bounce-uid-do-not-export",
+            fingerprint="c" * 64,
+            raw_sha256="d" * 64,
+            original_message_id="original-id-do-not-export",
+            command_id=command.id,
+            recipient="subject@example.test",
+            action="failed",
+            status_code="5.1.1",
+            diagnostic="bounce-diagnostic-do-not-export",
+            permanent=True,
+            observed_at=now,
+            matched=True,
+            evidence={"secret": "bounce-evidence-do-not-export"},
+        )
+        self.session.add_all(
+            [
+                account,
+                user,
+                profile,
+                message,
+                unrelated,
+                tenant_two_profile,
+                tenant_two,
+                command,
+                attempt,
+                reconciliation,
+                bounce,
+            ]
+        )
+        self.session.commit()
+        self.provider = MailDsarProvider()
+        self.subject = DsarSubjectRef(
+            membership_id=user.id, email="subject@example.test"
+        )
+
+    def tearDown(self):
+        self.session.close()
+        self.engine.dispose()
+
+    def test_manifest_and_minimized_tenant_scoped_search(self):
+        self.assertIn(
+            MAIL_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}
+        )
+        self.assertIsInstance(
+            manifest.capability_factories[MAIL_DSAR_CAPABILITY](None), DsarProvider
+        )
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self.subject
+        )
+        self.assertTrue(
+            {
+                "mail_server_profile",
+                "mailbox_message_index",
+                "mail_delivery_command",
+                "mail_delivery_attempt",
+                "mail_delivery_reconciliation",
+                "mail_bounce_observation",
+            }.issubset({r.resource_type for r in records})
+        )
+        serialized = repr([record.to_dict() for record in records])
+        for hidden in (
+            "other@example.test",
+            "Unrelated Person",
+            "message-other",
+            "Unrelated message do not export",
+            "message-tenant-2",
+            "Tenant two message do not export",
+            "smtp-secret-do-not-export",
+            "smtp-user-do-not-export",
+            "smtp-cipher-do-not-export",
+            "imap-secret-do-not-export",
+            "imap-user-do-not-export",
+            "imap-cipher-do-not-export",
+            "INBOX-secret-do-not-export",
+            "uid-secret-do-not-export",
+            "message-locator-do-not-export",
+            "idempotency-do-not-export",
+            "recipient-cipher-do-not-export",
+            "message-cipher-do-not-export",
+            "worker-do-not-export",
+            "diagnostic-do-not-export",
+            "private-reference-do-not-export",
+            "private-note-do-not-export",
+            "bounce-folder-do-not-export",
+            "bounce-uid-do-not-export",
+            "original-id-do-not-export",
+            "bounce-diagnostic-do-not-export",
+            "bounce-evidence-do-not-export",
+        ):
+            self.assertNotIn(hidden, serialized)
+
+    def test_conflicting_email_fails_closed(self):
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                email="subject@example.test",
+                external_references={"mail.email": "other@example.test"},
+            ),
+        )
+        self.assertEqual((), records)
+
+    def test_plan_preserves_evidence_and_executes_nothing(self):
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self.subject
+        )
+        actions = self.provider.plan_erasure(
+            self.session, tenant_id="tenant-1", subject=self.subject, records=records
+        )
+        self.assertTrue({"retain", "manual_review"}.issubset({a.kind for a in actions}))
+        self.assertFalse(any(action.executable for action in actions))
+
+    def test_core_workflow_discovers_active_and_skips_disabled_provider(self):
+        request = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-MAIL-1",
+            request_kind="access_and_erasure",
+            subject=self.subject,
+            purpose="Authorized request",
+            legal_basis="GDPR",
+            due_at=None,
+            requested_by_account_id="officer",
+        )
+        self.session.commit()
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider),
+            row=request,
+            expected_revision=1,
+        )
+        self.assertEqual(["mail"], request.coverage["covered_modules"])
+        plan_data_subject_erasure(
+            self.session,
+            registry=_Registry(self.provider),
+            row=request,
+            expected_revision=2,
+        )
+        self.assertFalse(
+            any(action["executable"] for action in request.erasure_plan["actions"])
+        )
+        disabled = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-MAIL-OFF",
+            request_kind="access",
+            subject=self.subject,
+            purpose="Coverage",
+            legal_basis=None,
+            due_at=None,
+            requested_by_account_id="officer",
+        )
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider, active=False),
+            row=disabled,
+            expected_revision=1,
+        )
+        self.assertEqual(
+            [MAIL_DSAR_CAPABILITY], disabled.coverage["inactive_provider_capabilities"]
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_manifest.py

@@ -9,6 +9,15 @@ from govoplan_mail.backend.manifest import _mail_retirement_provider, get_manife
 
 
 class MailManifestTests(unittest.TestCase):
+    def test_mail_quick_access_can_return_exact_message_references(self) -> None:
+        frontend = get_manifest().frontend
+        self.assertIsNotNone(frontend)
+        tool = next(
+            item for item in frontend.quick_access_tools if item.id == "mail.messages"  # type: ignore[union-attr]
+        )
+        self.assertIn("select", tool.modes)
+        self.assertEqual(("mail.message",), tool.returned_reference_kinds)
+
     def test_manifest_declares_optional_addresses_lookup(self) -> None:
         manifest = get_manifest()
 
@@ -16,6 +25,7 @@ class MailManifestTests(unittest.TestCase):
         self.assertEqual(manifest.id, "mail")
         self.assertIn("addresses", manifest.optional_dependencies)
         self.assertIn("addresses.lookup", {interface.name for interface in manifest.requires_interfaces})
+        self.assertIn("addresses.contact_writer", {interface.name for interface in manifest.requires_interfaces})
         self.assertIn(
             {
                 "name": "campaigns.access",
@@ -62,6 +72,7 @@ class MailManifestTests(unittest.TestCase):
                 "mail.workflow.read-mailbox",
                 "mail.reference.credentials-egress-retirement",
                 "mail.reference.campaign-delivery-contract",
+                "mail.address-book-integration",
             }.issubset(topics)
         )
         ownership = topics["mail.profile-ownership-and-consumers"]

tests/test_smtp_helpers.py

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

webui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/mail-webui",
-  "version": "0.1.18",
+  "version": "0.1.19",
   "private": true,
   "type": "module",
   "main": "src/index.ts",
@@ -26,7 +26,7 @@
     }
   },
   "scripts": {
-    "test:mail-ui": "rm -rf .mail-test-build && mkdir -p .mail-test-build && printf '{\"type\":\"commonjs\"}\\n' > .mail-test-build/package.json && tsc -p tsconfig.mail-tests.json && node .mail-test-build/tests/mailbox-display.test.js && node .mail-test-build/tests/mailbox-folders.test.js && node .mail-test-build/tests/mailbox-launch.test.js && node .mail-test-build/tests/mail-profile-editor-model.test.js && node .mail-test-build/tests/mail-policy-validation.test.js && node scripts/test-mailbox-icon-button-structure.mjs && node scripts/test-interface-pattern-language.mjs"
+    "test:mail-ui": "rm -rf .mail-test-build && mkdir -p .mail-test-build && printf '{\"type\":\"commonjs\"}\\n' > .mail-test-build/package.json && tsc -p tsconfig.mail-tests.json && node .mail-test-build/tests/mailbox-display.test.js && node .mail-test-build/tests/mailbox-folders.test.js && node .mail-test-build/tests/mailbox-launch.test.js && node .mail-test-build/tests/mail-profile-editor-model.test.js && node .mail-test-build/tests/mail-policy-validation.test.js && node .mail-test-build/tests/mail-address-integration.test.js && node scripts/test-mailbox-icon-button-structure.mjs && node scripts/test-interface-pattern-language.mjs"
   },
   "devDependencies": {
     "typescript": "^5.7.2"

webui/src/api/mail.ts

@@ -65,6 +65,41 @@ export type MailAddressLookupResponse = {
   candidates: MailAddressLookupCandidate[];
 };
 
+export type MailAddressWriteTarget = {
+  address_book_id: string;
+  address_book_label?: string | null;
+  operation: string;
+  allowed: boolean;
+  reason: string;
+  message: string;
+  scope_type?: string | null;
+  scope_id?: string | null;
+  source_kind?: string | null;
+  read_only: boolean;
+  required_scopes: string[];
+  provenance: Record<string, unknown>;
+};
+
+export type MailAddressWriteTargetResponse = {
+  available: boolean;
+  targets: MailAddressWriteTarget[];
+};
+
+export type MailContactCreatePayload = {
+  address_book_id: string;
+  display_name?: string | null;
+  email: string;
+};
+
+export type MailContactCreateResponse = {
+  contact_id: string;
+  address_book_id: string;
+  display_name: string;
+  email?: string | null;
+  source_kind: string;
+  provenance: Record<string, unknown>;
+};
+
 export type MailMailboxAttachment = {
   filename?: string | null;
   content_type: string;
@@ -209,6 +244,20 @@ export async function lookupMailAddresses(settings: ApiSettings, query: string,
   return apiFetch<MailAddressLookupResponse>(settings, apiPath("/api/v1/mail/address-lookup", { query, limit }));
 }
 
+export async function listMailAddressWriteTargets(settings: ApiSettings): Promise<MailAddressWriteTargetResponse> {
+  return apiFetch<MailAddressWriteTargetResponse>(settings, "/api/v1/mail/address-write-targets");
+}
+
+export async function createMailAddressContact(
+  settings: ApiSettings,
+  payload: MailContactCreatePayload
+): Promise<MailContactCreateResponse> {
+  return apiFetch<MailContactCreateResponse>(settings, "/api/v1/mail/address-contacts", {
+    method: "POST",
+    body: JSON.stringify(payload)
+  });
+}
+
 export async function listMailServerProfiles(settings: ApiSettings, includeInactive = false, campaignId?: string): Promise<MailServerProfile[]> {
   return apiGetList<MailServerProfile, "profiles">(settings, "/api/v1/mail/profiles", "profiles", {
     include_inactive: includeInactive ? true : undefined,

webui/src/features/mail/MailQuickAccess.tsx

@@ -1,19 +1,24 @@
-import { useCallback } from "react";
-import { ExternalLink, FilePenLine, Mail, Pencil } from "lucide-react";
+import { useCallback, useRef, useState } from "react";
+import { ExternalLink, FilePenLine, Mail, Pencil, X } from "lucide-react";
 import { Link } from "react-router";
 import {
   DashboardWidgetList,
+  Button,
   DismissibleAlert,
+  EmailAddressInput,
   LoadingFrame,
   quickAccessLaunchState,
   useDashboardWidgetData,
+  type MailboxAddress,
   type QuickAccessToolRenderContext
 } from "@govoplan/core-webui";
 import {
   bootstrapMailbox,
   listMailServerProfiles,
+  lookupMailAddresses,
   type MailMailboxMessageSummary
 } from "../../api/mail";
+import { mailLookupSuggestions, mailtoHref } from "./mailAddressIntegration";
 import {
   mailboxDraftsLaunchPath,
   mailboxMessageLaunchPath
@@ -30,10 +35,22 @@ type MailQuickAccessData = {
 
 type Props = Pick<
   QuickAccessToolRenderContext,
-  "settings" | "launchContext" | "close"
+  "settings" | "launchContext" | "complete" | "cancel" | "close"
 >;
 
-export default function MailQuickAccess({ settings, launchContext, close }: Props) {
+export default function MailQuickAccess({
+  settings,
+  launchContext,
+  complete,
+  cancel,
+  close
+}: Props) {
+  const [composing, setComposing] = useState(false);
+  const [recipients, setRecipients] = useState<MailboxAddress[]>([]);
+  const [suggestions, setSuggestions] = useState<MailboxAddress[]>([]);
+  const [lookupAvailable, setLookupAvailable] = useState<boolean | null>(null);
+  const [lookupError, setLookupError] = useState("");
+  const lookupRequestRef = useRef(0);
   const load = useCallback(async (): Promise<MailQuickAccessData> => {
     const profiles = await listMailServerProfiles(settings);
     const profile = profiles.find((item) => item.is_active && item.imap);
@@ -50,10 +67,39 @@ export default function MailQuickAccess({ settings, launchContext, close }: Prop
     };
   }, [settings]);
   const { data, loading, error } = useDashboardWidgetData(load, 0);
+  const selectingForCase = launchContext.activeObject?.ownerModule === "cases"
+    && launchContext.activeObject.kind === "case";
+
+  const lookupRecipients = useCallback(async (query: string) => {
+    const request = ++lookupRequestRef.current;
+    const normalized = query.trim();
+    if (!normalized) {
+      setSuggestions([]);
+      setLookupError("");
+      return;
+    }
+    try {
+      const response = await lookupMailAddresses(settings, normalized, 12);
+      if (request !== lookupRequestRef.current) return;
+      setLookupAvailable(response.available);
+      setSuggestions(mailLookupSuggestions(response.candidates));
+      setLookupError("");
+    } catch (lookupFailure) {
+      if (request !== lookupRequestRef.current) return;
+      setSuggestions([]);
+      setLookupError(lookupFailure instanceof Error ? lookupFailure.message : String(lookupFailure));
+    }
+  }, [settings]);
 
   return (
     <LoadingFrame loading={loading} label="i18n:govoplan-mail.loading_messages.4294022c">
       {error ? <DismissibleAlert tone="warning" resetKey={error}>{error}</DismissibleAlert> : null}
+      {selectingForCase ? (
+        <p className="muted small-note">
+          Select an authorized exact message for {launchContext.activeObject?.label}.
+          Mail content remains in Mail and access is checked again when opened.
+        </p>
+      ) : null}
       <DashboardWidgetList
         emptyText={data?.available ? "i18n:govoplan-mail.no_messages_in_this_folder.5c7fa25d" : "i18n:govoplan-mail.select_an_imap_profile.5445648c"}
         items={(data?.messages ?? []).map((message) => ({
@@ -62,15 +108,63 @@ export default function MailQuickAccess({ settings, launchContext, close }: Prop
           detail: message.from_header || data?.profileName,
           meta: formatMessageDate(message.date),
           leading: <Mail size={17} aria-hidden="true" />,
-          to: mailboxMessageLaunchPath(data!.profileId!, message),
-          state: quickAccessLaunchState(launchContext),
-          onClick: close
+          to: selectingForCase
+            ? undefined
+            : mailboxMessageLaunchPath(data!.profileId!, message),
+          state: selectingForCase ? undefined : quickAccessLaunchState(launchContext),
+          onClick: selectingForCase ? () => complete({
+            contractVersion: "1",
+            outcome: "completed",
+            action: "selected",
+            reference: {
+              ownerModule: "mail",
+              kind: "message",
+              objectId: `${data!.profileId!}:${message.folder}:${message.uid}`,
+              tenantId: launchContext.tenantId,
+              label: message.subject || "Mail message",
+              version: message.uid,
+              path: mailboxMessageLaunchPath(data!.profileId!, message)
+            }
+          }) : close
         }))}
       />
-      <div className="dashboard-contribution-footer">
-        <a className="btn btn-secondary" href="mailto:" onClick={close}>
-          <Pencil size={15} aria-hidden="true" /> i18n:govoplan-mail.compose
+      {!selectingForCase && composing ? (
+        <div className="mail-quick-compose" aria-label="i18n:govoplan-mail.compose">
+          <label>i18n:govoplan-mail.recipients</label>
+          <EmailAddressInput
+            value={recipients}
+            onChange={setRecipients}
+            suggestions={suggestions}
+            onSuggestionQueryChange={(query) => void lookupRecipients(query)}
+            compact
+            interfaceId="mail.quick-access.compose.recipients"
+            helpModuleId="mail"
+            helpTopicId="mail.address-book-integration"
+          />
+          {lookupAvailable === false ? (
+            <p className="form-help">i18n:govoplan-mail.address_suggestions_unavailable</p>
+          ) : null}
+          {lookupError ? <DismissibleAlert tone="warning" resetKey={lookupError}>{lookupError}</DismissibleAlert> : null}
+          <div className="button-row compact-actions">
+            <button type="button" className="btn btn-secondary" onClick={() => setComposing(false)}>
+              i18n:govoplan-mail.cancel.77dfd213
+            </button>
+            <a className="btn btn-primary" href={mailtoHref(recipients)} onClick={close}>
+              <Pencil size={15} aria-hidden="true" /> i18n:govoplan-mail.open_mail_application
             </a>
+          </div>
+        </div>
+      ) : null}
+      {selectingForCase ? (
+        <div className="dashboard-contribution-footer">
+          <Button onClick={() => cancel("user")}>
+            <X size={15} aria-hidden="true" /> Cancel selection
+          </Button>
+        </div>
+      ) : <div className="dashboard-contribution-footer">
+        <button type="button" className="btn btn-secondary" onClick={() => setComposing((current) => !current)} aria-expanded={composing}>
+          <Pencil size={15} aria-hidden="true" /> i18n:govoplan-mail.compose
+        </button>
         {data?.profileId && data.draftsFolder ? (
           <Link
             className="btn btn-secondary"
@@ -89,7 +183,7 @@ export default function MailQuickAccess({ settings, launchContext, close }: Prop
         >
           <ExternalLink size={15} aria-hidden="true" /> i18n:govoplan-mail.open_mail
         </Link>
-      </div>
+      </div>}
     </LoadingFrame>
   );
 }

webui/src/features/mail/MailboxPage.tsx

@@ -1,5 +1,5 @@
 import { useEffect, useMemo, useRef, useState } from "react";
-import { Activity, ChevronRight, Database, Home, Mail, MailOpen, Paperclip, RefreshCw, Search, X } from "lucide-react";
+import { Activity, Check, ChevronRight, Database, Home, Mail, MailOpen, Paperclip, RefreshCw, Search, UserPlus, X } from "lucide-react";
 import { useLocation } from "react-router";
 import { ToolbarGroup, ActionToolbar,
   ActionBlockerHint,
@@ -21,9 +21,12 @@ import { ToolbarGroup, ActionToolbar,
 } from "@govoplan/core-webui";
 import {
   bootstrapMailbox,
+  createMailAddressContact,
   getMailboxMessage,
+  listMailAddressWriteTargets,
   listMailboxMessages,
   listMailServerProfiles,
+  type MailAddressWriteTarget,
   type MailImapFolderResponse,
   type MailMailboxMessageDetail,
   type MailMailboxMessageSummary,
@@ -32,6 +35,7 @@ import {
 import { buildMailboxFolderTree, findFolderNodeId, folderAncestorIds, type MailFolderNode } from "./mailboxFolders";
 import { isMailboxMessageRead, mailboxSyncState, type MailboxSyncProvenance } from "./mailboxDisplay";
 import { mailboxLaunchFolder, parseMailboxLaunch, type MailboxLaunch } from "./mailboxLaunch";
+import { mailboxHeaderAddresses } from "./mailAddressIntegration";
 
 const MAILBOX_DOCUMENTATION = {
   topicId: "mail.workflow.read-mailbox",
@@ -628,6 +632,8 @@ export default function MailboxPage({ settings, auth }: { settings: ApiSettings;
               }))}
               emptyText={previewEmptyText} />
 
+            <MailboxContactActions settings={settings} message={selectedMessage} />
+
           </div>
         </section>
 
@@ -641,6 +647,146 @@ export default function MailboxPage({ settings, auth }: { settings: ApiSettings;
 
 }
 
+function MailboxContactActions({
+  settings,
+  message
+}: {
+  settings: ApiSettings;
+  message: MailMailboxMessageDetail | null;
+}) {
+  const [available, setAvailable] = useState(false);
+  const [loaded, setLoaded] = useState(false);
+  const [targets, setTargets] = useState<MailAddressWriteTarget[]>([]);
+  const [selectedTargetId, setSelectedTargetId] = useState("");
+  const [creatingEmail, setCreatingEmail] = useState("");
+  const [addedEmails, setAddedEmails] = useState<Set<string>>(() => new Set());
+  const [error, setError] = useState("");
+  const [success, setSuccess] = useState("");
+
+  useEffect(() => {
+    let active = true;
+    setLoaded(false);
+    void listMailAddressWriteTargets(settings)
+      .then((response) => {
+        if (!active) return;
+        const writable = response.targets.filter((target) => target.allowed);
+        setAvailable(response.available);
+        setTargets(response.targets);
+        setSelectedTargetId((current) => writable.some((target) => target.address_book_id === current)
+          ? current
+          : writable[0]?.address_book_id || "");
+        setError("");
+      })
+      .catch((loadError) => {
+        if (!active) return;
+        setAvailable(false);
+        setTargets([]);
+        setError(loadError instanceof Error ? loadError.message : String(loadError));
+      })
+      .finally(() => {
+        if (active) setLoaded(true);
+      });
+    return () => { active = false; };
+  }, [settings.apiBaseUrl, settings.apiKey, settings.accessToken]);
+
+  useEffect(() => {
+    setAddedEmails(new Set());
+    setError("");
+    setSuccess("");
+  }, [message?.folder, message?.uid]);
+
+  if (!message || !loaded || !available) return null;
+
+  const writableTargets = targets.filter((target) => target.allowed);
+  const blockedTargets = targets.filter((target) => !target.allowed);
+  const addresses = uniqueMailboxAddresses([
+    ...mailboxHeaderAddresses(message.from_header),
+    ...mailboxHeaderAddresses(message.to_header),
+    ...mailboxHeaderAddresses(message.cc_header)
+  ]);
+
+  async function addContact(address: { name?: string | null; email: string }) {
+    if (!selectedTargetId || creatingEmail) return;
+    setCreatingEmail(address.email);
+    setError("");
+    setSuccess("");
+    try {
+      const result = await createMailAddressContact(settings, {
+        address_book_id: selectedTargetId,
+        display_name: address.name || address.email,
+        email: address.email
+      });
+      setAddedEmails((current) => new Set(current).add(address.email));
+      setSuccess(i18nMessage("i18n:govoplan-mail.contact_added", { value0: result.display_name }));
+    } catch (createError) {
+      setError(createError instanceof Error ? createError.message : String(createError));
+    } finally {
+      setCreatingEmail("");
+    }
+  }
+
+  return (
+    <section className="mailbox-contact-actions" aria-label="i18n:govoplan-mail.address_book_actions">
+      <h4>i18n:govoplan-mail.address_book_actions</h4>
+      {error ? <DismissibleAlert tone="warning" resetKey={error}>{error}</DismissibleAlert> : null}
+      {success ? <DismissibleAlert tone="success" resetKey={success}>{success}</DismissibleAlert> : null}
+      {writableTargets.length > 0 ? (
+        <label className="mailbox-contact-target">
+          <span>i18n:govoplan-mail.save_contacts_to</span>
+          <select value={selectedTargetId} onChange={(event) => setSelectedTargetId(event.target.value)}>
+            {writableTargets.map((target) => (
+              <option key={target.address_book_id} value={target.address_book_id}>
+                {target.address_book_label || target.address_book_id}
+              </option>
+            ))}
+          </select>
+        </label>
+      ) : (
+        <p className="form-help">i18n:govoplan-mail.no_writable_address_book</p>
+      )}
+      <div className="mailbox-contact-candidates">
+        {addresses.map((address) => {
+          const added = addedEmails.has(address.email);
+          return (
+            <Button
+              key={address.email}
+              className="compact"
+              disabled={!selectedTargetId || Boolean(creatingEmail) || added}
+              disabledReason={!selectedTargetId ? blockedTargets[0]?.message || "i18n:govoplan-mail.no_writable_address_book" : undefined}
+              onClick={() => void addContact(address)}
+            >
+              {added ? <Check size={15} aria-hidden="true" /> : <UserPlus size={15} aria-hidden="true" />}
+              {added ? "i18n:govoplan-mail.contact_added_short" : i18nMessage("i18n:govoplan-mail.add_value_to_contacts", { value0: address.name || address.email })}
+            </Button>
+          );
+        })}
+      </div>
+      {blockedTargets.length > 0 ? (
+        <details className="mailbox-contact-policy">
+          <summary>i18n:govoplan-mail.unavailable_address_books</summary>
+          <ul>
+            {blockedTargets.map((target) => (
+              <li key={target.address_book_id}>
+                <strong>{target.address_book_label || target.address_book_id}</strong>: {target.message}
+              </li>
+            ))}
+          </ul>
+        </details>
+      ) : null}
+    </section>
+  );
+}
+
+function uniqueMailboxAddresses<T extends { email: string }>(addresses: T[]): T[] {
+  const seen = new Set<string>();
+  return addresses.filter((address) => {
+    const email = address.email.toLocaleLowerCase();
+    if (seen.has(email)) return false;
+    seen.add(email);
+    return true;
+  });
+}
+
 
 function mailboxMessageKey(folder: string, uid: string): string {
   return `${folder || "INBOX"}::${uid}`;

webui/src/features/mail/mailAddressIntegration.ts

@@ -0,0 +1,44 @@
+type MailAddressLookupCandidateLike = {
+  display_name: string;
+  email?: string | null;
+};
+
+export type MailAddressValue = {
+  name?: string | null;
+  email: string;
+};
+
+const EMAIL_PATTERN = /([^<>;,\s]+@[^<>;,\s]+)/g;
+
+export function mailLookupSuggestions(candidates: readonly MailAddressLookupCandidateLike[]): MailAddressValue[] {
+  const seen = new Set<string>();
+  const suggestions: MailAddressValue[] = [];
+  for (const candidate of candidates) {
+    const email = String(candidate.email ?? "").trim().toLocaleLowerCase();
+    if (!email || seen.has(email)) continue;
+    seen.add(email);
+    suggestions.push({ name: candidate.display_name || email, email });
+  }
+  return suggestions;
+}
+
+export function mailboxHeaderAddresses(value?: string | null): MailAddressValue[] {
+  const input = String(value ?? "").trim();
+  if (!input) return [];
+  const results: MailAddressValue[] = [];
+  const seen = new Set<string>();
+  for (const match of input.matchAll(EMAIL_PATTERN)) {
+    const email = match[1]?.replace(/[)>]+$/, "").toLocaleLowerCase();
+    if (!email || seen.has(email)) continue;
+    seen.add(email);
+    const prefix = input.slice(Math.max(0, input.lastIndexOf(",", match.index) + 1), match.index).trim();
+    const name = prefix.replace(/[<"']/g, "").trim() || undefined;
+    results.push({ name, email });
+  }
+  return results;
+}
+
+export function mailtoHref(recipients: readonly MailAddressValue[]): string {
+  const addresses = recipients.map((recipient) => recipient.email.trim()).filter(Boolean);
+  return `mailto:${addresses.map(encodeURIComponent).join(",")}`;
+}

webui/src/i18n/generatedTranslations.ts

@@ -96,6 +96,16 @@ export const generatedTranslations: PlatformTranslations = {
     "i18n:govoplan-mail.mail_server_profiles.b1726682": "Mail server profiles",
     "i18n:govoplan-mail.mail.92379cbb": "Mail",
     "i18n:govoplan-mail.compose": "Compose",
+    "i18n:govoplan-mail.recipients": "Recipients",
+    "i18n:govoplan-mail.address_suggestions_unavailable": "Address-book suggestions are unavailable. You can still enter an email address manually.",
+    "i18n:govoplan-mail.open_mail_application": "Open mail application",
+    "i18n:govoplan-mail.address_book_actions": "Address-book actions",
+    "i18n:govoplan-mail.save_contacts_to": "Save contacts to",
+    "i18n:govoplan-mail.no_writable_address_book": "No writable address book is available for your account.",
+    "i18n:govoplan-mail.contact_added": "{value0} was added to contacts.",
+    "i18n:govoplan-mail.contact_added_short": "Added",
+    "i18n:govoplan-mail.add_value_to_contacts": "Add {value0} to contacts",
+    "i18n:govoplan-mail.unavailable_address_books": "Unavailable address books and policy reasons",
     "i18n:govoplan-mail.open_mail": "Open Mail",
     "i18n:govoplan-mail.quick_access_description": "Recent mailbox messages and mail actions.",
     "i18n:govoplan-mail.mailbox_folders_could_not_be_loaded.c3e3880e": "Mailbox folders could not be loaded.",
@@ -296,6 +306,16 @@ export const generatedTranslations: PlatformTranslations = {
     "i18n:govoplan-mail.mail_server_profiles.b1726682": "Mail server profiles",
     "i18n:govoplan-mail.mail.92379cbb": "Mail",
     "i18n:govoplan-mail.compose": "Verfassen",
+    "i18n:govoplan-mail.recipients": "Empfänger",
+    "i18n:govoplan-mail.address_suggestions_unavailable": "Adressbuchvorschläge sind nicht verfügbar. Eine E-Mail-Adresse kann weiterhin manuell eingegeben werden.",
+    "i18n:govoplan-mail.open_mail_application": "Mail-Anwendung öffnen",
+    "i18n:govoplan-mail.address_book_actions": "Adressbuchaktionen",
+    "i18n:govoplan-mail.save_contacts_to": "Kontakte speichern in",
+    "i18n:govoplan-mail.no_writable_address_book": "Für dieses Konto ist kein beschreibbares Adressbuch verfügbar.",
+    "i18n:govoplan-mail.contact_added": "{value0} wurde zu den Kontakten hinzugefügt.",
+    "i18n:govoplan-mail.contact_added_short": "Hinzugefügt",
+    "i18n:govoplan-mail.add_value_to_contacts": "{value0} zu Kontakten hinzufügen",
+    "i18n:govoplan-mail.unavailable_address_books": "Nicht verfügbare Adressbücher und Richtliniengründe",
     "i18n:govoplan-mail.open_mail": "Mail öffnen",
     "i18n:govoplan-mail.quick_access_description": "Aktuelle Posteingangsnachrichten und Mail-Aktionen.",
     "i18n:govoplan-mail.mailbox_folders_could_not_be_loaded.c3e3880e": "Mailbox folders could not be loaded.",

webui/src/styles/mail-profiles.css

@@ -627,6 +627,56 @@
   padding: 18px;
 }
 
+.mail-quick-compose {
+  display: grid;
+  gap: 10px;
+  padding: 12px;
+  margin-top: 10px;
+  border: var(--border-line);
+  border-radius: var(--radius-md);
+  background: var(--panel-soft);
+}
+
+.mail-quick-compose > label,
+.mailbox-contact-target > span {
+  color: var(--text-strong);
+  font-size: 12px;
+  font-weight: 700;
+}
+
+.mailbox-contact-actions {
+  display: grid;
+  gap: 10px;
+  margin-top: 16px;
+  padding-top: 16px;
+  border-top: var(--border-line);
+}
+
+.mailbox-contact-actions h4 {
+  margin: 0;
+}
+
+.mailbox-contact-target {
+  display: grid;
+  gap: 5px;
+}
+
+.mailbox-contact-candidates {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 8px;
+}
+
+.mailbox-contact-policy {
+  color: var(--muted);
+  font-size: 12px;
+}
+
+.mailbox-contact-policy ul {
+  margin: 8px 0 0;
+  padding-left: 20px;
+}
+
 @media (max-width: 1280px) {
   .mailbox-shell.file-manager-shell {
     grid-template-columns: minmax(220px, 280px) minmax(0, 1fr);

webui/tests/mail-address-integration.test.ts

@@ -0,0 +1,51 @@
+import {
+  mailboxHeaderAddresses,
+  mailLookupSuggestions,
+  mailtoHref
+} from "../src/features/mail/mailAddressIntegration";
+
+function assertEqual(actual: unknown, expected: unknown): void {
+  if (actual !== expected) throw new Error(`expected ${String(expected)}, got ${String(actual)}`);
+}
+
+function assertDeepEqual(actual: unknown, expected: unknown): void {
+  const actualJson = JSON.stringify(actual);
+  const expectedJson = JSON.stringify(expected);
+  if (actualJson !== expectedJson) throw new Error(`expected ${expectedJson}, got ${actualJson}`);
+}
+
+assertDeepEqual(
+  mailLookupSuggestions([
+    {
+      display_name: "Ada Lovelace",
+      email: "Ada@Example.Test"
+    },
+    {
+      display_name: "Duplicate",
+      email: "ada@example.test"
+    },
+    {
+      display_name: "No email",
+      email: null
+    }
+  ]),
+  [{ name: "Ada Lovelace", email: "ada@example.test" }]
+);
+
+assertDeepEqual(
+  mailboxHeaderAddresses('Ada Lovelace <ada@example.test>, "Grace Hopper" <grace@example.test>'),
+  [
+    { name: "Ada Lovelace", email: "ada@example.test" },
+    { name: "Grace Hopper", email: "grace@example.test" }
+  ]
+);
+
+assertEqual(
+  mailtoHref([
+    { name: "Ada Lovelace", email: "ada@example.test" },
+    { email: "grace@example.test" }
+  ]),
+  "mailto:ada%40example.test,grace%40example.test"
+);
+
+console.log("mail address integration tests passed");

webui/tsconfig.mail-tests.json

@@ -22,10 +22,12 @@
     "tests/mailbox-launch.test.ts",
     "tests/mail-profile-editor-model.test.ts",
     "tests/mail-policy-validation.test.ts",
+    "tests/mail-address-integration.test.ts",
     "src/features/mail/mailboxDisplay.ts",
     "src/features/mail/mailboxFolders.ts",
     "src/features/mail/mailboxLaunch.ts",
     "src/features/mail/mailProfileEditorModel.ts",
-    "src/features/mail/mailPolicyValidation.ts"
+    "src/features/mail/mailPolicyValidation.ts",
+    "src/features/mail/mailAddressIntegration.ts"
   ]
 }