diff --git a/compose.yaml b/compose.yaml
index d4ebaabc..394b4ab1 100644
--- a/compose.yaml
+++ b/compose.yaml
@@ -617,7 +617,19 @@ services:
       - ./src/jmap-email/jmap_email:/app/jmap_email
       - ./src/jmap-email/tests:/app/tests
       - ./src/jmap-email/pyproject.toml:/app/pyproject.toml
+      # Hypothesis' example database. Without it the container starts with
+      # no memory of past failures, so Phase.reuse has nothing to replay
+      # and an intermittent find stays intermittent.
+      #
+      # A named volume rather than a bind mount: a bind mount to a path
+      # that does not exist yet — and it never does on a fresh clone,
+      # since .hypothesis is gitignored — is created by Docker as *root*,
+      # which then blocks `git clean -fdx` and any pytest run against a
+      # local venv. This is a cache, so nothing needs to read it from the
+      # host.
+      - jmap-email-hypothesis:/app/.hypothesis
 
 volumes:
+  jmap-email-hypothesis:
   objectstorage-data:
   frontend-node-modules:
diff --git a/src/backend/core/admin.py b/src/backend/core/admin.py
index d3bdd84a..2fc363e1 100644
--- a/src/backend/core/admin.py
+++ b/src/backend/core/admin.py
@@ -3,7 +3,6 @@
 
 import json
 import logging
-import mimetypes
 
 from django import forms
 from django.contrib import admin, messages
@@ -1379,8 +1378,14 @@ class BlobAdmin(admin.ModelAdmin):
             # POST-only download endpoint and 405.
             return redirect("..")
 
-        extension = mimetypes.guess_extension(blob.content_type or "") or ""
-        filename = f"blob-{blob.id}{extension}"
+        # No extension. This is a debugging download of an opaque blob, not
+        # a user-facing attachment: the ``Content-Type`` below already tells
+        # the browser what it is, and the blob id is the only part of the
+        # name anyone needs. Inferring one would either re-import the
+        # stdlib ``mimetypes`` table (whose answers vary with the host
+        # image) or reuse the attachment table (deliberately short, so most
+        # types would come back bare anyway).
+        filename = f"blob-{blob.id}"
         response = HttpResponse(
             content, content_type=blob.content_type or "application/octet-stream"
         )
diff --git a/src/backend/core/api/serializers.py b/src/backend/core/api/serializers.py
index 8a18500f..f4787954 100644
--- a/src/backend/core/api/serializers.py
+++ b/src/backend/core/api/serializers.py
@@ -21,6 +21,7 @@ from core.mda.dispatch_webhooks import (
     VALID_FORMATS,
 )
 from core.mda.inline_images import extract_inline_images_html
+from core.services.attachments import get_attachment_display_name
 from core.services.blob_gc import schedule_for_gc
 from core.services.identity import keycloak as keycloak_service
 from core.services.importer.channel import merged_state
@@ -1191,7 +1192,9 @@ class MessageSerializer(serializers.ModelSerializer):
                 stripped_attachments.append(
                     {
                         "blobId": f"msg_{instance.id}_{index}",
-                        "name": attachment.get("name") or "unnamed",
+                        "name": get_attachment_display_name(
+                            attachment.get("name"), attachment.get("type")
+                        ),
                         "size": attachment["size"],
                         "type": attachment["type"],
                         "cid": attachment.get("cid"),
diff --git a/src/backend/core/api/viewsets/blob.py b/src/backend/core/api/viewsets/blob.py
index a169a930..d23bbf92 100644
--- a/src/backend/core/api/viewsets/blob.py
+++ b/src/backend/core/api/viewsets/blob.py
@@ -24,6 +24,7 @@ from rest_framework.viewsets import ViewSet
 
 from core import enums, models
 from core.api import permissions, utils
+from core.services.attachments import get_attachment_display_name
 from core.services.blob_gc import upload_and_reserve_blob
 
 # Number of leading bytes inspected by python-magic on the preview endpoint.
@@ -185,7 +186,7 @@ class BlobViewSet(ViewSet):
 
         Returns:
             A dict with keys `content` (bytes), `declared_type` (str),
-            `filename` (str), `size` (int).
+            `filename` (str, never empty), `size` (int).
 
         Raises:
             ParseError: malformed `msg_*` ID.
@@ -202,7 +203,14 @@ class BlobViewSet(ViewSet):
             return {
                 "content": attachment["content"],
                 "declared_type": attachment["type"],
-                "filename": attachment["name"],
+                # The parser reports ``None`` for a part with no filename.
+                # Synthesize the same display name the message serializer
+                # exposes for it, so the download lands under the name the
+                # user was shown — and so ``Content-Disposition`` is built
+                # from a real string either way.
+                "filename": get_attachment_display_name(
+                    attachment["name"], attachment["type"]
+                ),
                 "size": attachment["size"],
             }
 
diff --git a/src/backend/core/api/viewsets/drive.py b/src/backend/core/api/viewsets/drive.py
index 7cf7da85..ae7483f5 100644
--- a/src/backend/core/api/viewsets/drive.py
+++ b/src/backend/core/api/viewsets/drive.py
@@ -23,6 +23,7 @@ from rest_framework.views import APIView
 from core import models
 from core.api import utils
 from core.api.serializers import PartialDriveItemSerializer
+from core.services.attachments import get_attachment_display_name
 
 logger = logging.getLogger(__name__)
 
@@ -155,9 +156,19 @@ class DriveAPIView(APIView):
             "Content-Type": "application/json",
         }
 
+        # The parser reports ``None`` for a part with no filename. Resolve
+        # the same display name the message serializer and blob download
+        # expose, so the Drive dedup search keeps its ``title`` filter (a
+        # ``None`` param would be dropped by ``requests``, matching an
+        # unrelated file on size alone) and the created file carries a
+        # real filename.
+        filename = get_attachment_display_name(attachment["name"], attachment["type"])
+
         # Check if file already exists in Drive (get_or_create pattern)
         try:
-            existing_item = self._find_existing_drive_item(attachment, auth_headers)
+            existing_item = self._find_existing_drive_item(
+                attachment, auth_headers, filename
+            )
         except requests.exceptions.RequestException:
             logger.exception("Failed to search Drive for existing file")
             return Response(
@@ -170,7 +181,7 @@ class DriveAPIView(APIView):
 
         # File doesn't exist, create it
         try:
-            return self._create_drive_item(attachment, auth_headers)
+            return self._create_drive_item(attachment, auth_headers, filename)
         except requests.exceptions.RequestException:
             logger.exception("Failed to create file in Drive")
             return Response(
@@ -178,7 +189,7 @@ class DriveAPIView(APIView):
                 data={"error": "Failed to create file in Drive"},
             )
 
-    def _find_existing_drive_item(self, attachment, headers):
+    def _find_existing_drive_item(self, attachment, headers, filename):
         """Search for an existing file in Drive matching the attachment name and size.
 
         Raises RequestException on network/server errors so callers don't
@@ -189,7 +200,7 @@ class DriveAPIView(APIView):
             params={
                 "is_creator_me": True,
                 "type": "file",
-                "title": attachment["name"],
+                "title": filename,
             },
             headers=headers,
             timeout=5,
@@ -202,13 +213,13 @@ class DriveAPIView(APIView):
 
         return None
 
-    def _create_drive_item(self, attachment, headers):
+    def _create_drive_item(self, attachment, headers, filename):
         """Create a new file in Drive and upload its content."""
         response = requests.post(
             f"{self.drive_external_api}/items/",
             json={
                 "type": "file",
-                "filename": attachment["name"],
+                "filename": filename,
             },
             headers=headers,
             timeout=5,
diff --git a/src/backend/core/api/viewsets/inbound/widget.py b/src/backend/core/api/viewsets/inbound/widget.py
index 80a63d63..f268b52f 100644
--- a/src/backend/core/api/viewsets/inbound/widget.py
+++ b/src/backend/core/api/viewsets/inbound/widget.py
@@ -7,7 +7,7 @@ from urllib.parse import urlparse
 from django.conf import settings
 
 from drf_spectacular.utils import extend_schema
-from jmap_email import compose_email, parse_address
+from jmap_email import ComposeError, compose_email, parse_address
 from rest_framework import status, viewsets
 from rest_framework.authentication import BaseAuthentication
 from rest_framework.decorators import action
@@ -18,7 +18,7 @@ from rest_framework.throttling import SimpleRateThrottle
 from core import enums, models
 from core.api.permissions import IsAuthenticated
 from core.mda.inbound import deliver_inbound_message
-from core.mda.utils import current_sent_at
+from core.mda.utils import COMPOSE_OPTIONS, current_sent_at
 
 logger = logging.getLogger(__name__)
 
@@ -240,10 +240,31 @@ class InboundWidgetViewSet(viewsets.GenericViewSet):
             "textBody": [{"content": message_text}],
         }
 
+        # ``parse_address`` above accepts an RFC 6531 address — it is a
+        # valid one — but the composer cannot carry a non-ASCII local part
+        # over the 7-bit SMTP we emit, and refuses. That is a property of
+        # the address this caller typed into a public form, so it is a 400
+        # like the shape check above, not a 500.
+        try:
+            raw_mime = compose_email(
+                parsed_email,
+                prepend_headers=prepend_headers,
+                options=COMPOSE_OPTIONS,
+            )
+        except ComposeError:
+            logger.info(
+                "Widget submission rejected: cannot compose MIME for sender "
+                "at domain %r",
+                sender_email.split("@", 1)[-1],
+            )
+            return Response(
+                {"detail": "Invalid email format"}, status=status.HTTP_400_BAD_REQUEST
+            )
+
         delivered = deliver_inbound_message(
             target_email,
             parsed_email,
-            compose_email(parsed_email, prepend_headers=prepend_headers),
+            raw_mime,
             channel=channel,
             envelope={
                 "origin": enums.InboundOrigin.WIDGET,
diff --git a/src/backend/core/api/viewsets/send.py b/src/backend/core/api/viewsets/send.py
index e0cb9ee8..08479e69 100644
--- a/src/backend/core/api/viewsets/send.py
+++ b/src/backend/core/api/viewsets/send.py
@@ -10,6 +10,7 @@ from drf_spectacular.utils import (
     extend_schema,
     inline_serializer,
 )
+from jmap_email import ComposeError
 from rest_framework import exceptions as drf_exceptions
 from rest_framework import serializers as drf_serializers
 from rest_framework import status
@@ -147,13 +148,28 @@ class SendMessageView(APIView):
                     "You do not have permission to send as this mailbox."
                 )
 
-            prepared = prepare_outbound_message(
-                mailbox_sender,
-                message,
-                request.data.get("textBody"),
-                request.data.get("htmlBody"),
-                request.user,
-            )
+            # A recipient the composer cannot put on the wire — a non-ASCII
+            # local part needing SMTPUTF8, a malformed addr-spec — is a
+            # property of the draft, not a server fault, so it is a 400
+            # rather than the 500 an escaping ComposeError would give.
+            try:
+                prepared = prepare_outbound_message(
+                    mailbox_sender,
+                    message,
+                    request.data.get("textBody"),
+                    request.data.get("htmlBody"),
+                    request.user,
+                )
+            except ComposeError as e:
+                logger.info(
+                    "Send rejected for message %s: cannot compose MIME (%s)",
+                    message_id,
+                    type(e).__name__,
+                )
+                raise drf_exceptions.ValidationError(
+                    "This message cannot be sent: one of its addresses or "
+                    "attachments cannot be represented on the wire."
+                ) from e
             if not prepared:
                 raise drf_exceptions.APIException(
                     "Failed to prepare message for sending.",
diff --git a/src/backend/core/management/commands/send_mail.py b/src/backend/core/management/commands/send_mail.py
index bf457106..63a9e4d8 100644
--- a/src/backend/core/management/commands/send_mail.py
+++ b/src/backend/core/management/commands/send_mail.py
@@ -18,12 +18,12 @@ import logging
 
 from django.core.management.base import BaseCommand, CommandError
 
-from jmap_email import compose_email, parse_address
+from jmap_email import ComposeError, compose_email, parse_address
 
 from core import models
 from core.mda.outbound import send_outbound_email
 from core.mda.signing import sign_message_dkim
-from core.mda.utils import current_sent_at, generate_mime_id
+from core.mda.utils import COMPOSE_OPTIONS, current_sent_at, generate_mime_id
 
 logger = logging.getLogger(__name__)
 
@@ -136,8 +136,12 @@ class Command(BaseCommand):
             "messageId": [mime_id],
         }
 
-        # Compose the email
-        raw_mime = compose_email(mime_data)
+        # Compose the email. A malformed addr-spec surfaces here rather
+        # than at parse time, so it gets the same CommandError treatment.
+        try:
+            raw_mime = compose_email(mime_data, options=COMPOSE_OPTIONS)
+        except ComposeError as e:
+            raise CommandError(f"Cannot compose message: {e}") from e
 
         # Sign the message with DKIM (only if mailbox exists)
         dkim_signature_header = None
diff --git a/src/backend/core/mda/draft.py b/src/backend/core/mda/draft.py
index 5fad83cc..25c32623 100644
--- a/src/backend/core/mda/draft.py
+++ b/src/backend/core/mda/draft.py
@@ -12,6 +12,10 @@ import rest_framework as drf
 
 from core import enums, models
 from core.api.utils import get_attachment_from_blob_id
+from core.services.attachments import (
+    UNNAMED_ATTACHMENT_STEM,
+    get_attachment_display_name,
+)
 from core.services.blob_gc import release_upload, schedule_for_gc
 
 logger = logging.getLogger(__name__)
@@ -88,7 +92,7 @@ def _get_or_create_attachment_from_message_blob(
         The created Attachment or None if processing failed
     """
     blob_id = attachment_data.get("blobId")
-    name = attachment_data.get("name", "unnamed")
+    name = attachment_data.get("name")
     cid = attachment_data.get("cid")
 
     try:
@@ -99,9 +103,17 @@ def _get_or_create_attachment_from_message_blob(
         if not cid:
             cid = parsed_attachment.get("cid")
 
-        # Use name from parsed attachment if not provided
-        if name == "unnamed":
-            name = parsed_attachment.get("name", "unnamed")
+        # Forwarding round-trips our own serializer output back to us.
+        # No name, or the bare placeholder stem, both mean "none supplied".
+        # Guarded by ``not parsed_name``: "unnamed" is also a real filename.
+        parsed_name = parsed_attachment.get("name")
+        if not name or (name == UNNAMED_ATTACHMENT_STEM and not parsed_name):
+            name = parsed_name
+
+        # Sanitizes and, for a part with no name at all, synthesizes one
+        # with an extension inferred from the MIME type. ``Attachment.name``
+        # is NOT NULL, so this is what keeps the save from failing.
+        name = get_attachment_display_name(name, parsed_attachment.get("type"))
 
         # Atomic: the Blob INSERT and the Attachment INSERT must be
         # visible together so the GC sweep never sees the blob row
@@ -158,7 +170,7 @@ def _get_or_create_attachment_from_blob(
         The created/existing Attachment or None if processing failed
     """
     blob_id = attachment_data.get("blobId")
-    name = attachment_data.get("name", "unnamed")
+    name = attachment_data.get("name")
     cid = attachment_data.get("cid")
 
     try:
@@ -169,6 +181,8 @@ def _get_or_create_attachment_from_blob(
         # Try to get the blob
         blob = models.Blob.objects.get(id=blob_id)
 
+        name = get_attachment_display_name(name, blob.content_type)
+
         # Provenance check: the user attaching this blob must have
         # either an active upload reservation tied to this mailbox
         # (the JMAP upload-then-attach window) or an existing
diff --git a/src/backend/core/mda/inbound_tasks.py b/src/backend/core/mda/inbound_tasks.py
index d6d71b9b..3bb514c4 100644
--- a/src/backend/core/mda/inbound_tasks.py
+++ b/src/backend/core/mda/inbound_tasks.py
@@ -8,11 +8,13 @@ iterate, and turn the final ``Decision`` into a task return value.
 
 # pylint: disable=unused-argument, broad-exception-raised, broad-exception-caught
 
+from datetime import timedelta
 from typing import Any, Dict, Optional
 
 from django.conf import settings
 from django.core.cache import cache
 from django.db import transaction
+from django.db.models import Q
 from django.utils import timezone
 
 from celery.exceptions import SoftTimeLimitExceeded
@@ -66,6 +68,49 @@ _INBOUND_TASK_SOFT_TIME_LIMIT = max(_INBOUND_TASK_TIME_LIMIT - 60, 1)
 # sweep can retry.
 _INBOUND_TASK_LOCK_TTL = _INBOUND_TASK_TIME_LIMIT + 60
 
+# How often a held row is re-attempted, as a function of how long it has
+# already been failing: ``(minimum age, interval between attempts)``, in
+# increasing age order. A row picks the last band whose minimum age it has
+# reached.
+#
+# Both inputs are existing columns — ``created_at`` for the age, ``updated_at``
+# for the previous attempt (every attempt stamps it) — so this needs no schema
+# change. Outbound does the same thing with an explicit ``retry_count`` /
+# ``retry_at`` pair on the recipient row (``RETRY_INTERVALS`` in
+# ``outbound.py``); deriving the interval from age instead of counting attempts
+# gives the same spacing without the columns.
+#
+# The point is that a dependency outage is not cheaper to survive by asking
+# more often. A flat retry every 5 minutes turns one degraded webhook provider
+# into ~600 pipeline runs per message over the deferral window, each re-POSTing
+# to the same dead endpoint; the schedule below covers 48h in ~28 attempts.
+_RETRY_BACKOFF = [
+    (timedelta(0), timedelta(minutes=5)),
+    (timedelta(minutes=30), timedelta(minutes=15)),
+    (timedelta(hours=2), timedelta(hours=1)),
+    (timedelta(hours=8), timedelta(hours=4)),
+]
+
+
+def _due_for_retry_q(now) -> Q:
+    """Rows whose next attempt is due, per ``_RETRY_BACKOFF``.
+
+    One OR'd band per entry, each bounded to its own age range so the bands
+    are disjoint and a row matches exactly one. A band is satisfied when the
+    row's last attempt (``updated_at``) is at least that band's interval old.
+
+    A freshly-queued row has ``updated_at == created_at``, so the first band
+    also supplies the "don't touch it for 5 minutes" grace the immediate
+    ``.delay()`` dispatch needs to do its work unraced.
+    """
+    due = Q()
+    for index, (min_age, interval) in enumerate(_RETRY_BACKOFF):
+        band = Q(created_at__lte=now - min_age, updated_at__lte=now - interval)
+        if index + 1 < len(_RETRY_BACKOFF):
+            band &= Q(created_at__gt=now - _RETRY_BACKOFF[index + 1][0])
+        due |= band
+    return due
+
 
 def _is_selfcheck(parsed_email: JmapEmail, recipient_email: str) -> bool:
     """Strict envelope match for the configured self-probe.
@@ -192,8 +237,18 @@ def _retry_or_abandon(
         inbound_message.id,
         age,
     )
-    # Keep the row (and its bytes) — stamp it terminally failed so the sweep
-    # skips it instead of deleting and losing the only copy of the mail.
+    return _abandon(inbound_message, reason)
+
+
+def _abandon(inbound_message: models.InboundMessage, reason: str) -> Dict[str, Any]:
+    """Stamp a message terminally failed so the sweep stops retrying it.
+
+    Keeps the row and its bytes — the ``blob`` is the only copy of the
+    mail, so an operator can still inspect and replay it from the admin.
+
+    Called directly rather than via ``_retry_or_abandon`` for
+    deterministic failures, which no number of retries can clear.
+    """
     inbound_message.error_message = reason
     inbound_message.abandoned_at = timezone.now()
     inbound_message.save(update_fields=["error_message", "abandoned_at", "updated_at"])
@@ -264,13 +319,28 @@ def process_inbound_message_task(self, inbound_message_id: str):
                 "error": "abandoned",
             }
 
+        # Stamp the attempt before doing any work. ``updated_at`` is what the
+        # sweep backs off from, so a run that outlives a sweep interval (the
+        # soft limit is 9 min, the interval 5) must not be selected again while
+        # it is still going: those dispatches would only bounce off the lock
+        # above and burn the batch. Costs one UPDATE on the happy path, on a
+        # row that is about to be deleted anyway.
+        inbound_message.save(update_fields=["updated_at"])
+
         raw_data_bytes = inbound_message.get_raw_bytes()
         parsed_email = parse_email(raw_data_bytes)
         if parsed_email is None:
-            # A deterministic parse failure never succeeds on retry —
-            # route through ``_retry_or_abandon`` so it's bounded by the
-            # deferral window instead of looping on every 5-min sweep.
-            return _retry_or_abandon(inbound_message, "Failed to parse email message")
+            # Deterministic: the parse is a pure function of the stored
+            # bytes, so deferring would just repeat it for 48h.
+            #
+            # ``error`` rather than ``warning``: abandoned rows are hard
+            # deleted by the 7-day purge, so a parser regression would
+            # otherwise drop real mail with no alert.
+            logger.error(
+                "Inbound message %s could not be parsed; abandoning (not retryable)",
+                inbound_message.id,
+            )
+            return _abandon(inbound_message, "Failed to parse email message")
 
         mailbox = inbound_message.mailbox
         recipient_email = str(mailbox)
@@ -598,52 +668,83 @@ def process_inbound_message_task(self, inbound_message_id: str):
 
 
 @celery_app.task(bind=True)
-def process_inbound_messages_queue_task(self, batch_size: int = 10):
-    """Retry processing of inbound messages that are older than 5 minutes.
+def process_inbound_messages_queue_task(
+    self, chunk_size: int = 500, max_dispatch: int = 2000
+):
+    """Re-dispatch inbound messages whose next retry is due.
 
     This task only handles retries for messages that may have failed or gotten stuck.
     Regular messages are processed immediately when created via process_inbound_message_task.delay().
 
+    Due-ness is per ``_RETRY_BACKOFF``, so a row that keeps failing is asked
+    less and less often. Ordering by ``updated_at`` takes the
+    least-recently-attempted first: with a flat schedule and ``created_at``
+    ordering, a backlog larger than the batch let the oldest rows hold every
+    slot forever while newer ones aged out to the deferral window without ever
+    being retried once. It also means a run cut short by ``max_dispatch``
+    drops the rows that have waited least, never the same rows every tick.
+
+    Dispatches the whole due set rather than one truncated slice: a backlog
+    bigger than the slice used to cap retry throughput regardless of how much
+    capacity the workers had. ``max_dispatch`` bounds a single run so one tick
+    cannot enqueue an unbounded burst; hitting it is logged, never silent.
+
+    Streamed with ``.iterator()`` (server-side cursor) over a ``values_list``
+    of ids, so neither the row count nor the message size lands in worker RAM
+    — the same idiom as ``re_store_blobs``. One snapshot, so a row whose
+    ``updated_at`` is stamped by its worker mid-sweep can neither be handed
+    back nor skipped.
+
     Args:
-        batch_size: Number of messages to process in this batch
+        chunk_size: Rows fetched per round trip
+        max_dispatch: Maximum messages re-dispatched in one run
 
     Returns:
         dict: A dictionary with processing results
     """
-    # Only retry messages older than 5 minutes
-    retry_threshold = timezone.now() - timezone.timedelta(minutes=5)
-    old_messages = models.InboundMessage.objects.filter(
-        created_at__lt=retry_threshold,
-        # Terminally-failed rows are kept for inspection/replay but must
-        # not be retried — otherwise the poison message loops the pipeline
-        # (and re-fires every user webhook) every 5 minutes forever.
-        abandoned_at__isnull=True,
-    ).order_by("created_at")[:batch_size]
-
-    total = len(old_messages)
-    if total == 0:
-        return {
-            "success": True,
-            "processed": 0,
-            "total": 0,
-        }
+    due = (
+        models.InboundMessage.objects.filter(
+            _due_for_retry_q(timezone.now()),
+            # Terminally-failed rows are kept for inspection/replay but must
+            # not be retried — otherwise the poison message loops the pipeline
+            # (and re-fires every user webhook) every 5 minutes forever.
+            abandoned_at__isnull=True,
+        )
+        .order_by("updated_at")
+        .values_list("id", flat=True)[: max_dispatch + 1]
+    )
 
     processed = 0
     errors = 0
+    total = 0
+    capped = False
 
-    for inbound_message in old_messages:
+    for row_id in due.iterator(chunk_size=chunk_size):
+        if total >= max_dispatch:
+            # The extra row the slice asked for: more work is due than this
+            # run will take.
+            capped = True
+            break
+        total += 1
         try:
             # Trigger async task for each old message (retry)
-            process_inbound_message_task.delay(str(inbound_message.id))
+            process_inbound_message_task.delay(str(row_id))
             processed += 1
         except Exception as e:
             logger.exception(
                 "Error queuing inbound message %s for retry: %s",
-                inbound_message.id,
+                row_id,
                 e,
             )
             errors += 1
 
+    if capped:
+        logger.warning(
+            "Inbound retry sweep stopped at its %d-message cap — more were due; "
+            "the queue is not draining within one run",
+            max_dispatch,
+        )
+
     return {
         "success": True,
         "processed": processed,
diff --git a/src/backend/core/mda/inline_images.py b/src/backend/core/mda/inline_images.py
index 276ad944..6d95deff 100644
--- a/src/backend/core/mda/inline_images.py
+++ b/src/backend/core/mda/inline_images.py
@@ -19,6 +19,8 @@ import re
 import typing
 import uuid
 
+from core.services.attachments import guess_mime_extension
+
 logger = logging.getLogger(__name__)
 
 # Matches src="data:<mime>;base64,<data>" in HTML img tags
@@ -31,15 +33,6 @@ _MD_BASE64_IMG_RE = re.compile(
     r"(!\[[^\]]*\]\()data:(image/[a-zA-Z0-9.+-]+);base64,([A-Za-z0-9+/\n\r =]+)(\))"
 )
 
-# Common image MIME types → file extensions for the synthesized filename.
-_MIME_TO_EXT = {
-    "image/png": "png",
-    "image/jpeg": "jpg",
-    "image/gif": "gif",
-    "image/webp": "webp",
-    "image/svg+xml": "svg",
-}
-
 
 def _resolve_image(
     content: bytes,
@@ -60,8 +53,8 @@ def _resolve_image(
         return known_images[digest]
 
     cid = str(uuid.uuid4())
-    ext = _MIME_TO_EXT.get(content_type)
-    filename = f"{cid}.{ext}" if ext else cid
+    ext = guess_mime_extension(content_type)
+    filename = f"{cid}{ext}" if ext else cid
 
     images.append(
         {
diff --git a/src/backend/core/mda/outbound.py b/src/backend/core/mda/outbound.py
index 47eac9be..fe4b7f8a 100644
--- a/src/backend/core/mda/outbound.py
+++ b/src/backend/core/mda/outbound.py
@@ -13,6 +13,7 @@ from django.utils import timezone
 
 import rest_framework as drf
 from jmap_email import (
+    ComposeError,
     compose_email,
     find_header,
     first_address_email,
@@ -30,7 +31,7 @@ from core.mda.outbound_direct import send_message_via_mx
 from core.mda.replies import make_forward, make_reply
 from core.mda.signing import sign_message_dkim, verify_message_dkim
 from core.mda.smtp import send_smtp_mail
-from core.mda.utils import current_sent_at
+from core.mda.utils import COMPOSE_OPTIONS, current_sent_at
 from core.services.blob_gc import schedule_for_gc
 from core.services.dns.check import check_spf_status
 from core.services.throttle import check_and_increment_throttle
@@ -223,6 +224,7 @@ def compose_and_sign_mime(
         mime_data,
         in_reply_to=message.parent.mime_id if message.parent else None,
         prepend_headers=prepend_headers,
+        options=COMPOSE_OPTIONS,
     )
 
     # Bcc/Cc-only send: the composed MIME has no To header. Add the empty-group
@@ -451,6 +453,13 @@ def prepare_outbound_message(
             message.attachments.all().delete()
     except drf.exceptions.ValidationError:
         raise
+    except ComposeError:
+        # A draft the composer refuses (unrepresentable address,
+        # undecodable attachment) is the caller's to turn into a 4xx —
+        # swallowing it into ``False`` would surface as a 500. Re-raising
+        # before the broad handler also keeps the raw addr-spec embedded
+        # in the exception message out of logger.exception/Sentry.
+        raise
     except Exception:
         logger.exception("Failed to compose MIME for message %s", message.id)
         return False
diff --git a/src/backend/core/mda/selfcheck.py b/src/backend/core/mda/selfcheck.py
index 55c88b29..cfb8d7ce 100644
--- a/src/backend/core/mda/selfcheck.py
+++ b/src/backend/core/mda/selfcheck.py
@@ -10,7 +10,7 @@ from typing import Optional, Tuple
 from django.conf import settings
 from django.utils import timezone
 
-from jmap_email import body_text_joined
+from jmap_email import ComposeError, body_text_joined
 
 from core import models
 from core.enums import is_delivered
@@ -82,9 +82,14 @@ def create_and_send_draft(
         bcc_emails=bcc_emails or [],
     )
 
-    # Prepare the outbound message
-    if not prepare_outbound_message(from_mailbox, message, text_body, html_body):
-        raise SelfCheckError("Failed to prepare outbound message")
+    # Prepare the outbound message. ``ComposeError`` now escapes
+    # ``prepare_outbound_message`` (the API views turn it into a 400);
+    # here every failure is equally a selfcheck failure.
+    try:
+        if not prepare_outbound_message(from_mailbox, message, text_body, html_body):
+            raise SelfCheckError("Failed to prepare outbound message")
+    except ComposeError as exc:
+        raise SelfCheckError("Failed to compose outbound message") from exc
 
     # Send the message synchronously
     send_message(message, force_mta_out=True)
diff --git a/src/backend/core/mda/utils.py b/src/backend/core/mda/utils.py
index a2ebba43..e13e3fb6 100644
--- a/src/backend/core/mda/utils.py
+++ b/src/backend/core/mda/utils.py
@@ -13,6 +13,8 @@ Three groups of helpers live here:
 - :func:`current_sent_at` — single source of truth for the
   ``sentAt`` ISO-8601 string outbound paths stamp on the JMAP dict
   they hand to :func:`jmap_email.compose_email`.
+- :data:`COMPOSE_OPTIONS` — the compose policy every path that
+  produces MIME for us shares.
 """
 
 import re
@@ -22,10 +24,33 @@ from email.utils import make_msgid
 
 from django.utils import timezone
 
-from jmap_email import body_part_text, decode_rfc2047_header
+from jmap_email import ComposeOptions, body_part_text, decode_rfc2047_header
 from jmap_email.types import JmapEmail
 
+# Compose policy shared by every path that builds MIME for us.
+#
+# ``idna_encode_domains``: we send over plain 7-bit SMTP (``core.mda.smtp``
+# passes no ``mail_options``), and an A-label is the only form that carries
+# there. Without it such addresses never send.
+#
+# A non-ASCII *local part* stays unsendable — that needs SMTPUTF8, which we
+# never negotiate — and raises ``InvalidAddressError`` for callers to
+# handle. RFC 6530 provides no downgrade, so there is no fallback variant.
+COMPOSE_OPTIONS = ComposeOptions(idna_encode_domains=True)
+
+# Archive reconstruction (PST import) additionally keeps Bcc: the list was
+# already in the source file, so dropping it would lose data the user owns.
+# ``allow_smtputf8`` for the same reason — an EAI address (non-ASCII local
+# part, RFC 6530) is legal in an Exchange archive, and the reconstructed
+# .eml is stored in the user's own mailbox, never retransmitted over SMTP;
+# refusing it would exclude the message from the import.
+ARCHIVE_COMPOSE_OPTIONS = ComposeOptions(
+    idna_encode_domains=True, emit_bcc=True, allow_smtputf8=True
+)
+
 __all__ = [
+    "ARCHIVE_COMPOSE_OPTIONS",
+    "COMPOSE_OPTIONS",
     "SNIPPET_MAX_LENGTH",
     "current_sent_at",
     "generate_mime_id",
diff --git a/src/backend/core/models.py b/src/backend/core/models.py
index eeb704e3..237e742e 100644
--- a/src/backend/core/models.py
+++ b/src/backend/core/models.py
@@ -2162,6 +2162,10 @@ class Message(BaseModel):
 
     # Internal cache for parsed data
     _parsed_email_cache: Optional[JmapEmail] = None
+    # True once ``get_parsed_data`` has run and the blob failed to load or
+    # parse. Distinguishes "no content" from "content we can no longer read",
+    # which the empty cache alone cannot express.
+    _parse_failed: bool = False
 
     class Meta:
         db_table = "messages_message"
@@ -2179,7 +2183,8 @@ class Message(BaseModel):
         Use the helpers in :mod:`jmap_email` (``first_address``,
         ``first_msgid``, ``find_header``, …) for null-safe access
         patterns over the list-typed fields. Returns ``{}`` when
-        there's no blob or parsing fails.
+        there's no blob or parsing fails; the two are told apart by
+        ``has_unreadable_content``.
         """
         if self._parsed_email_cache is not None:
             return self._parsed_email_cache
@@ -2201,14 +2206,37 @@ class Message(BaseModel):
                 logger.warning(
                     "Failed to load blob content for message %s: %s", self.id, exc
                 )
+                self._parse_failed = True
                 self._parsed_email_cache = {}
                 return self._parsed_email_cache
             parsed = parse_email(raw, body_values=False)
+            if parsed is None:
+                # The bytes are intact but the parser refuses them — a
+                # message stored under looser rules than the ones in force
+                # now. Rendering it as an empty message would be a silent
+                # lie, so flag it for the UI.
+                logger.warning(
+                    "Stored message %s is no longer parseable (%d bytes)",
+                    self.id,
+                    len(raw),
+                )
+                self._parse_failed = True
             self._parsed_email_cache = parsed if parsed is not None else {}
         else:
             self._parsed_email_cache = {}
         return self._parsed_email_cache
 
+    def has_unreadable_content(self) -> bool:
+        """True when this message has a blob we cannot turn into content.
+
+        Either the blob failed to load (decompression / decryption /
+        integrity check) or the parser refused it. Both leave every parsed
+        field empty, so consumers that would otherwise render a blank
+        message must show an error instead.
+        """
+        self.get_parsed_data()
+        return self._parse_failed
+
     def get_parsed_field(self, field_name: str) -> Any:
         """Get a parsed field from the parsed JMAP Email object."""
         return (self.get_parsed_data() or {}).get(field_name)
@@ -2265,6 +2293,11 @@ class Message(BaseModel):
             # The divergent envelope RCPT TO (alias/BCC/catch-all) recorded by
             # ``_record_divergent_rcpt`` — surfaces the recipient's own alias.
             result["rcpt_to"] = postmark["rcpt_to"]
+        if self._parse_failed:
+            # Set by the ``get_parsed_data`` call above. Not a pipeline verdict
+            # like the rest, but it rides the same channel because it needs the
+            # same banner: every field the UI would render is empty.
+            result["unreadable"] = "true"
         return result
 
     def generate_mime_id(self) -> str:
diff --git a/src/backend/core/services/attachments.py b/src/backend/core/services/attachments.py
new file mode 100644
index 00000000..bfcdf26b
--- /dev/null
+++ b/src/backend/core/services/attachments.py
@@ -0,0 +1,172 @@
+"""Attachment display names.
+
+A MIME part may legitimately carry no filename — no ``filename`` in
+``Content-Disposition``, no ``name`` in ``Content-Type`` — and the JMAP
+parser faithfully reports ``None`` for it (RFC 8621: ``name`` is
+``String | null``). Every consumer on our side needs a string, so the
+placeholder is ours to synthesize; ``jmap_email`` deliberately stops at
+sanitizing the names that do exist.
+
+Everything a user-visible attachment name depends on lives here, so the
+serializer, the blob download endpoint and the draft builder all
+synthesize the *same* name for the same part.
+"""
+
+from jmap_email import sanitize_filename
+
+# Filename stem used for attachments whose MIME part carries no filename.
+# Matched by the frontend, which swaps in its own name for some types —
+# see the calendar-invite download button in
+# ``features/layouts/components/thread-view/components/calendar-invite``.
+UNNAMED_ATTACHMENT_STEM = "unnamed"
+
+# Must track ``Attachment.name``'s ``max_length``;
+# ``tests/services/test_attachments.py`` asserts the two stay equal. Names
+# are truncated to it rather than rejected: ``full_clean`` on save would
+# otherwise roll back an entire draft over one overlong name.
+ATTACHMENT_NAME_MAX_LENGTH = 255
+
+# MIME type → extension for the synthesized name.
+#
+# Spelled out rather than deferred to the stdlib ``mimetypes``, which
+# completes its table at import time from system files (``/etc/mime.types``
+# and friends). That makes its answers a property of the host image: the
+# same nameless part could be stored as ``unnamed.jpg`` by the web
+# container and ``unnamed.jpeg`` by the worker, and a CI base-image change
+# could silently rewrite what users download. These names are persisted in
+# ``Attachment.name``, so the mapping is pinned here instead.
+#
+# An unlisted type yields no extension at all — a bare ``unnamed`` is
+# honest, whereas guessing wrong hands the OS a file it opens with the
+# wrong application. Add entries as real mail brings them in.
+_MIME_EXTENSIONS = {
+    # Images
+    "image/jpeg": ".jpg",
+    "image/jpg": ".jpg",  # non-standard, but common in the wild
+    "image/pjpeg": ".jpg",
+    "image/png": ".png",
+    "image/gif": ".gif",
+    "image/webp": ".webp",
+    "image/bmp": ".bmp",
+    "image/tiff": ".tiff",
+    "image/svg+xml": ".svg",
+    "image/heic": ".heic",
+    "image/heif": ".heif",
+    "image/avif": ".avif",
+    "image/x-icon": ".ico",
+    "image/vnd.microsoft.icon": ".ico",
+    # Text and data
+    "text/plain": ".txt",
+    "text/html": ".html",
+    "text/csv": ".csv",
+    "text/markdown": ".md",
+    "text/xml": ".xml",
+    "text/rtf": ".rtf",
+    "application/json": ".json",
+    # ``mimetypes`` answers ``.xsl`` here — an XSLT stylesheet, a different
+    # format from a generic XML document.
+    "application/xml": ".xml",
+    # Calendar invites are the most frequent nameless part of all, and the
+    # standard vCard type is the one the stdlib table misses (it knows only
+    # the legacy ``text/x-vcard``).
+    "text/calendar": ".ics",
+    "text/vcard": ".vcf",
+    "text/x-vcard": ".vcf",
+    # Documents
+    "application/pdf": ".pdf",
+    "application/rtf": ".rtf",
+    "application/msword": ".doc",
+    "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
+    "application/vnd.ms-excel": ".xls",
+    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
+    "application/vnd.ms-powerpoint": ".ppt",
+    "application/vnd.openxmlformats-officedocument.presentationml.presentation": ".pptx",
+    "application/vnd.oasis.opendocument.text": ".odt",
+    "application/vnd.oasis.opendocument.spreadsheet": ".ods",
+    "application/vnd.oasis.opendocument.presentation": ".odp",
+    "application/vnd.ms-outlook": ".msg",
+    "application/epub+zip": ".epub",
+    # Archives — including the Outlook/Windows spelling of ZIP.
+    "application/zip": ".zip",
+    "application/x-zip-compressed": ".zip",
+    "application/gzip": ".gz",
+    "application/x-gzip": ".gz",
+    "application/x-tar": ".tar",
+    "application/x-7z-compressed": ".7z",
+    "application/vnd.rar": ".rar",
+    "application/x-rar-compressed": ".rar",
+    "application/x-bzip2": ".bz2",
+    "application/x-apple-diskimage": ".dmg",
+    # Mail
+    "message/rfc822": ".eml",
+    # Signatures and S/MIME. ``application/pkcs7-mime`` over mail carries a
+    # signed/encrypted message, not the bare certificate ``mimetypes``
+    # assumes with ``.p7c``.
+    "application/pgp-signature": ".asc",
+    "application/pgp-encrypted": ".asc",
+    "application/pkcs7-mime": ".p7m",
+    "application/pkcs7-signature": ".p7s",
+    "application/x-pkcs7-signature": ".p7s",
+    # Audio and video
+    "audio/mpeg": ".mp3",
+    "audio/mp4": ".m4a",
+    "audio/ogg": ".ogg",
+    "audio/wav": ".wav",
+    "audio/x-wav": ".wav",
+    "audio/aac": ".aac",
+    "audio/amr": ".amr",
+    "video/mp4": ".mp4",
+    "video/mpeg": ".mpeg",
+    "video/quicktime": ".mov",
+    "video/webm": ".webm",
+    "video/x-msvideo": ".avi",
+    "video/3gpp": ".3gp",
+    # Deliberately no extension: the generic "some bytes" type says nothing
+    # about the format, and ``.bin`` would only look like it did. Listed so
+    # the intent is explicit rather than a fall-through.
+    "application/octet-stream": "",
+}
+
+
+def guess_mime_extension(content_type):
+    """Return the file extension (with leading dot) for a MIME type, or ``""``.
+
+    Accepts a full ``Content-Type`` header value — parameters (charset,
+    boundary, name…) are dropped before lookup. Unknown types yield
+    ``""``. The mapping is a fixed table, so the answer is the same on
+    every host; see ``_MIME_EXTENSIONS`` for why that matters.
+    """
+    if not content_type:
+        return ""
+
+    mime_type = content_type.split(";", maxsplit=1)[0].strip().lower()
+    return _MIME_EXTENSIONS.get(mime_type, "")
+
+
+def get_attachment_display_name(name, content_type=None):
+    """Return a non-empty, sanitized filename for an attachment part.
+
+    ``name`` is whatever we hold for the part — the parser's ``name``
+    (``None`` for a nameless part), or a client-supplied one. When it is
+    absent or sanitizes away to nothing, the stem is synthesized and the
+    extension inferred from *content_type*, so the recipient's OS can
+    still open the file.
+
+    Sanitizing is a near no-op for names the parser reported (it already
+    applied the same pass) and a real guard for client-supplied ones,
+    which reach us straight off the wire: the draft endpoint takes
+    ``attachments[].name`` without serializer validation, so without the
+    length cap here one overlong name fails ``full_clean`` and rolls back
+    the whole draft save with a 400.
+    """
+    # ``isinstance`` rather than truthiness: the draft endpoint takes
+    # ``attachments[].name`` straight off the wire with no serializer
+    # validation, so a client can send a number or a list. Those are
+    # truthy but unsubscriptable, and ``sanitize_filename`` would raise
+    # on them; treat anything that is not a string as no name at all.
+    if isinstance(name, str) and name:
+        sanitized = sanitize_filename(name, max_length=ATTACHMENT_NAME_MAX_LENGTH)
+        if sanitized:
+            return sanitized
+
+    return f"{UNNAMED_ATTACHMENT_STEM}{guess_mime_extension(content_type)}"
diff --git a/src/backend/core/services/exporter/tasks.py b/src/backend/core/services/exporter/tasks.py
index a5efa091..50a7aca2 100644
--- a/src/backend/core/services/exporter/tasks.py
+++ b/src/backend/core/services/exporter/tasks.py
@@ -18,7 +18,7 @@ from sentry_sdk import capture_exception
 
 from core.api.utils import generate_presigned_url
 from core.mda.inbound import deliver_inbound_message
-from core.mda.utils import current_sent_at
+from core.mda.utils import COMPOSE_OPTIONS, current_sent_at
 from core.models import Label, Mailbox, Message, ThreadAccess
 
 from messages.celery_app import app as celery_app
@@ -702,7 +702,7 @@ This file is in MBOX format and can be imported into most email clients.
         "textBody": [{"partId": "1", "type": "text/plain", "content": body_text}],
         "htmlBody": [{"partId": "2", "type": "text/html", "content": body_html}],
     }
-    raw_data = compose_email(notification)
+    raw_data = compose_email(notification, options=COMPOSE_OPTIONS)
     parsed_email = parse_email(raw_data)
     if parsed_email is None:
         # We just composed this; failing to parse it back means the
diff --git a/src/backend/core/services/importer/imap.py b/src/backend/core/services/importer/imap.py
index 12f41010..8dd60619 100644
--- a/src/backend/core/services/importer/imap.py
+++ b/src/backend/core/services/importer/imap.py
@@ -86,7 +86,9 @@ def decode_imap_utf7(s):
         return decoded_bytes.decode("utf-16-be")
 
     try:
-        return re.sub(r"&([^-]*)-", decode_match, s)
+        # ``[^-&]`` not ``[^-]``: ``&`` is not in the base64 alphabet, and
+        # excluding it stops a ``&&&&…`` name backtracking quadratically.
+        return re.sub(r"&([^-&]*)-", decode_match, s)
     except Exception as e:  # pylint: disable=broad-exception-caught
         # A malformed UTF-7 folder name must not abort the whole import (this
         # runs while building the folder map, outside any per-folder guard) —
diff --git a/src/backend/core/services/importer/pst.py b/src/backend/core/services/importer/pst.py
index dba8d7e1..84fe7068 100644
--- a/src/backend/core/services/importer/pst.py
+++ b/src/backend/core/services/importer/pst.py
@@ -23,6 +23,7 @@ import pypff
 from jmap_email import compose_email, is_valid_msg_id, parse_address, parse_addresses
 from jmap_email.types import EmailAddress
 
+from core.mda.utils import ARCHIVE_COMPOSE_OPTIONS
 from core.services.s3_seekable import BUFFER_NONE, S3SeekableReader
 
 from .channel import ImportCancelled
@@ -1276,7 +1277,7 @@ def reconstruct_eml(
     # PST is an archive: the Bcc list was in the original source file and we
     # are reconstructing the .eml for storage in the user's own mailbox, not
     # retransmitting. Preserve it.
-    return compose_email(jmap_data, keep_bcc=True)
+    return compose_email(jmap_data, options=ARCHIVE_COMPOSE_OPTIONS)
 
 
 def _find_ipm_subtree(pst_file):
diff --git a/src/backend/core/tests/api/test_draft_attachments.py b/src/backend/core/tests/api/test_draft_attachments.py
index 64c69fc7..33e7612e 100644
--- a/src/backend/core/tests/api/test_draft_attachments.py
+++ b/src/backend/core/tests/api/test_draft_attachments.py
@@ -875,6 +875,92 @@ Content-ID: <{cid}>
 """
         return email_template.encode("utf-8"), cid, image_content
 
+    @pytest.fixture
+    def make_multipart_email_with_attachment(self, user_mailbox):
+        """Build a multipart email carrying a single attachment part.
+
+        With ``filename`` left to ``None`` the part carries no filename at all —
+        no ``filename`` in ``Content-Disposition``, no ``name`` in
+        ``Content-Type`` — so the parser reports ``name`` as ``None``. Seen in
+        the wild on messages forwarded by other clients.
+        """
+
+        def _make(filename=None):
+            recipient_email = f"{user_mailbox.local_part}@{user_mailbox.domain.name}"
+            boundary = "------------boundary123456789"
+            image_content = b"fake-gif-content-for-testing"
+            disposition = "attachment"
+            if filename is not None:
+                disposition += f'; filename="{filename}"'
+
+            email_template = f"""From: sender@example.com
+To: {recipient_email}
+Subject: Original message with a nameless attachment
+Message-ID: <original-msg-{uuid.uuid4()}@example.com>
+MIME-Version: 1.0
+Content-Type: multipart/mixed; boundary="{boundary}"
+
+--{boundary}
+Content-Type: text/plain; charset="UTF-8"
+Content-Transfer-Encoding: 7bit
+
+This is the original message body.
+
+--{boundary}
+Content-Type: image/gif
+Content-Transfer-Encoding: base64
+Content-Disposition: {disposition}
+Content-ID: <cid123@example.com>
+
+{base64.b64encode(image_content).decode()}
+--{boundary}--
+"""
+            return email_template.encode("utf-8"), image_content
+
+        return _make
+
+    @pytest.fixture
+    def make_received_message_with_attachment(
+        self, user_mailbox, make_multipart_email_with_attachment
+    ):
+        """Build a received message holding a single attachment part."""
+
+        def _make(filename=None):
+            mime_content, image_content = make_multipart_email_with_attachment(filename)
+            return (
+                self._store_received_message(user_mailbox, mime_content),
+                image_content,
+            )
+
+        return _make
+
+    @staticmethod
+    def _store_received_message(user_mailbox, mime_content):
+        """Persist a received message from its raw MIME content."""
+        thread = factories.ThreadFactory(subject="Nameless attachment")
+        factories.ThreadAccessFactory(
+            thread=thread, mailbox=user_mailbox, role=ThreadAccessRoleChoices.EDITOR
+        )
+        sender = factories.ContactFactory(
+            mailbox=user_mailbox, email="sender@example.com", name="Sender"
+        )
+        blob = factories.BlobFactory(
+            mailbox=user_mailbox,
+            content=mime_content,
+            content_type="message/rfc822",
+        )
+        message = factories.MessageFactory(
+            thread=thread,
+            sender=sender,
+            is_draft=False,
+            is_sender=False,
+            subject="Nameless attachment",
+            blob=blob,
+            has_attachments=True,
+        )
+
+        return message
+
     @pytest.fixture
     def received_message_with_attachment(
         self, user_mailbox, multipart_email_with_attachment
@@ -1005,6 +1091,155 @@ Content-ID: <{cid}>
         assert not new_blob_id.startswith("msg_")
         uuid.UUID(new_blob_id)  # Should not raise
 
+    def test_draft_create_with_forwarded_nameless_attachment(
+        self, api_client, user_mailbox, make_received_message_with_attachment
+    ):
+        """Forwarding an attachment whose MIME part has no filename succeeds.
+
+        Regression: the parser reports ``name`` as ``None`` for such a part, the
+        API echoed the "unnamed" placeholder back, and the draft builder then
+        wrote that ``None`` straight into the NOT NULL ``Attachment.name`` —
+        answering 400 "This field cannot be null". The synthesized name now
+        carries the extension inferred from the MIME type.
+        """
+        client, _ = api_client
+        message, image_content = make_received_message_with_attachment()
+
+        # What the API exposes for that part is what the client sends back.
+        list_response = client.get(
+            reverse("messages-detail", kwargs={"id": str(message.id)})
+        )
+        assert list_response.status_code == status.HTTP_200_OK
+        exposed = list_response.data["attachments"][0]
+        assert exposed["name"] == "unnamed.gif"
+
+        response = client.post(
+            reverse("draft-message"),
+            {
+                "senderId": str(user_mailbox.id),
+                "parentId": str(message.id),
+                "subject": "Fwd: Nameless attachment",
+                "draftBody": json.dumps({"text": "Forwarding this message"}),
+                "to": ["recipient@example.com"],
+                "attachments": [{"blobId": exposed["blobId"], "name": exposed["name"]}],
+            },
+            format="json",
+        )
+
+        assert response.status_code == status.HTTP_201_CREATED
+
+        draft = models.Message.objects.get(id=response.data["id"])
+        attachment = draft.attachments.get()
+        assert attachment.name == "unnamed.gif"
+        assert attachment.blob.get_content() == image_content
+
+    def test_download_nameless_attachment_serves_the_name_the_user_saw(
+        self, api_client, make_received_message_with_attachment
+    ):
+        """Downloading a nameless part serves it under its displayed name.
+
+        The other half of the same round-trip: the blob endpoint reads the
+        parsed ``name`` too, so it was handing ``None`` to
+        ``content_disposition_header`` and emitting a bare ``attachment``
+        with no filename — leaving the browser to save the file under the
+        blob id, extensionless, while the UI listed it as ``unnamed.gif``.
+        """
+        client, _ = api_client
+        message, image_content = make_received_message_with_attachment()
+
+        list_response = client.get(
+            reverse("messages-detail", kwargs={"id": str(message.id)})
+        )
+        assert list_response.status_code == status.HTTP_200_OK
+        exposed = list_response.data["attachments"][0]
+        assert exposed["name"] == "unnamed.gif"
+
+        response = client.get(
+            reverse("blob-download", kwargs={"pk": exposed["blobId"]})
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response["Content-Disposition"] == 'attachment; filename="unnamed.gif"'
+        assert response.content == image_content
+
+    def test_draft_create_with_forwarded_attachment_actually_named_unnamed(
+        self, api_client, user_mailbox, make_received_message_with_attachment
+    ):
+        """A part genuinely named "unnamed" keeps that name when forwarded.
+
+        The draft builder treats "unnamed" as the placeholder older clients
+        echo back for a nameless part. It must not do so when the part really
+        carries that filename, or forwarding would rename the user's file.
+        """
+        client, _ = api_client
+        message, _ = make_received_message_with_attachment("unnamed")
+
+        list_response = client.get(
+            reverse("messages-detail", kwargs={"id": str(message.id)})
+        )
+        assert list_response.status_code == status.HTTP_200_OK
+        exposed = list_response.data["attachments"][0]
+        assert exposed["name"] == "unnamed"
+
+        response = client.post(
+            reverse("draft-message"),
+            {
+                "senderId": str(user_mailbox.id),
+                "parentId": str(message.id),
+                "subject": "Fwd: Attachment named unnamed",
+                "draftBody": json.dumps({"text": "Forwarding this message"}),
+                "to": ["recipient@example.com"],
+                "attachments": [{"blobId": exposed["blobId"], "name": exposed["name"]}],
+            },
+            format="json",
+        )
+
+        assert response.status_code == status.HTTP_201_CREATED
+
+        draft = models.Message.objects.get(id=response.data["id"])
+        assert draft.attachments.get().name == "unnamed"
+
+    def test_draft_create_with_forwarded_attachment_overlong_name(
+        self, api_client, user_mailbox, make_received_message_with_attachment
+    ):
+        """A client-sent name longer than 255 chars is truncated, not rejected.
+
+        The draft endpoint takes ``attachments[].name`` without serializer
+        validation; ``Attachment.name`` is a ``CharField(max_length=255)``
+        enforced by ``full_clean`` on save. Without sanitization the whole
+        draft save was rolled back with a 400 over one overlong name.
+        """
+        client, _ = api_client
+        message, _ = make_received_message_with_attachment("original.gif")
+
+        list_response = client.get(
+            reverse("messages-detail", kwargs={"id": str(message.id)})
+        )
+        assert list_response.status_code == status.HTTP_200_OK
+        exposed = list_response.data["attachments"][0]
+
+        response = client.post(
+            reverse("draft-message"),
+            {
+                "senderId": str(user_mailbox.id),
+                "parentId": str(message.id),
+                "subject": "Fwd: Overlong attachment name",
+                "draftBody": json.dumps({"text": "Forwarding this message"}),
+                "to": ["recipient@example.com"],
+                "attachments": [
+                    {"blobId": exposed["blobId"], "name": "a" * 300 + ".gif"}
+                ],
+            },
+            format="json",
+        )
+
+        assert response.status_code == status.HTTP_201_CREATED
+
+        draft = models.Message.objects.get(id=response.data["id"])
+        attachment = draft.attachments.get()
+        assert len(attachment.name) == 255
+        assert attachment.name.endswith(".gif")
+
     def test_draft_forward_attachments_inline_image_preserves_cid(
         self, api_client, user_mailbox, received_message_with_inline_image
     ):
diff --git a/src/backend/core/tests/api/test_drive.py b/src/backend/core/tests/api/test_drive.py
index c1b1a58f..dfd47fa5 100644
--- a/src/backend/core/tests/api/test_drive.py
+++ b/src/backend/core/tests/api/test_drive.py
@@ -83,6 +83,55 @@ Test file content for Drive upload.
     return mailbox, message
 
 
+@pytest.fixture
+def mailbox_with_nameless_attachment(api_client_with_user):
+    """Create a message whose attachment part declares no filename at all."""
+    _, user = api_client_with_user
+
+    mailbox = factories.MailboxFactory()
+    factories.MailboxAccessFactory(
+        mailbox=mailbox,
+        user=user,
+        role=MailboxRoleChoices.EDITOR,
+    )
+
+    thread = factories.ThreadFactory()
+    factories.ThreadAccessFactory(
+        thread=thread,
+        mailbox=mailbox,
+        role=ThreadAccessRoleChoices.EDITOR,
+    )
+
+    # No ``filename`` in Content-Disposition, no ``name`` in Content-Type:
+    # the parser reports ``name`` as None for this part.
+    raw_mime_content = b"""From: sender@example.com
+To: recipient@example.com
+Subject: Test message with nameless attachment
+MIME-Version: 1.0
+Content-Type: multipart/mixed; boundary="boundary-string"
+
+--boundary-string
+Content-Type: text/plain; charset="utf-8"
+
+This is a test message.
+
+--boundary-string
+Content-Type: text/plain
+Content-Disposition: attachment
+
+Nameless file content for Drive upload.
+--boundary-string--
+"""
+
+    message = factories.MessageFactory(
+        thread=thread,
+        raw_mime=raw_mime_content,
+        has_attachments=True,
+    )
+
+    return mailbox, message
+
+
 class TestDriveAPIView:
     """Tests for the Drive API View endpoints."""
 
@@ -436,6 +485,75 @@ Test file content for Drive upload without access.
         # Verify upload-ended confirmation
         assert f"items/{file_id}/upload-ended/" in responses.calls[3].request.url
 
+    @responses.activate
+    @patch(
+        "lasuite.oidc_login.middleware.RefreshOIDCAccessToken.is_expired",
+        return_value=False,
+    )
+    def test_api_third_party_drive_post_nameless_attachment(
+        self, _mock, api_client_with_user, mailbox_with_nameless_attachment
+    ):
+        """A nameless attachment gets the synthesized display name end to end.
+
+        With a raw ``None`` name, ``requests`` would drop the ``title``
+        param from the dedup search — matching an unrelated Drive file on
+        size alone — and the creation call would post ``"filename": null``.
+        """
+        client, _ = api_client_with_user
+        _, message = mailbox_with_nameless_attachment
+
+        blob_id = f"msg_{message.id}_0"
+        file_id = str(uuid.uuid4())
+        presigned_url = "http://s3.test/presigned-upload-url"
+
+        responses.add(
+            responses.GET,
+            "http://drive.test/external_api/v1.0/items/",
+            json={"count": 0, "next": None, "previous": None, "results": []},
+            status=status.HTTP_200_OK,
+        )
+        responses.add(
+            responses.POST,
+            "http://drive.test/external_api/v1.0/items/",
+            json={
+                "id": file_id,
+                "title": "unnamed.txt",
+                "type": "file",
+                "policy": presigned_url,
+                "created_at": "2024-01-01T00:00:00Z",
+                "updated_at": "2024-01-01T00:00:00Z",
+            },
+            status=status.HTTP_200_OK,
+        )
+        responses.add(responses.PUT, presigned_url, status=status.HTTP_200_OK)
+        responses.add(
+            responses.POST,
+            f"http://drive.test/external_api/v1.0/items/{file_id}/upload-ended/",
+            json={
+                "id": file_id,
+                "title": "unnamed.txt",
+                "type": "file",
+                "created_at": "2024-01-01T00:00:00Z",
+                "updated_at": "2024-01-01T00:00:00Z",
+            },
+            status=status.HTTP_200_OK,
+        )
+
+        response = client.post(
+            reverse("drive"),
+            {"blob_id": blob_id},
+            format="json",
+        )
+
+        assert response.status_code == status.HTTP_201_CREATED
+
+        # The dedup search must keep its title filter, with the same
+        # synthesized name the message serializer exposes.
+        assert "title=unnamed.txt" in responses.calls[0].request.url
+
+        file_creation_body = json.loads(responses.calls[1].request.body)
+        assert file_creation_body["filename"] == "unnamed.txt"
+
     @responses.activate
     @patch(
         "lasuite.oidc_login.middleware.RefreshOIDCAccessToken.is_expired",
diff --git a/src/backend/core/tests/api/test_inbound_widget.py b/src/backend/core/tests/api/test_inbound_widget.py
index 1b1fb4bd..23265dfb 100644
--- a/src/backend/core/tests/api/test_inbound_widget.py
+++ b/src/backend/core/tests/api/test_inbound_widget.py
@@ -272,6 +272,58 @@ class TestInboundWidgetDeliver:
         assert response.status_code == status.HTTP_400_BAD_REQUEST
         assert response.json() == {"detail": "Invalid email format"}
 
+    def test_inbound_widget_deliver_non_ascii_local_part(self, api_client, channel):
+        """An RFC 6531 local part is a *valid* address, so ``parse_address``
+        accepts it and the shape check above lets it through — but we emit
+        7-bit SMTP and never negotiate SMTPUTF8, so the composer refuses it.
+        That is a property of what this public form was given, so it must be
+        a 400 like any other unusable address, not a 500."""
+        response = api_client.post(
+            "/api/v1.0/inbound/widget/deliver/",
+            data={"email": "josé@example.com", "textBody": "hello"},
+            HTTP_X_CHANNEL_ID=str(channel.id),
+        )
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.json() == {"detail": "Invalid email format"}
+
+    def test_compose_rejection_log_carries_only_the_domain(self, api_client, channel):
+        """The rejection is logged so operators can see widgets failing,
+        but the submitter's address is user data on a public endpoint —
+        the local part must not land in the log. Same convention the
+        send_mail command already uses.
+
+        Asserts on the logger call rather than ``caplog``: this logger does
+        not propagate to root under the project's logging config, so a
+        ``caplog.text`` assertion would pass whether or not the address
+        leaked.
+        """
+        with patch.object(widget_module, "logger") as mock_logger:
+            api_client.post(
+                "/api/v1.0/inbound/widget/deliver/",
+                data={"email": "josé.secret@example.com", "textBody": "hello"},
+                HTTP_X_CHANNEL_ID=str(channel.id),
+            )
+
+        assert mock_logger.info.called
+        logged = " ".join(
+            str(a) for call in mock_logger.info.call_args_list for a in call.args
+        )
+        assert "example.com" in logged
+        assert "josé.secret" not in logged
+
+    def test_inbound_widget_deliver_idn_domain_is_accepted(self, api_client, channel):
+        """The other half: an IDN *domain* has an exact ASCII wire form, so
+        ``COMPOSE_OPTIONS`` converts it and the submission goes through
+        rather than being rejected alongside the case above."""
+        response = api_client.post(
+            "/api/v1.0/inbound/widget/deliver/",
+            data={"email": "contact@exemplé.fr", "textBody": "hello"},
+            HTTP_X_CHANNEL_ID=str(channel.id),
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+
     def test_inbound_widget_deliver_missing_message(self, api_client, channel):
         """Test deliver with missing message."""
         data = {"email": "sender@example.com"}
diff --git a/src/backend/core/tests/api/test_message_attachments_serializer.py b/src/backend/core/tests/api/test_message_attachments_serializer.py
index 5af5d831..4a8c0756 100644
--- a/src/backend/core/tests/api/test_message_attachments_serializer.py
+++ b/src/backend/core/tests/api/test_message_attachments_serializer.py
@@ -34,9 +34,28 @@ def test_get_attachments_preserves_present_name():
         pytest.param({"name": "", "size": 1, "type": "text/plain"}, id="empty-name"),
     ],
 )
-def test_get_attachments_falls_back_to_unnamed(attachment):
-    """A MIME part with no usable ``filename`` falls back to the "unnamed"
-    sentinel so consumers never receive a null/empty name (regression: a null
-    name crashed the frontend calendar-invite download button)."""
+def test_get_attachments_synthesizes_a_name(attachment):
+    """A MIME part with no usable ``filename`` gets a synthesized name, so
+    consumers never receive a null/empty one (regression: a null name crashed
+    the frontend calendar-invite download button). The extension comes from
+    the part's MIME type — ``text/plain`` here."""
     result = _serialize_parsed_attachments([attachment])
-    assert result[0]["name"] == "unnamed"
+    assert result[0]["name"] == "unnamed.txt"
+
+
+@pytest.mark.parametrize(
+    ("content_type", "expected"),
+    [
+        pytest.param("image/gif", "unnamed.gif", id="gif"),
+        pytest.param('text/calendar; charset="utf-8"', "unnamed.ics", id="parameters"),
+        pytest.param("application/x-unknown", "unnamed", id="unknown-type"),
+    ],
+)
+def test_get_attachments_passes_the_content_type_through(content_type, expected):
+    """The serializer hands the part's type to the naming policy, so a
+    forwarded nameless part stays openable by the recipient's OS. The mapping
+    itself is covered in ``core/tests/services/test_attachments.py``."""
+    result = _serialize_parsed_attachments(
+        [{"name": None, "size": 1, "type": content_type}]
+    )
+    assert result[0]["name"] == expected
diff --git a/src/backend/core/tests/api/test_send_message_signature.py b/src/backend/core/tests/api/test_send_message_signature.py
index fdd7e603..c74a8991 100644
--- a/src/backend/core/tests/api/test_send_message_signature.py
+++ b/src/backend/core/tests/api/test_send_message_signature.py
@@ -9,6 +9,7 @@ from django.test import override_settings
 from django.urls import reverse
 
 import pytest
+from jmap_email import ComposeError, InvalidAddressError
 from rest_framework import status
 from rest_framework.test import APIClient
 
@@ -550,3 +551,100 @@ class TestSendMessageSecurity:
                 if "SendMessageView" in getattr(cb, "__qualname__", "")
             ]
             assert len(send_callbacks) == 1
+
+
+class TestSendMessageComposeFailure:
+    """A draft the composer cannot put on the wire is a 400, not a 500.
+
+    ``prepare_outbound_message`` composes synchronously inside the request,
+    so a ``ComposeError`` — a non-ASCII local part needing SMTPUTF8, a
+    malformed addr-spec, an undecodable attachment — escaped as an
+    unhandled exception. That is a property of the draft the user built,
+    so it belongs in the 4xx family with a message they can act on.
+    """
+
+    def test_compose_error_escapes_prepare_outbound_message(
+        self, user, mailbox_access, mailbox, draft_message
+    ):
+        """The error must cross ``prepare_outbound_message``, not be
+        swallowed there.
+
+        ``prepare_outbound_message`` wraps composition in a broad
+        ``except Exception: return False`` whose ``False`` the view turns
+        into a 500 — so mocking ``prepare_outbound_message`` itself cannot
+        prove the 400 works. Run the real function and fail at its
+        composition boundary instead.
+        """
+        client = APIClient()
+        client.force_authenticate(user=user)
+
+        with (
+            patch("core.mda.outbound.compose_and_sign_mime") as mock_compose,
+            patch("core.mda.outbound.logger") as mock_outbound_logger,
+        ):
+            mock_compose.side_effect = InvalidAddressError(
+                "recipient needs SMTPUTF8: 'josé@example.com'"
+            )
+            response = client.post(
+                reverse("send-message"),
+                {
+                    "messageId": str(draft_message.id),
+                    "textBody": "hello",
+                    "htmlBody": "<p>hello</p>",
+                    "senderId": str(mailbox.id),
+                },
+                format="json",
+            )
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        # The broad handler in ``prepare_outbound_message`` must not have
+        # caught it: its ``logger.exception`` line would carry the raw
+        # addr-spec into logs/Sentry.
+        assert not mock_outbound_logger.exception.called
+
+    def test_compose_error_is_a_400(self, user, mailbox_access, mailbox, draft_message):
+        client = APIClient()
+        client.force_authenticate(user=user)
+
+        with patch("core.api.viewsets.send.prepare_outbound_message") as mock_prepare:
+            mock_prepare.side_effect = ComposeError("bad addr-spec")
+            response = client.post(
+                reverse("send-message"),
+                {
+                    "messageId": str(draft_message.id),
+                    "textBody": "hello",
+                    "htmlBody": "<p>hello</p>",
+                    "senderId": str(mailbox.id),
+                },
+                format="json",
+            )
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    def test_compose_error_log_does_not_carry_the_message_body(
+        self, user, mailbox_access, mailbox, draft_message
+    ):
+        """The rejection is logged for operators; the log line must not
+        become a copy of the user's content."""
+        client = APIClient()
+        client.force_authenticate(user=user)
+
+        with patch("core.api.viewsets.send.prepare_outbound_message") as mock_prepare:
+            mock_prepare.side_effect = ComposeError("bad addr-spec")
+            with patch("core.api.viewsets.send.logger") as mock_logger:
+                client.post(
+                    reverse("send-message"),
+                    {
+                        "messageId": str(draft_message.id),
+                        "textBody": "SECRET-BODY-CONTENT",
+                        "htmlBody": "<p>SECRET-BODY-CONTENT</p>",
+                        "senderId": str(mailbox.id),
+                    },
+                    format="json",
+                )
+
+        logged = " ".join(
+            str(a) for call in mock_logger.info.call_args_list for a in call.args
+        )
+        assert mock_logger.info.called
+        assert "SECRET-BODY-CONTENT" not in logged
diff --git a/src/backend/core/tests/commands/test_send_mail_compose_error.py b/src/backend/core/tests/commands/test_send_mail_compose_error.py
new file mode 100644
index 00000000..f65eddc0
--- /dev/null
+++ b/src/backend/core/tests/commands/test_send_mail_compose_error.py
@@ -0,0 +1,54 @@
+"""``send_mail`` must report a compose failure as a CommandError.
+
+``compose_email`` is strict: a malformed addr-spec, or a non-ASCII local
+part needing SMTPUTF8, raises rather than emitting something unroutable.
+Unhandled, that reached the operator as a bare traceback — while the very
+same class of problem caught one step earlier (``parse_address``) already
+produced a clean ``CommandError``. Same failure, same reporting.
+"""
+
+from unittest.mock import patch
+
+from django.core.management import call_command
+from django.core.management.base import CommandError
+
+import pytest
+from jmap_email import InvalidAddressError
+
+pytestmark = pytest.mark.django_db
+
+
+def test_compose_error_becomes_a_command_error():
+    with patch(
+        "core.management.commands.send_mail.compose_email",
+        side_effect=InvalidAddressError("Non-ASCII local-part in 'to'"),
+    ):
+        with pytest.raises(CommandError, match="Cannot compose message"):
+            call_command(
+                "send_mail",
+                "--from",
+                "sender@example.com",
+                "--to",
+                "recipient@example.com",
+                "--subject",
+                "hi",
+                "--body",
+                "hello",
+            )
+
+
+def test_invalid_address_still_reported_before_compose():
+    """The pre-existing path is unchanged: a shape failure is caught by
+    ``parse_address`` and never reaches the composer."""
+    with pytest.raises(CommandError, match="Invalid recipient email address"):
+        call_command(
+            "send_mail",
+            "--from",
+            "sender@example.com",
+            "--to",
+            "not-an-address",
+            "--subject",
+            "hi",
+            "--body",
+            "hello",
+        )
diff --git a/src/backend/core/tests/importer/test_pst_import.py b/src/backend/core/tests/importer/test_pst_import.py
index 3c4c5977..970e54ae 100644
--- a/src/backend/core/tests/importer/test_pst_import.py
+++ b/src/backend/core/tests/importer/test_pst_import.py
@@ -300,6 +300,30 @@ class TestReconstructEml:
         assert len(text_parts) >= 1
         assert "Hello world" in text_parts[0].get_payload(decode=True).decode()
 
+    def test_reconstruct_with_eai_address(self):
+        """A non-ASCII local part (RFC 6530 EAI) is legal in an Exchange
+        archive. The reconstructed .eml is stored in the user's own
+        mailbox, never retransmitted over SMTP, so the composer must
+        accept the address — refusing it would silently exclude the
+        message from the import."""
+        transport = (
+            "From: José <josé@exemple.fr>\r\n"
+            "To: recipient@example.com\r\n"
+            "Subject: EAI sender\r\n"
+            "Date: Mon, 26 May 2025 10:00:00 +0000\r\n"
+        )
+        msg = _make_message(
+            subject="EAI sender",
+            transport_headers=transport,
+            plain_text_body="Hello EAI",
+        )
+        eml_bytes = reconstruct_eml(msg)
+
+        assert "josé@exemple.fr".encode() in eml_bytes
+        parsed = email.message_from_bytes(eml_bytes)
+        text_parts = [p for p in parsed.walk() if p.get_content_type() == "text/plain"]
+        assert "Hello EAI" in text_parts[0].get_payload(decode=True).decode()
+
     def test_reconstruct_preserves_rfc5322_date(self):
         """Test that RFC 5322 date from transport headers is preserved correctly."""
         transport = (
diff --git a/src/backend/core/tests/mda/test_arc.py b/src/backend/core/tests/mda/test_arc.py
index 637d909b..26339318 100644
--- a/src/backend/core/tests/mda/test_arc.py
+++ b/src/backend/core/tests/mda/test_arc.py
@@ -38,17 +38,19 @@ class TestArcResult:
 
     def test_empty_allowlist_trusts_nothing(self):
         # Fail closed: no allowlist -> nothing trusted, and no parse/verify.
-        with patch("core.mda.arc._outermost_sealer") as mock_outer, patch(
-            "core.mda.arc.arc_verify"
-        ) as mock_verify:
+        with (
+            patch("core.mda.arc._outermost_sealer") as mock_outer,
+            patch("core.mda.arc.arc_verify") as mock_verify,
+        ):
             out = arc.arc_result(b"raw", set())
         assert out == {"trusted": False, "sealer": None, "aar": None, "dnsfail": False}
         mock_outer.assert_not_called()
         mock_verify.assert_not_called()
 
     def test_verify_exception_untrusted(self):
-        with self._outer("relay.example"), patch(
-            "core.mda.arc.arc_verify", side_effect=Exception("boom")
+        with (
+            self._outer("relay.example"),
+            patch("core.mda.arc.arc_verify", side_effect=Exception("boom")),
         ):
             out = arc.arc_result(b"raw", {"relay.example"})
         assert out["dnsfail"] is False
@@ -82,9 +84,10 @@ class TestArcFastPath:
     """Q1: full crypto/DNS verification runs ONLY for a claimed-trusted sealer."""
 
     def test_unlisted_sealer_skips_verification(self):
-        with patch(
-            "core.mda.arc._outermost_sealer", return_value=("evil.net", 1)
-        ), patch("core.mda.arc.arc_verify") as mock_verify:
+        with (
+            patch("core.mda.arc._outermost_sealer", return_value=("evil.net", 1)),
+            patch("core.mda.arc.arc_verify") as mock_verify,
+        ):
             out = arc.arc_result(b"raw", {"relay.example"})
         assert out["trusted"] is False
         assert out["sealer"] == "evil.net"
@@ -92,9 +95,10 @@ class TestArcFastPath:
         mock_verify.assert_not_called()
 
     def test_no_chain_skips_verification(self):
-        with patch(
-            "core.mda.arc._outermost_sealer", return_value=(None, 0)
-        ), patch("core.mda.arc.arc_verify") as mock_verify:
+        with (
+            patch("core.mda.arc._outermost_sealer", return_value=(None, 0)),
+            patch("core.mda.arc.arc_verify") as mock_verify,
+        ):
             out = arc.arc_result(b"raw", {"relay.example"})
         assert out["trusted"] is False
         assert out["sealer"] is None
@@ -102,9 +106,10 @@ class TestArcFastPath:
 
     def test_overlong_chain_skips_verification(self):
         # One past the cap (_MAX_ARC_INSTANCES = 20) is refused without verifying.
-        with patch(
-            "core.mda.arc._outermost_sealer", return_value=("relay.example", 21)
-        ), patch("core.mda.arc.arc_verify") as mock_verify:
+        with (
+            patch("core.mda.arc._outermost_sealer", return_value=("relay.example", 21)),
+            patch("core.mda.arc.arc_verify") as mock_verify,
+        ):
             out = arc.arc_result(b"raw", {"relay.example"})
         assert out["trusted"] is False
         mock_verify.assert_not_called()
@@ -112,11 +117,12 @@ class TestArcFastPath:
     def test_at_cap_still_verifies(self):
         # Exactly at the cap is still verified.
         results = [{"ams-domain": b"relay.example", "aar-value": b"x"}]
-        with patch(
-            "core.mda.arc._outermost_sealer", return_value=("relay.example", 20)
-        ), patch(
-            "core.mda.arc.arc_verify", return_value=(arc.CV_Pass, results, "ok")
-        ) as mock_verify:
+        with (
+            patch("core.mda.arc._outermost_sealer", return_value=("relay.example", 20)),
+            patch(
+                "core.mda.arc.arc_verify", return_value=(arc.CV_Pass, results, "ok")
+            ) as mock_verify,
+        ):
             out = arc.arc_result(b"raw", {"relay.example"})
         assert out["trusted"] is True
         mock_verify.assert_called_once()
diff --git a/src/backend/core/tests/mda/test_inbound_nonretryable.py b/src/backend/core/tests/mda/test_inbound_nonretryable.py
new file mode 100644
index 00000000..409e2758
--- /dev/null
+++ b/src/backend/core/tests/mda/test_inbound_nonretryable.py
@@ -0,0 +1,70 @@
+"""Deterministic inbound failures must not be re-dispatched for 48 hours.
+
+A parse failure fails identically on every attempt, so the deferral window
+buys nothing: abandon on the first one. The failures that *are* worth
+retrying, and how long they're held, live in ``test_inbound_retry_backoff``.
+"""
+# ``process_inbound_message_task`` is a bound Celery task; calling it
+# directly is how the other task tests drive it, and pylint cannot see
+# that ``self`` is already bound.
+# pylint: disable=unused-argument, no-value-for-parameter
+
+from unittest.mock import patch
+
+import pytest
+
+from core import factories, models
+from core.mda.inbound_tasks import process_inbound_message_task
+
+
+def _inbound(mailbox, content=b"raw"):
+    blob = factories.BlobFactory(
+        mailbox=mailbox, content=content, content_type="message/rfc822"
+    )
+    return models.InboundMessage.objects.create(mailbox=mailbox, blob=blob)
+
+
+@pytest.mark.django_db
+class TestUnparseableIsAbandonedImmediately:
+    """A parse failure is deterministic — abandon on the first attempt."""
+
+    def test_parse_failure_stamps_abandoned_at(self):
+        """The stamp is what stops the 5-min sweep re-dispatching it."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(mailbox)
+
+        with patch("core.mda.inbound_tasks.parse_email", return_value=None):
+            result = process_inbound_message_task(str(inbound.id))
+
+        assert result["error"] == "abandoned"
+        inbound.refresh_from_db()
+        assert inbound.abandoned_at is not None, (
+            "an unparseable message must be stamped terminally failed, not "
+            "left live for the 5-min sweep to retry for 48h"
+        )
+        assert inbound.error_message == "Failed to parse email message"
+
+    def test_abandoned_row_is_excluded_from_the_retry_sweep(self):
+        """The stamp is what stops the loop — the sweep filters on
+        ``abandoned_at__isnull=True``."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(mailbox)
+
+        with patch("core.mda.inbound_tasks.parse_email", return_value=None):
+            process_inbound_message_task(str(inbound.id))
+
+        live = models.InboundMessage.objects.filter(abandoned_at__isnull=True)
+        assert inbound.id not in {row.id for row in live}
+
+    def test_the_blob_is_kept_for_replay(self):
+        """Abandoning must never delete the row; the blob is the only
+        copy of the mail."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(mailbox, content=b"the only copy")
+
+        with patch("core.mda.inbound_tasks.parse_email", return_value=None):
+            process_inbound_message_task(str(inbound.id))
+
+        inbound.refresh_from_db()
+        assert inbound.blob is not None
+        assert inbound.blob.get_content() == b"the only copy"
diff --git a/src/backend/core/tests/mda/test_inbound_retry_backoff.py b/src/backend/core/tests/mda/test_inbound_retry_backoff.py
new file mode 100644
index 00000000..ad3dbb79
--- /dev/null
+++ b/src/backend/core/tests/mda/test_inbound_retry_backoff.py
@@ -0,0 +1,351 @@
+"""How a held inbound message is paced and bounded.
+
+Two halves of one policy: ``_RETRY_BACKOFF`` decides *when* the sweep asks
+again, ``DEFERRAL_MAX_AGE`` decides when it stops asking. Selection is derived
+from two existing columns — ``created_at`` for the age, ``updated_at`` for the
+last attempt — so there is no per-row retry state to migrate. What these pin: a
+row is not re-dispatched before its band's interval has passed, the interval
+grows with age, a large backlog cannot be monopolised by its oldest rows, and
+an expensive-but-transient failure is held for the full window rather than cut
+short.
+"""
+# The task functions are bound Celery tasks; calling them directly is how the
+# other task tests drive them.
+# pylint: disable=unused-argument, no-value-for-parameter
+
+from unittest.mock import patch
+
+from django.utils import timezone
+
+import pytest
+from celery.exceptions import SoftTimeLimitExceeded
+
+from core import factories, models
+from core.mda.inbound_pipeline import DEFERRAL_MAX_AGE
+from core.mda.inbound_tasks import (
+    _INBOUND_TASK_SOFT_TIME_LIMIT,
+    _RETRY_BACKOFF,
+    process_inbound_message_task,
+    process_inbound_messages_queue_task,
+)
+
+TIMEDELTA_RIGHT_NOW = timezone.timedelta(0)
+
+
+def _inbound(mailbox, age=TIMEDELTA_RIGHT_NOW, since_attempt=TIMEDELTA_RIGHT_NOW):
+    """A queued row aged ``age`` whose last attempt was ``since_attempt`` ago."""
+    blob = factories.BlobFactory(
+        mailbox=mailbox, content=b"raw", content_type="message/rfc822"
+    )
+    inbound = models.InboundMessage.objects.create(mailbox=mailbox, blob=blob)
+    now = timezone.now()
+    models.InboundMessage.objects.filter(id=inbound.id).update(
+        created_at=now - age, updated_at=now - since_attempt
+    )
+    inbound.refresh_from_db()
+    return inbound
+
+
+def _dispatched():
+    """Run the sweep, returning the ids it re-dispatched."""
+    with patch(
+        "core.mda.inbound_tasks.process_inbound_message_task.delay"
+    ) as delay_mock:
+        process_inbound_messages_queue_task()
+    return {call.args[0] for call in delay_mock.call_args_list}
+
+
+@pytest.mark.django_db
+class TestRetryBackoff:
+    """Due-ness follows ``_RETRY_BACKOFF``."""
+
+    def test_freshly_queued_row_is_left_alone(self):
+        """The immediate ``.delay()`` dispatch owns the first few minutes."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(
+            mailbox, age=TIMEDELTA_RIGHT_NOW, since_attempt=TIMEDELTA_RIGHT_NOW
+        )
+        assert str(inbound.id) not in _dispatched()
+
+    def test_young_row_is_retried_on_the_short_interval(self):
+        """Under 30 min old, a 5-minute-old attempt is due again."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(
+            mailbox,
+            age=timezone.timedelta(minutes=10),
+            since_attempt=timezone.timedelta(minutes=6),
+        )
+        assert str(inbound.id) in _dispatched()
+
+    def test_old_row_is_not_retried_on_the_short_interval(self):
+        """The regression this whole change is about: a row that has been
+        failing for hours must NOT be re-dispatched every 5 minutes."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(
+            mailbox,
+            age=timezone.timedelta(hours=6),
+            since_attempt=timezone.timedelta(minutes=6),
+        )
+        assert str(inbound.id) not in _dispatched()
+
+    def test_old_row_is_retried_once_its_longer_interval_elapses(self):
+        """Backed off, not abandoned — it still gets attempts, just fewer."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(
+            mailbox,
+            age=timezone.timedelta(hours=6),
+            since_attempt=timezone.timedelta(minutes=90),
+        )
+        assert str(inbound.id) in _dispatched()
+
+    def test_intervals_grow_with_age(self):
+        """Guards the table itself: a later band must wait longer than an
+        earlier one, or the schedule isn't a backoff."""
+        intervals = [interval for _, interval in _RETRY_BACKOFF]
+        assert intervals == sorted(intervals)
+        assert intervals[0] < intervals[-1]
+        ages = [min_age for min_age, _ in _RETRY_BACKOFF]
+        assert ages == sorted(ages)
+
+    def test_a_running_row_is_not_re_dispatched(self):
+        """``process_inbound_message_task`` stamps ``updated_at`` before doing
+        any work, so a run that outlives a sweep interval doesn't have its
+        batch slot burned by dispatches that only bounce off its lock."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(
+            mailbox,
+            age=timezone.timedelta(minutes=20),
+            since_attempt=timezone.timedelta(minutes=6),
+        )
+        assert str(inbound.id) in _dispatched()
+
+        # Simulate the attempt stamp taken at lock acquisition.
+        models.InboundMessage.objects.filter(id=inbound.id).update(
+            updated_at=timezone.now()
+        )
+        assert str(inbound.id) not in _dispatched()
+
+    def test_oldest_rows_cannot_monopolise_the_batch(self):
+        """With ``created_at`` ordering and a flat schedule, a backlog larger
+        than the batch let the oldest rows hold every slot forever while newer
+        ones aged out untried. Backing them off frees the slot: the old rows
+        below are not due, so a single-slot batch goes to the young one."""
+        mailbox = factories.MailboxFactory()
+        old = [
+            _inbound(
+                mailbox,
+                age=timezone.timedelta(hours=6),
+                since_attempt=timezone.timedelta(minutes=6),
+            )
+            for _ in range(3)
+        ]
+        young = _inbound(
+            mailbox,
+            age=timezone.timedelta(minutes=20),
+            since_attempt=timezone.timedelta(minutes=6),
+        )
+
+        dispatched = _dispatched()
+        assert dispatched == {str(young.id)}
+        assert not dispatched & {str(row.id) for row in old}
+
+    def test_least_recently_attempted_goes_first(self):
+        """Among rows that are all due, the one that has waited longest since
+        its last attempt is dispatched first — so a run cut short by the batch
+        cap starves nobody."""
+        mailbox = factories.MailboxFactory()
+        recent = _inbound(
+            mailbox,
+            age=timezone.timedelta(minutes=20),
+            since_attempt=timezone.timedelta(minutes=6),
+        )
+        longest_waiting = _inbound(
+            mailbox,
+            age=timezone.timedelta(minutes=20),
+            since_attempt=timezone.timedelta(minutes=25),
+        )
+
+        with patch(
+            "core.mda.inbound_tasks.process_inbound_message_task.delay"
+        ) as delay_mock:
+            process_inbound_messages_queue_task()
+
+        dispatched = [call.args[0] for call in delay_mock.call_args_list]
+        assert dispatched == [str(longest_waiting.id), str(recent.id)]
+
+    def test_a_backlog_larger_than_a_chunk_is_fully_dispatched(self):
+        """The sweep streams the whole due set instead of dropping everything
+        past the first slice until the next tick."""
+        mailbox = factories.MailboxFactory()
+        due = [
+            _inbound(
+                mailbox,
+                age=timezone.timedelta(minutes=20),
+                since_attempt=timezone.timedelta(minutes=6),
+            )
+            for _ in range(7)
+        ]
+
+        with patch(
+            "core.mda.inbound_tasks.process_inbound_message_task.delay"
+        ) as delay_mock:
+            result = process_inbound_messages_queue_task(chunk_size=2)
+
+        dispatched = [call.args[0] for call in delay_mock.call_args_list]
+        assert set(dispatched) == {str(row.id) for row in due}
+        # Each row dispatched exactly once, not once per chunk round trip.
+        assert len(dispatched) == len(due)
+        assert result["total"] == len(due)
+
+    def test_the_dispatch_cap_bounds_one_run(self):
+        """A run cannot enqueue an unbounded burst."""
+        mailbox = factories.MailboxFactory()
+        for _ in range(7):
+            _inbound(
+                mailbox,
+                age=timezone.timedelta(minutes=20),
+                since_attempt=timezone.timedelta(minutes=6),
+            )
+
+        with patch(
+            "core.mda.inbound_tasks.process_inbound_message_task.delay"
+        ) as delay_mock:
+            result = process_inbound_messages_queue_task(chunk_size=2, max_dispatch=4)
+
+        assert len(delay_mock.call_args_list) == 4
+        assert result["total"] == 4
+
+    def test_an_exactly_full_run_is_not_reported_as_capped(self):
+        """The cap is detected by fetching one row past the limit, so a run
+        landing exactly on ``max_dispatch`` must not warn."""
+        mailbox = factories.MailboxFactory()
+        for _ in range(4):
+            _inbound(
+                mailbox,
+                age=timezone.timedelta(minutes=20),
+                since_attempt=timezone.timedelta(minutes=6),
+            )
+
+        with (
+            patch("core.mda.inbound_tasks.process_inbound_message_task.delay"),
+            patch("core.mda.inbound_tasks.logger.warning") as warn_mock,
+        ):
+            result = process_inbound_messages_queue_task(max_dispatch=4)
+
+        assert result["total"] == 4
+        assert not warn_mock.called
+
+    def test_rows_sharing_a_timestamp_are_not_skipped(self):
+        """Ordering by a non-unique column must not drop rows that tie on it."""
+        mailbox = factories.MailboxFactory()
+        rows = [
+            _inbound(
+                mailbox,
+                age=timezone.timedelta(minutes=20),
+                since_attempt=timezone.timedelta(minutes=6),
+            )
+            for _ in range(5)
+        ]
+        # Every row attempted at the exact same instant.
+        same_instant = timezone.now() - timezone.timedelta(minutes=6)
+        models.InboundMessage.objects.filter(id__in=[row.id for row in rows]).update(
+            updated_at=same_instant
+        )
+
+        with patch(
+            "core.mda.inbound_tasks.process_inbound_message_task.delay"
+        ) as delay_mock:
+            process_inbound_messages_queue_task(chunk_size=2)
+
+        dispatched = [call.args[0] for call in delay_mock.call_args_list]
+        assert set(dispatched) == {str(row.id) for row in rows}
+        assert len(dispatched) == len(rows)
+
+    def test_only_ids_are_fetched(self):
+        """Nothing here should be proportional to message size: the sweep must
+        not materialise model instances (or touch blobs) to dispatch ids."""
+        mailbox = factories.MailboxFactory()
+        _inbound(
+            mailbox,
+            age=timezone.timedelta(minutes=20),
+            since_attempt=timezone.timedelta(minutes=6),
+        )
+
+        with (
+            patch("core.mda.inbound_tasks.process_inbound_message_task.delay"),
+            patch.object(
+                models.InboundMessage, "get_raw_bytes", side_effect=AssertionError
+            ),
+        ):
+            result = process_inbound_messages_queue_task()
+
+        assert result["processed"] == 1
+
+    def test_abandoned_rows_stay_excluded(self):
+        """Unchanged by the rewrite: a poison row must never be re-dispatched."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(
+            mailbox,
+            age=timezone.timedelta(hours=6),
+            since_attempt=timezone.timedelta(hours=6),
+        )
+        models.InboundMessage.objects.filter(id=inbound.id).update(
+            abandoned_at=timezone.now()
+        )
+        assert str(inbound.id) not in _dispatched()
+
+
+@pytest.mark.django_db
+class TestSoftTimeLimitIsCaughtAndBounded:
+    """``SoftTimeLimitExceeded`` has its own ``except`` branch, ahead of the
+    generic one: it must bail out gracefully — release the lock, hold the row
+    — rather than propagate and let the hard limit kill the task mid-flight.
+    The window it uses is the generic one; a slow dependency is transient, and
+    the backoff schedule rather than a shorter deadline is what keeps those
+    retries affordable."""
+
+    def _run_with_timeout(self, inbound):
+        with patch(
+            "core.mda.inbound_tasks.parse_email",
+            side_effect=SoftTimeLimitExceeded(),
+        ):
+            return process_inbound_message_task(str(inbound.id))
+
+    def test_recent_message_is_still_held_for_retry(self):
+        """The timeout is caught, not propagated, and the row is held: the
+        transient case (a slow webhook chain recovering) gets its retries."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(mailbox)
+
+        result = self._run_with_timeout(inbound)
+
+        assert result["error"] == "retry"
+        inbound.refresh_from_db()
+        assert inbound.abandoned_at is None
+
+    def test_message_held_past_an_hour_is_not_abandoned(self):
+        """Regression guard on a deliberate removal: this path once had its
+        own 1h window, so an hour of provider slowness lost the message."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(mailbox, age=timezone.timedelta(hours=6))
+
+        result = self._run_with_timeout(inbound)
+
+        assert result["error"] == "retry"
+        inbound.refresh_from_db()
+        assert inbound.abandoned_at is None
+
+    def test_message_past_the_deferral_window_is_abandoned(self):
+        """Still bounded, and the reason names the limit that was hit, which
+        is what tells an operator this was a timeout and not a crash."""
+        mailbox = factories.MailboxFactory()
+        inbound = _inbound(
+            mailbox, age=DEFERRAL_MAX_AGE + timezone.timedelta(minutes=5)
+        )
+
+        result = self._run_with_timeout(inbound)
+
+        assert result["error"] == "abandoned"
+        inbound.refresh_from_db()
+        assert inbound.abandoned_at is not None
+        assert str(_INBOUND_TASK_SOFT_TIME_LIMIT) in inbound.error_message
diff --git a/src/backend/core/tests/mda/test_spam_processing.py b/src/backend/core/tests/mda/test_spam_processing.py
index 58bac02a..073bc598 100644
--- a/src/backend/core/tests/mda/test_spam_processing.py
+++ b/src/backend/core/tests/mda/test_spam_processing.py
@@ -1085,14 +1085,15 @@ class TestProcessInboundMessagesQueueTask:
         """Test that the queue processing task triggers individual message processing."""
         mailbox = factories.MailboxFactory()
 
-        # Create multiple pending messages older than 5 minutes (for retry processing)
+        # Create multiple pending messages due for a retry. Both timestamps are
+        # backdated: ``created_at`` picks the backoff band, ``updated_at`` is
+        # the last attempt the band's interval is measured from.
         old_time = timezone.now() - timezone.timedelta(minutes=6)
         for i in range(3):
             # Distinct bytes per row so blob dedup doesn't collapse them.
             inbound_message = _queue_inbound(mailbox, f"Content {i}".encode())
-            # Update created_at to make it old enough for retry
             models.InboundMessage.objects.filter(id=inbound_message.id).update(
-                created_at=old_time
+                created_at=old_time, updated_at=old_time
             )
 
         # Call the bound task directly using .run() method
diff --git a/src/backend/core/tests/models/test_message_get_parsed_data.py b/src/backend/core/tests/models/test_message_get_parsed_data.py
index da00b2d5..0eae4c35 100644
--- a/src/backend/core/tests/models/test_message_get_parsed_data.py
+++ b/src/backend/core/tests/models/test_message_get_parsed_data.py
@@ -59,6 +59,63 @@ class TestMessageGetParsedData:
             assert not message.get_parsed_data()
 
 
+@pytest.mark.django_db
+class TestUnreadableContent:
+    """A stored blob that no longer parses must reach the UI as an error,
+    not as an empty message."""
+
+    def _message_with_blob(self, content):
+        fake_blob = MagicMock()
+        if isinstance(content, Exception):
+            fake_blob.get_content.side_effect = content
+        else:
+            fake_blob.get_content.return_value = content
+        message = models.Message()
+        message.id = "test-id"
+        message.postmark = None
+        return message, patch.object(
+            models.Message, "blob", new_callable=PropertyMock, return_value=fake_blob
+        )
+
+    def test_header_over_the_cap_is_reported_unreadable(self):
+        """The retroactive case: bytes accepted under looser rules that the
+        parser now refuses outright."""
+        raw = b"From: a@example.com\r\nSubject: " + b"x" * 200_000 + b"\r\n\r\nbody\r\n"
+        message, blob_patch = self._message_with_blob(raw)
+        with blob_patch:
+            assert message.get_parsed_data() == {}
+            assert message.has_unreadable_content()
+            assert message.get_stmsg_headers()["unreadable"] == "true"
+
+    def test_blob_read_failure_is_reported_unreadable(self):
+        """Decompression / decryption / integrity failure renders the same
+        blank message, so it takes the same banner."""
+        message, blob_patch = self._message_with_blob(ValueError("bad blob"))
+        with blob_patch:
+            assert message.has_unreadable_content()
+            assert message.get_stmsg_headers()["unreadable"] == "true"
+
+    def test_readable_message_is_not_flagged(self):
+        """A message that parses carries no marker."""
+        message, blob_patch = self._message_with_blob(
+            b"From: a@example.com\r\nSubject: hi\r\n\r\nbody\r\n"
+        )
+        with blob_patch:
+            assert not message.has_unreadable_content()
+            assert "unreadable" not in message.get_stmsg_headers()
+
+    def test_blobless_message_is_not_flagged(self):
+        """A draft skeleton has no content to fail on — banner would be a
+        false alarm on every empty draft."""
+        message = models.Message()
+        message.postmark = None
+        with patch.object(
+            models.Message, "blob", new_callable=PropertyMock, return_value=None
+        ):
+            assert not message.has_unreadable_content()
+            assert "unreadable" not in message.get_stmsg_headers()
+
+
 class TestGetStmsgHeaders:
     """``get_stmsg_headers`` unions legacy baked ``X-StMsg-*`` bytes with the
     structured ``postmark``; the structured value wins on overlap."""
diff --git a/src/backend/core/tests/services/test_attachments.py b/src/backend/core/tests/services/test_attachments.py
new file mode 100644
index 00000000..d043b65f
--- /dev/null
+++ b/src/backend/core/tests/services/test_attachments.py
@@ -0,0 +1,137 @@
+"""Tests for :mod:`core.services.attachments` — the display-name policy."""
+
+import mimetypes
+
+import pytest
+
+from core.models import Attachment
+from core.services.attachments import (
+    ATTACHMENT_NAME_MAX_LENGTH,
+    UNNAMED_ATTACHMENT_STEM,
+    get_attachment_display_name,
+    guess_mime_extension,
+)
+
+
+def test_name_cap_matches_the_model_field():
+    """The truncation limit must be the column's, or saves start failing.
+
+    ``get_attachment_display_name`` truncates so ``full_clean`` never
+    rejects a draft over one overlong attachment name. That only holds
+    while the constant tracks the field it is protecting.
+    """
+    assert ATTACHMENT_NAME_MAX_LENGTH == Attachment._meta.get_field("name").max_length
+
+
+@pytest.mark.parametrize(
+    ("content_type", "expected"),
+    [
+        pytest.param("image/gif", ".gif", id="image"),
+        pytest.param("application/pdf", ".pdf", id="document"),
+        pytest.param("text/calendar", ".ics", id="calendar-invite"),
+        pytest.param("text/vcard", ".vcf", id="standard-vcard"),
+        pytest.param("application/x-zip-compressed", ".zip", id="outlook-zip"),
+        pytest.param("application/pkcs7-mime", ".p7m", id="smime-message"),
+        pytest.param("application/xml", ".xml", id="xml"),
+        # Deliberately extensionless: the generic binary type names no format.
+        pytest.param("application/octet-stream", "", id="generic-binary"),
+        pytest.param("application/x-unknown", "", id="unknown-type"),
+        pytest.param("", "", id="empty"),
+        pytest.param(None, "", id="none"),
+    ],
+)
+def test_guess_mime_extension(content_type, expected):
+    assert guess_mime_extension(content_type) == expected
+
+
+def test_guess_mime_extension_drops_parameters():
+    assert guess_mime_extension('text/calendar; charset="utf-8"') == ".ics"
+
+
+def test_guess_mime_extension_is_case_insensitive():
+    assert guess_mime_extension("Application/PDF") == ".pdf"
+
+
+def test_guess_mime_extension_does_not_consult_the_host(monkeypatch):
+    """The table is self-contained: no ``mimetypes``, no ``/etc/mime.types``.
+
+    The stdlib table is completed at import time from system files, which
+    would make a stored ``Attachment.name`` a property of the host image.
+    Poison ``mimetypes`` to prove we never reach it.
+    """
+
+    def _explode(*args, **kwargs):
+        raise AssertionError("guess_mime_extension consulted the stdlib table")
+
+    monkeypatch.setattr(mimetypes, "guess_extension", _explode)
+    monkeypatch.setattr(mimetypes, "init", _explode)
+
+    assert guess_mime_extension("image/png") == ".png"
+    assert guess_mime_extension("application/x-unknown") == ""
+
+
+class TestGetAttachmentDisplayName:
+    """A part always resolves to a non-empty, storable name."""
+
+    def test_keeps_a_usable_name(self):
+        assert get_attachment_display_name("report.pdf", "application/pdf") == (
+            "report.pdf"
+        )
+
+    @pytest.mark.parametrize(
+        "name",
+        [
+            pytest.param(None, id="none"),
+            pytest.param("", id="empty"),
+            # Sanitizes away to nothing — the fallback has to catch this too.
+            pytest.param("...", id="sanitizes-to-empty"),
+        ],
+    )
+    def test_synthesizes_a_name_with_the_inferred_extension(self, name):
+        assert get_attachment_display_name(name, "image/gif") == "unnamed.gif"
+
+    def test_synthesizes_a_bare_stem_for_an_unknown_type(self):
+        assert (
+            get_attachment_display_name(None, "application/x-unknown")
+            == UNNAMED_ATTACHMENT_STEM
+        )
+
+    def test_synthesizes_a_bare_stem_without_a_type(self):
+        assert get_attachment_display_name(None) == UNNAMED_ATTACHMENT_STEM
+
+    def test_sanitizes_a_client_supplied_name(self):
+        """Names off the wire reach us unvalidated — path traversal included."""
+        assert get_attachment_display_name("../../etc/passwd", "text/plain") == (
+            "passwd"
+        )
+
+    def test_truncates_to_the_column_width_keeping_the_extension(self):
+        name = get_attachment_display_name("a" * 300 + ".gif", "image/gif")
+        assert len(name) == ATTACHMENT_NAME_MAX_LENGTH
+        assert name.endswith(".gif")
+
+
+@pytest.mark.parametrize(
+    "name",
+    [
+        pytest.param(123, id="int"),
+        pytest.param(["a.txt"], id="list"),
+        pytest.param({"n": "a.txt"}, id="dict"),
+        pytest.param(True, id="bool"),
+        pytest.param(object(), id="object"),
+    ],
+)
+def test_truthy_non_string_name_falls_back_instead_of_raising(name):
+    """``attachments[].name`` reaches us straight off the wire — the draft
+    endpoint takes it with no serializer validation — so a client can send
+    a number or a list. Those are truthy, and ``sanitize_filename`` slices
+    its argument, so a truthiness check let them through to a TypeError and
+    a 500. Anything that is not a string is treated as no name at all.
+    """
+    assert get_attachment_display_name(name, "image/png") == "unnamed.png"
+
+
+def test_string_names_are_unaffected():
+    assert get_attachment_display_name("report.pdf", "application/pdf") == "report.pdf"
+    assert get_attachment_display_name("", "image/png") == "unnamed.png"
+    assert get_attachment_display_name(None, "image/png") == "unnamed.png"
diff --git a/src/backend/pyproject.toml b/src/backend/pyproject.toml
index 68eb9575..f5be6413 100644
--- a/src/backend/pyproject.toml
+++ b/src/backend/pyproject.toml
@@ -53,7 +53,7 @@ dependencies = [
     "factory_boy==3.3.3",
     "gunicorn==25.1.0",
     "icalendar==7.0.3",
-    "jmap-email==0.1.0",
+    "jmap-email==0.3.0",
     "jsonschema==4.26.0",
     "nested-multipart-parser==1.6.0",
     "openai==2.21.0",
@@ -163,6 +163,8 @@ extra-standard-library = ["tomllib"]
 "jmap_email.composer".msg = "Import from the top-level `jmap_email` package (shapes: `jmap_email.types`)."
 "jmap_email.options".msg = "Import from the top-level `jmap_email` package (shapes: `jmap_email.types`)."
 "jmap_email.preview".msg = "Import from the top-level `jmap_email` package (shapes: `jmap_email.types`)."
+"jmap_email.addresses".msg = "Import from the top-level `jmap_email` package (shapes: `jmap_email.types`)."
+"jmap_email.filenames".msg = "Import from the top-level `jmap_email` package (shapes: `jmap_email.types`)."
 
 [tool.ruff.lint.per-file-ignores]
 "**/tests/*" = ["S", "SLF"]
diff --git a/src/backend/uv.lock b/src/backend/uv.lock
index f2eaa85b..7e494467 100644
--- a/src/backend/uv.lock
+++ b/src/backend/uv.lock
@@ -1013,11 +1013,14 @@ wheels = [
 
 [[package]]
 name = "jmap-email"
-version = "0.1.0"
+version = "0.3.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e5/65/86143692dc38e57ea950d85f6a7d66ea9df9a3b64debd8d258cb8ba2cb4d/jmap_email-0.1.0.tar.gz", hash = "sha256:04d906c22aff25ac62819bda04f467bc9f65e041be4a1f96bb5439a54c4d2de1", size = 149817, upload-time = "2026-06-09T13:41:36.923Z" }
+dependencies = [
+    { name = "idna" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/94/ca/821eee73857d072e78160e71de0f777d0a7551c11f473d656018d1ecb422/jmap_email-0.3.0.tar.gz", hash = "sha256:d559085692bb6694631baf5ab3f75843bc3fbf44dc03bc889b4c82aa2f583f01", size = 252066, upload-time = "2026-08-05T22:14:07.872Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/67/bf/41fd7b86859391bc112114162ae9d21d1a98c09986dfdc0872132e19dbf7/jmap_email-0.1.0-py3-none-any.whl", hash = "sha256:a32d8b49f5f33ae27de1279a1d402c1714c85c35a1b55844251c985a32ff1498", size = 59633, upload-time = "2026-06-09T13:41:34.395Z" },
+    { url = "https://files.pythonhosted.org/packages/37/f1/b90eb401d3155eedf36384c6840730f87b8730302317e955e97d060c32da/jmap_email-0.3.0-py3-none-any.whl", hash = "sha256:27f30fa35782c4f4fd214387484c85cba46db68c5ccf82e1aa4daddd881cd0f9", size = 98681, upload-time = "2026-08-05T22:14:06.55Z" },
 ]
 
 [[package]]
@@ -1269,7 +1272,7 @@ requires-dist = [
     { name = "httpx", extras = ["http2"], specifier = "==0.28.1" },
     { name = "hypothesis", marker = "extra == 'dev'", specifier = "==6.151.9" },
     { name = "icalendar", specifier = "==7.0.3" },
-    { name = "jmap-email", specifier = "==0.1.0" },
+    { name = "jmap-email", specifier = "==0.3.0" },
     { name = "jsonschema", specifier = "==4.26.0" },
     { name = "libpff-python", specifier = "==20231205" },
     { name = "nested-multipart-parser", specifier = "==1.6.0" },
diff --git a/src/frontend/public/locales/common/en-US.json b/src/frontend/public/locales/common/en-US.json
index 8ab8efd0..b05dd2e9 100644
--- a/src/frontend/public/locales/common/en-US.json
+++ b/src/frontend/public/locales/common/en-US.json
@@ -850,6 +850,7 @@
   "This file looks suspicious": "This file looks suspicious",
   "This is the mailbox owner, its access cannot be modified.": "This is the mailbox owner, its access cannot be modified.",
   "This is the only admin of this mailbox, you cannot therefore modify its access.": "This is the only admin of this mailbox, you cannot therefore modify its access.",
+  "This message could not be read and cannot be displayed. Its content is still stored and support can retrieve it.": "This message could not be read and cannot be displayed. Its content is still stored and support can retrieve it.",
   "This message failed sender authentication and is likely a forgery. Do not trust it.": "This message failed sender authentication and is likely a forgery. Do not trust it.",
   "This message has {{count}} attachments_one": "This message has one attachment",
   "This message has {{count}} attachments_other": "This message has {{count}} attachments",
diff --git a/src/frontend/public/locales/common/fr-FR.json b/src/frontend/public/locales/common/fr-FR.json
index c9ac8f5f..0da16b4c 100644
--- a/src/frontend/public/locales/common/fr-FR.json
+++ b/src/frontend/public/locales/common/fr-FR.json
@@ -938,6 +938,7 @@
   "This file looks suspicious": "Ce fichier semble suspect",
   "This is the mailbox owner, its access cannot be modified.": "Il s'agit du propriétaire de la boîte aux lettres, son accès ne peut pas être modifié.",
   "This is the only admin of this mailbox, you cannot therefore modify its access.": "C'est le seul administrateur de cette boîte aux lettres, vous ne pouvez donc pas modifier son accès.",
+  "This message could not be read and cannot be displayed. Its content is still stored and support can retrieve it.": "Ce message n'a pas pu être lu et ne peut pas être affiché. Son contenu est toujours conservé et l'équipe support peut le récupérer.",
   "This message failed sender authentication and is likely a forgery. Do not trust it.": "L'authentification de l'expéditeur a échoué pour ce message, qui est probablement frauduleux. Ne lui faites pas confiance.",
   "This message has {{count}} attachments_one": "Ce message a une pièce jointe",
   "This message has {{count}} attachments_many": "Ce message a {{count}} pièces jointes",
diff --git a/src/frontend/public/locales/common/nl-NL.json b/src/frontend/public/locales/common/nl-NL.json
index ede9dff7..7c63f36c 100644
--- a/src/frontend/public/locales/common/nl-NL.json
+++ b/src/frontend/public/locales/common/nl-NL.json
@@ -842,6 +842,7 @@
   "This file looks suspicious": "",
   "This is the mailbox owner, its access cannot be modified.": "",
   "This is the only admin of this mailbox, you cannot therefore modify its access.": "Dit is de enige admin van deze mailbox, u kunt daarom de toegang niet wijzigen.",
+  "This message could not be read and cannot be displayed. Its content is still stored and support can retrieve it.": "Dit bericht kon niet worden gelezen en kan niet worden weergegeven. De inhoud is nog steeds opgeslagen en de supportafdeling kan deze ophalen.",
   "This message failed sender authentication and is likely a forgery. Do not trust it.": "",
   "This message has {{count}} attachments_one": "Deze e-mail heeft een bijlage",
   "This message has {{count}} attachments_other": "Dit bericht heeft {{count}} bijlagen",
diff --git a/src/frontend/public/locales/common/ru-RU.json b/src/frontend/public/locales/common/ru-RU.json
index b810fa3d..6f946041 100644
--- a/src/frontend/public/locales/common/ru-RU.json
+++ b/src/frontend/public/locales/common/ru-RU.json
@@ -770,6 +770,7 @@
   "This file looks suspicious": "Этот файл выглядит подозрительным",
   "This is the mailbox owner, its access cannot be modified.": "Это владелец почтового ящика, его доступ не может быть изменён.",
   "This is the only admin of this mailbox, you cannot therefore modify its access.": "Это единственный администратор данного почтового ящика, поэтому вы не можете изменить доступ к нему.",
+  "This message could not be read and cannot be displayed. Its content is still stored and support can retrieve it.": "Это сообщение не удалось прочитать, и оно не может быть отображено. Его содержимое по-прежнему хранится, и команда поддержки может его извлечь.",
   "This message failed sender authentication and is likely a forgery. Do not trust it.": "Это сообщение не прошло аутентификацию отправителя и, скорее всего, это подделка. Не доверяйте ему.",
   "This message has {{count}} attachments_one": "Это сообщение содержит одно вложение",
   "This message has {{count}} attachments_other": "Это сообщение содержит {{count}} вложений",
diff --git a/src/frontend/public/locales/common/uk-UA.json b/src/frontend/public/locales/common/uk-UA.json
index b6a7aaf2..c72ad69d 100644
--- a/src/frontend/public/locales/common/uk-UA.json
+++ b/src/frontend/public/locales/common/uk-UA.json
@@ -770,6 +770,7 @@
   "This file looks suspicious": "Цей файл виглядає підозріло",
   "This is the mailbox owner, its access cannot be modified.": "Це власник поштової скриньки, доступ до неї не можна змінити.",
   "This is the only admin of this mailbox, you cannot therefore modify its access.": "Це єдиний адміністратор цієї поштової скриньки, тому ви не можете змінити доступ до неї.",
+  "This message could not be read and cannot be displayed. Its content is still stored and support can retrieve it.": "Це повідомлення не вдалося прочитати, і його неможливо відобразити. Його вміст досі зберігається, і служба підтримки може його отримати.",
   "This message failed sender authentication and is likely a forgery. Do not trust it.": "Це повідомлення не пройшло перевірку автентичності відправника і, ймовірно, є підробкою. Не довіряйте йому.",
   "This message has {{count}} attachments_one": "Це повідомлення має одне вкладення",
   "This message has {{count}} attachments_other": "Це повідомлення має {{count}} вкладень",
diff --git a/src/frontend/src/features/layouts/components/thread-view/components/thread-message/renderers/text_html.security.test.ts b/src/frontend/src/features/layouts/components/thread-view/components/thread-message/renderers/text_html.security.test.ts
new file mode 100644
index 00000000..d4094089
--- /dev/null
+++ b/src/frontend/src/features/layouts/components/thread-view/components/thread-message/renderers/text_html.security.test.ts
@@ -0,0 +1,99 @@
+/**
+ * Adversarial coverage for the text/html body path.
+ *
+ * The body is attacker-controlled: anyone can email it. Defence is in two
+ * layers — DOMPurify, then the sandboxed CSP iframe it is mounted in.
+ * Both are asserted here, because each is load-bearing for a different
+ * class: DOMPurify for script execution, the frame for layout, forms and
+ * remote loads (which survive sanitisation by design).
+ */
+import { describe, it, expect } from "vitest";
+import { renderTextHtml } from "./text_html";
+
+const render = (html: string) => renderTextHtml(html, new Map());
+
+describe("layer 1 — sanitiser blocks script execution", () => {
+  it.each([
+    ["script tag", `<script>alert(1)</script>`],
+    ["img onerror", `<img src=x onerror=alert(1)>`],
+    ["svg onload", `<svg onload=alert(1)>`],
+    ["input autofocus onfocus", `<input autofocus onfocus=alert(1)>`],
+    ["iframe srcdoc", `<iframe srcdoc="<script>alert(1)</script>"></iframe>`],
+    ["object data", `<object data="//evil.example/x"></object>`],
+    ["embed", `<embed src="//evil.example/x">`],
+  ])("neutralises %s", (_label, payload) => {
+    const out = render(payload);
+    expect(out).not.toMatch(/<script/i);
+    expect(out).not.toMatch(/\son(error|load|focus|click)\s*=/i);
+    expect(out).not.toMatch(/<i?frame|<object|<embed/i);
+  });
+
+  it.each([
+    ["javascript:", `<a href="javascript:alert(1)">x</a>`],
+    ["data:text/html", `<a href="data:text/html,<script>alert(1)</script>">x</a>`],
+    ["vbscript:", `<a href="vbscript:msgbox(1)">x</a>`],
+  ])("strips dangerous URL scheme %s", (_label, payload) => {
+    expect(render(payload)).not.toMatch(/javascript:|vbscript:|data:text\/html/i);
+  });
+
+  it("forces safe rel/target on links", () => {
+    const out = render(`<a href="https://example.com">x</a>`);
+    expect(out).toContain('rel="noopener noreferrer"');
+    expect(out).toContain('target="_blank"');
+  });
+
+  it("drops tracking pixels", () => {
+    expect(render(`<img src="https://e.example/p.gif" width="1" height="1">`))
+      .not.toMatch(/<img/i);
+    expect(render(`<img src="https://e.example/p.gif" style="display:none">`))
+      .not.toMatch(/<img/i);
+  });
+});
+
+describe("layer 2 — the frame contains what the sanitiser lets through", () => {
+  /**
+   * These payloads survive DOMPurify on purpose: a message may legitimately
+   * use inline styles, and stripping every layout property would mangle
+   * ordinary mail. They are contained by the iframe instead — `position:
+   * fixed` resolves against the iframe viewport, forms cannot submit
+   * without `allow-forms`, and remote loads are refused by `img-src`.
+   *
+   * Asserted so that if anyone moves this body out of the frame, or
+   * loosens the sandbox/CSP, these become live vulnerabilities and this
+   * test says so.
+   */
+  it("documents that layout and form markup do survive sanitisation", () => {
+    expect(render(`<div style="position:fixed;top:0">x</div>`)).toMatch(/position/i);
+    expect(render(`<form action="//evil.example"><input name="pw"></form>`)).toMatch(/<input/i);
+    expect(render(`<img srcset="//evil.example/t.png 1x" width="99" height="99">`))
+      .toMatch(/srcset/i);
+  });
+
+  it("the mount is a sandboxed iframe that cannot run scripts or submit forms", async () => {
+    const src = await import("fs").then((fs) =>
+      fs.readFileSync(
+        "src/features/layouts/components/thread-view/components/thread-message/thread-message-body.tsx",
+        "utf8"
+      )
+    );
+    const sandbox = src.match(/sandbox="([^"]+)"/)?.[1] ?? "";
+    expect(sandbox).not.toContain("allow-scripts");
+    expect(sandbox).not.toContain("allow-forms");
+    expect(src).toMatch(/srcDoc=\{wrappedHtml\}/);
+  });
+
+  it("the frame CSP forbids scripts and non-proxied remote loads", async () => {
+    const src = await import("fs").then((fs) =>
+      fs.readFileSync(
+        "src/features/layouts/components/thread-view/components/thread-message/thread-message-body.tsx",
+        "utf8"
+      )
+    );
+    expect(src).toMatch(/"script-src 'none'"/);
+    expect(src).toMatch(/"default-src 'none'"/);
+    expect(src).toMatch(/"connect-src 'none'"/);
+    // Remote images only via our own origin/API — this is what stops a
+    // srcset or background attribute leaking a read receipt.
+    expect(src).toMatch(/img-src 'self' data: \$\{getApiOrigin\(\)\}/);
+  });
+});
diff --git a/src/frontend/src/features/layouts/components/thread-view/components/thread-message/thread-message-header.tsx b/src/frontend/src/features/layouts/components/thread-view/components/thread-message/thread-message-header.tsx
index 12df252b..2584be85 100644
--- a/src/frontend/src/features/layouts/components/thread-view/components/thread-message/thread-message-header.tsx
+++ b/src/frontend/src/features/layouts/components/thread-view/components/thread-message/thread-message-header.tsx
@@ -46,6 +46,11 @@ const ThreadMessageHeader = ({
     // the inbound deferral window expired. Warn prominently.
     const processingFailed = Boolean(message.stmsg_headers?.['processing-failed']);
 
+    // The stored message exists but the server cannot turn it back into
+    // content, so every field below renders blank. Say so rather than
+    // showing an empty message.
+    const isUnreadable = Boolean(message.stmsg_headers?.['unreadable']);
+
     // Spam scanning flagged the message as probable spam but below the Junk
     // threshold, so it was delivered to the inbox with a graded marker
     // ('possible' < 'likely'). Show an inline caution banner.
@@ -178,6 +183,13 @@ const ThreadMessageHeader = ({
                             </div>
                         </Banner>
                     )}
+                    {isUnreadable && (
+                        <Banner type="error" compact fullWidth>
+                            <div className="thread-message__header-banner__content">
+                                <p>{t("This message could not be read and cannot be displayed. Its content is still stored and support can retrieve it.")}</p>
+                            </div>
+                        </Banner>
+                    )}
                     {processingFailed && (
                         <Banner type="error" compact fullWidth>
                             <div className="thread-message__header-banner__content">
diff --git a/src/frontend/src/features/utils/unquote-message/constants.ts b/src/frontend/src/features/utils/unquote-message/constants.ts
index ebbcb73e..e843bd7c 100644
--- a/src/frontend/src/features/utils/unquote-message/constants.ts
+++ b/src/frontend/src/features/utils/unquote-message/constants.ts
@@ -42,65 +42,69 @@ export const REPLY_PATTERNS = [
 
   // ==================== Main Language Patterns ====================
   // English - On DATE, NAME <EMAIL> wrote:
-  /^>*-*\s*((on|in a message dated)\s.+\s.+?(wrote|sent)\s*:)\s?-*/im,
+  /^>*-*[^\S\r\n]{0,20}((on|in a message dated)\s.{1,500}\s.{1,500}?(wrote|sent)\s*:)\s?-*/im,
   // French - Le DATE, NAME a écrit:
-  /^>*-*\s*((le)\s.+\s.+?(écrit)\s*:)\s?/im,
+  /^>*-*[^\S\r\n]{0,20}((le)\s.{1,500}\s.{1,500}?(écrit)\s*:)\s?/im,
   // Spanish - El DATE, NAME escribió:
-  /^>*-*\s*((el)\s.+\s.+?(escribió)\s*:)\s?/im,
+  /^>*-*[^\S\r\n]{0,20}((el)\s.{1,500}\s.{1,500}?(escribió)\s*:)\s?/im,
   // Italian - Il DATE, NAME scritto:
-  /^>*-*\s*((il)\s.+\s.+?(scritto)\s*:)\s?/im,
+  /^>*-*[^\S\r\n]{0,20}((il)\s.{1,500}\s.{1,500}?(scritto)\s*:)\s?/im,
   // Portuguese - Em DATE, NAME escreveu:
-  /^>*-*\s*((em)\s.+\s.+?(escreveu)\s*:)\s?/im,
+  /^>*-*[^\S\r\n]{0,20}((em)\s.{1,500}\s.{1,500}?(escreveu)\s*:)\s?/im,
   // German - Am DATE schrieb NAME <EMAIL>:
-  /^\s*(am\s.+\s)schrieb.+\s?(\[|<).+(\]|>):/im,
+  /^[^\S\r\n]{0,20}(am\s.{1,500}\s)schrieb.{1,500}\s?(\[|<).{1,500}(\]|>):/im,
   // Dutch - Op DATE, schreef NAME <EMAIL>:
-  /^\s*(op\s[\s\S]{1,500}?(schreef|verzond|geschreven)[^\r\n]+:)/im,
+  /^[^\S\r\n]{0,20}(op\s[\s\S]{1,500}?(schreef|verzond|geschreven)[^\r\n]+:)/im,
   // Polish - W dniu DATE, NAME pisze|napisał:
-  /^\s*((w\sdniu|dnia)\s[\s\S]{1,500}?(pisze|napisał(\(a\))?):)/im,
+  /^[^\S\r\n]{0,20}((w\sdniu|dnia)\s[\s\S]{1,500}?(pisze|napisał(\(a\))?):)/im,
   // Swedish/Danish - Den DATE skrev NAME <EMAIL>:
-  /^\s*(den|d.)?\s?.+\s?skrev\s?".+"\s*[\[|<].+[\]|>]\s?:/im,
+  // `[^\S\r\n]` (horizontal whitespace) instead of `\s`: with the m flag
+  // an unbounded `\s` quantifier crosses newlines and backtracks once
+  // per line start — quadratic in the number of lines of the body.
+  /^[^\S\r\n]{0,20}(den|d.)?\s?.{1,500}\s?skrev\s?".{1,500}"[^\S\r\n]{0,20}[\[|<].{1,500}[\]|>]\s?:/im,
   // Vietnamese - Vào DATE đã viết NAME <EMAIL>:
-  /^\s*(vào\s.+\sđã viết\s.+:)/im,
+  /^[^\S\r\n]{0,20}(vào\s.{1,500}\sđã viết\s.{1,500}:)/im,
   // Finnish - pe DATE NAME <EMAIL> kirjoitti:
-  /^\s*(pe\s.+\s.+kirjoitti:)/im,
+  /^[^\S\r\n]{0,20}(pe\s.{1,500}\s.{1,500}kirjoitti:)/im,
   // Chinese - 在 DATE, TIME, NAME 写道：
   /^(在[\s\S]{1,500}写道：)/m,
 
   // ==================== Outlook 2019 Patterns ====================
-  // Outlook 2019 (Norwegian)
-  /^\s?.+\s*[\[|<].+[\]|>]\s?skrev følgende den\s?.+\s?:/m,
+  // Outlook 2019 (Norwegian) — horizontal whitespace only, see the
+  // Swedish/Danish pattern above.
+  /^\s?.{1,500}[^\S\r\n]{0,20}[\[|<].{1,500}[\]|>]\s?skrev følgende den\s?.{1,500}\s?:/m,
   // Outlook 2019 (Czech)
-  /^\s?dne\s?.+\,\s?.+\s*[\[|<].+[\]|>]\s?napsal\(a\)\s?:/im,
+  /^\s?dne\s?.{1,500}\,\s?.{1,500}\s*[\[|<].{1,500}[\]|>]\s?napsal\(a\)\s?:/im,
   // Outlook 2019 (Russian)
-  /^\s?.+\s?пользователь\s?".+"\s*[\[|<].+[\]|>]\s?написал\s?:/im,
+  /^\s?.{1,500}\s?пользователь\s?".{1,500}"\s*[\[|<].{1,500}[\]|>]\s?написал\s?:/im,
   // Outlook 2019 (Slovak)
-  /^\s?.+\s?používateľ\s?.+\s*\([\[|<].+[\]|>]\)\s?napísal\s?:/im,
+  /^\s?.{1,500}\s?používateľ\s?.{1,500}\s*\([\[|<].{1,500}[\]|>]\)\s?napísal\s?:/im,
   // Outlook 2019 (Swedish)
-  /\s?Den\s?.+\s?skrev\s?".+"\s*[\[|<].+[\]|>]\s?följande\s?:/m,
+  /\s?Den\s?.{1,500}\s?skrev\s?".{1,500}"\s*[\[|<].{1,500}[\]|>]\s?följande\s?:/m,
   // Outlook 2019 (Turkish)
-  /^\s?".+"\s*[\[|<].+[\]|>]\,\s?.+\s?tarihinde şunu yazdı\s?:/im,
+  /^\s?".{1,500}"\s*[\[|<].{1,500}[\]|>]\,\s?.{1,500}\s?tarihinde şunu yazdı\s?:/im,
   // Outlook 2019 (Hungarian)
-  /^\s?.+\s?időpontban\s?.+\s*[\[|<|(].+[\]|>|)]\s?ezt írta\s?:/im,
+  /^\s?.{1,500}\s?időpontban\s?.{1,500}\s*[\[|<|(].{1,500}[\]|>|)]\s?ezt írta\s?:/im,
 
   // ==================== Additional Patterns ====================
   // NAME <EMAIL> schrieb:
-  /^(.+\s<.+>\sschrieb\s?:)/im,
+  /^(.{1,500}\s<.{1,500}>\sschrieb\s?:)/im,
   // NAME on DATE wrote:
-  /^(.+\son.*at.*wrote:)/im,
+  /^(.{1,500}\son.{0,500}at.{0,500}wrote:)/im,
   // "From: NAME <EMAIL>" (multiple languages)
-  /^\s*((from|van|de|von|da)\s?:.+\s?\n?\s*(\[|<).+(\]|>))/im,
+  /^[^\S\r\n]{0,20}((from|van|de|von|da)\s?:.{1,500}\s?\n?[^\S\r\n]{0,20}(\[|<).{1,500}(\]|>))/im,
 
   // ==================== Date Starting Patterns ====================
   // Korean - DATE TIME NAME 작성:
-  /^(20[0-9]{2}\..+\s작성:)$/m,
+  /^(20[0-9]{2}\..{1,500}\s작성:)$/m,
   // Japanese - DATE TIME、NAME のメッセージ:
-  /^(20[0-9]{2}\/.+のメッセージ:)/m,
+  /^(20[0-9]{2}\/.{1,500}のメッセージ:)/m,
   // ISO Date format - 20YY-MM-DD HH:II GMT+01:00 NAME <EMAIL>:
-  /^(20[0-9]{2})-([0-9]{2}).([0-9]{2}).([0-9]{2}):([0-9]{2})\n?(.*)>:/m,
+  /^(20[0-9]{2})-([0-9]{2}).([0-9]{2}).([0-9]{2}):([0-9]{2})\n?(.{0,500})>:/m,
   // European Date format - DD.MM.20YY HH:II NAME <EMAIL>
-  /^([0-9]{2}).([0-9]{2}).(20[0-9]{2})(.*)(([0-9]{2}).([0-9]{2}))(.*)"\s*<(.*)>\s*:/m,
+  /^([0-9]{2}).([0-9]{2}).(20[0-9]{2})(.{0,500})(([0-9]{2}).([0-9]{2}))(.{0,500})"\s*<(.{0,500})>\s*:/m,
   // Time first format - HH:II, DATE, NAME <EMAIL>:
-  /^[0-9]{2}:[0-9]{2}(.*)[0-9]{4}(.*)"\s*<(.*)>\s*:/m,
+  /^[0-9]{2}:[0-9]{2}(.{0,500})[0-9]{4}(.{0,500})"\s*<(.{0,500})>\s*:/m,
   // Russian format - 02.04.2012 14:20 пользователь "bob@example.com" <bob@xxx.mailgun.org> написал:
   /(\d+\/\d+\/\d+|\d+\.\d+\.\d+)[^\r\n]{0,500}\s\S+@\S+:/,
   // ISO 8601 with timezone - 2014-10-17 11:28 GMT+03:00 Bob <bob@example.com>:
@@ -111,7 +115,7 @@ export const REPLY_PATTERNS = [
   // ==================== Dash Delimiter Patterns ====================
   // Original Message delimiter (multi-language)
   new RegExp(
-    `^>?\\s*-{3,12}\\s*(` +
+    `^>?[^\\S\\r\\n]{0,20}-{3,12}\\s*(` +
       `original message|` +
       `reply message|` +
       `original text|` +
@@ -137,10 +141,10 @@ export const REPLY_PATTERNS = [
     "im"
   ),
   // Generic separators
-  /\r?\n\s*_{5,}\s*\r?\n/,
-  /\r?\n\s*-{5,}\s*\r?\n/,
+  /\r?\n[^\S\r\n]{0,20}_{5,}\s*\r?\n/,
+  /\r?\n[^\S\r\n]{0,20}-{5,}\s*\r?\n/,
   // Quote markers with ">" at line start
-  /\r?\n\s*>+\s*.+\r?\n/,
+  /\r?\n[^\S\r\n]{0,20}>+\s*.{1,500}\r?\n/,
   // Legacy patterns for backward compatibility
-  /\r?\n\s*From:\s+.+?\r?\n\s*Sent:\s+.+?\r?\n\s*To:\s+.+?\r?\n\s*Subject:\s+.+?\r?\n/i,
+  /\r?\n[^\S\r\n]{0,20}From:\s+.{1,500}?\r?\n\s*Sent:\s+.{1,500}?\r?\n\s*To:\s+.{1,500}?\r?\n\s*Subject:\s+.{1,500}?\r?\n/i,
 ];
diff --git a/src/frontend/src/features/utils/unquote-message/index.test.ts b/src/frontend/src/features/utils/unquote-message/index.test.ts
index d2c93244..c5bef866 100644
--- a/src/frontend/src/features/utils/unquote-message/index.test.ts
+++ b/src/frontend/src/features/utils/unquote-message/index.test.ts
@@ -1055,3 +1055,55 @@ On 2024-01-15, alice@example.com wrote:
     });
   });
 });
+
+describe("quote-pattern complexity", () => {
+  /**
+   * The attribution patterns run against the whole body, and their
+   * wildcards were only implicitly bounded by line length — so a body
+   * that is one very long line made them backtrack quadratically. At
+   * 195 KiB that was ~16s of blocked main thread, and
+   * MAX_INCOMING_EMAIL_SIZE is 10 MiB, so a max-size message froze the
+   * tab. Opening the email was the whole exploit.
+   *
+   * The bound is ~100x the fixed cost, so it catches a change in the
+   * exponent without being flaky about machine speed.
+   */
+  it("does not blow up on a single very long line", () => {
+    const body = "a <b> ".repeat(40000); // ~234 KiB, one line
+    const start = performance.now();
+    new UnquoteMessage("", body).getText();
+    expect(performance.now() - start).toBeLessThan(2000);
+  });
+
+  /**
+   * Bounding the intra-line wildcards is not enough on its own: a
+   * whitespace quantifier that crosses newlines (`\s*` under the `m`
+   * flag) backtracks once per line start, so a body of blank or
+   * space-only lines is quadratic in the number of lines — no long
+   * line required. Both shapes must stay under the same budget. The
+   * leading "x" defeats the all-whitespace early return: a real
+   * message needs a single visible character to reach the patterns.
+   */
+  it("does not blow up on a body of blank lines", () => {
+    const body = "x" + "\n".repeat(60000); // ~59 KiB, only newlines
+    const start = performance.now();
+    new UnquoteMessage("", body).getText();
+    expect(performance.now() - start).toBeLessThan(2000);
+  });
+
+  it("does not blow up on a body of whitespace-only lines", () => {
+    const body = "x\n" + (" ".repeat(79) + "\n").repeat(700); // ~55 KiB
+    const start = performance.now();
+    new UnquoteMessage("", body).getText();
+    expect(performance.now() - start).toBeLessThan(2000);
+  });
+
+  it("still finds an attribution line after long content", () => {
+    const body =
+      "a <b> ".repeat(5000) +
+      "\nOn Mon, 8 Jun 2026 at 14:30, Alice <alice@example.com> wrote:\n> quoted";
+    const result = new UnquoteMessage("", body).getText();
+    expect(result.hadQuotes).toBe(true);
+    expect(result.content).not.toContain("> quoted");
+  });
+});
diff --git a/src/mta-in/tests/test_security.py b/src/mta-in/tests/test_security.py
index 00ffe9d6..75f37067 100644
--- a/src/mta-in/tests/test_security.py
+++ b/src/mta-in/tests/test_security.py
@@ -147,16 +147,42 @@ def test_expn_disabled():
         b"\r.\r\n",
         # bare LF + bare LF EOD
         b"\n.\n",
+        # CRLF then dot then bare LF
+        b"\r\n.\n",
+        # CRLF then dot then bare CR
+        b"\r\n.\r",
+        # the doubled-CR form Postfix accepted (SEC Consult, 2023)
+        b"\r\r\n.\r\r\n",
+        # NUL ahead of the terminator, to unstick naive scanners
+        b"\x00\r\n.\r\n",
+    ],
+    ids=[
+        "LF-dot-CRLF",
+        "CR-dot-CRLF",
+        "LF-dot-LF",
+        "CRLF-dot-LF",
+        "CRLF-dot-CR",
+        "CRCRLF-dot-CRCRLF",
+        "NUL-CRLF-dot-CRLF",
     ],
-    ids=["LF-dot-CRLF", "CR-dot-CRLF", "LF-dot-LF"],
 )
-def test_smtp_smuggling_does_not_split_messages(mock_api_server, smuggle_bytes):
+def test_smtp_smuggling_does_not_split_messages(
+    mock_api_server, smuggle_bytes, mta_impl
+):
     """A smuggling EOD variant must NOT split the envelope into two messages.
 
     The MDA must see at most ONE delivery, and the "smuggled" MAIL FROM/RCPT
     TO must appear as text inside that single message body — never as a
     separately-delivered envelope to an attacker-chosen recipient.
     """
+    if mta_impl == "postfix" and smuggle_bytes.startswith(b"\x00"):
+        # Unlike the other variants, this payload embeds a *genuine*
+        # RFC 5321 terminator after the NUL. Postfix accepts and
+        # normalizes NUL bytes (see test_nul_byte_in_body_rejected), so
+        # it legitimately ends DATA there and reads what follows as a
+        # pipelined second transaction from the directly-connected
+        # client — ordinary submission, not smuggling.
+        pytest.skip("NUL-tolerant MTA: embedded CRLF.CRLF is a real terminator")
     mock_api_server.add_mailbox("victim@example.com")
     # Register the smuggled recipient too: otherwise an actual split would be
     # rejected at RCPT by the MDA (mailbox not found) and the test would