campaign sending prototype

src/govoplan_mail/backend/manifest.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from pathlib import Path
 
-from govoplan_core.core.modules import FrontendModule, MigrationSpec, ModuleContext, ModuleManifest, PermissionDefinition, RoleTemplate
+from govoplan_core.core.modules import FrontendModule, MigrationSpec, ModuleContext, ModuleManifest, NavItem, PermissionDefinition, RoleTemplate
 from govoplan_core.db.base import Base
 from govoplan_mail.backend.db import models as mail_models  # noqa: F401 - populate Mail ORM metadata
 
@@ -25,6 +25,7 @@ PERMISSIONS = (
     _permission("mail:profile:read", "View mail profiles", "Inspect reusable SMTP/IMAP profile metadata."),
     _permission("mail:profile:use", "Use mail profiles", "Select an approved mail profile for delivery."),
     _permission("mail:profile:test", "Test mail profiles", "Run SMTP/IMAP connection tests."),
+    _permission("mail:mailbox:read", "Read mailboxes", "List IMAP folders and inspect messages without mutating mailbox state."),
     _permission("mail:profile:write", "Manage mail profiles", "Create and edit reusable mail profiles."),
     _permission("mail:secret:manage", "Manage mail secrets", "Create or replace stored SMTP/IMAP credentials."),
 )
@@ -38,6 +39,7 @@ ROLE_TEMPLATES = (
             "mail:profile:read",
             "mail:profile:use",
             "mail:profile:test",
+            "mail:mailbox:read",
             "mail:profile:write",
             "mail:secret:manage",
         ),
@@ -46,7 +48,7 @@ ROLE_TEMPLATES = (
         slug="mail_profile_user",
         name="Mail profile user",
         description="Use and test approved mail profiles without reading secrets.",
-        permissions=("mail:profile:read", "mail:profile:use", "mail:profile:test"),
+        permissions=("mail:profile:read", "mail:profile:use", "mail:profile:test", "mail:mailbox:read"),
     ),
 )
 
@@ -69,7 +71,12 @@ manifest = ModuleManifest(
     permissions=PERMISSIONS,
     route_factory=_mail_router,
     role_templates=ROLE_TEMPLATES,
-    frontend=FrontendModule(module_id="mail", package_name="@govoplan/mail-webui"),
+    nav_items=(NavItem(path="/mail", label="Mail", icon="mail", required_any=("mail:mailbox:read",), order=50),),
+    frontend=FrontendModule(
+        module_id="mail",
+        package_name="@govoplan/mail-webui",
+        nav_items=(NavItem(path="/mail", label="Mail", icon="mail", required_any=("mail:mailbox:read",), order=50),),
+    ),
     migration_spec=MigrationSpec(
         module_id="mail",
         metadata=Base.metadata,

src/govoplan_mail/backend/router.py

@@ -7,6 +7,11 @@ from govoplan_mail.backend.schemas import (
     MailImapFolderListResponse,
     MailImapFolderResponse,
     MailImapTestRequest,
+    MailMailboxAttachmentResponse,
+    MailMailboxMessageDetailResponse,
+    MailMailboxMessageListResponse,
+    MailMailboxMessageResponse,
+    MailMailboxMessageSummaryResponse,
     MailProfilePolicyResponse,
     MailProfilePolicyUpdateRequest,
     MailServerProfileCreateRequest,
@@ -30,7 +35,7 @@ from govoplan_mail.backend.mail_profiles import (
     smtp_config_from_profile,
     update_mail_server_profile,
 )
-from govoplan_mail.backend.sending.imap import list_imap_folders, test_imap_login
+from govoplan_mail.backend.sending.imap import ImapAppendError, ImapConfigurationError, get_imap_message, list_imap_folders, list_imap_messages, test_imap_login
 from govoplan_mail.backend.sending.smtp import test_smtp_login
 from sqlalchemy.orm import Session
 
@@ -61,6 +66,11 @@ def _require_profile_credentials_scope(principal: ApiPrincipal, scope_type: str)
         _require_scope(principal, "mail:secret:manage")
 
 
+def _require_mailbox_read_scope(principal: ApiPrincipal) -> None:
+    _require_scope(principal, "mail:mailbox:read")
+    _require_scope(principal, "mail:profile:use")
+
+
 def _require_policy_read_scope(principal: ApiPrincipal, scope_type: str) -> None:
     if scope_type == "system":
         _require_scope(principal, "system:settings:read")
@@ -83,6 +93,51 @@ def _profile_response(profile) -> MailServerProfileResponse:
     return MailServerProfileResponse.model_validate(profile_response_payload(profile))
 
 
+def _mailbox_summary_response(message) -> MailMailboxMessageSummaryResponse:
+    return MailMailboxMessageSummaryResponse(
+        uid=message.uid,
+        folder=message.folder,
+        subject=message.subject,
+        from_header=message.from_header,
+        to_header=message.to_header,
+        cc_header=message.cc_header,
+        date=message.date,
+        message_id=message.message_id,
+        flags=message.flags,
+        size_bytes=message.size_bytes,
+        body_preview=message.body_preview,
+        attachment_count=message.attachment_count,
+    )
+
+
+def _mailbox_detail_response(message) -> MailMailboxMessageDetailResponse:
+    return MailMailboxMessageDetailResponse(
+        uid=message.uid,
+        folder=message.folder,
+        subject=message.subject,
+        from_header=message.from_header,
+        to_header=message.to_header,
+        cc_header=message.cc_header,
+        date=message.date,
+        message_id=message.message_id,
+        flags=message.flags,
+        size_bytes=message.size_bytes,
+        body_preview=message.body_preview,
+        body_text=message.body_text,
+        body_html=message.body_html,
+        headers=message.headers,
+        attachments=[MailMailboxAttachmentResponse(filename=item.filename, content_type=item.content_type, size_bytes=item.size_bytes) for item in message.attachments],
+    )
+
+
+def _imap_config_for_profile(session: Session, *, tenant_id: str, profile_id: str):
+    profile = get_mail_server_profile(session, tenant_id=tenant_id, profile_id=profile_id, require_active=True)
+    imap = imap_config_from_profile(profile)
+    if imap is None:
+        raise MailProfileError("Mail-server profile has no IMAP configuration")
+    return imap
+
+
 @router.get("/profiles", response_model=MailServerProfileListResponse)
 def list_profiles(
     include_inactive: bool = False,
@@ -311,6 +366,83 @@ def list_profile_imap_folders(
         return MailImapFolderListResponse(ok=False, message=_safe_error_message(exc), folders=[], detected_sent_folder=None, details={"error_type": exc.__class__.__name__})
 
 
+@router.get("/profiles/{profile_id}/mailbox/folders", response_model=MailImapFolderListResponse)
+def list_profile_mailbox_folders(
+    profile_id: str,
+    principal: ApiPrincipal = Depends(get_api_principal),
+    session: Session = Depends(get_session),
+):
+    _require_mailbox_read_scope(principal)
+    try:
+        imap = _imap_config_for_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id)
+        result = list_imap_folders(imap_config=imap)
+        folders = [MailImapFolderResponse(name=item.name, flags=item.flags) for item in result.folders]
+        return MailImapFolderListResponse(ok=True, host=result.host, port=result.port, security=result.security, message=f"Found {len(folders)} IMAP folder(s).", folders=folders, detected_sent_folder=result.detected_sent_folder)
+    except MailProfileError as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+    except ImapConfigurationError as exc:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
+    except ImapAppendError as exc:
+        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc
+
+
+@router.get("/profiles/{profile_id}/mailbox/messages", response_model=MailMailboxMessageListResponse)
+def list_profile_mailbox_messages(
+    profile_id: str,
+    folder: str = Query(default="INBOX", min_length=1, max_length=255),
+    limit: int = Query(default=50, ge=1, le=100),
+    principal: ApiPrincipal = Depends(get_api_principal),
+    session: Session = Depends(get_session),
+):
+    _require_mailbox_read_scope(principal)
+    try:
+        imap = _imap_config_for_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id)
+        result = list_imap_messages(imap_config=imap, folder=folder, limit=limit)
+        return MailMailboxMessageListResponse(
+            profile_id=profile_id,
+            folder=result.folder,
+            host=result.host,
+            port=result.port,
+            security=result.security,
+            messages=[_mailbox_summary_response(message) for message in result.messages],
+        )
+    except MailProfileError as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+    except ImapConfigurationError as exc:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
+    except ImapAppendError as exc:
+        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc
+
+
+@router.get("/profiles/{profile_id}/mailbox/messages/{message_uid}", response_model=MailMailboxMessageResponse)
+def get_profile_mailbox_message(
+    profile_id: str,
+    message_uid: str,
+    folder: str = Query(default="INBOX", min_length=1, max_length=255),
+    principal: ApiPrincipal = Depends(get_api_principal),
+    session: Session = Depends(get_session),
+):
+    _require_mailbox_read_scope(principal)
+    try:
+        imap = _imap_config_for_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id)
+        result = get_imap_message(imap_config=imap, folder=folder, uid=message_uid)
+        return MailMailboxMessageResponse(
+            profile_id=profile_id,
+            folder=result.folder,
+            host=result.host,
+            port=result.port,
+            security=result.security,
+            message=_mailbox_detail_response(result.message),
+        )
+    except MailProfileError as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+    except ImapConfigurationError as exc:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
+    except ImapAppendError as exc:
+        status_code = status.HTTP_404_NOT_FOUND if "not found" in str(exc).casefold() else status.HTTP_502_BAD_GATEWAY
+        raise HTTPException(status_code=status_code, detail=str(exc)) from exc
+
+
 def _safe_error_message(exc: Exception) -> str:
     text = str(exc).strip()
     return text or exc.__class__.__name__

src/govoplan_mail/backend/schemas.py

@@ -125,3 +125,49 @@ class MailImapFolderListResponse(BaseModel):
     folders: list[MailImapFolderResponse] = Field(default_factory=list)
     detected_sent_folder: str | None = None
     details: dict[str, Any] = Field(default_factory=dict)
+
+class MailMailboxAttachmentResponse(BaseModel):
+    filename: str | None = None
+    content_type: str
+    size_bytes: int
+
+
+class MailMailboxMessageSummaryResponse(BaseModel):
+    uid: str
+    folder: str
+    subject: str | None = None
+    from_header: str | None = None
+    to_header: str | None = None
+    cc_header: str | None = None
+    date: str | None = None
+    message_id: str | None = None
+    flags: list[str] = Field(default_factory=list)
+    size_bytes: int | None = None
+    body_preview: str | None = None
+    attachment_count: int = 0
+
+
+class MailMailboxMessageDetailResponse(MailMailboxMessageSummaryResponse):
+    body_text: str | None = None
+    body_html: str | None = None
+    headers: dict[str, str] = Field(default_factory=dict)
+    attachments: list[MailMailboxAttachmentResponse] = Field(default_factory=list)
+
+
+class MailMailboxMessageListResponse(BaseModel):
+    profile_id: str
+    folder: str
+    host: str | None = None
+    port: int | None = None
+    security: str | None = None
+    messages: list[MailMailboxMessageSummaryResponse] = Field(default_factory=list)
+
+
+class MailMailboxMessageResponse(BaseModel):
+    profile_id: str
+    folder: str
+    host: str | None = None
+    port: int | None = None
+    security: str | None = None
+    message: MailMailboxMessageDetailResponse
+

src/govoplan_mail/backend/sending/imap.py

@@ -6,12 +6,18 @@ import socket
 import ssl
 import time
 from dataclasses import dataclass
+from email import policy
+from email.message import EmailMessage
+from email.parser import BytesParser
+from typing import Any
 
 from govoplan_mail.backend.config import ImapConfig, TransportSecurity
 from govoplan_mail.backend.dev.mock_mailbox import (
     MOCK_IMAP_FOLDERS,
     consume_fail_next_imap,
+    get_record,
     is_mock_imap_host,
+    list_records,
     record_imap_append,
 )
 
@@ -55,6 +61,66 @@ class ImapFolderListResult:
     detected_sent_folder: str | None = None
 
 
+@dataclass(frozen=True, slots=True)
+class ImapMailboxAttachmentInfo:
+    filename: str | None
+    content_type: str
+    size_bytes: int
+
+
+@dataclass(frozen=True, slots=True)
+class ImapMailboxMessageSummary:
+    uid: str
+    folder: str
+    subject: str | None
+    from_header: str | None
+    to_header: str | None
+    cc_header: str | None
+    date: str | None
+    message_id: str | None
+    flags: list[str]
+    size_bytes: int | None
+    body_preview: str | None
+    attachment_count: int
+
+
+@dataclass(frozen=True, slots=True)
+class ImapMailboxMessageDetail:
+    uid: str
+    folder: str
+    subject: str | None
+    from_header: str | None
+    to_header: str | None
+    cc_header: str | None
+    date: str | None
+    message_id: str | None
+    flags: list[str]
+    size_bytes: int | None
+    body_preview: str | None
+    body_text: str | None
+    body_html: str | None
+    headers: dict[str, str]
+    attachments: list[ImapMailboxAttachmentInfo]
+
+
+@dataclass(frozen=True, slots=True)
+class ImapMailboxMessageListResult:
+    host: str
+    port: int
+    security: str
+    folder: str
+    messages: list[ImapMailboxMessageSummary]
+
+
+@dataclass(frozen=True, slots=True)
+class ImapMailboxMessageResult:
+    host: str
+    port: int
+    security: str
+    folder: str
+    message: ImapMailboxMessageDetail
+
+
 @dataclass(frozen=True, slots=True)
 class ImapAppendResult:
     host: str
@@ -283,6 +349,226 @@ def list_imap_folders(*, imap_config: ImapConfig) -> ImapFolderListResult:
             pass
 
 
+def _parse_message(raw: bytes) -> EmailMessage:
+    return BytesParser(policy=policy.default).parsebytes(raw)
+
+
+def _header_text(message: EmailMessage, name: str) -> str | None:
+    value = message.get(name)
+    return str(value) if value is not None else None
+
+
+def _body_content(message: EmailMessage, prefer: tuple[str, ...]) -> str | None:
+    try:
+        body = message.get_body(preferencelist=prefer) if message.is_multipart() else message
+        if body is None:
+            return None
+        content = body.get_content()
+    except Exception:
+        return None
+    text = str(content).strip()
+    return text or None
+
+
+def _body_preview(message: EmailMessage) -> str | None:
+    text = _body_content(message, ("plain", "html"))
+    if not text:
+        return None
+    preview = re.sub(r"\s+", " ", text).strip()
+    return preview[:600] or None
+
+
+def _attachment_infos(message: EmailMessage) -> list[ImapMailboxAttachmentInfo]:
+    attachments: list[ImapMailboxAttachmentInfo] = []
+    for part in message.iter_attachments():
+        payload = part.get_payload(decode=True) or b""
+        attachments.append(
+            ImapMailboxAttachmentInfo(
+                filename=part.get_filename(),
+                content_type=part.get_content_type(),
+                size_bytes=len(payload),
+            )
+        )
+    return attachments
+
+
+def _message_summary_from_raw(*, uid: str, folder: str, raw: bytes, flags: list[str], size_bytes: int | None) -> ImapMailboxMessageSummary:
+    message = _parse_message(raw)
+    attachments = _attachment_infos(message)
+    return ImapMailboxMessageSummary(
+        uid=uid,
+        folder=folder,
+        subject=_header_text(message, "Subject"),
+        from_header=_header_text(message, "From"),
+        to_header=_header_text(message, "To"),
+        cc_header=_header_text(message, "Cc"),
+        date=_header_text(message, "Date"),
+        message_id=_header_text(message, "Message-ID"),
+        flags=flags,
+        size_bytes=size_bytes if size_bytes is not None else len(raw),
+        body_preview=_body_preview(message),
+        attachment_count=len(attachments),
+    )
+
+
+def _message_detail_from_raw(*, uid: str, folder: str, raw: bytes, flags: list[str], size_bytes: int | None) -> ImapMailboxMessageDetail:
+    message = _parse_message(raw)
+    attachments = _attachment_infos(message)
+    return ImapMailboxMessageDetail(
+        uid=uid,
+        folder=folder,
+        subject=_header_text(message, "Subject"),
+        from_header=_header_text(message, "From"),
+        to_header=_header_text(message, "To"),
+        cc_header=_header_text(message, "Cc"),
+        date=_header_text(message, "Date"),
+        message_id=_header_text(message, "Message-ID"),
+        flags=flags,
+        size_bytes=size_bytes if size_bytes is not None else len(raw),
+        body_preview=_body_preview(message),
+        body_text=_body_content(message, ("plain",)),
+        body_html=_body_content(message, ("html",)),
+        headers={str(key): str(value) for key, value in message.items()},
+        attachments=attachments,
+    )
+
+
+def _parse_fetch_response(data: list[Any] | tuple[Any, ...] | None) -> tuple[str | None, list[str], int | None, bytes]:
+    metadata_parts: list[str] = []
+    raw: bytes | None = None
+    for item in data or []:
+        if isinstance(item, tuple):
+            metadata_parts.append(_decode_item(item[0]))
+            if isinstance(item[1], bytes):
+                raw = item[1]
+        else:
+            metadata_parts.append(_decode_item(item))
+    metadata = " ".join(part for part in metadata_parts if part)
+    uid_match = re.search(r"\bUID\s+(\d+)", metadata, re.IGNORECASE)
+    size_match = re.search(r"\bRFC822\.SIZE\s+(\d+)", metadata, re.IGNORECASE)
+    flags_match = re.search(r"\bFLAGS\s+\(([^)]*)\)", metadata, re.IGNORECASE)
+    flags = flags_match.group(1).split() if flags_match else []
+    if raw is None:
+        raise ImapAppendError(f"IMAP message fetch returned no message body: {metadata!r}", temporary=True)
+    return (
+        uid_match.group(1) if uid_match else None,
+        flags,
+        int(size_match.group(1)) if size_match else None,
+        raw,
+    )
+
+
+def _select_readonly(client: imaplib.IMAP4, folder: str) -> None:
+    typ, data = client.select(folder, readonly=True)
+    if typ != "OK":
+        raise ImapAppendError(f"IMAP folder {folder!r} could not be opened read-only: {data!r}", temporary=False)
+
+
+def _fetch_message_by_uid(client: imaplib.IMAP4, uid: str) -> tuple[str, list[str], int | None, bytes]:
+    typ, data = client.uid("fetch", str(uid), "(UID FLAGS RFC822.SIZE BODY.PEEK[])")
+    if typ != "OK":
+        raise ImapAppendError(f"IMAP message {uid!r} fetch failed: {data!r}", temporary=True)
+    fetched_uid, flags, size_bytes, raw = _parse_fetch_response(data)
+    return fetched_uid or str(uid), flags, size_bytes, raw
+
+
+def _mock_record_folder(record: dict[str, Any]) -> str:
+    folder = str(record.get("folder") or "").strip()
+    if folder:
+        return folder
+    return "Sent" if record.get("kind") == "imap_append" else "INBOX"
+
+
+def _mock_folder_matches(record: dict[str, Any], folder: str) -> bool:
+    return _mock_record_folder(record).casefold() == folder.casefold()
+
+
+def _mock_raw_bytes(record: dict[str, Any]) -> bytes:
+    raw = record.get("raw_eml")
+    if isinstance(raw, str) and raw:
+        return raw.encode("utf-8", errors="replace")
+    headers = record.get("headers") if isinstance(record.get("headers"), dict) else {}
+    lines = [f"{key}: {value}" for key, value in headers.items()]
+    lines.append("")
+    lines.append(str(record.get("body_preview") or ""))
+    return "\n".join(lines).encode("utf-8", errors="replace")
+
+
+def list_imap_messages(*, imap_config: ImapConfig, folder: str = "INBOX", limit: int = 50) -> ImapMailboxMessageListResult:
+    """List mailbox messages without mutating read/unread state."""
+
+    host, port = _require_imap_config(imap_config)
+    folder = (folder or "INBOX").strip() or "INBOX"
+    limit = max(1, min(limit, 100))
+    if is_mock_imap_host(imap_config.host):
+        records = [record for record in list_records(limit=500) if _mock_folder_matches(record, folder)]
+        messages = [
+            _message_summary_from_raw(
+                uid=str(record.get("id") or ""),
+                folder=_mock_record_folder(record),
+                raw=_mock_raw_bytes(record),
+                flags=["\\Seen"] if record.get("kind") == "imap_append" else [],
+                size_bytes=int(record.get("size_bytes") or 0) or None,
+            )
+            for record in records[:limit]
+        ]
+        return ImapMailboxMessageListResult(host=host, port=port, security=imap_config.security.value, folder=folder, messages=messages)
+
+    client = _open_imap(imap_config)
+    try:
+        _select_readonly(client, folder)
+        typ, data = client.uid("search", None, "ALL")
+        if typ != "OK":
+            raise ImapAppendError(f"IMAP search failed for folder {folder!r}: {data!r}", temporary=True)
+        uid_bytes = data[0] if data else b""
+        uids = _decode_item(uid_bytes).split()
+        messages: list[ImapMailboxMessageSummary] = []
+        for uid in reversed(uids[-limit:]):
+            fetched_uid, flags, size_bytes, raw = _fetch_message_by_uid(client, uid)
+            messages.append(_message_summary_from_raw(uid=fetched_uid, folder=folder, raw=raw, flags=flags, size_bytes=size_bytes))
+        return ImapMailboxMessageListResult(host=host, port=port, security=imap_config.security.value, folder=folder, messages=messages)
+    finally:
+        try:
+            client.logout()
+        except Exception:
+            pass
+
+
+def get_imap_message(*, imap_config: ImapConfig, folder: str, uid: str) -> ImapMailboxMessageResult:
+    """Fetch one message body using BODY.PEEK in a read-only selected mailbox."""
+
+    host, port = _require_imap_config(imap_config)
+    folder = (folder or "INBOX").strip() or "INBOX"
+    uid = str(uid).strip()
+    if not uid:
+        raise ImapConfigurationError("Message UID is required")
+
+    if is_mock_imap_host(imap_config.host):
+        record = get_record(uid, include_raw=True)
+        if not record or not _mock_folder_matches(record, folder):
+            raise ImapAppendError(f"Mock mailbox message {uid!r} was not found in folder {folder!r}", temporary=False)
+        message = _message_detail_from_raw(
+            uid=uid,
+            folder=_mock_record_folder(record),
+            raw=_mock_raw_bytes(record),
+            flags=["\\Seen"] if record.get("kind") == "imap_append" else [],
+            size_bytes=int(record.get("size_bytes") or 0) or None,
+        )
+        return ImapMailboxMessageResult(host=host, port=port, security=imap_config.security.value, folder=folder, message=message)
+
+    client = _open_imap(imap_config)
+    try:
+        _select_readonly(client, folder)
+        fetched_uid, flags, size_bytes, raw = _fetch_message_by_uid(client, uid)
+        message = _message_detail_from_raw(uid=fetched_uid, folder=folder, raw=raw, flags=flags, size_bytes=size_bytes)
+        return ImapMailboxMessageResult(host=host, port=port, security=imap_config.security.value, folder=folder, message=message)
+    finally:
+        try:
+            client.logout()
+        except Exception:
+            pass
+
+
 def append_message_to_sent(
     message_bytes: bytes,
     *,

src/govoplan_mail/backend/sending/rate_limit.py

@@ -21,18 +21,23 @@ def _redis_client() -> Redis:
     return Redis.from_url(settings.redis_url, decode_responses=True)
 
 
+def _distributed_rate_limit_enabled() -> bool:
+    return bool(getattr(settings, "celery_enabled", False))
+
+
 def wait_for_rate_limit(*, key: str, messages_per_minute: int, enabled: bool = True) -> RateLimitDecision:
     """Throttle sends across worker processes using Redis when available.
 
     The implementation stores the next allowed send timestamp per key. A Redis
     lock keeps multiple Celery processes from reading/updating the timestamp at
-    the same time. If Redis is unavailable, it falls back to no distributed wait;
-    the per-container Celery concurrency still protects local development.
+    the same time. Direct local development runs do not have a broker, so Redis
+    is only used when Celery/worker mode is explicitly enabled. If Redis is
+    unavailable, it falls back to no distributed wait.
     """
 
     messages_per_minute = max(1, int(messages_per_minute or 1))
     gap = 60.0 / messages_per_minute
-    if not enabled:
+    if not enabled or not _distributed_rate_limit_enabled():
         return RateLimitDecision(key=key, messages_per_minute=messages_per_minute, gap_seconds=gap, waited_seconds=0.0)
 
     redis_key = f"multimailer:ratelimit:{key}:next_allowed"