feat(email): add consent-gated IMAP/SMTP email channel

EMAIL_ASSISTANT_E2E_GUIDE.md

@@ -0,0 +1,164 @@
+# Nanobot Email Assistant: End-to-End Guide
+
+This guide explains how to run nanobot as a real email assistant with explicit user permission and optional automatic replies.
+
+## 1. What This Feature Does
+
+- Read unread emails via IMAP.
+- Let the agent analyze/respond to email content.
+- Send replies via SMTP.
+- Enforce explicit owner consent before mailbox access.
+- Let you toggle automatic replies on or off.
+
+## 2. Permission Model (Required)
+
+`channels.email.consentGranted` is the hard permission gate.
+
+- `false`: nanobot must not access mailbox content and must not send email.
+- `true`: nanobot may read/send based on other settings.
+
+Only set `consentGranted: true` after the mailbox owner explicitly agrees.
+
+## 3. Auto-Reply Mode
+
+`channels.email.autoReplyEnabled` controls outbound automatic email replies.
+
+- `true`: inbound emails can receive automatic agent replies.
+- `false`: inbound emails can still be read/processed, but automatic replies are skipped.
+
+Use `autoReplyEnabled: false` when you want analysis-only mode.
+
+## 4. Required Account Setup (Gmail Example)
+
+1. Enable 2-Step Verification in Google account security settings.
+2. Create an App Password.
+3. Use this app password for both IMAP and SMTP auth.
+
+Recommended servers:
+- IMAP host/port: `imap.gmail.com:993` (SSL)
+- SMTP host/port: `smtp.gmail.com:587` (STARTTLS)
+
+## 5. Config Example
+
+Edit `~/.nanobot/config.json`:
+
+```json
+{
+  "channels": {
+    "email": {
+      "enabled": true,
+      "consentGranted": true,
+      "imapHost": "imap.gmail.com",
+      "imapPort": 993,
+      "imapUsername": "you@gmail.com",
+      "imapPassword": "${NANOBOT_EMAIL_IMAP_PASSWORD}",
+      "imapMailbox": "INBOX",
+      "imapUseSsl": true,
+      "smtpHost": "smtp.gmail.com",
+      "smtpPort": 587,
+      "smtpUsername": "you@gmail.com",
+      "smtpPassword": "${NANOBOT_EMAIL_SMTP_PASSWORD}",
+      "smtpUseTls": true,
+      "smtpUseSsl": false,
+      "fromAddress": "you@gmail.com",
+      "autoReplyEnabled": true,
+      "pollIntervalSeconds": 30,
+      "markSeen": true,
+      "allowFrom": ["trusted.sender@example.com"]
+    }
+  }
+}
+```
+
+## 6. Set Secrets via Environment Variables
+
+In the same shell before starting gateway:
+
+```bash
+read -s "NANOBOT_EMAIL_IMAP_PASSWORD?IMAP app password: "
+echo
+read -s "NANOBOT_EMAIL_SMTP_PASSWORD?SMTP app password: "
+echo
+export NANOBOT_EMAIL_IMAP_PASSWORD
+export NANOBOT_EMAIL_SMTP_PASSWORD
+```
+
+If you use one app password for both, enter the same value twice.
+
+## 7. Run and Verify
+
+Start:
+
+```bash
+cd /Users/kaijimima1234/Desktop/nanobot
+PYTHONPATH=/Users/kaijimima1234/Desktop/nanobot .venv/bin/nanobot gateway
+```
+
+Check channel status:
+
+```bash
+PYTHONPATH=/Users/kaijimima1234/Desktop/nanobot .venv/bin/nanobot channels status
+```
+
+Expected behavior:
+- `enabled=true + consentGranted=true + autoReplyEnabled=true`: read + auto reply.
+- `enabled=true + consentGranted=true + autoReplyEnabled=false`: read only, no auto reply.
+- `consentGranted=false`: no read, no send.
+
+## 8. Commands You Can Tell Nanobot
+
+Once gateway is running and email consent is enabled:
+
+1. Summarize yesterday's emails:
+
+```text
+summarize my yesterday email
+```
+
+or
+
+```text
+!email summary yesterday
+```
+
+2. Send an email to a friend:
+
+```text
+!email send friend@example.com | Subject here | Body here
+```
+
+or
+
+```text
+send email to friend@example.com subject: Subject here body: Body here
+```
+
+Notes:
+- Sending command always performs a direct send (manual action by you).
+- If `consentGranted` is `false`, send/read are blocked.
+- If `autoReplyEnabled` is `false`, automatic replies are disabled, but direct send command above still works.
+
+## 9. End-to-End Test Plan
+
+1. Send a test email from an allowed sender to your mailbox.
+2. Confirm nanobot receives and processes it.
+3. If `autoReplyEnabled=true`, confirm a reply is delivered.
+4. Set `autoReplyEnabled=false`, send another test email.
+5. Confirm no auto-reply is sent.
+6. Set `consentGranted=false`, send another test email.
+7. Confirm nanobot does not read/send.
+
+## 10. Security Notes
+
+- Never commit real passwords/tokens into git.
+- Prefer environment variables for secrets.
+- Keep `allowFrom` restricted whenever possible.
+- Rotate app passwords immediately if leaked.
+
+## 11. PR Checklist
+
+- [ ] `consentGranted` gating works for read/send.
+- [ ] `autoReplyEnabled` toggle works as documented.
+- [ ] README updated with new fields.
+- [ ] Tests pass (`pytest`).
+- [ ] No real credentials in tracked files.

nanobot/channels/email.py

@@ -0,0 +1,399 @@
+"""Email channel implementation using IMAP polling + SMTP replies."""
+
+import asyncio
+import html
+import imaplib
+import re
+import smtplib
+import ssl
+from datetime import date
+from email import policy
+from email.header import decode_header, make_header
+from email.message import EmailMessage
+from email.parser import BytesParser
+from email.utils import parseaddr
+from typing import Any
+
+from loguru import logger
+
+from nanobot.bus.events import OutboundMessage
+from nanobot.bus.queue import MessageBus
+from nanobot.channels.base import BaseChannel
+from nanobot.config.schema import EmailConfig
+
+
+class EmailChannel(BaseChannel):
+    """
+    Email channel.
+
+    Inbound:
+    - Poll IMAP mailbox for unread messages.
+    - Convert each message into an inbound event.
+
+    Outbound:
+    - Send responses via SMTP back to the sender address.
+    """
+
+    name = "email"
+    _IMAP_MONTHS = (
+        "Jan",
+        "Feb",
+        "Mar",
+        "Apr",
+        "May",
+        "Jun",
+        "Jul",
+        "Aug",
+        "Sep",
+        "Oct",
+        "Nov",
+        "Dec",
+    )
+
+    def __init__(self, config: EmailConfig, bus: MessageBus):
+        super().__init__(config, bus)
+        self.config: EmailConfig = config
+        self._last_subject_by_chat: dict[str, str] = {}
+        self._last_message_id_by_chat: dict[str, str] = {}
+        self._processed_uids: set[str] = set()
+
+    async def start(self) -> None:
+        """Start polling IMAP for inbound emails."""
+        if not self.config.consent_granted:
+            logger.warning(
+                "Email channel disabled: consent_granted is false. "
+                "Set channels.email.consentGranted=true after explicit user permission."
+            )
+            return
+
+        if not self._validate_config():
+            return
+
+        self._running = True
+        logger.info("Starting Email channel (IMAP polling mode)...")
+
+        poll_seconds = max(5, int(self.config.poll_interval_seconds))
+        while self._running:
+            try:
+                inbound_items = await asyncio.to_thread(self._fetch_new_messages)
+                for item in inbound_items:
+                    sender = item["sender"]
+                    subject = item.get("subject", "")
+                    message_id = item.get("message_id", "")
+
+                    if subject:
+                        self._last_subject_by_chat[sender] = subject
+                    if message_id:
+                        self._last_message_id_by_chat[sender] = message_id
+
+                    await self._handle_message(
+                        sender_id=sender,
+                        chat_id=sender,
+                        content=item["content"],
+                        metadata=item.get("metadata", {}),
+                    )
+            except Exception as e:
+                logger.error(f"Email polling error: {e}")
+
+            await asyncio.sleep(poll_seconds)
+
+    async def stop(self) -> None:
+        """Stop polling loop."""
+        self._running = False
+
+    async def send(self, msg: OutboundMessage) -> None:
+        """Send email via SMTP."""
+        if not self.config.consent_granted:
+            logger.warning("Skip email send: consent_granted is false")
+            return
+
+        force_send = bool((msg.metadata or {}).get("force_send"))
+        if not self.config.auto_reply_enabled and not force_send:
+            logger.info("Skip automatic email reply: auto_reply_enabled is false")
+            return
+
+        if not self.config.smtp_host:
+            logger.warning("Email channel SMTP host not configured")
+            return
+
+        to_addr = msg.chat_id.strip()
+        if not to_addr:
+            logger.warning("Email channel missing recipient address")
+            return
+
+        base_subject = self._last_subject_by_chat.get(to_addr, "nanobot reply")
+        subject = self._reply_subject(base_subject)
+        if msg.metadata and isinstance(msg.metadata.get("subject"), str):
+            override = msg.metadata["subject"].strip()
+            if override:
+                subject = override
+
+        email_msg = EmailMessage()
+        email_msg["From"] = self.config.from_address or self.config.smtp_username or self.config.imap_username
+        email_msg["To"] = to_addr
+        email_msg["Subject"] = subject
+        email_msg.set_content(msg.content or "")
+
+        in_reply_to = self._last_message_id_by_chat.get(to_addr)
+        if in_reply_to:
+            email_msg["In-Reply-To"] = in_reply_to
+            email_msg["References"] = in_reply_to
+
+        try:
+            await asyncio.to_thread(self._smtp_send, email_msg)
+        except Exception as e:
+            logger.error(f"Error sending email to {to_addr}: {e}")
+            raise
+
+    def _validate_config(self) -> bool:
+        missing = []
+        if not self.config.imap_host:
+            missing.append("imap_host")
+        if not self.config.imap_username:
+            missing.append("imap_username")
+        if not self.config.imap_password:
+            missing.append("imap_password")
+        if not self.config.smtp_host:
+            missing.append("smtp_host")
+        if not self.config.smtp_username:
+            missing.append("smtp_username")
+        if not self.config.smtp_password:
+            missing.append("smtp_password")
+
+        if missing:
+            logger.error(f"Email channel not configured, missing: {', '.join(missing)}")
+            return False
+        return True
+
+    def _smtp_send(self, msg: EmailMessage) -> None:
+        timeout = 30
+        if self.config.smtp_use_ssl:
+            with smtplib.SMTP_SSL(
+                self.config.smtp_host,
+                self.config.smtp_port,
+                timeout=timeout,
+            ) as smtp:
+                smtp.login(self.config.smtp_username, self.config.smtp_password)
+                smtp.send_message(msg)
+            return
+
+        with smtplib.SMTP(self.config.smtp_host, self.config.smtp_port, timeout=timeout) as smtp:
+            if self.config.smtp_use_tls:
+                smtp.starttls(context=ssl.create_default_context())
+            smtp.login(self.config.smtp_username, self.config.smtp_password)
+            smtp.send_message(msg)
+
+    def _fetch_new_messages(self) -> list[dict[str, Any]]:
+        """Poll IMAP and return parsed unread messages."""
+        return self._fetch_messages(
+            search_criteria=("UNSEEN",),
+            mark_seen=self.config.mark_seen,
+            dedupe=True,
+            limit=0,
+        )
+
+    def fetch_messages_between_dates(
+        self,
+        start_date: date,
+        end_date: date,
+        limit: int = 20,
+    ) -> list[dict[str, Any]]:
+        """
+        Fetch messages in [start_date, end_date) by IMAP date search.
+
+        This is used for historical summarization tasks (e.g. "yesterday").
+        """
+        if end_date <= start_date:
+            return []
+
+        return self._fetch_messages(
+            search_criteria=(
+                "SINCE",
+                self._format_imap_date(start_date),
+                "BEFORE",
+                self._format_imap_date(end_date),
+            ),
+            mark_seen=False,
+            dedupe=False,
+            limit=max(1, int(limit)),
+        )
+
+    def _fetch_messages(
+        self,
+        search_criteria: tuple[str, ...],
+        mark_seen: bool,
+        dedupe: bool,
+        limit: int,
+    ) -> list[dict[str, Any]]:
+        """Fetch messages by arbitrary IMAP search criteria."""
+        messages: list[dict[str, Any]] = []
+        mailbox = self.config.imap_mailbox or "INBOX"
+
+        if self.config.imap_use_ssl:
+            client = imaplib.IMAP4_SSL(self.config.imap_host, self.config.imap_port)
+        else:
+            client = imaplib.IMAP4(self.config.imap_host, self.config.imap_port)
+
+        try:
+            client.login(self.config.imap_username, self.config.imap_password)
+            status, _ = client.select(mailbox)
+            if status != "OK":
+                return messages
+
+            status, data = client.search(None, *search_criteria)
+            if status != "OK" or not data:
+                return messages
+
+            ids = data[0].split()
+            if limit > 0 and len(ids) > limit:
+                ids = ids[-limit:]
+            for imap_id in ids:
+                status, fetched = client.fetch(imap_id, "(BODY.PEEK[] UID)")
+                if status != "OK" or not fetched:
+                    continue
+
+                raw_bytes = self._extract_message_bytes(fetched)
+                if raw_bytes is None:
+                    continue
+
+                uid = self._extract_uid(fetched)
+                if dedupe and uid and uid in self._processed_uids:
+                    continue
+
+                parsed = BytesParser(policy=policy.default).parsebytes(raw_bytes)
+                sender = parseaddr(parsed.get("From", ""))[1].strip().lower()
+                if not sender:
+                    continue
+
+                subject = self._decode_header_value(parsed.get("Subject", ""))
+                date_value = parsed.get("Date", "")
+                message_id = parsed.get("Message-ID", "").strip()
+                body = self._extract_text_body(parsed)
+
+                if not body:
+                    body = "(empty email body)"
+
+                body = body[: self.config.max_body_chars]
+                content = (
+                    f"Email received.\n"
+                    f"From: {sender}\n"
+                    f"Subject: {subject}\n"
+                    f"Date: {date_value}\n\n"
+                    f"{body}"
+                )
+
+                metadata = {
+                    "message_id": message_id,
+                    "subject": subject,
+                    "date": date_value,
+                    "sender_email": sender,
+                    "uid": uid,
+                }
+                messages.append(
+                    {
+                        "sender": sender,
+                        "subject": subject,
+                        "message_id": message_id,
+                        "content": content,
+                        "metadata": metadata,
+                    }
+                )
+
+                if dedupe and uid:
+                    self._processed_uids.add(uid)
+
+                if mark_seen:
+                    client.store(imap_id, "+FLAGS", "\\Seen")
+        finally:
+            try:
+                client.logout()
+            except Exception:
+                pass
+
+        return messages
+
+    @classmethod
+    def _format_imap_date(cls, value: date) -> str:
+        """Format date for IMAP search (always English month abbreviations)."""
+        month = cls._IMAP_MONTHS[value.month - 1]
+        return f"{value.day:02d}-{month}-{value.year}"
+
+    @staticmethod
+    def _extract_message_bytes(fetched: list[Any]) -> bytes | None:
+        for item in fetched:
+            if isinstance(item, tuple) and len(item) >= 2 and isinstance(item[1], (bytes, bytearray)):
+                return bytes(item[1])
+        return None
+
+    @staticmethod
+    def _extract_uid(fetched: list[Any]) -> str:
+        for item in fetched:
+            if isinstance(item, tuple) and item and isinstance(item[0], (bytes, bytearray)):
+                head = bytes(item[0]).decode("utf-8", errors="ignore")
+                m = re.search(r"UID\s+(\d+)", head)
+                if m:
+                    return m.group(1)
+        return ""
+
+    @staticmethod
+    def _decode_header_value(value: str) -> str:
+        if not value:
+            return ""
+        try:
+            return str(make_header(decode_header(value)))
+        except Exception:
+            return value
+
+    @classmethod
+    def _extract_text_body(cls, msg: Any) -> str:
+        """Best-effort extraction of readable body text."""
+        if msg.is_multipart():
+            plain_parts: list[str] = []
+            html_parts: list[str] = []
+            for part in msg.walk():
+                if part.get_content_disposition() == "attachment":
+                    continue
+                content_type = part.get_content_type()
+                try:
+                    payload = part.get_content()
+                except Exception:
+                    payload_bytes = part.get_payload(decode=True) or b""
+                    charset = part.get_content_charset() or "utf-8"
+                    payload = payload_bytes.decode(charset, errors="replace")
+                if not isinstance(payload, str):
+                    continue
+                if content_type == "text/plain":
+                    plain_parts.append(payload)
+                elif content_type == "text/html":
+                    html_parts.append(payload)
+            if plain_parts:
+                return "\n\n".join(plain_parts).strip()
+            if html_parts:
+                return cls._html_to_text("\n\n".join(html_parts)).strip()
+            return ""
+
+        try:
+            payload = msg.get_content()
+        except Exception:
+            payload_bytes = msg.get_payload(decode=True) or b""
+            charset = msg.get_content_charset() or "utf-8"
+            payload = payload_bytes.decode(charset, errors="replace")
+        if not isinstance(payload, str):
+            return ""
+        if msg.get_content_type() == "text/html":
+            return cls._html_to_text(payload).strip()
+        return payload.strip()
+
+    @staticmethod
+    def _html_to_text(raw_html: str) -> str:
+        text = re.sub(r"<\s*br\s*/?>", "\n", raw_html, flags=re.IGNORECASE)
+        text = re.sub(r"<\s*/\s*p\s*>", "\n", text, flags=re.IGNORECASE)
+        text = re.sub(r"<[^>]+>", "", text)
+        return html.unescape(text)
+
+    def _reply_subject(self, base_subject: str) -> str:
+        subject = (base_subject or "").strip() or "nanobot reply"
+        prefix = self.config.subject_prefix or "Re: "
+        if subject.lower().startswith("re:"):
+            return subject
+        return f"{prefix}{subject}"

nanobot/channels/manager.py

@@ -77,6 +77,17 @@ class ChannelManager:
                 logger.info("Feishu channel enabled")
             except ImportError as e:
                 logger.warning(f"Feishu channel not available: {e}")
+
+        # Email channel
+        if self.config.channels.email.enabled:
+            try:
+                from nanobot.channels.email import EmailChannel
+                self.channels["email"] = EmailChannel(
+                    self.config.channels.email, self.bus
+                )
+                logger.info("Email channel enabled")
+            except ImportError as e:
+                logger.warning(f"Email channel not available: {e}")
     
     async def start_all(self) -> None:
         """Start WhatsApp channel and the outbound dispatcher."""

nanobot/config/schema.py

@@ -38,6 +38,36 @@ class DiscordConfig(BaseModel):
     gateway_url: str = "wss://gateway.discord.gg/?v=10&encoding=json"
     intents: int = 37377  # GUILDS + GUILD_MESSAGES + DIRECT_MESSAGES + MESSAGE_CONTENT
 
+class EmailConfig(BaseModel):
+    """Email channel configuration (IMAP inbound + SMTP outbound)."""
+    enabled: bool = False
+    consent_granted: bool = False  # Explicit owner permission to access mailbox data
+
+    # IMAP (receive)
+    imap_host: str = ""
+    imap_port: int = 993
+    imap_username: str = ""
+    imap_password: str = ""
+    imap_mailbox: str = "INBOX"
+    imap_use_ssl: bool = True
+
+    # SMTP (send)
+    smtp_host: str = ""
+    smtp_port: int = 587
+    smtp_username: str = ""
+    smtp_password: str = ""
+    smtp_use_tls: bool = True
+    smtp_use_ssl: bool = False
+    from_address: str = ""
+
+    # Behavior
+    auto_reply_enabled: bool = True  # If false, inbound email is read but no automatic reply is sent
+    poll_interval_seconds: int = 30
+    mark_seen: bool = True
+    max_body_chars: int = 12000
+    subject_prefix: str = "Re: "
+    allow_from: list[str] = Field(default_factory=list)  # Allowed sender email addresses
+
 
 class ChannelsConfig(BaseModel):
     """Configuration for chat channels."""
@@ -45,6 +75,7 @@ class ChannelsConfig(BaseModel):
     telegram: TelegramConfig = Field(default_factory=TelegramConfig)
     discord: DiscordConfig = Field(default_factory=DiscordConfig)
     feishu: FeishuConfig = Field(default_factory=FeishuConfig)
+    email: EmailConfig = Field(default_factory=EmailConfig)
 
 
 class AgentDefaults(BaseModel):

tests/test_email_channel.py

@@ -0,0 +1,311 @@
+from email.message import EmailMessage
+from datetime import date
+
+import pytest
+
+from nanobot.bus.events import OutboundMessage
+from nanobot.bus.queue import MessageBus
+from nanobot.channels.email import EmailChannel
+from nanobot.config.schema import EmailConfig
+
+
+def _make_config() -> EmailConfig:
+    return EmailConfig(
+        enabled=True,
+        consent_granted=True,
+        imap_host="imap.example.com",
+        imap_port=993,
+        imap_username="bot@example.com",
+        imap_password="secret",
+        smtp_host="smtp.example.com",
+        smtp_port=587,
+        smtp_username="bot@example.com",
+        smtp_password="secret",
+        mark_seen=True,
+    )
+
+
+def _make_raw_email(
+    from_addr: str = "alice@example.com",
+    subject: str = "Hello",
+    body: str = "This is the body.",
+) -> bytes:
+    msg = EmailMessage()
+    msg["From"] = from_addr
+    msg["To"] = "bot@example.com"
+    msg["Subject"] = subject
+    msg["Message-ID"] = "<m1@example.com>"
+    msg.set_content(body)
+    return msg.as_bytes()
+
+
+def test_fetch_new_messages_parses_unseen_and_marks_seen(monkeypatch) -> None:
+    raw = _make_raw_email(subject="Invoice", body="Please pay")
+
+    class FakeIMAP:
+        def __init__(self) -> None:
+            self.store_calls: list[tuple[bytes, str, str]] = []
+
+        def login(self, _user: str, _pw: str):
+            return "OK", [b"logged in"]
+
+        def select(self, _mailbox: str):
+            return "OK", [b"1"]
+
+        def search(self, *_args):
+            return "OK", [b"1"]
+
+        def fetch(self, _imap_id: bytes, _parts: str):
+            return "OK", [(b"1 (UID 123 BODY[] {200})", raw), b")"]
+
+        def store(self, imap_id: bytes, op: str, flags: str):
+            self.store_calls.append((imap_id, op, flags))
+            return "OK", [b""]
+
+        def logout(self):
+            return "BYE", [b""]
+
+    fake = FakeIMAP()
+    monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
+
+    channel = EmailChannel(_make_config(), MessageBus())
+    items = channel._fetch_new_messages()
+
+    assert len(items) == 1
+    assert items[0]["sender"] == "alice@example.com"
+    assert items[0]["subject"] == "Invoice"
+    assert "Please pay" in items[0]["content"]
+    assert fake.store_calls == [(b"1", "+FLAGS", "\\Seen")]
+
+    # Same UID should be deduped in-process.
+    items_again = channel._fetch_new_messages()
+    assert items_again == []
+
+
+def test_extract_text_body_falls_back_to_html() -> None:
+    msg = EmailMessage()
+    msg["From"] = "alice@example.com"
+    msg["To"] = "bot@example.com"
+    msg["Subject"] = "HTML only"
+    msg.add_alternative("<p>Hello<br>world</p>", subtype="html")
+
+    text = EmailChannel._extract_text_body(msg)
+    assert "Hello" in text
+    assert "world" in text
+
+
+@pytest.mark.asyncio
+async def test_start_returns_immediately_without_consent(monkeypatch) -> None:
+    cfg = _make_config()
+    cfg.consent_granted = False
+    channel = EmailChannel(cfg, MessageBus())
+
+    called = {"fetch": False}
+
+    def _fake_fetch():
+        called["fetch"] = True
+        return []
+
+    monkeypatch.setattr(channel, "_fetch_new_messages", _fake_fetch)
+    await channel.start()
+    assert channel.is_running is False
+    assert called["fetch"] is False
+
+
+@pytest.mark.asyncio
+async def test_send_uses_smtp_and_reply_subject(monkeypatch) -> None:
+    class FakeSMTP:
+        def __init__(self, _host: str, _port: int, timeout: int = 30) -> None:
+            self.timeout = timeout
+            self.started_tls = False
+            self.logged_in = False
+            self.sent_messages: list[EmailMessage] = []
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def starttls(self, context=None):
+            self.started_tls = True
+
+        def login(self, _user: str, _pw: str):
+            self.logged_in = True
+
+        def send_message(self, msg: EmailMessage):
+            self.sent_messages.append(msg)
+
+    fake_instances: list[FakeSMTP] = []
+
+    def _smtp_factory(host: str, port: int, timeout: int = 30):
+        instance = FakeSMTP(host, port, timeout=timeout)
+        fake_instances.append(instance)
+        return instance
+
+    monkeypatch.setattr("nanobot.channels.email.smtplib.SMTP", _smtp_factory)
+
+    channel = EmailChannel(_make_config(), MessageBus())
+    channel._last_subject_by_chat["alice@example.com"] = "Invoice #42"
+    channel._last_message_id_by_chat["alice@example.com"] = "<m1@example.com>"
+
+    await channel.send(
+        OutboundMessage(
+            channel="email",
+            chat_id="alice@example.com",
+            content="Acknowledged.",
+        )
+    )
+
+    assert len(fake_instances) == 1
+    smtp = fake_instances[0]
+    assert smtp.started_tls is True
+    assert smtp.logged_in is True
+    assert len(smtp.sent_messages) == 1
+    sent = smtp.sent_messages[0]
+    assert sent["Subject"] == "Re: Invoice #42"
+    assert sent["To"] == "alice@example.com"
+    assert sent["In-Reply-To"] == "<m1@example.com>"
+
+
+@pytest.mark.asyncio
+async def test_send_skips_when_auto_reply_disabled(monkeypatch) -> None:
+    class FakeSMTP:
+        def __init__(self, _host: str, _port: int, timeout: int = 30) -> None:
+            self.sent_messages: list[EmailMessage] = []
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def starttls(self, context=None):
+            return None
+
+        def login(self, _user: str, _pw: str):
+            return None
+
+        def send_message(self, msg: EmailMessage):
+            self.sent_messages.append(msg)
+
+    fake_instances: list[FakeSMTP] = []
+
+    def _smtp_factory(host: str, port: int, timeout: int = 30):
+        instance = FakeSMTP(host, port, timeout=timeout)
+        fake_instances.append(instance)
+        return instance
+
+    monkeypatch.setattr("nanobot.channels.email.smtplib.SMTP", _smtp_factory)
+
+    cfg = _make_config()
+    cfg.auto_reply_enabled = False
+    channel = EmailChannel(cfg, MessageBus())
+    await channel.send(
+        OutboundMessage(
+            channel="email",
+            chat_id="alice@example.com",
+            content="Should not send.",
+        )
+    )
+    assert fake_instances == []
+
+    await channel.send(
+        OutboundMessage(
+            channel="email",
+            chat_id="alice@example.com",
+            content="Force send.",
+            metadata={"force_send": True},
+        )
+    )
+    assert len(fake_instances) == 1
+    assert len(fake_instances[0].sent_messages) == 1
+
+
+@pytest.mark.asyncio
+async def test_send_skips_when_consent_not_granted(monkeypatch) -> None:
+    class FakeSMTP:
+        def __init__(self, _host: str, _port: int, timeout: int = 30) -> None:
+            self.sent_messages: list[EmailMessage] = []
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def starttls(self, context=None):
+            return None
+
+        def login(self, _user: str, _pw: str):
+            return None
+
+        def send_message(self, msg: EmailMessage):
+            self.sent_messages.append(msg)
+
+    called = {"smtp": False}
+
+    def _smtp_factory(host: str, port: int, timeout: int = 30):
+        called["smtp"] = True
+        return FakeSMTP(host, port, timeout=timeout)
+
+    monkeypatch.setattr("nanobot.channels.email.smtplib.SMTP", _smtp_factory)
+
+    cfg = _make_config()
+    cfg.consent_granted = False
+    channel = EmailChannel(cfg, MessageBus())
+    await channel.send(
+        OutboundMessage(
+            channel="email",
+            chat_id="alice@example.com",
+            content="Should not send.",
+            metadata={"force_send": True},
+        )
+    )
+    assert called["smtp"] is False
+
+
+def test_fetch_messages_between_dates_uses_imap_since_before_without_mark_seen(monkeypatch) -> None:
+    raw = _make_raw_email(subject="Status", body="Yesterday update")
+
+    class FakeIMAP:
+        def __init__(self) -> None:
+            self.search_args = None
+            self.store_calls: list[tuple[bytes, str, str]] = []
+
+        def login(self, _user: str, _pw: str):
+            return "OK", [b"logged in"]
+
+        def select(self, _mailbox: str):
+            return "OK", [b"1"]
+
+        def search(self, *_args):
+            self.search_args = _args
+            return "OK", [b"5"]
+
+        def fetch(self, _imap_id: bytes, _parts: str):
+            return "OK", [(b"5 (UID 999 BODY[] {200})", raw), b")"]
+
+        def store(self, imap_id: bytes, op: str, flags: str):
+            self.store_calls.append((imap_id, op, flags))
+            return "OK", [b""]
+
+        def logout(self):
+            return "BYE", [b""]
+
+    fake = FakeIMAP()
+    monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
+
+    channel = EmailChannel(_make_config(), MessageBus())
+    items = channel.fetch_messages_between_dates(
+        start_date=date(2026, 2, 6),
+        end_date=date(2026, 2, 7),
+        limit=10,
+    )
+
+    assert len(items) == 1
+    assert items[0]["subject"] == "Status"
+    # search(None, "SINCE", "06-Feb-2026", "BEFORE", "07-Feb-2026")
+    assert fake.search_args is not None
+    assert fake.search_args[1:] == ("SINCE", "06-Feb-2026", "BEFORE", "07-Feb-2026")
+    assert fake.store_calls == []