Merge pull request #935 from cbcoutinho/feat/mail-app-read-and-index

feat(mail): read and index Nextcloud Mail via the Mail OCS API

README.md

@@ -70,7 +70,7 @@ For Kubernetes, see [cbcoutinho/helm-charts](https://github.com/cbcoutinho/helm-
 
 - **110+ MCP Tools** - Comprehensive API coverage across 10 Nextcloud apps
 - **MCP Resources** - Structured data URIs for browsing Nextcloud data
-- **Semantic Search (Experimental)** - Optional vector-powered search for Notes, Files, News items, and Deck cards (requires Qdrant + Ollama)
+- **Semantic Search (Experimental)** - Optional vector-powered search for Notes, Files, News items, Deck cards, and Mail messages (requires Qdrant + Ollama)
 - **Document Processing** - OCR and text extraction from PDFs, DOCX, images with progress notifications
 - **Flexible Deployment** - Docker, Kubernetes ([Helm chart](https://github.com/cbcoutinho/helm-charts)), VM, or local installation
 - **Production-Ready Auth** - Basic Auth with app passwords; multi-user via Login Flow v2 — MCP clients authenticate via OAuth, the server handles Nextcloud app passwords transparently
@@ -90,9 +90,10 @@ For Kubernetes, see [cbcoutinho/helm-charts](https://github.com/cbcoutinho/helm-
 | **Tables** | 5 | Row operations on Nextcloud Tables |
 | **Sharing** | 10+ | Create and manage shares |
 | **News** | 8 | Feeds, folders, items, feed health monitoring |
+| **Mail** | 5 | Read-only: accounts, mailboxes, messages, attachments (via Mail app's OCS API) |
 | **Collectives** | 16 | Full CRUD on collectives, pages, and tags |
 | **Talk (spreed)** | 6 | List conversations, read/post messages, mark as read, list participants |
-| **Semantic Search** | 2+ | Vector search for Notes, Files, News items, and Deck cards (experimental, opt-in, requires infrastructure) |
+| **Semantic Search** | 2+ | Vector search for Notes, Files, News items, Deck cards, and Mail messages (experimental, opt-in, requires infrastructure) |
 
 Want to see another Nextcloud app supported? [Open an issue](https://github.com/cbcoutinho/nextcloud-mcp-server/issues) or contribute a pull request!
 

nextcloud_mcp_server/client/__init__.py

@@ -20,6 +20,7 @@ from .contacts import ContactsClient
 from .cookbook import CookbookClient
 from .deck import DeckClient
 from .groups import GroupsClient
+from .mail import MailClient
 from .news import NewsClient
 from .notes import NotesClient
 from .sharing import SharingClient
@@ -144,6 +145,7 @@ class NextcloudClient:
         self.collectives = CollectivesClient(self._client, username)
         self.deck = DeckClient(self._client, username)
         self.news = NewsClient(self._client, username)
+        self.mail = MailClient(self._client, username)
         self.talk = TalkClient(self._client, username)
         self.users = UsersClient(self._client, username)
         self.groups = GroupsClient(self._client, username)

nextcloud_mcp_server/client/mail.py

@@ -0,0 +1,200 @@
+"""Client for Nextcloud Mail app operations (read-only).
+
+Talks to the Mail app's OCS API under ``/ocs/v2.php/apps/mail/api/...``. The
+Mail app's *server* handles the IMAP connection on the user's behalf, so this
+client only ever speaks HTTP to Nextcloud — it never connects to IMAP/POP3
+itself. The read endpoints are ``#[NoCSRFRequired]`` + ``#[NoAdminRequired]``,
+so they are reachable with the same Basic-Auth app-password flow the other app
+clients use, provided the ``OCS-APIRequest`` header is sent.
+
+Prerequisites: the mail account must already be configured inside the Nextcloud
+Mail app (so the server has IMAP credentials), and the Mail app must expose the
+OCS API controllers (Mail 5.x / Nextcloud 32+).
+"""
+
+import logging
+from typing import Any
+from urllib.parse import quote
+
+from httpx import HTTPStatusError, RequestError, Response
+
+from .base import BaseNextcloudClient
+
+logger = logging.getLogger(__name__)
+
+
+class MailClient(BaseNextcloudClient):
+    """Read-only client for Nextcloud Mail app operations."""
+
+    app_name = "mail"
+    API_BASE = "/ocs/v2.php/apps/mail/api"
+
+    # OCS endpoints require this header; without it Nextcloud rejects the
+    # request (or redirects to a login page). ``format=json`` forces a JSON
+    # envelope rather than XML.
+    _OCS_HEADERS = {"OCS-APIRequest": "true", "Accept": "application/json"}
+
+    async def _ocs_get(self, path: str, *, params: dict[str, Any] | None = None) -> Any:
+        """GET an OCS endpoint and unwrap the ``ocs.data`` payload.
+
+        Args:
+            path: Path under ``API_BASE`` (e.g. ``/account/list``)
+            params: Optional query params (``format=json`` is added automatically)
+
+        Returns:
+            The ``data`` payload (a list, dict, or string depending on endpoint)
+        """
+        query: dict[str, Any] = {"format": "json"}
+        if params:
+            query.update(params)
+        response = await self._make_request(
+            "GET",
+            f"{self.API_BASE}{path}",
+            params=query,
+            headers=self._OCS_HEADERS,
+        )
+
+        # The Mail app being absent (or a misconfigured proxy) can return HTTP
+        # 200 with an HTML body; surface that as a network-style error rather
+        # than letting json() raise an opaque JSONDecodeError to the caller.
+        try:
+            body = response.json()
+        except ValueError as e:
+            raise RequestError(
+                f"Mail OCS returned a non-JSON response for {path}: {e}",
+                request=response.request,
+            ) from e
+
+        # Standard OCS envelope: {"ocs": {"meta": {...}, "data": <payload>}}.
+        # OCS can return HTTP 200 while signalling failure (e.g. 403/404) in
+        # ocs.meta.statuscode; re-raise those as an HTTPStatusError carrying the
+        # OCS code so callers' existing 404/403 handling applies uniformly
+        # instead of silently unwrapping data=null.
+        ocs = body.get("ocs", {}) if isinstance(body, dict) else {}
+        meta = ocs.get("meta", {})
+        # statuscode is spec'd as an int, but harden against a non-numeric value
+        # in a non-spec response rather than letting int() raise ValueError
+        # (which neither MCP-tool handler catches) — treat it as success.
+        try:
+            status_code = int(meta.get("statuscode", 200) or 200)
+        except (TypeError, ValueError):
+            status_code = 200
+        if status_code >= 400:
+            synthetic = Response(
+                status_code=status_code, request=response.request, content=b""
+            )
+            raise HTTPStatusError(
+                f"Mail OCS error {status_code} for {path}: {meta.get('message')}",
+                request=response.request,
+                response=synthetic,
+            )
+        return ocs.get("data")
+
+    # --- Accounts ---
+
+    async def list_accounts(self) -> list[dict[str, Any]]:
+        """List the user's configured mail accounts.
+
+        Returns:
+            List of account objects (keys: id, email, isDelegated, aliases)
+        """
+        data = await self._ocs_get("/account/list")
+        return data or []
+
+    # --- Mailboxes ---
+
+    async def get_mailboxes(self, account_id: int) -> list[dict[str, Any]]:
+        """List the mailboxes (folders) of an account.
+
+        Args:
+            account_id: Account ID (the ``id`` from :meth:`list_accounts`)
+
+        Returns:
+            List of mailbox objects. Note ``databaseId`` is the numeric mailbox
+            id needed by :meth:`list_messages` (``id`` is a base64 string).
+        """
+        data = await self._ocs_get("/mailboxes", params={"accountId": account_id})
+        return data or []
+
+    # --- Messages ---
+
+    async def list_messages(
+        self,
+        mailbox_id: int,
+        *,
+        cursor: int | None = None,
+        search_filter: str | None = None,
+        limit: int = 20,
+        view: str | None = None,
+    ) -> list[dict[str, Any]]:
+        """List message envelopes in a mailbox (newest first).
+
+        Reads DB-cached envelope metadata, so this is fast and does not hit
+        IMAP per request.
+
+        Args:
+            mailbox_id: Numeric mailbox id (``databaseId`` from get_mailboxes)
+            cursor: Pagination cursor (timestamp/id from a prior page)
+            search_filter: Optional search/filter query (maps to the OCS
+                ``filter`` query param; named to avoid shadowing ``builtins.filter``)
+            limit: Max messages to return. Clamped server-side to 1..100; a
+                missing limit collapses to 1 server-side, so always pass one.
+            view: ``"singleton"`` or ``"threaded"`` (default threaded)
+
+        Returns:
+            List of message summary objects (keys include databaseId, subject,
+            from, to, dateInt, flags, previewText, mailboxId).
+        """
+        # The Mail OCS API clamps limit to 1..100 server-side; do it here too so
+        # the contract is enforced at the client layer with a predictable value
+        # (a limit of 0 would otherwise collapse to 1 server-side).
+        params: dict[str, Any] = {"limit": min(max(1, limit), 100)}
+        if cursor is not None:
+            params["cursor"] = cursor
+        if search_filter is not None:
+            params["filter"] = search_filter
+        if view is not None:
+            params["view"] = view
+        data = await self._ocs_get(f"/mailboxes/{mailbox_id}/messages", params=params)
+        return data or []
+
+    async def get_message(self, message_id: int) -> dict[str, Any]:
+        """Get a single message with its full body.
+
+        The Mail app fetches the body from IMAP server-side and returns it as a
+        single ``body`` field (sanitized HTML when ``hasHtmlBody`` is true,
+        otherwise plain text). ``body`` may be absent on partial (206) responses
+        when S/MIME decryption fails.
+
+        Args:
+            message_id: Numeric message id (``databaseId`` from list_messages)
+
+        Returns:
+            Full message object.
+        """
+        data = await self._ocs_get(f"/message/{message_id}")
+        return data or {}
+
+    async def get_attachment(
+        self, message_id: int, attachment_id: str
+    ) -> dict[str, Any]:
+        """Get a single attachment's metadata and content.
+
+        The Mail OCS API returns the attachment as a JSON object (not a binary
+        download): keys ``name``, ``mime``, ``size``, ``content``.
+
+        Args:
+            message_id: Numeric message id
+            attachment_id: Attachment id (a string; from the message's
+                ``attachments`` array)
+
+        Returns:
+            Attachment object with name, mime, size, content.
+        """
+        # URL-encode the caller-supplied attachment id (defense-in-depth: keeps
+        # a value like "../.." from being normalised into a different path).
+        safe_attachment_id = quote(attachment_id, safe="")
+        data = await self._ocs_get(
+            f"/message/{message_id}/attachment/{safe_attachment_id}"
+        )
+        return data or {}

nextcloud_mcp_server/models/auth.py

@@ -74,6 +74,7 @@ ALL_SUPPORTED_SCOPES: frozenset[str] = frozenset(
         "sharing.write",
         "news.read",
         "news.write",
+        "mail.read",
         "collectives.read",
         "collectives.write",
     }

nextcloud_mcp_server/models/mail.py

@@ -0,0 +1,191 @@
+"""Pydantic models for Nextcloud Mail app responses (read-only)."""
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from .base import BaseResponse
+
+
+class MailAddress(BaseModel):
+    """An email address with an optional display label."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    label: str | None = Field(None, description="Display name")
+    email: str | None = Field(None, description="Email address")
+
+
+class MailAccount(BaseModel):
+    """A configured mail account (from the account/list endpoint)."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    id: int = Field(description="Account ID")
+    email: str = Field(description="Account email address")
+    is_delegated: bool = Field(
+        False, alias="isDelegated", description="Whether this is a delegated account"
+    )
+
+
+class MailMailbox(BaseModel):
+    """A mailbox (folder) within an account."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    # ``databaseId`` is the numeric id used by list_messages; ``id`` is a
+    # base64-encoded mailbox name (a string), so we expose the numeric one.
+    database_id: int = Field(alias="databaseId", description="Numeric mailbox ID")
+    name: str = Field(description="IMAP mailbox name (e.g. INBOX, Sent)")
+    display_name: str | None = Field(
+        None, alias="displayName", description="Human-readable mailbox name"
+    )
+    account_id: int = Field(alias="accountId", description="Parent account ID")
+    special_use: list[str] = Field(
+        default_factory=list,
+        alias="specialUse",
+        description="Special-use roles (e.g. inbox, sent, trash)",
+    )
+    unread: int = Field(0, description="Number of unread messages")
+
+
+class MailMessageFlags(BaseModel):
+    """IMAP flags on a message."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    seen: bool = False
+    flagged: bool = False
+    answered: bool = False
+    deleted: bool = False
+    draft: bool = False
+    forwarded: bool = False
+    has_attachments: bool = Field(False, alias="hasAttachments")
+    important: bool = False
+
+
+class MailAttachment(BaseModel):
+    """Metadata for a message attachment."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    # Attachment id is a string and may be null for inline/body-part messages.
+    id: str | None = Field(None, description="Attachment ID")
+    file_name: str | None = Field(
+        None, alias="fileName", description="Attachment file name"
+    )
+    mime: str | None = Field(None, description="MIME type")
+    size: int | None = Field(None, description="Size in bytes")
+    cid: str | None = Field(None, description="Content-ID (for inline attachments)")
+    disposition: str | None = Field(
+        None, description="Content disposition (attachment/inline)"
+    )
+
+
+class MailMessageSummary(BaseModel):
+    """Lightweight message envelope for mailbox listings."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    # ``databaseId`` is the numeric id passed to get_message.
+    database_id: int = Field(alias="databaseId", description="Numeric message ID")
+    uid: int | None = Field(None, description="IMAP UID")
+    subject: str | None = Field(None, description="Message subject")
+    date_int: int | None = Field(
+        None, alias="dateInt", description="Sent date as a Unix timestamp (seconds)"
+    )
+    from_: list[MailAddress] = Field(
+        default_factory=list, alias="from", description="Sender addresses"
+    )
+    to: list[MailAddress] = Field(
+        default_factory=list, description="Recipient addresses"
+    )
+    mailbox_id: int | None = Field(
+        None, alias="mailboxId", description="Parent mailbox ID"
+    )
+    preview_text: str | None = Field(
+        None, alias="previewText", description="Short preview snippet"
+    )
+    flags: MailMessageFlags | None = Field(None, description="IMAP flags")
+
+
+class MailMessage(BaseModel):
+    """Full message with body (from the message/{id} endpoint)."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    id: int = Field(description="Numeric message ID")
+    uid: int | None = Field(None, description="IMAP UID")
+    message_id: str | None = Field(
+        None, alias="messageId", description="RFC Message-ID header"
+    )
+    subject: str | None = Field(None, description="Message subject")
+    date_int: int | None = Field(
+        None, alias="dateInt", description="Sent date as a Unix timestamp (seconds)"
+    )
+    from_: list[MailAddress] = Field(
+        default_factory=list, alias="from", description="Sender addresses"
+    )
+    to: list[MailAddress] = Field(
+        default_factory=list, description="Recipient addresses"
+    )
+    cc: list[MailAddress] = Field(default_factory=list, description="CC addresses")
+    bcc: list[MailAddress] = Field(default_factory=list, description="BCC addresses")
+    has_html_body: bool = Field(
+        False, alias="hasHtmlBody", description="Whether the body is HTML"
+    )
+    body: str | None = Field(
+        None,
+        description="Rendered body (sanitized HTML if hasHtmlBody, else plain text)",
+    )
+    attachments: list[MailAttachment] = Field(
+        default_factory=list, description="Message attachments"
+    )
+
+
+# --- Response Models ---
+
+
+class ListAccountsResponse(BaseResponse):
+    """Response model for listing mail accounts."""
+
+    results: list[MailAccount] = Field(description="List of mail accounts")
+    total_count: int = Field(description="Total number of accounts")
+
+
+class ListMailboxesResponse(BaseResponse):
+    """Response model for listing mailboxes."""
+
+    results: list[MailMailbox] = Field(description="List of mailboxes")
+    total_count: int = Field(description="Total number of mailboxes")
+
+
+class ListMessagesResponse(BaseResponse):
+    """Response model for listing message envelopes."""
+
+    results: list[MailMessageSummary] = Field(description="List of message summaries")
+    total_count: int = Field(
+        description="Number of messages returned in this page (NOT the mailbox "
+        "total, which isn't known without a full scan); page with cursor and "
+        "stop on an empty result"
+    )
+    has_more: bool = Field(False, description="Whether more messages may exist")
+
+
+class GetMessageResponse(BaseResponse):
+    """Response model for getting a single full message."""
+
+    message: MailMessage = Field(description="Full message details")
+
+
+class GetAttachmentResponse(BaseResponse):
+    """Response model for getting a single attachment.
+
+    Intentionally does NOT nest ``MailAttachment``: the Mail OCS *get-attachment*
+    endpoint returns a different shape (``name``/``mime``/``size``/``content``)
+    than the attachment entries on a message listing, which ``MailAttachment``
+    models (``id``/``fileName``/``cid``/``disposition``, and no ``content``).
+    """
+
+    name: str | None = Field(None, description="Attachment file name")
+    mime: str | None = Field(None, description="MIME type")
+    size: int | None = Field(None, description="Size in bytes")
+    content: str | None = Field(None, description="Attachment content")

nextcloud_mcp_server/search/algorithms.py

@@ -76,6 +76,11 @@ class NextcloudClientProtocol(Protocol):
         """News client for accessing news item documents."""
         ...
 
+    @property
+    def mail(self) -> Any:
+        """Mail client for accessing mail message documents."""
+        ...
+
     # Top-level client helper (not a sub-client) used by verify-on-read to
     # gate file results on current vector-index tag membership.
     async def find_files_by_tag(

nextcloud_mcp_server/search/context.py

@@ -17,6 +17,7 @@ from nextcloud_mcp_server.models.deck import DeckCard
 from nextcloud_mcp_server.search.access_filter import build_ownership_filter
 from nextcloud_mcp_server.utils.validation import is_valid_nextcloud_doc_id
 from nextcloud_mcp_server.vector.html_processor import html_to_markdown
+from nextcloud_mcp_server.vector.mail_content import build_mail_content
 from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter
 from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
 
@@ -822,6 +823,22 @@ async def _fetch_document_text(
             if card.description:
                 content_parts.append(card.description)
             return "\n\n".join(content_parts)
+        elif doc_type == "mail_message":
+            # Mail message IDs are positive ASCII integers (MySQL AUTO_INCREMENT).
+            if not is_valid_nextcloud_doc_id(doc_id):
+                logger.warning(
+                    "Expected numeric mail_message doc_id, got %r — skipping document fetch",
+                    doc_id,
+                )
+                return None
+            # Reconstruct full content via the shared helper so chunk offsets
+            # match what the processor indexed (single source of truth).
+            message = await nc_client.mail.get_message(int(doc_id))
+            # Empty payload (OCS data=null with a <400 meta) -> skip context
+            # expansion, mirroring the processor's index-time guard.
+            if not message:
+                return None
+            return build_mail_content(message)
         else:
             logger.warning("Unsupported doc_type for context expansion: %s", doc_type)
             return None

nextcloud_mcp_server/search/verification.py

@@ -53,6 +53,7 @@ from nextcloud_mcp_server.search.algorithms import (
 )
 from nextcloud_mcp_server.utils.validation import is_valid_nextcloud_doc_id
 from nextcloud_mcp_server.vector.eviction import delete_document_points
+from nextcloud_mcp_server.vector.mail_content import MAIL_SCAN_MAX_PER_MAILBOX
 
 logger = logging.getLogger(__name__)
 
@@ -488,11 +489,121 @@ async def _verify_news_items(
     return accessible
 
 
+async def _verify_mail_messages(
+    client: NextcloudClientProtocol,
+    results: list[SearchResult],
+    semaphore: anyio.Semaphore,
+) -> set[str]:
+    """Verify mail messages with one DB-cached list per mailbox, then intersect.
+
+    ``mail.get_message`` triggers a server-side IMAP body fetch, so a per-result
+    verify would issue one IMAP FETCH per hit — multiple seconds for an active
+    inbox. Instead we batch by ``mailbox_id`` (propagated into ``result.metadata``
+    from the Qdrant payload) and call ``mail.list_messages`` once per mailbox,
+    which reads the Mail app's DB cache (no IMAP). A message is accessible iff it
+    is present in its mailbox's newest-N listing — the same window the scanner
+    indexes, so anything aged out of the window is being evicted anyway.
+
+    Failure policy mirrors ``_verify_news_items``: a definitive 403/404 for a
+    mailbox drops all its results (eviction reclaims); transient errors keep
+    them (fail-open). Results with no/!numeric ``mailbox_id`` or a non-numeric
+    ``doc_id`` are kept (fail-open; verification can't batch them).
+    """
+    # safe: cooperative concurrency, no lock needed (see verify_search_results)
+    accessible: set[str] = set()
+
+    # Partition results by mailbox so each mailbox is listed exactly once.
+    by_mailbox: dict[int, list[SearchResult]] = {}
+    for r in results:
+        mailbox_id = (r.metadata or {}).get("mailbox_id")
+        # No usable mailbox_id (legacy payload) — can't batch via the DB cache
+        # without an IMAP-triggering per-id fetch, so keep it (fail-open).
+        if mailbox_id is None:
+            logger.warning(
+                "Mail result %s has no mailbox_id; keeping (batch verification "
+                "skipped)",
+                r.id,
+            )
+            accessible.add(r.id)
+            continue
+        try:
+            mailbox_int = int(mailbox_id)
+        except (TypeError, ValueError):
+            logger.warning(
+                "Mail result %s has non-numeric mailbox_id %r; keeping "
+                "(batch verification skipped)",
+                r.id,
+                mailbox_id,
+            )
+            accessible.add(r.id)
+            continue
+        by_mailbox.setdefault(mailbox_int, []).append(r)
+
+    async def check_mailbox(mailbox_id: int, mb_results: list[SearchResult]) -> None:
+        async with semaphore:
+            try:
+                messages = await client.mail.list_messages(
+                    mailbox_id, limit=MAIL_SCAN_MAX_PER_MAILBOX
+                )
+            except HTTPStatusError as e:
+                if _is_definitive_404_or_403(e):
+                    # Mailbox/account gone — all its results are inaccessible.
+                    return
+                logger.warning(
+                    "Transient error listing mailbox %s for verification: %s %s; "
+                    "keeping its %d result(s)",
+                    mailbox_id,
+                    e.response.status_code,
+                    e,
+                    len(mb_results),
+                )
+                for r in mb_results:
+                    accessible.add(r.id)
+                return
+            except Exception as e:
+                logger.warning(
+                    "Unexpected error listing mailbox %s for verification: %s; "
+                    "keeping its %d result(s)",
+                    mailbox_id,
+                    e,
+                    len(mb_results),
+                )
+                for r in mb_results:
+                    accessible.add(r.id)
+                return
+
+        present_ids = {
+            str(m.get("databaseId"))
+            for m in messages
+            if m.get("databaseId") is not None
+        }
+        for r in mb_results:
+            if r.id in present_ids:
+                accessible.add(r.id)
+            elif not is_valid_nextcloud_doc_id(r.id):
+                # Malformed stored id can't match the numeric listing; keep it
+                # (fail-open) rather than drop a possibly-legitimate result —
+                # mirrors the notes/news posture.
+                logger.warning(
+                    "Malformed mail_message doc_id %r in verifier; keeping",
+                    r.id,
+                )
+                accessible.add(r.id)
+            # else: genuinely absent (deleted or aged out) -> drop + evict.
+
+    async with anyio.create_task_group() as tg:
+        for mailbox_id, mb_results in by_mailbox.items():
+            tg.start_soon(check_mailbox, mailbox_id, mb_results)
+
+    return accessible
+
+
 _VERIFIERS: dict[str, BatchVerifier] = {
     "note": _verify_notes,
     "file": _verify_files,
     "deck_card": _verify_deck_cards,
     "news_item": _verify_news_items,
+    "mail_message": _verify_mail_messages,
 }
 
 

nextcloud_mcp_server/server/__init__.py

@@ -7,6 +7,7 @@ from .collectives import configure_collectives_tools
 from .contacts import configure_contacts_tools
 from .cookbook import configure_cookbook_tools
 from .deck import configure_deck_tools
+from .mail import configure_mail_tools
 from .news import configure_news_tools
 from .notes import configure_notes_tools
 from .semantic import configure_semantic_tools
@@ -30,6 +31,7 @@ AVAILABLE_APPS: dict[str, Callable[[FastMCP], None]] = {
     "cookbook": configure_cookbook_tools,
     "deck": configure_deck_tools,
     "news": configure_news_tools,
+    "mail": configure_mail_tools,
     "talk": configure_talk_tools,
 }
 
@@ -40,6 +42,7 @@ __all__ = [
     "configure_contacts_tools",
     "configure_cookbook_tools",
     "configure_deck_tools",
+    "configure_mail_tools",
     "configure_news_tools",
     "configure_notes_tools",
     "configure_semantic_tools",

nextcloud_mcp_server/server/mail.py

@@ -0,0 +1,270 @@
+"""MCP tools for Nextcloud Mail app (read-only)."""
+
+import logging
+
+from httpx import HTTPStatusError, RequestError
+from mcp.server.fastmcp import Context, FastMCP
+from mcp.shared.exceptions import McpError
+from mcp.types import ErrorData, ToolAnnotations
+
+from nextcloud_mcp_server.auth import require_scopes
+from nextcloud_mcp_server.context import get_client
+from nextcloud_mcp_server.models.mail import (
+    GetAttachmentResponse,
+    GetMessageResponse,
+    ListAccountsResponse,
+    ListMailboxesResponse,
+    ListMessagesResponse,
+    MailAccount,
+    MailMailbox,
+    MailMessage,
+    MailMessageSummary,
+)
+from nextcloud_mcp_server.observability.metrics import instrument_tool
+
+logger = logging.getLogger(__name__)
+
+# Hard cap on inlined attachment content. The Mail OCS API returns the full
+# attachment body in the JSON response, which then has to fit in the host LLM's
+# context window; replace anything larger with a sentinel so a 20 MB design file
+# can't blow up the MCP response. Callers can still see the real size via the
+# message's attachment list.
+MAX_ATTACHMENT_CONTENT_BYTES = 5 * 1024 * 1024
+
+
+def _cap_attachment_content(content: str | None) -> str | None:
+    """Replace oversized attachment content with a size sentinel.
+
+    Measures UTF-8 byte length (what actually lands in the MCP response/LLM
+    context), not character count. Non-string content is returned unchanged.
+    """
+    if not isinstance(content, str):
+        return content
+    content_bytes = len(content.encode("utf-8"))
+    if content_bytes > MAX_ATTACHMENT_CONTENT_BYTES:
+        return (
+            f"[attachment too large to inline: {content_bytes} bytes "
+            f"(> {MAX_ATTACHMENT_CONTENT_BYTES})]"
+        )
+    return content
+
+
+def configure_mail_tools(mcp: FastMCP):
+    """Configure Mail app MCP tools (read-only)."""
+
+    @mcp.tool(
+        title="List Mail Accounts",
+        annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
+    )
+    @require_scopes("mail.read")
+    @instrument_tool
+    async def nc_mail_list_accounts(ctx: Context) -> ListAccountsResponse:
+        """List the user's configured mail accounts (requires mail.read scope)."""
+        client = await get_client(ctx)
+        try:
+            accounts_data = await client.mail.list_accounts()
+            accounts = [MailAccount(**a) for a in accounts_data]
+            return ListAccountsResponse(results=accounts, total_count=len(accounts))
+        except RequestError as e:
+            raise McpError(
+                ErrorData(code=-1, message=f"Network error listing accounts: {str(e)}")
+            )
+        except HTTPStatusError as e:
+            raise McpError(
+                ErrorData(
+                    code=-1,
+                    message=f"Failed to list accounts: {e.response.status_code}",
+                )
+            )
+
+    @mcp.tool(
+        title="List Mail Mailboxes",
+        annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
+    )
+    @require_scopes("mail.read")
+    @instrument_tool
+    async def nc_mail_list_mailboxes(
+        account_id: int, ctx: Context
+    ) -> ListMailboxesResponse:
+        """List the mailboxes (folders) of a mail account (requires mail.read scope).
+
+        Args:
+            account_id: Account ID (from nc_mail_list_accounts)
+
+        Returns:
+            ListMailboxesResponse with mailboxes. Use a mailbox's ``database_id``
+            with nc_mail_list_messages.
+        """
+        client = await get_client(ctx)
+        try:
+            mailboxes_data = await client.mail.get_mailboxes(account_id)
+            mailboxes = [MailMailbox(**m) for m in mailboxes_data]
+            return ListMailboxesResponse(results=mailboxes, total_count=len(mailboxes))
+        except RequestError as e:
+            raise McpError(
+                ErrorData(code=-1, message=f"Network error listing mailboxes: {str(e)}")
+            )
+        except HTTPStatusError as e:
+            raise McpError(
+                ErrorData(
+                    code=-1,
+                    message=f"Failed to list mailboxes: {e.response.status_code}",
+                )
+            )
+
+    @mcp.tool(
+        title="List Mail Messages",
+        annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
+    )
+    @require_scopes("mail.read")
+    @instrument_tool
+    async def nc_mail_list_messages(
+        mailbox_id: int,
+        ctx: Context,
+        cursor: int | None = None,
+        search_filter: str | None = None,
+        limit: int = 20,
+    ) -> ListMessagesResponse:
+        """List message envelopes in a mailbox, newest first (requires mail.read scope).
+
+        Reads cached envelope metadata (fast); does not fetch bodies. Use
+        nc_mail_get_message to fetch a full body.
+
+        Args:
+            mailbox_id: Numeric mailbox id (``database_id`` from nc_mail_list_mailboxes)
+            cursor: Pagination cursor from a prior page
+            search_filter: Optional search/filter query
+            limit: Max messages to return (1-100, default 20)
+
+        Returns:
+            ListMessagesResponse with message summaries. ``has_more`` is a
+            heuristic (true when exactly ``limit`` messages were returned), so it
+            can be a false positive when a mailbox holds exactly ``limit``
+            messages; page with ``cursor`` and stop on an empty result.
+        """
+        client = await get_client(ctx)
+        # Clamp to the same window the client/OCS API enforce so the has_more
+        # heuristic compares against the limit actually applied (a caller passing
+        # limit<=0 otherwise gets a misleading count).
+        effective_limit = min(max(1, limit), 100)
+        try:
+            messages_data = await client.mail.list_messages(
+                mailbox_id,
+                cursor=cursor,
+                search_filter=search_filter,
+                limit=effective_limit,
+            )
+            messages = [MailMessageSummary(**m) for m in messages_data]
+            return ListMessagesResponse(
+                results=messages,
+                total_count=len(messages),
+                has_more=len(messages) == effective_limit,
+            )
+        except RequestError as e:
+            raise McpError(
+                ErrorData(code=-1, message=f"Network error listing messages: {str(e)}")
+            )
+        except HTTPStatusError as e:
+            raise McpError(
+                ErrorData(
+                    code=-1,
+                    message=f"Failed to list messages: {e.response.status_code}",
+                )
+            )
+
+    @mcp.tool(
+        title="Get Mail Message",
+        annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
+    )
+    @require_scopes("mail.read")
+    @instrument_tool
+    async def nc_mail_get_message(message_id: int, ctx: Context) -> GetMessageResponse:
+        """Get a single mail message with its full body (requires mail.read scope).
+
+        The Mail app fetches the body from IMAP server-side.
+
+        Args:
+            message_id: Numeric message id (``database_id`` from nc_mail_list_messages)
+
+        Returns:
+            GetMessageResponse with the full message including body and attachments.
+            Attachments with ``id: null`` are inline body parts and cannot be
+            fetched via nc_mail_get_attachment (which requires a string id).
+        """
+        client = await get_client(ctx)
+        try:
+            message_data = await client.mail.get_message(message_id)
+            # An empty payload (OCS data=null with a 200 meta) would make
+            # MailMessage(**{}) raise an uncaught ValidationError; treat it as
+            # not-found instead.
+            if not message_data:
+                raise McpError(
+                    ErrorData(code=-1, message=f"Message {message_id} not found")
+                )
+            message = MailMessage(**message_data)
+            return GetMessageResponse(message=message)
+        except RequestError as e:
+            raise McpError(
+                ErrorData(
+                    code=-1,
+                    message=f"Network error getting message {message_id}: {str(e)}",
+                )
+            )
+        except HTTPStatusError as e:
+            if e.response.status_code == 404:
+                raise McpError(
+                    ErrorData(code=-1, message=f"Message {message_id} not found")
+                )
+            raise McpError(
+                ErrorData(
+                    code=-1,
+                    message=f"Failed to get message {message_id}: "
+                    f"{e.response.status_code}",
+                )
+            )
+
+    @mcp.tool(
+        title="Get Mail Attachment",
+        annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
+    )
+    @require_scopes("mail.read")
+    @instrument_tool
+    async def nc_mail_get_attachment(
+        message_id: int, attachment_id: str, ctx: Context
+    ) -> GetAttachmentResponse:
+        """Get a single mail attachment's metadata and content (requires mail.read scope).
+
+        Args:
+            message_id: Numeric message id
+            attachment_id: Attachment id (a string, from the message's attachments)
+
+        Returns:
+            GetAttachmentResponse with name, mime, size, and content. ``content``
+            is the attachment body as returned by the Mail OCS API; large
+            attachments produce a correspondingly large response, so prefer the
+            ``size`` from the message's attachment list before fetching.
+        """
+        client = await get_client(ctx)
+        try:
+            data = await client.mail.get_attachment(message_id, attachment_id)
+            return GetAttachmentResponse(
+                name=data.get("name"),
+                mime=data.get("mime"),
+                size=data.get("size"),
+                content=_cap_attachment_content(data.get("content")),
+            )
+        except RequestError as e:
+            raise McpError(
+                ErrorData(
+                    code=-1, message=f"Network error getting attachment: {str(e)}"
+                )
+            )
+        except HTTPStatusError as e:
+            if e.response.status_code == 404:
+                raise McpError(ErrorData(code=-1, message="Attachment not found"))
+            raise McpError(
+                ErrorData(
+                    code=-1,
+                    message=f"Failed to get attachment: {e.response.status_code}",
+                )
+            )

nextcloud_mcp_server/server/semantic.py

@@ -214,12 +214,12 @@ def configure_semantic_tools(mcp: FastMCP):
         understanding and keyword precision.
 
         Requires VECTOR_SYNC_ENABLED=true. Supports indexing of notes, files,
-        news items, and deck cards.
+        news items, deck cards, and mail messages.
 
         Args:
             query: Natural language or keyword search query
             limit: Maximum number of results to return (default: 10)
-            doc_types: Document types to search (e.g., ["note", "file", "deck_card", "news_item"]). None = search all indexed types (default)
+            doc_types: Document types to search (e.g., ["note", "file", "deck_card", "news_item", "mail_message"]). None = search all indexed types (default)
             score_threshold: Minimum fusion score (0-1, default: 0.0)
             fusion: Fusion algorithm: "rrf" (Reciprocal Rank Fusion, default) or "dbsf" (Distribution-Based Score Fusion)
                    RRF: Good general-purpose fusion using reciprocal ranks

nextcloud_mcp_server/vector/mail_content.py

@@ -0,0 +1,78 @@
+"""Shared reconstruction of mail-message content for indexing and context.
+
+The vector processor (index-time) and search context expansion (query-time)
+must build the *identical* text for a mail message so chunk offsets align.
+Keeping that logic here — rather than copy-pasted in both call sites — is the
+single source of truth for the reconstruction.
+"""
+
+from typing import Any
+
+from nextcloud_mcp_server.vector.html_processor import html_to_markdown
+
+# Newest-N messages indexed (and verified) per mailbox. This equals the Mail
+# OCS API's per-request maximum (it clamps ``limit`` to 1..100), so it cannot be
+# raised without adding cursor pagination — hence a documented constant rather
+# than a config knob that would silently cap at 100. Shared by the scanner
+# (index window) and the verifier (presence window) so they stay consistent.
+MAIL_SCAN_MAX_PER_MAILBOX = 100
+
+
+def format_mail_addresses(addrs: list[dict[str, Any]] | None) -> str:
+    """Render a list of {label, email} address objects as a display string.
+
+    ``None`` and address objects with neither ``label`` nor ``email`` are
+    skipped (yielding ``""`` for an all-empty list) — IMAP envelope addresses
+    effectively always carry at least an email, so this only drops malformed
+    entries rather than losing real recipients.
+    """
+    parts: list[str] = []
+    for addr in addrs or []:
+        label = addr.get("label")
+        email = addr.get("email")
+        if label and email and label != email:
+            parts.append(f"{label} <{email}>")
+        elif email:
+            parts.append(email)
+        elif label:
+            parts.append(label)
+    return ", ".join(parts)
+
+
+def build_mail_content(message: dict[str, Any]) -> str:
+    """Reconstruct the indexed text body for a mail message.
+
+    Layout (kept stable so index-time and query-time offsets match):
+        <subject>
+        From: <from>
+        To: <to>
+        Cc: <cc>          # only when non-empty
+        Bcc: <bcc>        # only when non-empty
+        <blank line>
+        <body>
+
+    Cc/Bcc are included so recipient-oriented queries ("emails where alice was
+    cc'd") can match. The body is the Mail OCS ``body`` field — sanitized HTML
+    when ``hasHtmlBody`` is set (converted to Markdown for embedding), otherwise
+    plain text.
+    """
+    subject = message.get("subject") or ""
+    from_str = format_mail_addresses(message.get("from"))
+    to_str = format_mail_addresses(message.get("to"))
+    cc_str = format_mail_addresses(message.get("cc"))
+    bcc_str = format_mail_addresses(message.get("bcc"))
+    raw_body = message.get("body") or ""
+    body_text = html_to_markdown(raw_body) if message.get("hasHtmlBody") else raw_body
+
+    content_parts = [subject]
+    if from_str:
+        content_parts.append(f"From: {from_str}")
+    if to_str:
+        content_parts.append(f"To: {to_str}")
+    if cc_str:
+        content_parts.append(f"Cc: {cc_str}")
+    if bcc_str:
+        content_parts.append(f"Bcc: {bcc_str}")
+    content_parts.append("")  # Blank line
+    content_parts.append(body_text)
+    return "\n".join(content_parts)

nextcloud_mcp_server/vector/processor.py

@@ -42,6 +42,7 @@ from nextcloud_mcp_server.observability.metrics import (
 from nextcloud_mcp_server.observability.tracing import trace_operation
 from nextcloud_mcp_server.search.pdf_highlighter import PDFHighlighter
 from nextcloud_mcp_server.usage import UsageEventStore
+from nextcloud_mcp_server.utils.validation import is_valid_nextcloud_doc_id
 from nextcloud_mcp_server.vector import payload_keys
 from nextcloud_mcp_server.vector._errors import format_exception_group
 from nextcloud_mcp_server.vector.dead_letter import (
@@ -53,6 +54,10 @@ from nextcloud_mcp_server.vector.document_chunker import (
     PageAwareChunker,
 )
 from nextcloud_mcp_server.vector.html_processor import html_to_markdown
+from nextcloud_mcp_server.vector.mail_content import (
+    build_mail_content,
+    format_mail_addresses,
+)
 from nextcloud_mcp_server.vector.placeholder import (
     delete_placeholder_point,
     update_placeholder_status,
@@ -840,6 +845,44 @@ async def _index_document(
             file_path = None
             content_bytes = None
             content_type = None
+        elif doc_task.doc_type == "mail_message":
+            # Fetch the full message via the Mail OCS API. The Mail app handles
+            # IMAP server-side; we only ever speak HTTP. build_mail_content is
+            # shared with search/context.py so index- and query-time text match.
+            # Guard the cast before the network call (consistent with the same
+            # doc_type in search/context.py) so a malformed queue record produces
+            # a specific error rather than a bare ValueError.
+            if not is_valid_nextcloud_doc_id(doc_task.doc_id):
+                raise ValueError(f"Invalid mail_message doc_id: {doc_task.doc_id!r}")
+            message = await nc_client.mail.get_message(int(doc_task.doc_id))
+            # An empty payload (OCS data=null with a <400 meta) would otherwise
+            # index a useless near-empty placeholder; fail loudly so the task
+            # dead-letters instead of corrupting the index.
+            if not message:
+                raise ValueError(
+                    f"mail_message {doc_task.doc_id!r} returned an empty payload"
+                )
+            content = build_mail_content(message)
+
+            subject = message.get("subject") or ""
+            title = subject
+            # Email is immutable; key change-detection on the message id so a
+            # re-index is a no-op unless the id changes.
+            etag = str(message.get("id") or doc_task.doc_id)
+            file_metadata = {
+                "subject": subject,
+                "from": format_mail_addresses(message.get("from")),
+                "to": format_mail_addresses(message.get("to")),
+                "cc": format_mail_addresses(message.get("cc")),
+                "bcc": format_mail_addresses(message.get("bcc")),
+                "date_int": message.get("dateInt"),
+                "has_attachments": bool(message.get("attachments")),
+                "account_id": (doc_task.metadata or {}).get("account_id"),
+                "mailbox_id": (doc_task.metadata or {}).get("mailbox_id"),
+            }
+            file_path = None
+            content_bytes = None
+            content_type = None
         elif doc_task.doc_type == "deck_card":
             # Fetch card from Deck API
             # Use metadata from scanner if available (O(1) lookup)
@@ -1604,6 +1647,22 @@ async def _index_document(
                         if doc_task.doc_type == "deck_card"
                         else {}
                     ),
+                    # Mail message-specific metadata
+                    **(
+                        {
+                            "subject": file_metadata.get("subject"),
+                            "from": file_metadata.get("from"),
+                            "to": file_metadata.get("to"),
+                            "cc": file_metadata.get("cc"),
+                            "bcc": file_metadata.get("bcc"),
+                            "date_int": file_metadata.get("date_int"),
+                            "has_attachments": file_metadata.get("has_attachments"),
+                            "account_id": file_metadata.get("account_id"),
+                            "mailbox_id": file_metadata.get("mailbox_id"),
+                        }
+                        if doc_task.doc_type == "mail_message"
+                        else {}
+                    ),
                     # Chunk bbox (PDF only) — normalized rectangles in [0,1]
                     # relative to page width/height. Replaces the legacy
                     # `highlighted_page_image` (Deck #76). The page number

nextcloud_mcp_server/vector/scanner.py

@@ -28,6 +28,7 @@ from nextcloud_mcp_server.server.tag_exclusion import (
     is_path_excluded,
 )
 from nextcloud_mcp_server.vector.dead_letter import is_dead_lettered
+from nextcloud_mcp_server.vector.mail_content import MAIL_SCAN_MAX_PER_MAILBOX
 from nextcloud_mcp_server.vector.placeholder import (
     query_document_metadata,
     write_placeholder_point,
@@ -48,7 +49,7 @@ logger = logging.getLogger(__name__)
 # same PR that adds a new indexed doc_type, or accept ghost-record exposure for
 # that type (see ADR-019).
 INDEXED_DOC_TYPES: frozenset[str] = frozenset(
-    {"note", "file", "deck_card", "news_item"}
+    {"note", "file", "deck_card", "news_item", "mail_message"}
 )
 
 
@@ -956,13 +957,36 @@ async def scan_user_documents(
                 user_id,
             )
 
+        # Scan Mail messages (newest per mailbox)
+        mail_queued = 0
+        if _should_scan("mail", "mail_message", enabled_apps, allowed):
+            try:
+                mail_queued = await scan_mail_messages(
+                    user_id=user_id,
+                    send_stream=send_stream,
+                    nc_client=nc_client,
+                    initial_sync=initial_sync,
+                    scan_id=scan_id,
+                )
+                queued += mail_queued
+            except Exception as e:
+                logger.warning("Failed to scan mail messages for %s: %s", user_id, e)
+        else:
+            logger.debug(
+                "[SCAN-%s] Mail app not enabled for %s; skipping mail messages",
+                scan_id,
+                user_id,
+            )
+
         if queued > 0:
             logger.info(
-                "Sent %s documents (%s files, %s news items, %s deck cards) for incremental sync: %s",
+                "Sent %s documents (%s files, %s news items, %s deck cards, "
+                "%s mail messages) for incremental sync: %s",
                 queued,
                 file_queued,
                 news_queued,
                 deck_queued,
+                mail_queued,
                 user_id,
             )
         else:
@@ -1332,6 +1356,291 @@ async def scan_news_items(
     return queued
 
 
+# Newest-N messages indexed per mailbox (= the Mail OCS per-request maximum;
+# imported from mail_content so the scanner index window and the search-time
+# verifier presence window stay identical). Older messages age out of the index
+# as newer ones arrive — the deletion-tracking pass below evicts them, the same
+# way it handles actually-deleted messages. Going beyond 100 would require
+# cursor pagination, so the value is fixed rather than configurable.
+#
+# Per-process record of (user_id, mailbox_id) for which the newest-N cap has
+# already been logged, so the "older mail not indexed" notice is emitted once at
+# info level (discoverable) rather than on every scan tick (which would flood
+# multi-tenant logs). Insertion-ordered dict + bounded eviction (mirrors
+# _consent_backstop_done) so a long-running multi-tenant process can't leak it.
+_mail_cap_logged: dict[tuple[str, int], None] = {}
+_MAIL_CAP_LOGGED_MAX = 50_000
+
+
+def _mark_mail_cap_logged(key: tuple[str, int]) -> bool:
+    """Record a one-shot cap-log marker; return True if this is the first time.
+
+    Evicts oldest-first to half capacity on overflow (a re-log after eviction is
+    a harmless info line), so the dedup set stays bounded.
+    """
+    if key in _mail_cap_logged:
+        return False
+    if len(_mail_cap_logged) >= _MAIL_CAP_LOGGED_MAX:
+        overage = len(_mail_cap_logged) - _MAIL_CAP_LOGGED_MAX // 2
+        for stale_key in list(_mail_cap_logged)[:overage]:
+            del _mail_cap_logged[stale_key]
+    _mail_cap_logged[key] = None
+    return True
+
+
+async def scan_mail_messages(
+    user_id: str,
+    send_stream: TaskProducer,
+    nc_client: NextcloudClient,
+    initial_sync: bool,
+    scan_id: int,
+) -> int:
+    """
+    Scan a user's Mail messages and queue changed messages for indexing.
+
+    Enumerates accounts → mailboxes → newest ``MAIL_SCAN_MAX_PER_MAILBOX``
+    messages per mailbox. Email is immutable, so a message's ``dateInt`` (sent
+    timestamp) is used as the change-detection ``modified_at`` — a message is
+    indexed once and not re-sent. Messages that drop out of the newest-N window
+    (or are deleted) are evicted via the deletion-tracking pass, keeping the
+    index bounded to recent mail.
+
+    The MCP server never speaks IMAP: listing reads the Mail app's DB-cached
+    envelopes, and the body fetch (in the processor) goes through the Mail app's
+    OCS API, which handles IMAP server-side.
+
+    Args:
+        user_id: User to scan
+        send_stream: Stream to send changed documents to processors
+        nc_client: Authenticated Nextcloud client
+        initial_sync: If True, send all documents (first-time sync)
+        scan_id: Scan identifier for logging
+
+    Returns:
+        Number of messages queued for processing
+    """
+    settings = get_settings()
+    queued = 0
+
+    # Get indexed mail message IDs from Qdrant (for deletion tracking)
+    indexed_message_ids: set[str] = set()
+    if not initial_sync:
+        qdrant_client = await get_qdrant_client()
+        points = await _scroll_all_points(
+            qdrant_client,
+            collection_name=settings.get_collection_name(),
+            scroll_filter=Filter(
+                must=[
+                    FieldCondition(key="user_id", match=MatchValue(value=user_id)),
+                    FieldCondition(
+                        key="doc_type", match=MatchValue(value="mail_message")
+                    ),
+                ]
+            ),
+            payload_fields=["doc_id"],
+        )
+        indexed_message_ids = {
+            str(point.payload["doc_id"])
+            for point in points
+            if point.payload is not None and "doc_id" in point.payload
+        }
+        logger.debug(
+            "Found %s indexed mail messages in Qdrant", len(indexed_message_ids)
+        )
+
+    # Enumerate accounts → mailboxes → newest-N messages.
+    accounts = await nc_client.mail.list_accounts()
+    nextcloud_message_ids: set[str] = set()
+    message_count = 0
+
+    for account in accounts:
+        account_id = account.get("id")
+        if account_id is None:
+            continue
+        try:
+            mailboxes = await nc_client.mail.get_mailboxes(account_id)
+        except Exception as e:
+            logger.warning(
+                "[SCAN-%s] Failed to list mailboxes for account %s: %s",
+                scan_id,
+                account_id,
+                e,
+            )
+            continue
+
+        for mailbox in mailboxes:
+            mailbox_id = mailbox.get("databaseId")
+            if mailbox_id is None:
+                continue
+            try:
+                messages = await nc_client.mail.list_messages(
+                    mailbox_id, limit=MAIL_SCAN_MAX_PER_MAILBOX
+                )
+            except Exception as e:
+                logger.warning(
+                    "[SCAN-%s] Failed to list messages for mailbox %s: %s",
+                    scan_id,
+                    mailbox_id,
+                    e,
+                )
+                continue
+
+            if len(messages) >= MAIL_SCAN_MAX_PER_MAILBOX and _mark_mail_cap_logged(
+                (user_id, mailbox_id)
+            ):
+                logger.info(
+                    "[SCAN-%s] Mailbox %s contains more than %s messages; only "
+                    "the newest %s are indexed (cursor pagination not yet "
+                    "implemented)",
+                    scan_id,
+                    mailbox_id,
+                    MAIL_SCAN_MAX_PER_MAILBOX,
+                    MAIL_SCAN_MAX_PER_MAILBOX,
+                )
+
+            for message in messages:
+                msg_db_id = message.get("databaseId")
+                if msg_db_id is None:
+                    continue
+                doc_id = str(msg_db_id)
+                nextcloud_message_ids.add(doc_id)
+                message_count += 1
+
+                modified_at = message.get("dateInt", 0) or 0
+                task_metadata: dict[str, int | str] = {
+                    "account_id": account_id,
+                    "mailbox_id": mailbox_id,
+                }
+
+                if initial_sync:
+                    await write_placeholder_point(
+                        doc_id=doc_id,
+                        doc_type="mail_message",
+                        user_id=user_id,
+                        modified_at=modified_at,
+                    )
+                    await send_stream.send(
+                        DocumentTask(
+                            user_id=user_id,
+                            doc_id=doc_id,
+                            doc_type="mail_message",
+                            operation="index",
+                            modified_at=modified_at,
+                            metadata=task_metadata,
+                        )
+                    )
+                    queued += 1
+                else:
+                    doc_key = (user_id, doc_id)
+                    if doc_key in _potentially_deleted:
+                        logger.debug(
+                            "Mail message %s reappeared, removing from deletion "
+                            "grace period",
+                            doc_id,
+                        )
+                        del _potentially_deleted[doc_key]
+
+                    existing_metadata = await query_document_metadata(
+                        doc_id=doc_id, doc_type="mail_message", user_id=user_id
+                    )
+
+                    needs_indexing = False
+                    if existing_metadata is None:
+                        needs_indexing = True
+                    elif existing_metadata.get("modified_at", 0) < modified_at:
+                        needs_indexing = True
+                    elif existing_metadata.get("status") == "failed":
+                        # A permanent processing failure — don't re-queue an
+                        # unchanged message that will just fail again; the
+                        # modified_at branch above retries once it changes
+                        # (mirrors the file scanner's failed-placeholder guard).
+                        logger.debug(
+                            "Skipping mail message %s: previous processing "
+                            "failed permanently",
+                            doc_id,
+                        )
+                    elif existing_metadata.get("is_placeholder", False):
+                        queued_at = existing_metadata.get("queued_at", 0)
+                        placeholder_age = time.time() - queued_at
+                        stale_threshold = settings.vector_sync_scan_interval * 5
+                        if placeholder_age > stale_threshold:
+                            logger.debug(
+                                "Found stale placeholder for mail message %s "
+                                "(age=%ss), requeuing",
+                                doc_id,
+                                format(placeholder_age, ".1f"),
+                            )
+                            needs_indexing = True
+
+                    if needs_indexing:
+                        await write_placeholder_point(
+                            doc_id=doc_id,
+                            doc_type="mail_message",
+                            user_id=user_id,
+                            modified_at=modified_at,
+                        )
+                        await send_stream.send(
+                            DocumentTask(
+                                user_id=user_id,
+                                doc_id=doc_id,
+                                doc_type="mail_message",
+                                operation="index",
+                                modified_at=modified_at,
+                                metadata=task_metadata,
+                            )
+                        )
+                        queued += 1
+
+    logger.info(
+        "[SCAN-%s] Found %s mail messages for %s",
+        scan_id,
+        message_count,
+        user_id,
+    )
+    record_vector_sync_scan(message_count)
+
+    # Check for deleted / aged-out messages (not initial sync)
+    if not initial_sync:
+        grace_period = settings.vector_sync_scan_interval * 1.5
+        current_time = time.time()
+
+        for doc_id in indexed_message_ids:
+            if doc_id not in nextcloud_message_ids:
+                doc_key = (user_id, doc_id)
+
+                if doc_key in _potentially_deleted:
+                    first_missing_time = _potentially_deleted[doc_key]
+                    time_missing = current_time - first_missing_time
+
+                    if time_missing >= grace_period:
+                        logger.info(
+                            "Mail message %s missing for %ss (>%ss grace period), "
+                            "sending deletion",
+                            doc_id,
+                            format(time_missing, ".1f"),
+                            format(grace_period, ".1f"),
+                        )
+                        await send_stream.send(
+                            DocumentTask(
+                                user_id=user_id,
+                                doc_id=doc_id,
+                                doc_type="mail_message",
+                                operation="delete",
+                                modified_at=0,
+                            )
+                        )
+                        queued += 1
+                        del _potentially_deleted[doc_key]
+                else:
+                    logger.debug(
+                        "Mail message %s missing for first time, starting grace period",
+                        doc_id,
+                    )
+                    _potentially_deleted[doc_key] = current_time
+
+    return queued
+
+
 async def scan_deck_cards(
     user_id: str,
     send_stream: TaskProducer,

tests/client/mail/test_mail_api.py

@@ -0,0 +1,262 @@
+"""Unit tests for MailClient API methods."""
+
+import logging
+from typing import Any
+
+import httpx
+import pytest
+
+from nextcloud_mcp_server.client.mail import MailClient
+from tests.client.conftest import create_mock_response
+
+logger = logging.getLogger(__name__)
+
+# Mark all tests in this module as unit tests
+pytestmark = pytest.mark.unit
+
+
+def _ocs_response(data: Any, status_code: int = 200) -> httpx.Response:
+    """Wrap a payload in the standard OCS envelope."""
+    return create_mock_response(
+        status_code=status_code,
+        json_data={
+            "ocs": {
+                "meta": {"status": "ok", "statuscode": status_code, "message": "OK"},
+                "data": data,
+            }
+        },
+    )
+
+
+async def test_list_accounts_unwraps_ocs_envelope(mocker):
+    """list_accounts returns the ocs.data payload."""
+    mock_response = _ocs_response(
+        [
+            {"id": 1, "email": "alice@example.com", "isDelegated": False},
+            {"id": 2, "email": "bob@example.com", "isDelegated": False},
+        ]
+    )
+    mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
+    mock_make_request = mocker.patch.object(
+        MailClient, "_make_request", return_value=mock_response
+    )
+
+    client = MailClient(mock_client, "testuser")
+    accounts = await client.list_accounts()
+
+    assert len(accounts) == 2
+    assert accounts[0]["id"] == 1
+    assert accounts[0]["email"] == "alice@example.com"
+
+    # Correct URL, OCS header, and format=json param.
+    args, kwargs = mock_make_request.call_args
+    assert args == ("GET", "/ocs/v2.php/apps/mail/api/account/list")
+    assert kwargs["headers"]["OCS-APIRequest"] == "true"
+    assert kwargs["params"]["format"] == "json"
+
+
+async def test_get_mailboxes_passes_account_id(mocker):
+    """get_mailboxes sends accountId and unwraps the list."""
+    mock_response = _ocs_response(
+        [
+            {
+                "databaseId": 10,
+                "id": "SU5CT1g=",
+                "name": "INBOX",
+                "displayName": "INBOX",
+                "accountId": 1,
+                "specialUse": ["inbox"],
+                "unread": 3,
+            }
+        ]
+    )
+    mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
+    mock_make_request = mocker.patch.object(
+        MailClient, "_make_request", return_value=mock_response
+    )
+
+    client = MailClient(mock_client, "testuser")
+    mailboxes = await client.get_mailboxes(account_id=1)
+
+    assert len(mailboxes) == 1
+    assert mailboxes[0]["databaseId"] == 10
+    assert mailboxes[0]["specialUse"] == ["inbox"]
+
+    args, kwargs = mock_make_request.call_args
+    assert args == ("GET", "/ocs/v2.php/apps/mail/api/mailboxes")
+    assert kwargs["params"]["accountId"] == 1
+
+
+async def test_list_messages_builds_params(mocker):
+    """list_messages forwards limit/cursor/filter/view query params."""
+    mock_response = _ocs_response(
+        [
+            {
+                "databaseId": 100,
+                "subject": "Hello",
+                "dateInt": 1700000000,
+                "from": [{"label": "Alice", "email": "alice@example.com"}],
+                "to": [{"label": "Bob", "email": "bob@example.com"}],
+                "mailboxId": 10,
+            }
+        ]
+    )
+    mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
+    mock_make_request = mocker.patch.object(
+        MailClient, "_make_request", return_value=mock_response
+    )
+
+    client = MailClient(mock_client, "testuser")
+    messages = await client.list_messages(
+        10, cursor=42, search_filter="hello", limit=50, view="threaded"
+    )
+
+    assert len(messages) == 1
+    assert messages[0]["databaseId"] == 100
+
+    args, kwargs = mock_make_request.call_args
+    assert args == ("GET", "/ocs/v2.php/apps/mail/api/mailboxes/10/messages")
+    assert kwargs["params"]["limit"] == 50
+    assert kwargs["params"]["cursor"] == 42
+    assert kwargs["params"]["filter"] == "hello"
+    assert kwargs["params"]["view"] == "threaded"
+
+
+async def test_list_messages_omits_optional_params(mocker):
+    """Optional params are omitted when not supplied; limit always present."""
+    mock_response = _ocs_response([])
+    mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
+    mock_make_request = mocker.patch.object(
+        MailClient, "_make_request", return_value=mock_response
+    )
+
+    client = MailClient(mock_client, "testuser")
+    await client.list_messages(10)
+
+    _, kwargs = mock_make_request.call_args
+    params = kwargs["params"]
+    assert params["limit"] == 20  # default
+    assert "cursor" not in params
+    assert "filter" not in params
+    assert "view" not in params
+
+
+async def test_get_message_unwraps_full_message(mocker):
+    """get_message returns the full message dict."""
+    mock_response = _ocs_response(
+        {
+            "id": 100,
+            "subject": "Hello",
+            "hasHtmlBody": True,
+            "body": "<p>Hi there</p>",
+            "from": [{"label": "Alice", "email": "alice@example.com"}],
+            "attachments": [
+                {
+                    "id": "1.2",
+                    "fileName": "doc.pdf",
+                    "mime": "application/pdf",
+                    "size": 1024,
+                }
+            ],
+        }
+    )
+    mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
+    mock_make_request = mocker.patch.object(
+        MailClient, "_make_request", return_value=mock_response
+    )
+
+    client = MailClient(mock_client, "testuser")
+    message = await client.get_message(100)
+
+    assert message["id"] == 100
+    assert message["hasHtmlBody"] is True
+    assert message["attachments"][0]["fileName"] == "doc.pdf"
+
+    args, _ = mock_make_request.call_args
+    assert args == ("GET", "/ocs/v2.php/apps/mail/api/message/100")
+
+
+async def test_get_attachment_unwraps_json(mocker):
+    """get_attachment returns the JSON attachment object (not a binary download)."""
+    mock_response = _ocs_response(
+        {"name": "doc.pdf", "mime": "application/pdf", "size": 1024, "content": "abc"}
+    )
+    mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
+    mock_make_request = mocker.patch.object(
+        MailClient, "_make_request", return_value=mock_response
+    )
+
+    client = MailClient(mock_client, "testuser")
+    attachment = await client.get_attachment(100, "1.2")
+
+    assert attachment["name"] == "doc.pdf"
+    assert attachment["content"] == "abc"
+
+    args, _ = mock_make_request.call_args
+    assert args == ("GET", "/ocs/v2.php/apps/mail/api/message/100/attachment/1.2")
+
+
+async def test_get_attachment_url_encodes_attachment_id(mocker):
+    """A traversal-style attachment_id is percent-encoded in the URL path."""
+    mock_response = _ocs_response({"name": "x", "content": "y"})
+    mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
+    mock_make_request = mocker.patch.object(
+        MailClient, "_make_request", return_value=mock_response
+    )
+
+    client = MailClient(mock_client, "testuser")
+    await client.get_attachment(100, "../../evil")
+
+    args, _ = mock_make_request.call_args
+    # The "/" and ".." are encoded, so they can't escape the attachment path.
+    assert args == (
+        "GET",
+        "/ocs/v2.php/apps/mail/api/message/100/attachment/..%2F..%2Fevil",
+    )
+
+
+async def test_empty_data_returns_empty_list(mocker):
+    """A null ocs.data payload degrades to an empty list for list endpoints."""
+    mock_response = _ocs_response(None)
+    mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
+    mocker.patch.object(MailClient, "_make_request", return_value=mock_response)
+
+    client = MailClient(mock_client, "testuser")
+    assert await client.list_accounts() == []
+
+
+async def test_ocs_meta_failure_raises_httpstatuserror(mocker):
+    """HTTP 200 with an OCS meta failure code is re-raised as HTTPStatusError.
+
+    The synthetic response carries the OCS statuscode so callers' 404/403
+    handling applies (e.g. nc_mail_get_message maps 404 to 'not found').
+    """
+    mock_response = create_mock_response(
+        status_code=200,
+        json_data={
+            "ocs": {
+                "meta": {"status": "failure", "statuscode": 404, "message": "nope"},
+                "data": None,
+            }
+        },
+    )
+    mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
+    mocker.patch.object(MailClient, "_make_request", return_value=mock_response)
+
+    client = MailClient(mock_client, "testuser")
+    with pytest.raises(httpx.HTTPStatusError) as excinfo:
+        await client.get_message(100)
+    assert excinfo.value.response.status_code == 404
+
+
+async def test_non_json_response_raises_requesterror(mocker):
+    """A non-JSON 200 body (e.g. Mail app absent) raises a RequestError."""
+    mock_response = create_mock_response(
+        status_code=200, content=b"<html>not found</html>"
+    )
+    mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
+    mocker.patch.object(MailClient, "_make_request", return_value=mock_response)
+
+    client = MailClient(mock_client, "testuser")
+    with pytest.raises(httpx.RequestError):
+        await client.list_accounts()

tests/unit/search/test_verification.py

@@ -12,11 +12,13 @@ from nextcloud_mcp_server.search.algorithms import SearchResult
 from nextcloud_mcp_server.search.verification import (
     _verify_deck_cards,
     _verify_files,
+    _verify_mail_messages,
     _verify_news_items,
     _verify_notes,
     get_supported_doc_types,
     verify_search_results,
 )
+from nextcloud_mcp_server.vector.mail_content import MAIL_SCAN_MAX_PER_MAILBOX
 from nextcloud_mcp_server.vector.scanner import INDEXED_DOC_TYPES
 
 # ---------------------------------------------------------------------------
@@ -222,6 +224,134 @@ async def test_verify_notes_string_doc_id_matches_production(mocker):
     notes_client.get_note.assert_awaited_once_with(42)
 
 
+# ---------------------------------------------------------------------------
+# Mail verifier (per-id, mirrors the note verifier)
+# ---------------------------------------------------------------------------
+
+
+def _mail_result(doc_id, mailbox_id=10):
+    """A mail_message SearchResult carrying mailbox_id in its metadata."""
+    return _make_result(
+        doc_id, doc_type="mail_message", metadata={"mailbox_id": mailbox_id}
+    )
+
+
+@pytest.mark.unit
+async def test_verify_mail_batches_one_list_per_mailbox(mocker):
+    """The verifier lists each mailbox once (DB cache, not per-message IMAP)."""
+    list_messages = mocker.AsyncMock(
+        return_value=[{"databaseId": 10}, {"databaseId": 30}]
+    )
+    mail_client = SimpleNamespace(list_messages=list_messages)
+    client = SimpleNamespace(mail=mail_client, username="alice")
+
+    result = await _verify_mail_messages(
+        client,
+        [_mail_result(10), _mail_result(20), _mail_result(30)],  # all mailbox 10
+        _sem(),
+    )
+    # 10 and 30 present; 20 absent (deleted/aged out) -> dropped.
+    assert result == {"10", "30"}
+    # One DB-cached list call for the single mailbox, not one per result.
+    list_messages.assert_awaited_once_with(10, limit=MAIL_SCAN_MAX_PER_MAILBOX)
+
+
+@pytest.mark.unit
+async def test_verify_mail_404_drops_mailbox(mocker):
+    mail_client = SimpleNamespace(
+        list_messages=mocker.AsyncMock(side_effect=_http_error(404))
+    )
+    client = SimpleNamespace(mail=mail_client, username="alice")
+
+    result = await _verify_mail_messages(client, [_mail_result(42)], _sem())
+    assert result == set()
+
+
+@pytest.mark.unit
+async def test_verify_mail_403_drops_mailbox(mocker):
+    mail_client = SimpleNamespace(
+        list_messages=mocker.AsyncMock(side_effect=_http_error(403))
+    )
+    client = SimpleNamespace(mail=mail_client, username="alice")
+
+    result = await _verify_mail_messages(client, [_mail_result(42)], _sem())
+    assert result == set()
+
+
+@pytest.mark.unit
+async def test_verify_mail_transient_5xx_keeps(mocker):
+    mail_client = SimpleNamespace(
+        list_messages=mocker.AsyncMock(side_effect=_http_error(503))
+    )
+    client = SimpleNamespace(mail=mail_client, username="alice")
+
+    result = await _verify_mail_messages(client, [_mail_result(42)], _sem())
+    assert result == {"42"}
+
+
+@pytest.mark.unit
+async def test_verify_mail_missing_mailbox_id_keeps(mocker):
+    """A result without a usable mailbox_id is kept without any network call."""
+    list_messages = mocker.AsyncMock()
+    mail_client = SimpleNamespace(list_messages=list_messages)
+    client = SimpleNamespace(mail=mail_client, username="alice")
+
+    result = await _verify_mail_messages(
+        client, [_make_result(42, doc_type="mail_message")], _sem()
+    )
+    assert result == {"42"}
+    list_messages.assert_not_awaited()
+
+
+@pytest.mark.unit
+async def test_verify_mail_non_numeric_mailbox_id_keeps(mocker):
+    """A non-numeric mailbox_id in metadata is kept without a list call."""
+    list_messages = mocker.AsyncMock()
+    mail_client = SimpleNamespace(list_messages=list_messages)
+    client = SimpleNamespace(mail=mail_client, username="alice")
+
+    result = await _verify_mail_messages(
+        client,
+        [_make_result(42, doc_type="mail_message", metadata={"mailbox_id": "bad"})],
+        _sem(),
+    )
+    assert result == {"42"}
+    list_messages.assert_not_awaited()
+
+
+@pytest.mark.unit
+async def test_verify_mail_non_numeric_id_kept_when_mailbox_listed(mocker):
+    """A malformed doc_id can't match the numeric listing, so it's kept."""
+    mail_client = SimpleNamespace(
+        list_messages=mocker.AsyncMock(return_value=[{"databaseId": 99}])
+    )
+    client = SimpleNamespace(mail=mail_client, username="alice")
+
+    result = await _verify_mail_messages(client, [_mail_result("not-a-number")], _sem())
+    assert result == {"not-a-number"}
+
+
+@pytest.mark.unit
+async def test_verify_mail_partitions_distinct_mailboxes(mocker):
+    """Results in different mailboxes each get their own list call."""
+
+    async def list_messages(mailbox_id, *, limit):
+        return {10: [{"databaseId": 1}], 20: [{"databaseId": 2}]}[mailbox_id]
+
+    mail_client = SimpleNamespace(
+        list_messages=mocker.AsyncMock(side_effect=list_messages)
+    )
+    client = SimpleNamespace(mail=mail_client, username="alice")
+
+    result = await _verify_mail_messages(
+        client,
+        [_mail_result(1, mailbox_id=10), _mail_result(2, mailbox_id=20)],
+        _sem(),
+    )
+    assert result == {"1", "2"}
+    assert mail_client.list_messages.await_count == 2
+
+
 # ---------------------------------------------------------------------------
 # News batch verifier
 # ---------------------------------------------------------------------------

tests/unit/test_mail_models.py

@@ -0,0 +1,109 @@
+"""Unit tests for Mail Pydantic models (alias mapping from the OCS API)."""
+
+import pytest
+
+from nextcloud_mcp_server.models.mail import (
+    GetMessageResponse,
+    ListAccountsResponse,
+    MailAccount,
+    MailMailbox,
+    MailMessage,
+    MailMessageSummary,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def test_account_maps_is_delegated_alias():
+    account = MailAccount(**{"id": 1, "email": "a@example.com", "isDelegated": True})
+    assert account.id == 1
+    assert account.is_delegated is True
+
+
+def test_mailbox_maps_camelcase_aliases():
+    mailbox = MailMailbox(
+        **{
+            "databaseId": 10,
+            "id": "SU5CT1g=",
+            "name": "INBOX",
+            "displayName": "Inbox",
+            "accountId": 1,
+            "specialUse": ["inbox"],
+            "unread": 5,
+        }
+    )
+    assert mailbox.database_id == 10
+    assert mailbox.account_id == 1
+    assert mailbox.display_name == "Inbox"
+    assert mailbox.special_use == ["inbox"]
+    assert mailbox.unread == 5
+
+
+def test_message_summary_maps_from_and_dateint():
+    summary = MailMessageSummary(
+        **{
+            "databaseId": 100,
+            "subject": "Hello",
+            "dateInt": 1700000000,
+            "from": [{"label": "Alice", "email": "alice@example.com"}],
+            "to": [{"email": "bob@example.com"}],
+            "mailboxId": 10,
+            "previewText": "snippet",
+            "flags": {"seen": True, "hasAttachments": True},
+        }
+    )
+    assert summary.database_id == 100
+    assert summary.date_int == 1700000000
+    assert summary.from_[0].label == "Alice"
+    assert summary.to[0].email == "bob@example.com"
+    assert summary.mailbox_id == 10
+    assert summary.preview_text == "snippet"
+    assert summary.flags is not None
+    assert summary.flags.seen is True
+    assert summary.flags.has_attachments is True
+
+
+def test_full_message_maps_body_and_attachments():
+    message = MailMessage(
+        **{
+            "id": 100,
+            "subject": "Hello",
+            "hasHtmlBody": True,
+            "body": "<p>Hi</p>",
+            "from": [{"label": "Alice", "email": "alice@example.com"}],
+            "attachments": [
+                {
+                    "id": "1.2",
+                    "fileName": "doc.pdf",
+                    "mime": "application/pdf",
+                    "size": 1024,
+                }
+            ],
+        }
+    )
+    assert message.id == 100
+    assert message.has_html_body is True
+    assert message.body == "<p>Hi</p>"
+    assert message.attachments[0].file_name == "doc.pdf"
+    assert message.attachments[0].id == "1.2"
+
+
+def test_message_tolerates_missing_optional_fields():
+    """A 206 partial response may omit the body."""
+    message = MailMessage(**{"id": 100})
+    assert message.id == 100
+    assert message.body is None
+    assert message.has_html_body is False
+    assert message.attachments == []
+
+
+def test_response_models_wrap_results():
+    resp = ListAccountsResponse(
+        results=[MailAccount(id=1, email="a@example.com")], total_count=1
+    )
+    assert resp.success is True
+    assert resp.total_count == 1
+    assert resp.results[0].email == "a@example.com"
+
+    msg_resp = GetMessageResponse(message=MailMessage(id=5, subject="Hi"))
+    assert msg_resp.message.id == 5

tests/unit/test_mail_server_helpers.py

@@ -0,0 +1,38 @@
+"""Unit tests for server/mail.py helper logic."""
+
+import pytest
+
+from nextcloud_mcp_server.server.mail import (
+    MAX_ATTACHMENT_CONTENT_BYTES,
+    _cap_attachment_content,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def test_small_content_passes_through():
+    assert _cap_attachment_content("hello") == "hello"
+
+
+def test_none_content_passes_through():
+    assert _cap_attachment_content(None) is None
+
+
+def test_oversized_content_replaced_with_byte_sentinel():
+    oversized = "a" * (MAX_ATTACHMENT_CONTENT_BYTES + 1)
+    result = _cap_attachment_content(oversized)
+    assert result != oversized
+    assert "too large to inline" in result
+    # Reports the actual UTF-8 byte count, not character count.
+    assert f"{MAX_ATTACHMENT_CONTENT_BYTES + 1} bytes" in result
+
+
+def test_multibyte_counted_in_bytes_not_chars():
+    # Each "€" is 3 UTF-8 bytes; a string just under the byte cap in characters
+    # can still exceed it in bytes.
+    char_count = (MAX_ATTACHMENT_CONTENT_BYTES // 3) + 1
+    content = "€" * char_count
+    # Under the cap by character count, over it by byte count -> capped.
+    assert len(content) <= MAX_ATTACHMENT_CONTENT_BYTES
+    assert len(content.encode("utf-8")) > MAX_ATTACHMENT_CONTENT_BYTES
+    assert "too large to inline" in _cap_attachment_content(content)

tests/unit/vector/test_mail_content.py

@@ -0,0 +1,79 @@
+"""Unit tests for the shared mail content reconstruction.
+
+``build_mail_content`` is the single source of truth for index-time and
+query-time chunk offsets; these tests pin the exact layout so a change to the
+separators or header order can't silently misalign every indexed message.
+"""
+
+import pytest
+
+from nextcloud_mcp_server.vector.mail_content import (
+    build_mail_content,
+    format_mail_addresses,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def test_format_addresses_variants():
+    assert (
+        format_mail_addresses([{"label": "Alice", "email": "alice@example.com"}])
+        == "Alice <alice@example.com>"
+    )
+    # Email only, label only, label==email, and multiple joined by ", ".
+    assert format_mail_addresses([{"email": "bob@example.com"}]) == "bob@example.com"
+    assert format_mail_addresses([{"label": "Ops"}]) == "Ops"
+    assert format_mail_addresses([{"label": "x@y.z", "email": "x@y.z"}]) == "x@y.z"
+    assert format_mail_addresses(None) == ""
+    assert (
+        format_mail_addresses([{"email": "a@x.io"}, {"label": "B", "email": "b@x.io"}])
+        == "a@x.io, B <b@x.io>"
+    )
+
+
+def test_build_mail_content_plain_text_layout():
+    message = {
+        "subject": "Hello",
+        "from": [{"label": "Alice", "email": "alice@example.com"}],
+        "to": [{"email": "bob@example.com"}],
+        "hasHtmlBody": False,
+        "body": "Hi there.",
+    }
+    assert build_mail_content(message) == (
+        "Hello\nFrom: Alice <alice@example.com>\nTo: bob@example.com\n\nHi there."
+    )
+
+
+def test_build_mail_content_includes_cc_and_bcc_when_present():
+    message = {
+        "subject": "Sync",
+        "from": [{"email": "a@x.io"}],
+        "to": [{"email": "b@x.io"}],
+        "cc": [{"email": "c@x.io"}],
+        "bcc": [{"email": "d@x.io"}],
+        "hasHtmlBody": False,
+        "body": "body",
+    }
+    assert build_mail_content(message) == (
+        "Sync\nFrom: a@x.io\nTo: b@x.io\nCc: c@x.io\nBcc: d@x.io\n\nbody"
+    )
+
+
+def test_build_mail_content_converts_html_body():
+    message = {
+        "subject": "HTML",
+        "from": [{"email": "a@x.io"}],
+        "hasHtmlBody": True,
+        "body": "<p>Hello <strong>world</strong></p>",
+    }
+    result = build_mail_content(message)
+    # Header preserved; body converted to markdown (no raw tags).
+    assert result.startswith("HTML\nFrom: a@x.io\n\n")
+    assert "<p>" not in result
+    assert "world" in result
+
+
+def test_build_mail_content_tolerates_empty_fields():
+    # No subject/addresses/body (e.g. a 206 partial) -> just the blank-line +
+    # empty body, with no spurious header lines.
+    assert build_mail_content({}) == "\n\n"

tests/unit/vector/test_scanner_consent_backstop.py

@@ -104,7 +104,9 @@ async def test_noop_when_allowed_is_none(monkeypatch):
 
 async def test_noop_when_all_text_types_allowed(monkeypatch):
     send = AsyncMock()
-    allowed = frozenset({"note", "news_item", "deck_card", "file"})
+    # Derive from INDEXED_DOC_TYPES so a newly-indexed text type doesn't make
+    # this "all allowed" set silently incomplete (and trip the backstop).
+    allowed = frozenset(scanner_module.INDEXED_DOC_TYPES)
     queued = await _enqueue_deletes_for_disabled_types(
         "alice", _producer(send), allowed, 1
     )

tests/unit/vector/test_scanner_mail.py

@@ -0,0 +1,252 @@
+"""Unit tests for the mail-message scanner (initial-sync path).
+
+The incremental path depends on live Qdrant lookups (``_scroll_all_points`` /
+``query_document_metadata``); these tests cover the initial-sync enumeration —
+accounts → mailboxes → newest-N messages — which is the bulk of the new logic
+and needs no Qdrant.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from nextcloud_mcp_server.vector import scanner as scanner_module
+from nextcloud_mcp_server.vector.scanner import DocumentTask, scan_mail_messages
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture(autouse=True)
+def _clear_scanner_module_state():
+    """Isolate the module-global grace-period / cap-log dicts per test."""
+    scanner_module._potentially_deleted.clear()
+    scanner_module._mail_cap_logged.clear()
+    yield
+    scanner_module._potentially_deleted.clear()
+    scanner_module._mail_cap_logged.clear()
+
+
+def _patch_incremental(mocker, *, indexed_ids, existing_metadata, interval=1):
+    """Patch the Qdrant-facing helpers for the incremental scan path."""
+    mocker.patch.object(scanner_module, "get_qdrant_client", new=AsyncMock())
+    mocker.patch.object(
+        scanner_module,
+        "_scroll_all_points",
+        new=AsyncMock(
+            return_value=[
+                SimpleNamespace(payload={"doc_id": doc_id}) for doc_id in indexed_ids
+            ]
+        ),
+    )
+    mocker.patch.object(
+        scanner_module,
+        "query_document_metadata",
+        new=AsyncMock(return_value=existing_metadata),
+    )
+    mocker.patch.object(scanner_module, "write_placeholder_point", new=AsyncMock())
+    mocker.patch.object(scanner_module, "record_vector_sync_scan")
+    mocker.patch.object(
+        scanner_module,
+        "get_settings",
+        return_value=MagicMock(vector_sync_scan_interval=interval),
+    )
+
+
+class _CollectingStream:
+    """Minimal TaskProducer stand-in that records sent DocumentTasks."""
+
+    def __init__(self) -> None:
+        self.tasks: list[DocumentTask] = []
+
+    async def send(self, task: DocumentTask) -> None:
+        self.tasks.append(task)
+
+
+async def test_initial_sync_enumerates_accounts_mailboxes_messages(mocker):
+    nc_client = MagicMock()
+    nc_client.mail.list_accounts = AsyncMock(return_value=[{"id": 1}])
+    nc_client.mail.get_mailboxes = AsyncMock(
+        return_value=[{"databaseId": 10}, {"databaseId": 11}]
+    )
+
+    async def list_messages(mailbox_id, *, limit):
+        if mailbox_id == 10:
+            return [
+                {"databaseId": 100, "dateInt": 1700000000},
+                {"databaseId": 101, "dateInt": 1700000001},
+            ]
+        return [{"databaseId": 200, "dateInt": 1700000002}]
+
+    nc_client.mail.list_messages = AsyncMock(side_effect=list_messages)
+
+    placeholder = mocker.patch.object(
+        scanner_module, "write_placeholder_point", new=AsyncMock()
+    )
+    mocker.patch.object(scanner_module, "record_vector_sync_scan")
+
+    stream = _CollectingStream()
+    queued = await scan_mail_messages(
+        user_id="alice",
+        send_stream=stream,
+        nc_client=nc_client,
+        initial_sync=True,
+        scan_id=1,
+    )
+
+    assert queued == 3
+    assert len(stream.tasks) == 3
+    # All are mail_message index tasks carrying account/mailbox metadata.
+    assert {t.doc_id for t in stream.tasks} == {"100", "101", "200"}
+    assert all(t.doc_type == "mail_message" for t in stream.tasks)
+    assert all(t.operation == "index" for t in stream.tasks)
+    t100 = next(t for t in stream.tasks if t.doc_id == "100")
+    assert t100.modified_at == 1700000000
+    assert t100.metadata == {"account_id": 1, "mailbox_id": 10}
+    # A placeholder is written per message before queueing.
+    assert placeholder.await_count == 3
+    # The per-mailbox cap is passed through.
+    nc_client.mail.list_messages.assert_any_await(
+        10, limit=scanner_module.MAIL_SCAN_MAX_PER_MAILBOX
+    )
+
+
+async def test_initial_sync_skips_mailbox_on_list_error(mocker):
+    """A failing mailbox is logged and skipped; other mailboxes still index."""
+    nc_client = MagicMock()
+    nc_client.mail.list_accounts = AsyncMock(return_value=[{"id": 1}])
+    nc_client.mail.get_mailboxes = AsyncMock(
+        return_value=[{"databaseId": 10}, {"databaseId": 11}]
+    )
+
+    async def list_messages(mailbox_id, *, limit):
+        if mailbox_id == 10:
+            raise RuntimeError("imap hiccup")
+        return [{"databaseId": 200, "dateInt": 1700000002}]
+
+    nc_client.mail.list_messages = AsyncMock(side_effect=list_messages)
+    mocker.patch.object(scanner_module, "write_placeholder_point", new=AsyncMock())
+    mocker.patch.object(scanner_module, "record_vector_sync_scan")
+
+    stream = _CollectingStream()
+    queued = await scan_mail_messages(
+        user_id="alice",
+        send_stream=stream,
+        nc_client=nc_client,
+        initial_sync=True,
+        scan_id=1,
+    )
+
+    assert queued == 1
+    assert {t.doc_id for t in stream.tasks} == {"200"}
+
+
+async def test_no_accounts_queues_nothing(mocker):
+    nc_client = MagicMock()
+    nc_client.mail.list_accounts = AsyncMock(return_value=[])
+    mocker.patch.object(scanner_module, "write_placeholder_point", new=AsyncMock())
+    mocker.patch.object(scanner_module, "record_vector_sync_scan")
+
+    stream = _CollectingStream()
+    queued = await scan_mail_messages(
+        user_id="alice",
+        send_stream=stream,
+        nc_client=nc_client,
+        initial_sync=True,
+        scan_id=1,
+    )
+
+    assert queued == 0
+    assert stream.tasks == []
+
+
+def _single_message_client(messages):
+    nc_client = MagicMock()
+    nc_client.mail.list_accounts = AsyncMock(return_value=[{"id": 1}])
+    nc_client.mail.get_mailboxes = AsyncMock(return_value=[{"databaseId": 10}])
+    nc_client.mail.list_messages = AsyncMock(return_value=messages)
+    return nc_client
+
+
+async def test_incremental_new_message_queued(mocker):
+    """A message absent from Qdrant (no existing metadata) is queued to index."""
+    _patch_incremental(mocker, indexed_ids=[], existing_metadata=None)
+    nc_client = _single_message_client([{"databaseId": 100, "dateInt": 1700000000}])
+
+    stream = _CollectingStream()
+    queued = await scan_mail_messages(
+        user_id="alice",
+        send_stream=stream,
+        nc_client=nc_client,
+        initial_sync=False,
+        scan_id=1,
+    )
+
+    assert queued == 1
+    assert [(t.doc_id, t.operation) for t in stream.tasks] == [("100", "index")]
+
+
+async def test_incremental_reappeared_message_clears_grace(mocker):
+    """A message back in Nextcloud is removed from the deletion grace period."""
+    # Already indexed and up-to-date, so it won't be re-queued.
+    _patch_incremental(
+        mocker, indexed_ids=["100"], existing_metadata={"modified_at": 1700000000}
+    )
+    scanner_module._potentially_deleted[("alice", "100")] = 123.0
+    nc_client = _single_message_client([{"databaseId": 100, "dateInt": 1700000000}])
+
+    stream = _CollectingStream()
+    queued = await scan_mail_messages(
+        user_id="alice",
+        send_stream=stream,
+        nc_client=nc_client,
+        initial_sync=False,
+        scan_id=1,
+    )
+
+    assert queued == 0
+    assert stream.tasks == []
+    assert ("alice", "100") not in scanner_module._potentially_deleted
+
+
+async def test_incremental_deletes_after_grace_period(mocker):
+    """An indexed message gone from Nextcloud past the grace period is deleted."""
+    _patch_incremental(mocker, indexed_ids=["999"], existing_metadata=None)
+    # Seed the grace period far in the past so the delta exceeds grace_period.
+    scanner_module._potentially_deleted[("alice", "999")] = 0.0
+    # Mailbox now returns no messages, so 999 is missing.
+    nc_client = _single_message_client([])
+
+    stream = _CollectingStream()
+    queued = await scan_mail_messages(
+        user_id="alice",
+        send_stream=stream,
+        nc_client=nc_client,
+        initial_sync=False,
+        scan_id=1,
+    )
+
+    assert queued == 1
+    assert [(t.doc_id, t.operation) for t in stream.tasks] == [("999", "delete")]
+    assert ("alice", "999") not in scanner_module._potentially_deleted
+
+
+async def test_incremental_first_missing_starts_grace(mocker):
+    """A newly-missing indexed message enters the grace period (no delete yet)."""
+    _patch_incremental(mocker, indexed_ids=["999"], existing_metadata=None)
+    # Not previously seen as missing, and the mailbox now returns no messages.
+    nc_client = _single_message_client([])
+
+    stream = _CollectingStream()
+    queued = await scan_mail_messages(
+        user_id="alice",
+        send_stream=stream,
+        nc_client=nc_client,
+        initial_sync=False,
+        scan_id=1,
+    )
+
+    # First miss only starts the grace period — nothing queued, nothing deleted.
+    assert queued == 0
+    assert stream.tasks == []
+    assert ("alice", "999") in scanner_module._potentially_deleted