From f8405b88f020481d7bd608ecd2db70f849bc37e5 Mon Sep 17 00:00:00 2001
From: Konst <constkolesnyak@gmail.com>
Date: Sun, 19 Jul 2026 00:37:44 +0200
Subject: [PATCH] sources: add a generic IMAP source
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds an IMAP mailbox source for providers the Gmail source cannot reach
(the existing one goes through the gog CLI / Gmail API, so a plain IMAP
mailbox — GMX, Fastmail, a company server — has no path in today). One
source per mailbox, each with an independent cursor
(<UIDVALIDITY>:<max_uid>), registered from sync.imap.

Some mail carries the part worth acting on only as an inline image — a
scan, a photo, a rendered document — so the body text is useless to the
inbox consumer. sync.imap.match singles those messages out by sender or by
an image's Content-ID/filename, and sync.imap.vision runs a multimodal pass
over the image at ingest time so what lands in the inbox is plain text.
match.only_matched drops everything else at the source, turning the mailbox
into a single-purpose notifier rather than a second inbox.

Both blocks are inert by default: with no match rules configured nothing is
singled out, no model is ever called, and this is a plain mailbox reader.

The vision prompt and the answer key are deliberately configured as a pair:
the prompt tells the model which label to emit and the parser reads the line
after exactly that label, so changing one without the other would silently
yield unknown_answer every time — a failure that looks like the model
degrading rather than a config mistake. Two tests pin the coupling, one
moving both (works) and one moving only the prompt (degrades as expected).

Disabled by default; passwords are read from sync.imap.passwords keyed by
username so no credential belongs in the tracked config. An account with no
password is skipped with a warning instead of failing the whole sync.

25 tests pass.
---
 config.example.yaml       |  11 +
 docs/config.md            |  17 ++
 docs/sources.md           |  81 +++++++
 nerve/config.py           | 135 +++++++++++
 nerve/sources/imap.py     | 497 ++++++++++++++++++++++++++++++++++++++
 nerve/sources/registry.py |  59 +++++
 tests/test_imap_source.py | 437 +++++++++++++++++++++++++++++++++
 7 files changed, 1237 insertions(+)
 create mode 100644 nerve/sources/imap.py
 create mode 100644 tests/test_imap_source.py

diff --git a/config.example.yaml b/config.example.yaml
index 1ab39970..c32be920 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -100,6 +100,17 @@ sync:
     enabled: true
     accounts: []                 # Set in config.local.yaml
     schedule: "*/15 * * * *"
+  # Plain IMAP mailboxes — for providers the Gmail source can't reach.
+  # One source per account. Passwords go in config.local.yaml under
+  # sync.imap.passwords, keyed by username. See docs/sources.md for the
+  # optional pass that reads mail whose payload is only legible as an image.
+  # imap:
+  #   enabled: false
+  #   accounts:
+  #     - host: imap.example.net
+  #       username: me@example.net
+  #       label: mailbox           # source name becomes imap:mailbox
+  #   schedule: "*/30 * * * *"
   github:
     enabled: true
     schedule: "*/15 * * * *"
diff --git a/docs/config.md b/docs/config.md
index 74deb176..52c829f9 100644
--- a/docs/config.md
+++ b/docs/config.md
@@ -218,6 +218,23 @@ Sources pull data from external services on a schedule. See [sources.md](sources
 | `sync.gmail.schedule` | cron | `*/15 * * * *` | Fetch frequency |
 | `sync.gmail.keyring_password` | string | - | gog keyring password |
 
+**IMAP-specific** (see [sources.md](sources.md) for the image pass):
+
+| Key | Type | Default | Description |
+|-----|------|---------|-------------|
+| `sync.imap.enabled` | bool | `false` | Enable IMAP mailboxes |
+| `sync.imap.accounts` | list | `[]` | Mailboxes: `host`, `username`, `label`, `port`, `mailbox` |
+| `sync.imap.passwords` | dict | `{}` | Password per username — keep in `config.local.yaml` |
+| `sync.imap.schedule` | cron | `*/30 * * * *` | Fetch frequency |
+| `sync.imap.initial_lookback_days` | int | `1` | `SINCE` window on first run |
+| `sync.imap.match.sender_contains` | list | `[]` | Substrings matched against `From:` |
+| `sync.imap.match.attachment_contains` | list | `[]` | Substrings matched against an image's Content-ID / filename |
+| `sync.imap.match.only_matched` | bool | `false` | Drop everything that did not match |
+| `sync.imap.vision.enabled` | bool | `false` | Run a multimodal pass over a matched message's image |
+| `sync.imap.vision.model` | string | *(empty)* | Falls back to `memory.fast_model` |
+| `sync.imap.vision.prompt` | string | *(empty)* | What to ask about the image — required when vision is enabled |
+| `sync.imap.vision.answer_key` | string | *(empty)* | Label the prompt asks the model to emit; empty = first non-empty line |
+
 **GitHub-specific:**
 
 | Key | Type | Default | Description |
diff --git a/docs/sources.md b/docs/sources.md
index f24a0749..a9411d71 100644
--- a/docs/sources.md
+++ b/docs/sources.md
@@ -72,6 +72,70 @@ A persistent cron job (`inbox-processor`) runs every 15 minutes:
 - **Two-step fetch:** Search returns metadata only; body + `internalDate` are fetched per-message via `gog gmail get` (up to 5 concurrent)
 - **Default schedule:** `*/15 * * * *` (every 15 min)
 
+### IMAP
+- **Adapter:** `nerve/sources/imap.py` — plain `imaplib`, no CLI or provider API
+- **Scope:** Any IMAP mailbox the Gmail source cannot reach (that one goes through the `gog` CLI / Gmail API), e.g. GMX, Fastmail, a company server
+- **Instances:** One source per configured account (`imap:<label>`), each with an independent cursor
+- **Cursor:** `<UIDVALIDITY>:<max_uid>` — UIDs are only stable within a UIDVALIDITY, so a server-side reset is detected and re-baselined instead of trusted
+- **First run:** `SINCE` window of `initial_lookback_days` (default 1)
+- **Subsequent runs:** `UID SEARCH <last+1>:*`, filtered client-side (the `N:*` range always returns at least the highest UID)
+- **Blocking I/O:** the whole IMAP conversation runs in a worker thread via `asyncio.to_thread`
+- **Credentials:** passwords live in `sync.imap.passwords` keyed by username (put them in `config.local.yaml`); an account with no password is skipped with a warning instead of failing the sync
+- **Optional image pass:** see below
+- **Default schedule:** `*/30 * * * *` (every 30 min)
+- **Disabled by default**
+
+#### Reading mail that is only legible as an image
+
+Some senders put the part you actually care about in an inline image — a
+scan, a photo, a rendered document — so the body text is useless to an
+agent. `sync.imap.match` singles those messages out and `sync.imap.vision`
+runs a multimodal model over the image at ingest time, so the inbox
+consumer gets plain text.
+
+Both are inert by default: with no match rules, this is a plain mailbox
+source and no model is ever called.
+
+`vision.prompt` and `vision.answer_key` are a **matched pair** — the prompt
+tells the model which label to emit, and the parser reads the line after
+exactly that label. Change one without the other and every message silently
+comes back as `unknown_answer`, which looks like the model degrading rather
+than a config mistake. Leave `answer_key` empty to just take the first
+non-empty line.
+
+```yaml
+sync:
+  imap:
+    enabled: true
+    accounts:
+      - host: imap.example.net
+        username: me@example.net
+        label: scans              # source name becomes imap:scans
+    passwords:                    # config.local.yaml
+      me@example.net: "..."
+    match:
+      sender_contains: ["scans@example.net"]   # substrings of From:
+      attachment_contains: ["scan"]            # substrings of Content-ID / filename
+      only_matched: true                       # drop everything else at the source
+    vision:
+      enabled: true
+      model: ""                   # empty = memory.fast_model
+      prompt: |
+        Read the scan. Answer with EXACTLY one line:
+        Sender: <company or person, or "unreadable">
+      answer_key: "Sender:"
+      unknown_answer: "unreadable"
+      summary: "[{label}] scan from {answer}"
+      content: "{vision}\n\nSubject: {subject}\nDate: {date}"
+      summary_unknown: "[{label}] unreadable scan"
+      content_unknown: "The image could not be read.\n\nSubject: {subject}"
+```
+
+Wording templates accept `{label} {answer} {vision} {subject} {sender}
+{date} {body}`, where `{answer}` is the parsed line and `{vision}` the full
+model reply. An unknown placeholder logs a warning and leaves the template
+visible rather than failing the fetch.
+
 ### GitHub
 - **Adapter:** `nerve/sources/github.py` — uses `gh api notifications` CLI
 - **Cursor:** ISO 8601 timestamp of the newest notification's `updated_at`
@@ -131,6 +195,23 @@ sync:
     batch_size: 20
     condense: true                # Strip boilerplate + Haiku extraction for long emails
 
+  imap:
+    enabled: false                # Off by default
+    accounts:                     # One source per mailbox, each with own cursor
+      - host: imap.example.net
+        username: me@example.net
+        label: mailbox            # Source name becomes imap:mailbox
+        port: 993
+        mailbox: INBOX
+    passwords:                    # Keyed by username (in config.local.yaml)
+      me@example.net: "..."
+    schedule: "*/30 * * * *"
+    batch_size: 20
+    initial_lookback_days: 1
+    condense: false
+    match: {}                     # See "Reading mail that is only legible as an image"
+    vision: {}
+
   github:
     enabled: true
     schedule: "*/15 * * * *"
diff --git a/nerve/config.py b/nerve/config.py
index 92b775ed..fdccc887 100644
--- a/nerve/config.py
+++ b/nerve/config.py
@@ -368,6 +368,139 @@ def from_dict(cls, d: dict) -> GmailSyncConfig:
         )
 
 
+@dataclass
+class ImapMatchConfig:
+    """Which IMAP messages are singled out for the optional image pass.
+
+    Both lists are case-insensitive substrings and both default to empty, so
+    an unconfigured source never singles anything out and stays a plain
+    mailbox reader.
+    """
+
+    # Matched against the decoded From: header.
+    sender_contains: list[str] = field(default_factory=list)
+    # Matched against an inline image's Content-ID / filename. A hit also
+    # picks that image over the merely largest one.
+    attachment_contains: list[str] = field(default_factory=list)
+    # Drop everything that did not match, turning the mailbox into a
+    # single-purpose notifier instead of a second inbox.
+    only_matched: bool = False
+
+    @classmethod
+    def from_dict(cls, d: dict) -> ImapMatchConfig:
+        return cls(
+            sender_contains=[str(s) for s in d.get("sender_contains", [])],
+            attachment_contains=[str(s) for s in d.get("attachment_contains", [])],
+            only_matched=bool(d.get("only_matched", False)),
+        )
+
+
+@dataclass
+class ImapVisionConfig:
+    """Multimodal pass over an inline image in matched messages.
+
+    For mail whose payload is only legible in an image — a scan, a photo, a
+    rendered document — the model reads it at ingest time so downstream
+    consumers get plain text.
+
+    ``prompt`` and ``answer_key`` are a matched pair: the prompt tells the
+    model which label to emit, and the parser reads the line after exactly
+    that label. Changing one without the other silently yields
+    ``unknown_answer`` every time, so they live together here. Leave
+    ``answer_key`` empty to take the first non-empty line of the answer.
+
+    Wording templates accept ``{label} {answer} {vision} {subject} {sender}
+    {date} {body}``; ``{answer}`` is the parsed line and ``{vision}`` the full
+    model reply.
+    """
+
+    enabled: bool = False
+    # Defaults to memory.fast_model when left empty.
+    model: str = ""
+    prompt: str = ""
+    answer_key: str = ""
+    unknown_answer: str = "unreadable"
+    summary: str = "[{label}] {answer}"
+    summary_unknown: str = "[{label}] {subject}"
+    content: str = (
+        "{vision}\n\n"
+        "Subject: {subject}\n"
+        "From: {sender}\n"
+        "Date: {date}"
+    )
+    content_unknown: str = (
+        "The image could not be read (missing or unreadable).\n\n"
+        "Subject: {subject}\n"
+        "From: {sender}\n"
+        "Date: {date}"
+    )
+
+    @classmethod
+    def from_dict(cls, d: dict) -> ImapVisionConfig:
+        base = cls()
+        return cls(
+            enabled=bool(d.get("enabled", base.enabled)),
+            model=str(d.get("model", base.model)),
+            prompt=str(d.get("prompt", base.prompt)),
+            answer_key=str(d.get("answer_key", base.answer_key)),
+            unknown_answer=str(d.get("unknown_answer", base.unknown_answer)),
+            summary=str(d.get("summary", base.summary)),
+            summary_unknown=str(d.get("summary_unknown", base.summary_unknown)),
+            content=str(d.get("content", base.content)),
+            content_unknown=str(d.get("content_unknown", base.content_unknown)),
+        )
+
+
+@dataclass
+class ImapAccountConfig:
+    """One IMAP mailbox. The password lives in config.local.yaml under
+    ``sync.imap.passwords[<username>]``, never here."""
+    host: str
+    username: str
+    label: str
+    port: int = 993
+    mailbox: str = "INBOX"
+
+    @classmethod
+    def from_dict(cls, d: dict) -> ImapAccountConfig:
+        return cls(
+            host=str(d["host"]),
+            username=str(d["username"]),
+            label=str(d.get("label") or d["username"].split("@")[0]),
+            port=int(d.get("port", 993)),
+            mailbox=str(d.get("mailbox", "INBOX")),
+        )
+
+
+@dataclass
+class ImapSyncConfig:
+    enabled: bool = False
+    accounts: list[ImapAccountConfig] = field(default_factory=list)
+    passwords: dict[str, str] = field(default_factory=dict)
+    schedule: str = "*/30 * * * *"
+    batch_size: int = 20
+    initial_lookback_days: int = 1
+    condense: bool = False
+    condense_prompt: str = ""
+    match: ImapMatchConfig = field(default_factory=ImapMatchConfig)
+    vision: ImapVisionConfig = field(default_factory=ImapVisionConfig)
+
+    @classmethod
+    def from_dict(cls, d: dict) -> ImapSyncConfig:
+        return cls(
+            enabled=bool(d.get("enabled", False)),
+            accounts=[ImapAccountConfig.from_dict(a) for a in d.get("accounts", [])],
+            passwords=dict(d.get("passwords", {})),
+            schedule=str(d.get("schedule", "*/30 * * * *")),
+            batch_size=int(d.get("batch_size", 20)),
+            initial_lookback_days=int(d.get("initial_lookback_days", 1)),
+            condense=bool(d.get("condense", False)),
+            condense_prompt=str(d.get("condense_prompt", "")),
+            match=ImapMatchConfig.from_dict(d.get("match", {})),
+            vision=ImapVisionConfig.from_dict(d.get("vision", {})),
+        )
+
+
 @dataclass
 class GitHubSyncConfig:
     enabled: bool = True
@@ -562,6 +695,7 @@ def from_dict(cls, d: dict) -> CodexSyncConfig:
 class SyncConfig:
     telegram: TelegramSyncConfig = field(default_factory=TelegramSyncConfig)
     gmail: GmailSyncConfig = field(default_factory=GmailSyncConfig)
+    imap: ImapSyncConfig = field(default_factory=ImapSyncConfig)
     github: GitHubSyncConfig = field(default_factory=GitHubSyncConfig)
     github_events: GitHubEventsSyncConfig = field(default_factory=GitHubEventsSyncConfig)
     github_repos: GitHubReposSyncConfig = field(default_factory=GitHubReposSyncConfig)
@@ -574,6 +708,7 @@ def from_dict(cls, d: dict) -> SyncConfig:
         return cls(
             telegram=TelegramSyncConfig.from_dict(d.get("telegram", {})),
             gmail=GmailSyncConfig.from_dict(d.get("gmail", {})),
+            imap=ImapSyncConfig.from_dict(d.get("imap", {})),
             github=GitHubSyncConfig.from_dict(d.get("github", {})),
             github_events=GitHubEventsSyncConfig.from_dict(d.get("github_events", {})),
             github_repos=GitHubReposSyncConfig.from_dict(d.get("github_repos", {})),
diff --git a/nerve/sources/imap.py b/nerve/sources/imap.py
new file mode 100644
index 00000000..0aa23b25
--- /dev/null
+++ b/nerve/sources/imap.py
@@ -0,0 +1,497 @@
+"""Generic IMAP source — fetches emails from any IMAP server.
+
+Each ImapSource instance handles ONE mailbox with its own cursor. The
+registry creates one instance per configured account, so each gets
+independent cursor tracking in the DB.
+
+Cursor semantics: ``<UIDVALIDITY>:<max_uid>``. IMAP UIDs are only stable
+within a given UIDVALIDITY; if the server resets it, we detect the mismatch
+and re-baseline from a SINCE lookback window instead of trusting old UIDs.
+
+imaplib is blocking, so the whole IMAP conversation runs in a worker thread
+via ``asyncio.to_thread``.
+
+Optional image pass
+-------------------
+
+Some mail carries its payload only as an inline image — a scan, a photo, a
+rendered document — so the text an agent would act on is not in the body at
+all. For those messages the source can run a multimodal model over the image
+at ingest time and fold the answer into the record, so downstream consumers
+get plain text.
+
+Which messages that applies to, what to ask the model, and how to word the
+result are entirely configuration (``sync.imap.match`` / ``sync.imap.vision``).
+With no match rules configured nothing is singled out and this is a plain
+IMAP mailbox source. See ``docs/sources.md`` for a worked example.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import email
+import imaplib
+import logging
+from datetime import datetime, timedelta, timezone
+from email.header import decode_header, make_header
+from email.message import Message
+from typing import Any, Callable, Sequence
+
+from nerve.sources.base import Source
+from nerve.sources.gmail import _html_to_text, _parse_to_epoch
+from nerve.sources.models import FetchResult, SourceRecord
+
+logger = logging.getLogger(__name__)
+
+
+class ImapSource(Source):
+    """Generic IMAP mailbox source for a single account."""
+
+    def __init__(
+        self,
+        *,
+        host: str,
+        username: str,
+        password: str,
+        label: str,
+        port: int = 993,
+        mailbox: str = "INBOX",
+        initial_lookback_days: int = 1,
+        match: Any | None = None,
+        vision: Any | None = None,
+        vision_client_factory: Callable[[], Any] | None = None,
+    ):
+        self.host = host
+        self.port = port
+        self.username = username
+        self._password = password
+        self.mailbox = mailbox
+        self.label = label
+        self.source_name = f"imap:{label}"
+        self.initial_lookback_days = max(1, int(initial_lookback_days))
+        self._vision_client_factory = vision_client_factory
+
+        # Match rules decide which messages are "of interest"; the vision
+        # block decides what to do with them. Both default to inert, so an
+        # unconfigured source is a plain mailbox reader.
+        if match is None or vision is None:
+            from nerve.config import ImapMatchConfig, ImapVisionConfig
+
+            match = ImapMatchConfig() if match is None else match
+            vision = ImapVisionConfig() if vision is None else vision
+        self.match = match
+        self.vision = vision
+
+    async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
+        """Fetch new messages since cursor, running the image pass on matches."""
+        try:
+            parsed, next_cursor = await asyncio.to_thread(
+                self._imap_fetch_blocking, cursor, limit,
+            )
+        except Exception as e:
+            logger.error("IMAP error for %s: %s", self.source_name, e)
+            return FetchResult(records=[], next_cursor=cursor, has_more=False)
+
+        records: list[SourceRecord] = []
+        for msg in parsed:
+            # only_matched: drop everything the rules did not single out
+            # before it ever reaches the inbox.
+            if self.match.only_matched and not msg.get("matched"):
+                logger.debug(
+                    "IMAP %s: dropping unmatched message %s (only_matched)",
+                    self.source_name, msg.get("id"),
+                )
+                continue
+            if msg.get("matched") and self.vision.enabled:
+                summary, content = await self._build_vision_record(msg)
+            elif msg.get("matched"):
+                summary, content = self._vision_fallback(msg)
+            else:
+                summary, content = self._plain_record(msg)
+
+            records.append(SourceRecord(
+                id=msg["id"],
+                source=self.source_name,
+                record_type="imap_message",
+                summary=summary,
+                content=content,
+                timestamp=msg["timestamp"],
+                metadata={
+                    "account": self.username,
+                    "label": self.label,
+                    "mailbox": self.mailbox,
+                    "matched": bool(msg.get("matched")),
+                    "from": msg.get("from", ""),
+                },
+            ))
+
+        return FetchResult(records=records, next_cursor=next_cursor, has_more=False)
+
+    # ------------------------------------------------------------------
+    # Blocking IMAP conversation (runs in a worker thread)
+    # ------------------------------------------------------------------
+
+    def _imap_fetch_blocking(
+        self, cursor: str | None, limit: int,
+    ) -> tuple[list[dict], str | None]:
+        """Connect, select the mailbox, and parse new messages.
+
+        Returns (parsed_messages, next_cursor). Runs entirely in a worker
+        thread — no async here.
+        """
+        M = imaplib.IMAP4_SSL(self.host, self.port)
+        try:
+            M.login(self.username, self._password)
+            M.select(self.mailbox, readonly=True)
+
+            uidvalidity, uidnext = self._mailbox_status(M)
+
+            prev_validity, last_uid = _parse_cursor(cursor)
+            incremental = (
+                cursor is not None
+                and prev_validity == uidvalidity
+                and last_uid is not None
+            )
+
+            if incremental:
+                typ, data = M.uid("search", None, f"{last_uid + 1}:*")
+            else:
+                since = (
+                    datetime.now(timezone.utc)
+                    - timedelta(days=self.initial_lookback_days)
+                ).strftime("%d-%b-%Y")
+                typ, data = M.uid("search", None, "SINCE", since)
+
+            if typ != "OK" or not data or not data[0]:
+                # Nothing matched. Baseline the cursor so we don't re-scan.
+                baseline = last_uid if incremental else (uidnext - 1 if uidnext else 0)
+                return [], f"{uidvalidity}:{max(0, baseline)}"
+
+            uids = [int(u) for u in data[0].split()]
+            # The `N:*` range trick always returns at least the highest UID even
+            # when nothing is newer — filter defensively against the cursor.
+            if incremental and last_uid is not None:
+                uids = [u for u in uids if u > last_uid]
+            uids.sort()
+            if not uids:
+                return [], f"{uidvalidity}:{last_uid}" if last_uid is not None else \
+                    f"{uidvalidity}:{max(0, (uidnext - 1) if uidnext else 0)}"
+
+            # Cap to the newest `limit` messages, but track the true max UID so
+            # the cursor advances past everything we saw.
+            max_uid = uids[-1]
+            if len(uids) > limit:
+                uids = uids[-limit:]
+
+            parsed: list[dict] = []
+            for uid in uids:
+                try:
+                    parsed.append(self._fetch_one(M, uid, uidvalidity))
+                except Exception as e:
+                    logger.warning(
+                        "IMAP %s: failed to parse uid %d: %s",
+                        self.source_name, uid, e,
+                    )
+
+            return parsed, f"{uidvalidity}:{max_uid}"
+        finally:
+            try:
+                M.logout()
+            except Exception:
+                pass
+
+    def _mailbox_status(self, M: imaplib.IMAP4_SSL) -> tuple[int, int]:
+        """Return (UIDVALIDITY, UIDNEXT) for the selected mailbox."""
+        uidvalidity = 0
+        uidnext = 0
+        try:
+            typ, data = M.status(self.mailbox, "(UIDVALIDITY UIDNEXT)")
+            if typ == "OK" and data and data[0]:
+                text = data[0].decode() if isinstance(data[0], bytes) else str(data[0])
+                uidvalidity = _extract_status_int(text, "UIDVALIDITY")
+                uidnext = _extract_status_int(text, "UIDNEXT")
+        except Exception as e:
+            logger.warning("IMAP %s: STATUS failed: %s", self.source_name, e)
+        return uidvalidity, uidnext
+
+    def _fetch_one(
+        self, M: imaplib.IMAP4_SSL, uid: int, uidvalidity: int,
+    ) -> dict:
+        """Fetch and parse a single message by UID into a raw dict."""
+        typ, data = M.uid("fetch", str(uid), "(RFC822)")
+        if typ != "OK" or not data or not isinstance(data[0], tuple):
+            raise ValueError(f"empty fetch for uid {uid}")
+
+        raw_bytes = data[0][1]
+        msg: Message = email.message_from_bytes(raw_bytes)
+
+        subject = _decode_hdr(msg.get("Subject", "(no subject)"))
+        sender = _decode_hdr(msg.get("From", "?"))
+        date_str = msg.get("Date", "")
+        epoch = _parse_to_epoch(date_str)
+        timestamp = (
+            datetime.fromtimestamp(epoch, tz=timezone.utc).isoformat()
+            if epoch
+            else datetime.now(timezone.utc).isoformat()
+        )
+
+        body, image, media_type, attachment_hit = _extract_parts(
+            msg, self.match.attachment_contains,
+        )
+
+        sender_l = sender.lower()
+        matched = attachment_hit or any(
+            h.lower() in sender_l for h in self.match.sender_contains if h
+        )
+
+        return {
+            "id": f"{uidvalidity}-{uid}",
+            "subject": subject,
+            "from": sender,
+            "date": date_str,
+            "timestamp": timestamp,
+            "body": body,
+            "matched": bool(matched),
+            "image": image,
+            "media_type": media_type,
+        }
+
+    # ------------------------------------------------------------------
+    # Record builders
+    # ------------------------------------------------------------------
+
+    async def _build_vision_record(self, msg: dict) -> tuple[str, str]:
+        """Run the image pass on a matched message; fall back if it can't."""
+        image = msg.get("image")
+        media_type = msg.get("media_type") or "image/png"
+        if not image or not self._vision_client_factory or not self.vision.model:
+            return self._vision_fallback(msg)
+
+        try:
+            vision_text = await self._analyze_image(image, media_type)
+        except Exception as e:
+            logger.warning(
+                "IMAP %s: image analysis failed: %s", self.source_name, e,
+            )
+            return self._vision_fallback(msg)
+
+        if not vision_text:
+            return self._vision_fallback(msg)
+
+        v = self.vision
+        # answer_key is the label the prompt asked the model to emit; with no
+        # key configured, take the first non-empty line of the answer.
+        answer = (
+            _first_line_after(vision_text, v.answer_key)
+            if v.answer_key
+            else _first_nonempty_line(vision_text)
+        ) or v.unknown_answer
+        fields = self._template_fields(msg, vision=vision_text, answer=answer)
+        return _format(v.summary, fields), _format(v.content, fields)
+
+    def _vision_fallback(self, msg: dict) -> tuple[str, str]:
+        v = self.vision
+        fields = self._template_fields(
+            msg, vision="", answer=v.unknown_answer,
+        )
+        return _format(v.summary_unknown, fields), _format(v.content_unknown, fields)
+
+    def _template_fields(self, msg: dict, *, vision: str, answer: str) -> dict:
+        """Placeholders available to every configured wording template."""
+        return {
+            "label": self.label,
+            "answer": answer,
+            "vision": vision,
+            "subject": msg.get("subject", ""),
+            "sender": msg.get("from", ""),
+            "date": msg.get("date", ""),
+            "body": msg.get("body", ""),
+        }
+
+    def _plain_record(self, msg: dict) -> tuple[str, str]:
+        subject = msg.get("subject", "(no subject)")
+        sender = msg.get("from", "?")
+        summary = f"[{self.label}] {subject} — from {sender}"
+        content = (
+            f"Subject: {subject}\n"
+            f"From: {sender}\n"
+            f"Date: {msg.get('date', '')}\n\n"
+            f"{msg.get('body', '')}"
+        )
+        return summary, content
+
+    async def _analyze_image(self, image: bytes, media_type: str) -> str:
+        """Send the image to the multimodal model and return its text."""
+        import base64
+
+        client = self._vision_client_factory()
+        b64 = base64.standard_b64encode(image).decode("ascii")
+        try:
+            resp = await client.messages.create(
+                model=self.vision.model,
+                max_tokens=300,
+                messages=[{
+                    "role": "user",
+                    "content": [
+                        {
+                            "type": "image",
+                            "source": {
+                                "type": "base64",
+                                "media_type": media_type,
+                                "data": b64,
+                            },
+                        },
+                        {"type": "text", "text": self.vision.prompt},
+                    ],
+                }],
+            )
+        finally:
+            close = getattr(client, "close", None)
+            if close:
+                try:
+                    await close()
+                except Exception:
+                    pass
+
+        parts = []
+        for block in getattr(resp, "content", []) or []:
+            text = getattr(block, "text", None)
+            if text:
+                parts.append(text)
+        return "\n".join(parts).strip()
+
+
+# ---------------------------------------------------------------------------
+# Module-level helpers
+# ---------------------------------------------------------------------------
+
+def _parse_cursor(cursor: str | None) -> tuple[int, int | None]:
+    """Parse a ``<UIDVALIDITY>:<max_uid>`` cursor. Returns (validity, uid)."""
+    if not cursor:
+        return 0, None
+    try:
+        validity_s, uid_s = cursor.split(":", 1)
+        return int(validity_s), int(uid_s)
+    except (ValueError, AttributeError):
+        return 0, None
+
+
+def _extract_status_int(text: str, key: str) -> int:
+    """Pull an integer value for *key* out of an IMAP STATUS response line."""
+    import re
+
+    m = re.search(rf"{key}\s+(\d+)", text)
+    return int(m.group(1)) if m else 0
+
+
+def _decode_hdr(raw: str | None) -> str:
+    """Decode an RFC 2047 encoded email header into a plain string."""
+    if not raw:
+        return ""
+    try:
+        return str(make_header(decode_header(raw)))
+    except Exception:
+        return str(raw)
+
+
+def _part_text(part: Message) -> str:
+    """Decode a text MIME part to a string using its declared charset."""
+    payload = part.get_payload(decode=True)
+    if payload is None:
+        return ""
+    charset = part.get_content_charset() or "utf-8"
+    try:
+        return payload.decode(charset, errors="replace")
+    except (LookupError, TypeError):
+        return payload.decode("utf-8", errors="replace")
+
+
+def _extract_parts(
+    msg: Message, attachment_hints: Sequence[str] = (),
+) -> tuple[str, bytes | None, str | None, bool]:
+    """Walk a message and extract body text + the most relevant image.
+
+    Returns ``(body_text, image_bytes, media_type, attachment_hit)``.
+    ``attachment_hit`` is True when an inline image's Content-ID or filename
+    contains one of *attachment_hints* — that image then wins over the merely
+    largest one, which is the fallback candidate.
+    """
+    text_plain: str | None = None
+    text_html: str | None = None
+    best_img: tuple[bytes, str] | None = None      # (bytes, media_type)
+    hinted_img: tuple[bytes, str] | None = None
+    hints = [h.lower() for h in attachment_hints if h]
+    attachment_hit = False
+
+    if msg.is_multipart():
+        for part in msg.walk():
+            if part.is_multipart():
+                continue
+            ctype = (part.get_content_type() or "").lower()
+            disp = str(part.get("Content-Disposition") or "").lower()
+
+            if ctype == "text/plain" and "attachment" not in disp and text_plain is None:
+                text_plain = _part_text(part)
+            elif ctype == "text/html" and text_html is None:
+                text_html = _part_text(part)
+            elif ctype.startswith("image/"):
+                payload = part.get_payload(decode=True)
+                if not payload:
+                    continue
+                cid = str(part.get("Content-ID") or "").lower()
+                fname = str(part.get_filename() or "").lower()
+                marker = f"{cid} {fname}"
+                # Track the largest image as a fallback candidate.
+                if best_img is None or len(payload) > len(best_img[0]):
+                    best_img = (payload, ctype)
+                if any(h in marker for h in hints):
+                    attachment_hit = True
+                    hinted_img = (payload, ctype)
+    else:
+        ctype = (msg.get_content_type() or "").lower()
+        if ctype == "text/html":
+            text_html = _part_text(msg)
+        else:
+            text_plain = _part_text(msg)
+
+    if text_plain:
+        body = text_plain
+    elif text_html:
+        body = _html_to_text(text_html)
+    else:
+        body = ""
+
+    image = hinted_img or best_img
+    if image:
+        return body, image[0], image[1], attachment_hit
+    return body, None, None, attachment_hit
+
+
+def _first_line_after(text: str, marker: str) -> str:
+    """Return the text following *marker* on the line where it appears."""
+    for line in text.splitlines():
+        if marker in line:
+            return line.split(marker, 1)[1].strip()
+    return ""
+
+
+def _first_nonempty_line(text: str) -> str:
+    """Return the first line of *text* that carries anything."""
+    for line in text.splitlines():
+        if line.strip():
+            return line.strip()
+    return ""
+
+
+def _format(template: str, fields: dict) -> str:
+    """Fill a configured wording template, surviving a bad placeholder.
+
+    A typo in config would otherwise raise mid-fetch and lose the whole
+    batch, so an unknown ``{placeholder}`` leaves the template visible
+    instead — the operator sees their own broken wording in the inbox.
+    """
+    try:
+        return template.format(**fields)
+    except (KeyError, IndexError, ValueError) as e:
+        logger.warning("IMAP: bad placeholder in configured template (%s)", e)
+        return template
diff --git a/nerve/sources/registry.py b/nerve/sources/registry.py
index 67f3504e..9197feac 100644
--- a/nerve/sources/registry.py
+++ b/nerve/sources/registry.py
@@ -83,6 +83,65 @@ def build_source_runners(
             ))
             logger.info("Registered source: %s (batch=%d)", source.source_name, gmail.batch_size)
 
+    # IMAP — one source per mailbox, each with an independent cursor.
+    # Passwords come from sync.imap.passwords (config.local.yaml), keyed by
+    # username, so no credential ever sits in the tracked config.
+    imap = config.sync.imap
+    if imap.enabled and imap.accounts:
+        import dataclasses
+
+        from nerve.sources.imap import ImapSource
+
+        # The image pass needs a prompt to send; without one it would ask the
+        # model nothing and label every message unreadable, so treat a missing
+        # prompt as "not configured" and say so once.
+        vision = dataclasses.replace(
+            imap.vision, model=imap.vision.model or config.memory.fast_model,
+        )
+        if vision.enabled and not vision.prompt:
+            logger.warning(
+                "sync.imap.vision.enabled is set but vision.prompt is empty — "
+                "the image pass stays off",
+            )
+            vision = dataclasses.replace(vision, enabled=False)
+
+        for account in imap.accounts:
+            password = imap.passwords.get(account.username, "")
+            if not password:
+                logger.warning(
+                    "Skipping IMAP source %s: no password for %s in "
+                    "sync.imap.passwords",
+                    account.label, account.username,
+                )
+                continue
+            source = ImapSource(
+                host=account.host,
+                port=account.port,
+                username=account.username,
+                password=password,
+                mailbox=account.mailbox,
+                label=account.label,
+                initial_lookback_days=imap.initial_lookback_days,
+                match=imap.match,
+                vision=vision,
+                vision_client_factory=condense_factory,
+            )
+            runners.append(SourceRunner(
+                source=source,
+                db=db,
+                batch_size=imap.batch_size,
+                condense=imap.condense,
+                condense_model=condense_model,
+                condense_prompt=imap.condense_prompt or "",
+                condense_client_factory=condense_factory,
+                ttl_days=ttl_days,
+            ))
+            logger.info(
+                "Registered source: %s (batch=%d, only_matched=%s, vision=%s)",
+                source.source_name, imap.batch_size,
+                imap.match.only_matched, vision.enabled,
+            )
+
     # GitHub (notifications)
     gh = config.sync.github
     if gh.enabled:
diff --git a/tests/test_imap_source.py b/tests/test_imap_source.py
new file mode 100644
index 00000000..38c4ad00
--- /dev/null
+++ b/tests/test_imap_source.py
@@ -0,0 +1,437 @@
+"""Unit tests for :class:`ImapSource`.
+
+These tests avoid the network entirely. They exercise:
+  * cursor parse/format helpers and template formatting,
+  * MIME part extraction + the configured match rules,
+  * the async record builders (image pass success, failure → fallback,
+    match with no image → fallback, plain email),
+by stubbing the blocking ``_imap_fetch_blocking`` step and the vision client.
+
+The scenario used throughout is a mailbox where some senders deliver their
+payload only as an inline scan, so the model has to read the image.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+from email.message import EmailMessage
+
+import pytest
+
+from nerve.config import ImapMatchConfig, ImapVisionConfig
+from nerve.sources.imap import (
+    ImapSource,
+    _decode_hdr,
+    _extract_parts,
+    _extract_status_int,
+    _first_line_after,
+    _first_nonempty_line,
+    _format,
+    _parse_cursor,
+)
+
+# 1x1 transparent PNG (valid, tiny) for the image tests.
+_PNG_BYTES = base64.b64decode(
+    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk"
+    "+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
+)
+
+_MATCH = ImapMatchConfig(
+    sender_contains=["scans@example.net"],
+    attachment_contains=["scan"],
+)
+
+_VISION = ImapVisionConfig(
+    enabled=True,
+    model="claude-haiku-4-5-20251001",
+    prompt="Read the scan. Answer with one line:\nSender: <name>",
+    answer_key="Sender:",
+    unknown_answer="unreadable",
+    summary="[{label}] scan from {answer}",
+    content="{vision}\n\nSubject: {subject}",
+    summary_unknown="[{label}] unreadable scan",
+    content_unknown="The image could not be read.\n\nSubject: {subject}",
+)
+
+
+# ---------------------------------------------------------------------------
+# Pure helpers
+# ---------------------------------------------------------------------------
+
+def test_parse_cursor():
+    assert _parse_cursor("123:456") == (123, 456)
+    assert _parse_cursor(None) == (0, None)
+    assert _parse_cursor("garbage") == (0, None)
+    assert _parse_cursor("123") == (0, None)
+
+
+def test_extract_status_int():
+    line = "INBOX (UIDVALIDITY 1781468017 UIDNEXT 42)"
+    assert _extract_status_int(line, "UIDVALIDITY") == 1781468017
+    assert _extract_status_int(line, "UIDNEXT") == 42
+    assert _extract_status_int(line, "MISSING") == 0
+
+
+def test_decode_hdr():
+    # RFC 2047 encoded-word (UTF-8 base64 for "Grüße")
+    assert _decode_hdr("=?UTF-8?B?R3LDvMOfZQ==?=") == "Grüße"
+    assert _decode_hdr(None) == ""
+    assert _decode_hdr("plain") == "plain"
+
+
+def test_first_line_after():
+    text = "Sender: Acme GmbH\nType: invoice"
+    assert _first_line_after(text, "Sender:") == "Acme GmbH"
+    assert _first_line_after(text, "Nope:") == ""
+
+
+def test_first_nonempty_line():
+    assert _first_nonempty_line("\n\n  Acme GmbH \nrest") == "Acme GmbH"
+    assert _first_nonempty_line("   \n\n") == ""
+
+
+def test_format_survives_a_bad_placeholder():
+    """A typo in configured wording must not lose the whole fetch batch."""
+    fields = {"label": "post", "answer": "Acme GmbH"}
+    assert _format("[{label}] {answer}", fields) == "[post] Acme GmbH"
+    # Unknown key: template comes back verbatim instead of raising.
+    assert _format("[{label}] {nope}", fields) == "[{label}] {nope}"
+
+
+# ---------------------------------------------------------------------------
+# MIME extraction + match rules
+# ---------------------------------------------------------------------------
+
+def _plain_email(from_addr: str, subject: str, body: str) -> EmailMessage:
+    m = EmailMessage()
+    m["From"] = from_addr
+    m["Subject"] = subject
+    m["Date"] = "Sat, 20 Jun 2026 13:34:40 +0000"
+    m.set_content(body)
+    return m
+
+
+def _email_with_image(
+    from_addr: str, *, cid: str, filename: str, payload: bytes = _PNG_BYTES,
+) -> EmailMessage:
+    m = _plain_email(from_addr, "Your document", "See attached")
+    m.add_related(
+        payload, maintype="image", subtype="png", cid=cid, filename=filename,
+    )
+    return m
+
+
+def test_extract_parts_plain_text():
+    m = _plain_email("a@b.com", "Hi", "Hello world")
+    body, image, mt, hit = _extract_parts(m, _MATCH.attachment_contains)
+    assert "Hello world" in body
+    assert image is None and mt is None and hit is False
+
+
+def test_extract_parts_hinted_image_by_cid():
+    m = _email_with_image(
+        "noreply@example.net", cid="<scan_12345>", filename="page1.png",
+    )
+    _body, image, mt, hit = _extract_parts(m, _MATCH.attachment_contains)
+    assert image == _PNG_BYTES
+    assert mt == "image/png"
+    assert hit is True
+
+
+def test_extract_parts_image_without_hint_is_not_a_hit():
+    """An image still comes back as a candidate, but it does not match."""
+    m = _email_with_image(
+        "noreply@example.net", cid="<logo_1>", filename="logo.png",
+    )
+    _body, image, _mt, hit = _extract_parts(m, _MATCH.attachment_contains)
+    assert image == _PNG_BYTES  # largest-image fallback candidate
+    assert hit is False
+
+
+def test_extract_parts_with_no_rules_never_hits():
+    m = _email_with_image(
+        "noreply@example.net", cid="<scan_12345>", filename="scan.png",
+    )
+    _body, _image, _mt, hit = _extract_parts(m, [])
+    assert hit is False
+
+
+# ---------------------------------------------------------------------------
+# Match evaluation through the real parse path
+# ---------------------------------------------------------------------------
+
+class _FakeIMAP:
+    """Minimal stand-in for imaplib.IMAP4_SSL.uid("fetch", ...)."""
+
+    def __init__(self, message: EmailMessage):
+        self._raw = message.as_bytes()
+
+    def uid(self, command, *args):
+        assert command == "fetch"
+        return "OK", [(b"1 (RFC822 {n}", self._raw)]
+
+
+def _parse(src: ImapSource, message: EmailMessage) -> dict:
+    return src._fetch_one(_FakeIMAP(message), 7, 99)
+
+
+def test_match_by_sender():
+    src = _src()
+    parsed = _parse(src, _plain_email("Scans <scans@example.net>", "Doc", "x"))
+    assert parsed["matched"] is True
+    assert parsed["id"] == "99-7"
+
+
+def test_match_by_sender_is_case_insensitive():
+    src = _src()
+    parsed = _parse(src, _plain_email("SCANS@EXAMPLE.NET", "Doc", "x"))
+    assert parsed["matched"] is True
+
+
+def test_match_by_attachment_hint():
+    src = _src()
+    parsed = _parse(
+        src,
+        _email_with_image("other@elsewhere.org", cid="<scan_1>", filename="a.png"),
+    )
+    assert parsed["matched"] is True
+    assert parsed["image"] == _PNG_BYTES
+
+
+def test_unmatched_message():
+    src = _src()
+    parsed = _parse(src, _plain_email("newsletter@elsewhere.org", "Ad", "buy"))
+    assert parsed["matched"] is False
+
+
+def test_no_rules_configured_matches_nothing():
+    """Empty match config = plain IMAP source, whatever the mail looks like."""
+    src = _src(match=ImapMatchConfig())
+    parsed = _parse(
+        src,
+        _email_with_image("scans@example.net", cid="<scan_1>", filename="scan.png"),
+    )
+    assert parsed["matched"] is False
+
+
+# ---------------------------------------------------------------------------
+# Record builders (async) with stubbed vision
+# ---------------------------------------------------------------------------
+
+class _FakeContentBlock:
+    def __init__(self, text):
+        self.text = text
+
+
+class _FakeResp:
+    def __init__(self, text):
+        self.content = [_FakeContentBlock(text)]
+
+
+class _FakeVisionClient:
+    def __init__(self, text=None, raise_exc=False):
+        self._text = text
+        self._raise = raise_exc
+        self.closed = False
+        self.last_kwargs = None
+
+        class _Messages:
+            async def create(_self, **kwargs):
+                self.last_kwargs = kwargs
+                if self._raise:
+                    raise RuntimeError("vision boom")
+                return _FakeResp(self._text)
+
+        self.messages = _Messages()
+
+    async def close(self):
+        self.closed = True
+
+
+def _src(**kw) -> ImapSource:
+    kw.setdefault("match", _MATCH)
+    kw.setdefault("vision", _VISION)
+    return ImapSource(
+        host="h", username="u@x", password="p", label="scans", **kw,
+    )
+
+
+def _matched_msg(**over) -> dict:
+    msg = {
+        "id": "1-2", "subject": "Your document", "from": "scans@example.net",
+        "date": "d", "timestamp": "2026-06-20T00:00:00+00:00",
+        "body": "b", "matched": True,
+        "image": _PNG_BYTES, "media_type": "image/png",
+    }
+    msg.update(over)
+    return msg
+
+
+@pytest.mark.asyncio
+async def test_vision_success():
+    client = _FakeVisionClient(text="Sender: Acme GmbH\nType: invoice")
+    src = _src(vision_client_factory=lambda: client)
+    summary, content = await src._build_vision_record(_matched_msg())
+    assert summary == "[scans] scan from Acme GmbH"
+    assert "Acme GmbH" in content
+    assert client.closed is True
+
+
+@pytest.mark.asyncio
+async def test_vision_failure_falls_back():
+    client = _FakeVisionClient(raise_exc=True)
+    src = _src(vision_client_factory=lambda: client)
+    summary, content = await src._build_vision_record(_matched_msg())
+    assert summary == "[scans] unreadable scan"
+    assert "could not be read" in content
+
+
+@pytest.mark.asyncio
+async def test_match_without_image_falls_back():
+    src = _src(vision_client_factory=lambda: _FakeVisionClient(text="x"))
+    summary, _ = await src._build_vision_record(
+        _matched_msg(image=None, media_type=None),
+    )
+    assert summary == "[scans] unreadable scan"
+
+
+@pytest.mark.asyncio
+async def test_empty_answer_key_takes_the_first_line():
+    client = _FakeVisionClient(text="Acme GmbH\nsecond line")
+    src = _src(
+        vision=ImapVisionConfig(**{**_VISION.__dict__, "answer_key": ""}),
+        vision_client_factory=lambda: client,
+    )
+    summary, _ = await src._build_vision_record(_matched_msg())
+    assert summary == "[scans] scan from Acme GmbH"
+
+
+@pytest.mark.asyncio
+async def test_fetch_builds_records_from_stub(monkeypatch):
+    """fetch() should route matched vs plain and produce SourceRecords."""
+    client = _FakeVisionClient(text="Sender: Acme GmbH\nType: invoice")
+    src = _src(vision_client_factory=lambda: client)
+
+    stub_messages = [
+        {
+            "id": "9-1", "subject": "Welcome", "from": "hello@example.org",
+            "date": "d", "timestamp": "2026-06-20T00:00:00+00:00", "body": "hi",
+            "matched": False, "image": None, "media_type": None,
+        },
+        _matched_msg(id="9-2"),
+    ]
+
+    monkeypatch.setattr(
+        src, "_imap_fetch_blocking", lambda cursor, limit: (stub_messages, "9:2"),
+    )
+
+    result = await src.fetch(cursor=None, limit=10)
+    assert result.next_cursor == "9:2"
+    assert len(result.records) == 2
+    plain, matched = result.records
+    assert plain.record_type == "imap_message"
+    assert plain.metadata["matched"] is False
+    assert "Welcome" in plain.summary
+    assert matched.metadata["matched"] is True
+    assert "Acme GmbH" in matched.summary
+
+
+@pytest.mark.asyncio
+async def test_vision_disabled_matched_message_uses_fallback(monkeypatch):
+    """With the image pass off, a match still gets its configured wording."""
+    src = _src(vision=ImapVisionConfig(**{**_VISION.__dict__, "enabled": False}))
+    monkeypatch.setattr(
+        src, "_imap_fetch_blocking",
+        lambda cursor, limit: ([_matched_msg(id="9-2")], "9:2"),
+    )
+    result = await src.fetch(cursor=None, limit=10)
+    assert result.records[0].summary == "[scans] unreadable scan"
+
+
+@pytest.mark.asyncio
+async def test_only_matched_drops_the_rest(monkeypatch):
+    """only_matched=True keeps only matches but still advances the cursor."""
+    client = _FakeVisionClient(text="Sender: Acme GmbH")
+    src = _src(
+        match=ImapMatchConfig(**{**_MATCH.__dict__, "only_matched": True}),
+        vision_client_factory=lambda: client,
+    )
+
+    stub_messages = [
+        {
+            "id": "9-1", "subject": "Ad", "from": "newsletter@elsewhere.org",
+            "date": "d", "timestamp": "2026-06-20T00:00:00+00:00", "body": "ad",
+            "matched": False, "image": None, "media_type": None,
+        },
+        _matched_msg(id="9-2"),
+    ]
+    monkeypatch.setattr(
+        src, "_imap_fetch_blocking", lambda cursor, limit: (stub_messages, "9:2"),
+    )
+
+    result = await src.fetch(cursor=None, limit=10)
+    # Only the match survives; cursor still advances past the dropped ad.
+    assert len(result.records) == 1
+    assert result.records[0].metadata["matched"] is True
+    assert result.next_cursor == "9:2"
+
+
+@pytest.mark.asyncio
+async def test_fetch_swallows_blocking_errors(monkeypatch):
+    src = _src(vision_client_factory=lambda: _FakeVisionClient(text="x"))
+
+    def _boom(cursor, limit):
+        raise RuntimeError("imap down")
+
+    monkeypatch.setattr(src, "_imap_fetch_blocking", _boom)
+    result = await src.fetch(cursor="5:5", limit=10)
+    assert result.records == []
+    assert result.next_cursor == "5:5"  # cursor preserved on error
+
+
+# ---------------------------------------------------------------------------
+# Prompt and answer key are a pair
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_prompt_and_answer_key_move_together():
+    """A reworded prompt must carry its answer_key along.
+
+    The two are coupled: the prompt tells the model which label to emit and
+    the parser reads the line after exactly that label.
+    """
+    vision = ImapVisionConfig.from_dict({
+        "enabled": True,
+        "model": "claude-haiku-4-5-20251001",
+        "prompt": "Wer ist der Absender? Antworte: Absender: <name>",
+        "answer_key": "Absender:",
+        "unknown_answer": "unlesbar",
+        "summary": "[{label}] Brief: {answer}",
+    })
+    client = _FakeVisionClient(text="Absender: Acme GmbH\nTyp: Rechnung")
+    src = _src(vision=vision, vision_client_factory=lambda: client)
+
+    summary, _content = await src._build_vision_record(_matched_msg())
+    assert summary == "[scans] Brief: Acme GmbH"
+    # The prompt actually sent is the configured one.
+    assert "Absender" in json.dumps(client.last_kwargs, ensure_ascii=False)
+
+
+@pytest.mark.asyncio
+async def test_mismatched_answer_key_degrades_to_unknown():
+    """The failure mode the coupling guards against, pinned explicitly."""
+    vision = ImapVisionConfig.from_dict({
+        "enabled": True,
+        "model": "claude-haiku-4-5-20251001",
+        "prompt": "Antworte: Absender: <name>",
+        # answer_key deliberately left pointing at the old label
+        "answer_key": "Sender:",
+        "unknown_answer": "unlesbar",
+        "summary": "[{label}] Brief: {answer}",
+    })
+    client = _FakeVisionClient(text="Absender: Acme GmbH")
+    src = _src(vision=vision, vision_client_factory=lambda: client)
+
+    summary, _ = await src._build_vision_record(_matched_msg())
+    assert "unlesbar" in summary