Merge remote-tracking branch 'origin/feat/WA1-email-notify'

apps/api/app/db/migrate.py

@@ -73,6 +73,7 @@ def reset_database(database_url: str | None = None) -> None:
         conn.execute(
             """
             DROP TABLE IF EXISTS task_run, outbox, approval, artifact,
+                email_suggestion, notification_log,
                 application, job_posting, cv_section, profile, schema_migrations
             CASCADE
             """

apps/api/app/imap_watch.py

@@ -0,0 +1,468 @@
+"""IMAP email watch: polls UNSEEN messages and classifies them.
+
+Uses stdlib imaplib (SSL). Enabled only when EMAIL_WATCH_ENABLED=true.
+Config: IMAP_HOST, IMAP_PORT, IMAP_USER, IMAP_PASS.
+
+Flow:
+1. Connect via IMAP SSL.
+2. Fetch UNSEEN messages since last poll.
+3. For each message, match sender domain + subject/body keywords to
+   open applications (status in sent/interviewing).
+4. Classify via LLM task 'email_classify' -> {classification, state_proposal, reason}.
+5. Insert email_suggestion rows (skip noise and dedupe by from+subject+day).
+"""
+
+from __future__ import annotations
+
+import email
+import email.utils
+import hashlib
+import imaplib
+import logging
+import os
+from datetime import datetime, timezone
+from typing import Any, Sequence
+
+from app.db import execute, fetch_all, fetch_one
+from app import llm
+
+logger = logging.getLogger(__name__)
+
+VALID_CLASSIFICATIONS = frozenset({
+    "interview_invite",
+    "rejection",
+    "question",
+    "noise",
+})
+
+# Keywords for cheap pre-matching before LLM classify
+INTERVIEW_KEYWORDS = ("interview", "invite", "meeting", "schedule", "call")
+REJECTION_KEYWORDS = ("regret", "unfortunately", "not moving", "rejection", "position has been filled")
+QUESTION_KEYWORDS = ("question", "clarif", "additional", "could you", "please provide")
+NOISE_KEYWORDS = ("newsletter", "unsubscribe", "promotion", "advert", "offer")
+
+
+def is_email_watch_enabled() -> bool:
+    """Check if email watch is enabled."""
+    return os.environ.get("EMAIL_WATCH_ENABLED", "false").lower() in (
+        "true",
+        "1",
+        "yes",
+    )
+
+
+def _extract_sender_domain(from_addr: str) -> str:
+    """Extract the domain from an email From header."""
+    parsed = email.utils.parseaddr(from_addr)
+    addr = parsed[1] or from_addr
+    parts = addr.split("@")
+    if len(parts) >= 2:
+        return parts[-1].lower().strip()
+    return ""
+
+
+def _build_snippet(body: str, max_len: int = 300) -> str:
+    """Truncate body to a snippet."""
+    body = body.replace("\r", " ").replace("\n", " ").strip()
+    if len(body) > max_len:
+        return body[:max_len] + "..."
+    return body
+
+
+def _parse_email_message(raw_bytes: bytes) -> dict[str, str]:
+    """Parse raw email bytes into a dict with from, subject, body."""
+    msg = email.message_from_bytes(raw_bytes)
+    from_addr = msg.get("From", "")
+    subject = msg.get("Subject", "")
+    date_str = msg.get("Date", "")
+
+    # Extract body (prefer plain text)
+    body = ""
+    if msg.is_multipart():
+        for part in msg.walk():
+            ct = part.get_content_type()
+            if ct == "text/plain":
+                payload = part.get_payload(decode=True)
+                if payload:
+                    body = payload.decode("utf-8", errors="replace")
+                    break
+        if not body:
+            for part in msg.walk():
+                if part.get_content_type().startswith("text/"):
+                    payload = part.get_payload(decode=True)
+                    if payload:
+                        body = payload.decode("utf-8", errors="replace")
+                        break
+    else:
+        payload = msg.get_payload(decode=True)
+        if payload:
+            body = payload.decode("utf-8", errors="replace")
+
+    return {
+        "from": from_addr,
+        "subject": subject,
+        "body": body,
+        "date": date_str,
+    }
+
+
+def _parse_date(date_str: str) -> datetime:
+    """Parse RFC 2822 date string to UTC datetime. Falls back to now()."""
+    if date_str:
+        try:
+            parsed = email.utils.parsedate_to_datetime(date_str)
+            if parsed is not None:
+                if parsed.tzinfo is None:
+                    parsed = parsed.replace(tzinfo=timezone.utc)
+                return parsed.astimezone(timezone.utc)
+        except (ValueError, TypeError):
+            pass
+    return datetime.now(timezone.utc)
+
+
+def match_application(
+    from_addr: str,
+    subject: str,
+    body: str,
+    applications: Sequence[dict[str, Any]],
+) -> dict[str, Any] | None:
+    """Match an email to an open application.
+
+    Match by:
+    1. Company name in subject/body
+    2. Sender domain in posting URL
+
+    Only matches applications with state in ('sent', 'interviewing').
+    """
+    sender_domain = _extract_sender_domain(from_addr)
+    text_lower = (subject + " " + body).lower()
+
+    for app_row in applications:
+        state = app_row.get("state", "")
+        if state not in ("sent", "interviewing"):
+            continue
+
+        company = (app_row.get("company") or "").lower()
+        title = (app_row.get("title") or "").lower()
+        posting_url = app_row.get("url") or ""
+
+        # Match 1: company name in subject/body
+        if company and len(company) > 2 and company in text_lower:
+            return app_row
+
+        # Match 2: sender domain in posting URL
+        if sender_domain and sender_domain in (posting_url or "").lower():
+            return app_row
+
+        # Match 3: title keywords in subject (looser)
+        if title and len(title) > 3:
+            title_words = [w for w in title.split() if len(w) > 3]
+            matches = sum(1 for w in title_words if w in text_lower)
+            if matches >= 2 and len(title_words) >= 2:
+                return app_row
+
+    return None
+
+
+def classify_email(subject: str, body: str) -> dict[str, Any]:
+    """Classify an email via LLM task 'email_classify'.
+
+    Returns {classification, state_proposal, reason}.
+    Falls back to inline keyword heuristics if LLM fails.
+    """
+    text_lower = (subject + " " + body).lower()
+
+    result = llm.run_task(
+        "email_classify",
+        f"Subject: {subject}\nBody: {body[:1000]}",
+    )
+
+    classification = result.get("classification", "noise")
+    state_proposal = result.get("state_proposal")
+    reason = result.get("reason", "")
+
+    # Validate
+    if classification not in VALID_CLASSIFICATIONS:
+        classification = "noise"
+
+    return {
+        "classification": classification,
+        "state_proposal": state_proposal,
+        "reason": reason,
+    }
+
+
+def is_duplicate(from_addr: str, subject: str, received_at: datetime) -> bool:
+    """Check if a similar email_suggestion already exists (same from+subject+day)."""
+    day_start = received_at.replace(hour=0, minute=0, second=0, microsecond=0)
+    row = fetch_one(
+        """
+        SELECT id FROM email_suggestion
+        WHERE mailbox_from = %s AND subject = %s
+          AND received_at >= %s AND received_at < %s + interval '1 day'
+        LIMIT 1
+        """,
+        (from_addr, subject, day_start, day_start),
+    )
+    return row is not None
+
+
+def create_email_suggestion(
+    application_id: str | None,
+    mailbox_from: str,
+    subject: str,
+    snippet: str,
+    classification: str,
+    state_proposal: str | None,
+    received_at: datetime,
+) -> dict[str, Any]:
+    """Insert an email_suggestion row."""
+    row = execute(
+        """
+        INSERT INTO email_suggestion
+            (application_id, mailbox_from, subject, snippet, classification, state_proposal, status, received_at)
+        VALUES
+            (%s, %s, %s, %s, %s, %s, 'pending', %s)
+        RETURNING *
+        """,
+        (application_id, mailbox_from, subject, snippet, classification, state_proposal, received_at),
+    )
+    if row is None:
+        raise RuntimeError("insert email_suggestion failed")
+    return _normalize_suggestion(row)
+
+
+def _normalize_suggestion(row: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "id": str(row["id"]),
+        "application_id": str(row["application_id"]) if row.get("application_id") else None,
+        "mailbox_from": row["mailbox_from"],
+        "subject": row["subject"],
+        "snippet": row["snippet"],
+        "classification": row["classification"],
+        "state_proposal": row.get("state_proposal"),
+        "status": row["status"],
+        "received_at": row["received_at"].isoformat() if row.get("received_at") else None,
+        "created_at": row["created_at"].isoformat() if row.get("created_at") else None,
+    }
+
+
+def list_pending_suggestions() -> list[dict[str, Any]]:
+    """Return all pending email_suggestions, newest first."""
+    rows = fetch_all(
+        """
+        SELECT * FROM email_suggestion
+        WHERE status = 'pending'
+        ORDER BY received_at DESC
+        """
+    )
+    return [_normalize_suggestion(r) for r in rows]
+
+
+def get_suggestion(suggestion_id: str) -> dict[str, Any] | None:
+    row = fetch_one("SELECT * FROM email_suggestion WHERE id = %s", (suggestion_id,))
+    if row is None:
+        return None
+    return _normalize_suggestion(row)
+
+
+def update_suggestion_status(
+    suggestion_id: str,
+    status: str,
+) -> dict[str, Any] | None:
+    row = execute(
+        "UPDATE email_suggestion SET status = %s WHERE id = %s RETURNING *",
+        (status, suggestion_id),
+    )
+    if row is None:
+        return None
+    return _normalize_suggestion(row)
+
+
+def get_open_applications() -> list[dict[str, Any]]:
+    """Return applications in sent/interviewing state for matching."""
+    rows = fetch_all(
+        """
+        SELECT a.*, j.company, j.title, j.url
+        FROM application a
+        JOIN job_posting j ON a.job_posting_id = j.id
+        WHERE a.state IN ('sent', 'interviewing')
+        """
+    )
+    result: list[dict[str, Any]] = []
+    for row in rows:
+        result.append({
+            "id": str(row["id"]),
+            "state": row["state"],
+            "company": row.get("company", ""),
+            "title": row.get("title", ""),
+            "url": row.get("url", ""),
+        })
+    return result
+
+
+# --- IMAP poll ---
+
+class FakeImap:
+    """Test double for imaplib IMAP4_SSL. No network.
+
+    Usage:
+        fake = FakeImap(messages=[(uid1, raw1), (uid2, raw2)])
+        poll_inbox(fake)  # uses fake instead of real connection
+    """
+
+    def __init__(self, messages: list[tuple[bytes, bytes]] | None = None) -> None:
+        # messages: list of (uid, raw_email_bytes)
+        self._messages = messages or []
+        self._seen_uids: set[bytes] = set()
+        self.selected = False
+
+    def select(self, mailbox: str = "INBOX") -> tuple[str, list[bytes]]:
+        self.selected = True
+        count = len(self._messages)
+        return ("OK", [str(count).encode()])
+
+    def search(self, charset: str | None, *criteria: str) -> tuple[str, list[bytes]]:
+        # Return uids of messages matching criteria (we keep it simple)
+        uids = [uid for uid, _ in self._messages]
+        return ("OK", [b" ".join(uids)])
+
+    def fetch(self, uid: bytes, parts: str) -> tuple[str, list[tuple[bytes, bytes]]]:
+        for msg_uid, raw in self._messages:
+            if msg_uid == uid:
+                return ("OK", [(uid, raw)])
+        return ("OK", [])
+
+    def store(self, uid: bytes, flags: str, flag_set: str) -> tuple[str, list[bytes]]:
+        self._seen_uids.add(uid)
+        return ("OK", [uid])
+
+    def close(self) -> tuple[str, list[bytes]]:
+        self.selected = False
+        return ("OK", [b""])
+
+    def logout(self) -> tuple[str, list[bytes]]:
+        return ("OK", [b"BYE"])
+
+
+def _connect_imap() -> Any:
+    """Connect to IMAP server using env config."""
+    host = os.environ.get("IMAP_HOST", "")
+    port = int(os.environ.get("IMAP_PORT", "993"))
+    user = os.environ.get("IMAP_USER", "")
+    password = os.environ.get("IMAP_PASS", "")
+
+    conn = imaplib.IMAP4_SSL(host, port)
+    conn.login(user, password)
+    return conn
+
+
+def poll_inbox(conn: Any = None) -> list[dict[str, Any]]:
+    """Poll the inbox for UNSEEN messages, classify, and create suggestions.
+
+    If conn is provided (e.g. FakeImap for tests), uses it instead of connecting.
+    Returns list of created suggestions.
+
+    No network is used when conn is a FakeImap.
+    """
+    created: list[dict[str, Any]] = []
+    own_conn = False
+
+    if conn is None:
+        conn = _connect_imap()
+        own_conn = True
+
+    try:
+        conn.select("INBOX")
+        status, data = conn.search(None, "UNSEEN")
+        if status != "OK":
+            logger.warning("imap_watch: search failed: %s", status)
+            return created
+
+        uids = []
+        if data and data[0]:
+            uids = data[0].split()
+
+        if not uids:
+            return created
+
+        # Get open applications for matching
+        applications = get_open_applications()
+
+        for uid in uids:
+            status, fetch_data = conn.fetch(uid, "(RFC822)")
+            if status != "OK" or not fetch_data:
+                continue
+
+            raw_bytes = b""
+            for item in fetch_data:
+                if isinstance(item, tuple) and len(item) >= 2:
+                    raw_bytes = item[1]
+                    break
+
+            if not raw_bytes:
+                continue
+
+            parsed = _parse_email_message(raw_bytes)
+            from_addr = parsed["from"]
+            subject = parsed["subject"]
+            body = parsed["body"]
+            received_at = _parse_date(parsed["date"])
+            snippet = _build_snippet(body)
+
+            # Dedupe: same from+subject+day
+            if is_duplicate(from_addr, subject, received_at):
+                logger.debug("imap_watch: dedupe skip: %s / %s", from_addr, subject)
+                continue
+
+            # Match to application
+            app_row = match_application(from_addr, subject, body, applications)
+
+            # Classify
+            classification_result = classify_email(subject, body)
+            classification = classification_result["classification"]
+            state_proposal = classification_result["state_proposal"]
+
+            # Skip pure noise (don't create suggestion rows)
+            if classification == "noise":
+                continue
+
+            suggestion = create_email_suggestion(
+                application_id=app_row["id"] if app_row else None,
+                mailbox_from=from_addr,
+                subject=subject,
+                snippet=snippet,
+                classification=classification,
+                state_proposal=state_proposal,
+                received_at=received_at,
+            )
+            created.append(suggestion)
+
+            # Send notification for interview invites
+            if classification == "interview_invite" and app_row:
+                from app.notify import send_notification
+                send_notification(
+                    "email_suggestion",
+                    f"Interview invite from {app_row.get('company', 'unknown')}",
+                    {
+                        "suggestion_id": suggestion["id"],
+                        "application_id": app_row["id"],
+                        "classification": classification,
+                    },
+                )
+
+            # Mark as seen
+            try:
+                conn.store(uid, "+FLAGS", "\\Seen")
+            except Exception:
+                logger.debug("imap_watch: store failed for uid %s", uid)
+
+    finally:
+        if own_conn:
+            try:
+                conn.close()
+                conn.logout()
+            except Exception:
+                pass
+
+    return created

apps/api/app/llm.py

@@ -123,6 +123,11 @@ MOCK_OUTPUTS: dict[str, dict[str, Any]] = {
             "posting requirements. Be specific."
         ),
     },
+    "email_classify": {
+        "classification": "interview_invite",
+        "state_proposal": "interviewing",
+        "reason": "The email mentions an interview invitation.",
+    },
 }
 
 

apps/api/app/main.py

@@ -37,11 +37,13 @@ from app.schemas import (
     CvSectionOut,
     CvSectionUpdate,
     DigestItem,
+    EmailSuggestionOut,
     ErrorOut,
     InterviewPrepResponse,
     JobPostingCreate,
     JobPostingOut,
     NudgeItem,
+    NotificationLogOut,
     OutboxOut,
     OutboxSendRequest,
     PostingsFetchRequest,
@@ -932,4 +934,109 @@ def seed_demo() -> Any:
         "postings": postings_count,
         "applications": apps_count,
         "sections": sections_count,
-    }
+    }
+
+
+# --- v1.1: Email Suggestions ---
+
+@app.get("/api/suggestions", response_model=list[EmailSuggestionOut])
+def get_suggestions() -> Any:
+    """Return all pending email suggestions, newest first."""
+    from app.imap_watch import list_pending_suggestions
+    return list_pending_suggestions()
+
+
+@app.post("/api/suggestions/{suggestion_id}/accept")
+def accept_suggestion(suggestion_id: str) -> Any:
+    """Accept a suggestion: apply state_proposal via guarded transition.
+
+    If the suggestion has an application_id and a state_proposal, apply the
+    state transition through the normal guard path. Updates last_activity_at.
+    Marks the suggestion as 'accepted'.
+    """
+    from app.imap_watch import get_suggestion, update_suggestion_status
+
+    suggestion = get_suggestion(suggestion_id)
+    if suggestion is None:
+        raise HTTPException(status_code=404, detail="Suggestion not found")
+
+    if suggestion["status"] != "pending":
+        raise HTTPException(
+            status_code=409,
+            detail={"code": "not_pending", "message": "Suggestion is not pending"},
+        )
+
+    app_id = suggestion.get("application_id")
+    state_proposal = suggestion.get("state_proposal")
+
+    if app_id and state_proposal:
+        # Apply guarded transition
+        app_row = repo_app.get_application(app_id)
+        if app_row is None:
+            raise HTTPException(status_code=404, detail="Linked application not found")
+
+        from_state = app_row["state"]
+        to_state = state_proposal
+
+        from app.statemachine import TransitionContext, check_transition, InvalidTransition
+
+        has_score = app_row.get("score") is not None
+        has_confirmed_approval = False
+        artifact_hash_match = False
+
+        ctx = TransitionContext(
+            application_id=app_id,
+            from_state=from_state,
+            to_state=to_state,
+            has_score=has_score,
+            has_confirmed_approval=has_confirmed_approval,
+            artifact_hash_match=artifact_hash_match,
+        )
+
+        try:
+            check_transition(ctx)
+        except InvalidTransition as exc:
+            raise HTTPException(
+                status_code=409,
+                detail={"code": "invalid_transition", "message": str(exc)},
+            )
+
+        updated = repo_app.update_application_state(app_id, to_state)
+        if updated is None:
+            raise HTTPException(status_code=500, detail="State update failed")
+
+    # Mark suggestion as accepted
+    result = update_suggestion_status(suggestion_id, "accepted")
+    if result is None:
+        raise HTTPException(status_code=500, detail="Failed to update suggestion")
+    return result
+
+
+@app.post("/api/suggestions/{suggestion_id}/dismiss")
+def dismiss_suggestion(suggestion_id: str) -> Any:
+    """Dismiss a suggestion (mark as dismissed)."""
+    from app.imap_watch import get_suggestion, update_suggestion_status
+
+    suggestion = get_suggestion(suggestion_id)
+    if suggestion is None:
+        raise HTTPException(status_code=404, detail="Suggestion not found")
+
+    if suggestion["status"] != "pending":
+        raise HTTPException(
+            status_code=409,
+            detail={"code": "not_pending", "message": "Suggestion is not pending"},
+        )
+
+    result = update_suggestion_status(suggestion_id, "dismissed")
+    if result is None:
+        raise HTTPException(status_code=500, detail="Failed to update suggestion")
+    return result
+
+
+# --- v1.1: Notification Log ---
+
+@app.get("/api/notifications/log", response_model=list[NotificationLogOut])
+def get_notification_log() -> Any:
+    """Return last 50 notification log entries."""
+    from app.notify import list_notification_log
+    return list_notification_log(limit=50)

apps/api/app/notify.py

@@ -0,0 +1,183 @@
+"""Notification channels: LogChannel + WebhookChannel.
+
+Protocol-based: NotificationChannel defines send(kind, text, data) -> bool.
+LogChannel writes to notification_log (delivered=true always).
+WebhookChannel POSTs to NOTIFY_WEBHOOK_URL (2xx=delivered, else error row).
+
+send_notification(kind, text, data) is the public entry point used by
+the scheduler and email watch modules.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from typing import Any, Protocol
+
+from app.db import execute, fetch_all
+
+logger = logging.getLogger(__name__)
+
+
+class NotificationChannel(Protocol):
+    """Protocol for notification delivery channels."""
+
+    def send(self, kind: str, text: str, data: dict[str, Any]) -> bool:
+        """Send a notification. Returns True if delivered, False otherwise."""
+        ...
+
+
+def _normalize_log_row(row: dict[str, Any]) -> dict[str, Any]:
+    payload = row.get("payload")
+    if payload is not None and not isinstance(payload, dict):
+        payload = json.loads(payload) if isinstance(payload, str) else payload
+    return {
+        "id": str(row["id"]),
+        "channel": row["channel"],
+        "kind": row["kind"],
+        "payload": payload or {},
+        "delivered": row["delivered"],
+        "error": row.get("error"),
+        "created_at": row["created_at"].isoformat() if row.get("created_at") else None,
+    }
+
+
+class LogChannel:
+    """Writes notification entries to notification_log with delivered=true."""
+
+    channel_name: str = "log"
+
+    def send(self, kind: str, text: str, data: dict[str, Any]) -> bool:
+        row = execute(
+            """
+            INSERT INTO notification_log (channel, kind, payload, delivered, error)
+            VALUES (%s, %s, %s, true, NULL)
+            RETURNING *
+            """,
+            (self.channel_name, kind, json.dumps({"text": text, **data})),
+        )
+        if row is None:
+            logger.error("LogChannel: insert notification_log failed")
+            return False
+        logger.info("LogChannel: delivered kind=%s", kind)
+        return True
+
+
+class WebhookChannel:
+    """POSTs notification payload to NOTIFY_WEBHOOK_URL.
+
+    On 2xx response: writes notification_log with delivered=true.
+    On non-2xx or exception: writes notification_log with delivered=false and error.
+    """
+
+    channel_name: str = "webhook"
+
+    def __init__(self, url: str | None = None) -> None:
+        self.url = url or os.environ.get("NOTIFY_WEBHOOK_URL", "")
+
+    def send(self, kind: str, text: str, data: dict[str, Any]) -> bool:
+        payload: dict[str, Any] = {"kind": kind, "text": text, "data": data}
+        error: str | None = None
+        delivered = False
+
+        if not self.url:
+            error = "NOTIFY_WEBHOOK_URL not configured"
+            logger.warning("WebhookChannel: %s", error)
+        else:
+            try:
+                import httpx
+
+                resp = httpx.post(self.url, json=payload, timeout=10)
+                if 200 <= resp.status_code < 300:
+                    delivered = True
+                else:
+                    error = f"HTTP {resp.status_code}: {resp.text[:200]}"
+                    logger.warning("WebhookChannel: %s", error)
+            except Exception as exc:
+                error = str(exc)
+                logger.warning("WebhookChannel: exception: %s", error)
+
+        row = execute(
+            """
+            INSERT INTO notification_log (channel, kind, payload, delivered, error)
+            VALUES (%s, %s, %s, %s, %s)
+            RETURNING *
+            """,
+            (
+                self.channel_name,
+                kind,
+                json.dumps(payload),
+                delivered,
+                error,
+            ),
+        )
+        if row is None:
+            logger.error("WebhookChannel: insert notification_log failed")
+        return delivered
+
+
+# --- Channel registry ---
+
+_channels: list[NotificationChannel] | None = None
+
+
+def get_channels() -> list[NotificationChannel]:
+    """Return the list of active notification channels.
+
+    LogChannel is always included.
+    WebhookChannel is included when NOTIFY_WEBHOOK_URL is set.
+    """
+    global _channels
+    if _channels is not None:
+        return _channels
+
+    channels: list[NotificationChannel] = [LogChannel()]
+    webhook_url = os.environ.get("NOTIFY_WEBHOOK_URL", "").strip()
+    if webhook_url:
+        channels.append(WebhookChannel())
+    _channels = channels
+    return channels
+
+
+def set_channels(channels: list[NotificationChannel] | None) -> None:
+    """Override channel list (for testing)."""
+    global _channels
+    _channels = channels
+
+
+def reset_channels() -> None:
+    """Reset to default (for testing)."""
+    global _channels
+    _channels = None
+
+
+def send_notification(kind: str, text: str, data: dict[str, Any] | None = None) -> None:
+    """Send a notification via all active channels.
+
+    kind: e.g. 'daily_digest', 'email_suggestion'
+    text: human-readable notification text
+    data: structured payload dict
+    """
+    chs = get_channels()
+    payload = data or {}
+    for ch in chs:
+        try:
+            ch.send(kind, text, payload)
+        except Exception:
+            logger.exception("send_notification: channel %s failed", type(ch).__name__)
+
+
+# --- Repository helpers ---
+
+def list_notification_log(limit: int = 50) -> list[dict[str, Any]]:
+    """Return recent notification_log rows, newest first."""
+    rows = fetch_all(
+        """
+        SELECT * FROM notification_log
+        ORDER BY created_at DESC
+        LIMIT %s
+        """,
+        (limit,),
+    )
+    return [_normalize_log_row(r) for r in rows]

apps/api/app/scheduler.py

@@ -1,7 +1,10 @@
-"""APScheduler integration: daily fetch + batch score job.
+"""APScheduler integration: daily fetch + batch score + imap poll + digest.
 
 Starts during app lifespan when SCHEDULER_ENABLED=true (default false).
-Runs a daily job at 07:00 that fetches postings and batch-scores pending applications.
+Jobs:
+  - daily_fetch_score at 07:00: fetch postings + batch score
+  - daily_digest at 07:30: send daily digest notification
+  - imap_poll every 15 min: poll inbox for new emails (gated by EMAIL_WATCH_ENABLED)
 """
 
 from __future__ import annotations
@@ -23,6 +26,15 @@ def is_scheduler_enabled() -> bool:
     )
 
 
+def is_email_watch_enabled() -> bool:
+    """Check if email watch is enabled via env."""
+    return os.environ.get("EMAIL_WATCH_ENABLED", "false").lower() in (
+        "true",
+        "1",
+        "yes",
+    )
+
+
 async def _daily_fetch_and_score() -> None:
     """Daily job: fetch postings and batch-score pending applications."""
     logger.info("Scheduler: running daily fetch + batch score")
@@ -56,6 +68,67 @@ async def _daily_fetch_and_score() -> None:
         logger.exception("Scheduler: daily job failed")
 
 
+async def _daily_digest() -> None:
+    """Daily digest: build /today payload text and send notification."""
+    logger.info("Scheduler: running daily digest")
+    try:
+        from app.notify import send_notification
+        from app.db import repo_app
+
+        digest_apps = repo_app.get_digest(limit=20)
+        nudge_apps = repo_app.get_nudge_applications()
+        pending = repo_app.count_pending_approvals()
+
+        lines: list[str] = []
+        lines.append("Daily Digest")
+        lines.append(f"Scored applications: {len(digest_apps)}")
+        if digest_apps:
+            lines.append("")
+            lines.append("Top opportunities:")
+            for item in digest_apps[:5]:
+                score = item.get("score")
+                score_str = f" (score: {int(score)})" if score else ""
+                lines.append(
+                    f"  - {item.get('title', '?')} at {item.get('company', '?')}{score_str}"
+                )
+
+        if nudge_apps:
+            lines.append("")
+            lines.append(f"Follow-up nudges: {len(nudge_apps)}")
+            for n in nudge_apps[:5]:
+                lines.append(
+                    f"  - {n.get('title', '?')} at {n.get('company', '?')}"
+                    f" ({n.get('days_since_sent', 0)} days since sent)"
+                )
+
+        if pending:
+            lines.append("")
+            lines.append(f"Pending approvals: {pending}")
+
+        text = "\n".join(lines)
+        send_notification("daily_digest", text, {
+            "digest_count": len(digest_apps),
+            "nudge_count": len(nudge_apps),
+            "pending_approvals": pending,
+        })
+        logger.info("Scheduler: daily digest sent")
+    except Exception:
+        logger.exception("Scheduler: daily digest failed")
+
+
+async def _imap_poll() -> None:
+    """IMAP poll job: check for new emails and create suggestions."""
+    if not is_email_watch_enabled():
+        return
+    logger.info("Scheduler: running imap poll")
+    try:
+        from app.imap_watch import poll_inbox
+        created = poll_inbox()
+        logger.info("Scheduler: imap poll created %s suggestions", len(created))
+    except Exception:
+        logger.exception("Scheduler: imap poll failed")
+
+
 def start_scheduler() -> None:
     """Start the APScheduler if enabled."""
     global _scheduler
@@ -66,6 +139,7 @@ def start_scheduler() -> None:
     try:
         from apscheduler.schedulers.asyncio import AsyncIOScheduler
         from apscheduler.triggers.cron import CronTrigger
+        from apscheduler.triggers.interval import IntervalTrigger
     except ImportError:
         logger.warning(
             "APScheduler not installed; scheduler will not start."
@@ -79,8 +153,20 @@ def start_scheduler() -> None:
         id="daily_fetch_score",
         replace_existing=True,
     )
+    _scheduler.add_job(
+        _daily_digest,
+        CronTrigger(hour=7, minute=30),
+        id="daily_digest",
+        replace_existing=True,
+    )
+    _scheduler.add_job(
+        _imap_poll,
+        IntervalTrigger(minutes=15),
+        id="imap_poll",
+        replace_existing=True,
+    )
     _scheduler.start()
-    logger.info("Scheduler started: daily fetch+score at 07:00")
+    logger.info("Scheduler started: daily fetch+score at 07:00, digest at 07:30, imap poll every 15 min")
 
 
 def stop_scheduler() -> None:

apps/api/app/schemas.py

@@ -319,4 +319,31 @@ class SeedDemoResponse(BaseModel):
     profile: str
     postings: int
     applications: int
-    sections: int
+    sections: int
+
+
+# --- v1.1: Email Suggestions ---
+
+class EmailSuggestionOut(BaseModel):
+    id: str
+    application_id: str | None = None
+    mailbox_from: str
+    subject: str
+    snippet: str
+    classification: str
+    state_proposal: str | None = None
+    status: str
+    received_at: str | None = None
+    created_at: str | None = None
+
+
+# --- v1.1: Notification Log ---
+
+class NotificationLogOut(BaseModel):
+    id: str
+    channel: str
+    kind: str
+    payload: dict[str, Any] = {}
+    delivered: bool
+    error: str | None = None
+    created_at: str | None = None

apps/api/migrations/003_email_notify.sql

@@ -0,0 +1,24 @@
+-- 003_email_notify.sql -- email suggestions + notification log
+
+CREATE TABLE IF NOT EXISTS email_suggestion (
+  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  application_id uuid REFERENCES application(id) ON DELETE SET NULL,
+  mailbox_from text NOT NULL,
+  subject text NOT NULL,
+  snippet text NOT NULL,
+  classification text NOT NULL CHECK (classification IN ('interview_invite','rejection','question','noise')),
+  state_proposal text,
+  status text NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','accepted','dismissed')),
+  received_at timestamptz NOT NULL,
+  created_at timestamptz DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS notification_log (
+  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  channel text NOT NULL,
+  kind text NOT NULL,
+  payload jsonb NOT NULL,
+  delivered boolean NOT NULL,
+  error text,
+  created_at timestamptz DEFAULT now()
+);

apps/api/tests/conftest.py

@@ -38,7 +38,7 @@ def _truncate_tables():
     with psycopg.connect(DATABASE_URL) as conn:
         conn.execute(
             """
-            TRUNCATE TABLE task_run, outbox, approval, artifact,
+            TRUNCATE TABLE notification_log, email_suggestion, task_run, outbox, approval, artifact,
                 application, job_posting, cv_section, profile
             RESTART IDENTITY CASCADE
             """

apps/api/tests/test_v11_email_notify.py

@@ -0,0 +1,682 @@
+"""Tests for v1.1: email watch, notifications, suggestions endpoints.
+
+Covers:
+- IMAP matching logic (company in subject, sender domain in URL, title keywords)
+- Classifier -> suggestion row
+- Noise dedupe (same from+subject+day)
+- Accept applies transition through guard
+- Dismiss marks suggestion
+- Webhook success/failure notification_log rows
+- LogChannel writes delivered=true
+- Daily digest payload shape
+- GET /suggestions, POST accept, POST dismiss, GET /notifications/log
+- FakeImap end-to-end poll
+"""
+
+from __future__ import annotations
+
+import email as email_mod
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.config import DATABASE_URL
+from app.db import repo_app
+from app.imap_watch import (
+    FakeImap,
+    _build_snippet,
+    _extract_sender_domain,
+    _parse_email_message,
+    classify_email,
+    create_email_suggestion,
+    is_duplicate,
+    match_application,
+    poll_inbox,
+)
+from app.notify import (
+    LogChannel,
+    WebhookChannel,
+    get_channels,
+    list_notification_log,
+    reset_channels,
+    send_notification,
+    set_channels,
+)
+
+
+@pytest.fixture()
+def client():
+    from app.main import app
+    return TestClient(app)
+
+
+# --- Helpers ---
+
+def _make_email(raw_from: str, subject: str, body: str, date: str = "") -> bytes:
+    """Build raw email bytes for FakeImap."""
+    msg = email_mod.message_from_string(
+        f"From: {raw_from}\r\n"
+        f"Subject: {subject}\r\n"
+        f"Date: {date or 'Mon, 01 Jul 2026 10:00:00 +0000'}\r\n"
+        f"\r\n"
+        f"{body}"
+    )
+    return msg.as_bytes()
+
+
+def _create_app_in_state(state: str = "sent", company: str = "TechCorp", url: str = "https://techcorp.com/jobs/1") -> dict:
+    """Create a posting + application, force state via SQL."""
+    posting = repo_app.create_job_posting(
+        source="manual_url",
+        url=url,
+        company=company,
+        title="Senior Python Developer",
+        location="Malmo",
+        description="",
+        raw={},
+    )
+    app_row = repo_app.create_application(posting["id"])
+    if state != "discovered":
+        repo_app.update_application_score(app_row["id"], 80, {"factors": {}})
+        if state in ("approved", "rejected"):
+            repo_app.update_application_state(app_row["id"], "scored")
+            repo_app.update_application_state(app_row["id"], state)
+        elif state == "sent":
+            repo_app.update_application_state(app_row["id"], "scored")
+            repo_app.update_application_state(app_row["id"], "approved")
+            repo_app.update_application_state(app_row["id"], "drafting")
+            # Bypass guard for test
+            from app.db import execute
+            execute(
+                "UPDATE application SET state = 'sent', last_activity_at = now() WHERE id = %s",
+                (app_row["id"],),
+            )
+        elif state == "interviewing":
+            repo_app.update_application_state(app_row["id"], "scored")
+            repo_app.update_application_state(app_row["id"], "approved")
+            repo_app.update_application_state(app_row["id"], "drafting")
+            from app.db import execute
+            execute(
+                "UPDATE application SET state = 'sent' WHERE id = %s",
+                (app_row["id"],),
+            )
+            execute(
+                "UPDATE application SET state = 'interviewing' WHERE id = %s",
+                (app_row["id"],),
+            )
+    return app_row
+
+
+# ========================================================================
+# IMAP matching logic (3 tests)
+# ========================================================================
+
+class TestImapMatching:
+    def test_match_by_company_name_in_subject(self):
+        """Email subject contains company name -> match."""
+        apps = [
+            {"id": "app1", "state": "sent", "company": "TechCorp", "title": "Python Dev", "url": "https://example.com"},
+        ]
+        result = match_application(
+            "recruiter@gmail.com",
+            "Interview at TechCorp",
+            "Please come for an interview.",
+            apps,
+        )
+        assert result is not None
+        assert result["id"] == "app1"
+
+    def test_match_by_sender_domain_in_url(self):
+        """Sender domain matches the posting URL -> match."""
+        apps = [
+            {"id": "app2", "state": "sent", "company": "Unknown", "title": "Dev", "url": "https://techcorp.com/careers/1"},
+        ]
+        result = match_application(
+            "hr@techcorp.com",
+            "Your application",
+            "We reviewed your application.",
+            apps,
+        )
+        assert result is not None
+        assert result["id"] == "app2"
+
+    def test_match_by_title_keywords(self):
+        """Email subject contains 2+ title words -> match."""
+        apps = [
+            {"id": "app3", "state": "interviewing", "company": "SomeCompany", "title": "Senior Python Developer", "url": "https://other.com"},
+        ]
+        result = match_application(
+            "someone@other.com",
+            "Senior Python position update",
+            "Regarding the developer role.",
+            apps,
+        )
+        assert result is not None
+        assert result["id"] == "app3"
+
+    def test_no_match_wrong_state(self):
+        """Applications in discovered state are not matched."""
+        apps = [
+            {"id": "app4", "state": "discovered", "company": "TechCorp", "title": "Dev", "url": "https://techcorp.com"},
+        ]
+        result = match_application(
+            "hr@techcorp.com",
+            "Interview at TechCorp",
+            "Come for an interview.",
+            apps,
+        )
+        assert result is None
+
+    def test_no_match_unrelated_email(self):
+        """Email unrelated to any application -> no match."""
+        apps = [
+            {"id": "app5", "state": "sent", "company": "TechCorp", "title": "Dev", "url": "https://techcorp.com"},
+        ]
+        result = match_application(
+            "newsletter@spam.com",
+            "Buy now!",
+            "Special offer for you.",
+            apps,
+        )
+        assert result is None
+
+
+# ========================================================================
+# Classifier -> row (3 tests)
+# ========================================================================
+
+class TestClassifierToRow:
+    def test_classify_returns_interview_invite(self):
+        """classify_email returns interview_invite from mock."""
+        result = classify_email("Interview invitation", "Please come for an interview next week.")
+        assert result["classification"] == "interview_invite"
+        assert result["state_proposal"] == "interviewing"
+        assert "reason" in result
+
+    def test_classify_falls_back_on_invalid_classification(self):
+        """Invalid classification from LLM falls back to noise."""
+        import app.llm as llm_mod
+        original = llm_mod.MOCK_OUTPUTS.get("email_classify", {}).copy()
+        try:
+            llm_mod.MOCK_OUTPUTS["email_classify"] = {"classification": "bogus", "state_proposal": None, "reason": "test"}
+            result = classify_email("test", "test")
+            assert result["classification"] == "noise"
+        finally:
+            llm_mod.MOCK_OUTPUTS["email_classify"] = original
+
+    def test_create_email_suggestion_row(self):
+        """create_email_suggestion inserts a row correctly."""
+        now = datetime.now(timezone.utc)
+        suggestion = create_email_suggestion(
+            application_id=None,
+            mailbox_from="hr@example.com",
+            subject="Interview invite",
+            snippet="Please come for an interview.",
+            classification="interview_invite",
+            state_proposal="interviewing",
+            received_at=now,
+        )
+        assert suggestion["id"] is not None
+        assert suggestion["mailbox_from"] == "hr@example.com"
+        assert suggestion["classification"] == "interview_invite"
+        assert suggestion["status"] == "pending"
+        assert suggestion["state_proposal"] == "interviewing"
+
+
+# ========================================================================
+# Noise dedupe (3 tests)
+# ========================================================================
+
+class TestNoiseDedupe:
+    def test_duplicate_detected_same_day(self):
+        """Same from+subject+day is flagged as duplicate."""
+        now = datetime.now(timezone.utc)
+        create_email_suggestion(
+            application_id=None,
+            mailbox_from="hr@example.com",
+            subject="Interview",
+            snippet="Come for an interview.",
+            classification="interview_invite",
+            state_proposal="interviewing",
+            received_at=now,
+        )
+        assert is_duplicate("hr@example.com", "Interview", now)
+
+    def test_different_subject_not_duplicate(self):
+        """Different subject -> not duplicate."""
+        now = datetime.now(timezone.utc)
+        create_email_suggestion(
+            application_id=None,
+            mailbox_from="hr@example.com",
+            subject="Interview",
+            snippet="Come.",
+            classification="interview_invite",
+            state_proposal="interviewing",
+            received_at=now,
+        )
+        assert not is_duplicate("hr@example.com", "Different Subject", now)
+
+    def test_different_sender_not_duplicate(self):
+        """Different sender -> not duplicate."""
+        now = datetime.now(timezone.utc)
+        create_email_suggestion(
+            application_id=None,
+            mailbox_from="hr@example.com",
+            subject="Interview",
+            snippet="Come.",
+            classification="interview_invite",
+            state_proposal="interviewing",
+            received_at=now,
+        )
+        assert not is_duplicate("other@example.com", "Interview", now)
+
+
+# ========================================================================
+# Accept applies transition through guard (3 tests)
+# ========================================================================
+
+class TestAcceptSuggestion:
+    def test_accept_applies_transition(self, client):
+        """POST /suggestions/{id}/accept transitions app from sent to interviewing."""
+        app_row = _create_app_in_state("sent", company="TechCorp")
+        suggestion = create_email_suggestion(
+            application_id=app_row["id"],
+            mailbox_from="hr@techcorp.com",
+            subject="Interview at TechCorp",
+            snippet="Please come in for an interview.",
+            classification="interview_invite",
+            state_proposal="interviewing",
+            received_at=datetime.now(timezone.utc),
+        )
+
+        resp = client.post(f"/api/suggestions/{suggestion['id']}/accept")
+        assert resp.status_code == 200
+        assert resp.json()["status"] == "accepted"
+
+        # Verify application state changed
+        updated_app = repo_app.get_application(app_row["id"])
+        assert updated_app["state"] == "interviewing"
+
+    def test_accept_invalid_transition_409(self, client):
+        """Accept with invalid transition (e.g. discovered -> interviewing) returns 409."""
+        posting = repo_app.create_job_posting(
+            source="manual_url", url="https://example.com/bad/1",
+            company="X", title="X", location=None, description="", raw={},
+        )
+        app_row = repo_app.create_application(posting["id"])
+
+        suggestion = create_email_suggestion(
+            application_id=app_row["id"],
+            mailbox_from="hr@example.com",
+            subject="Test",
+            snippet="Test",
+            classification="interview_invite",
+            state_proposal="interviewing",
+            received_at=datetime.now(timezone.utc),
+        )
+
+        resp = client.post(f"/api/suggestions/{suggestion['id']}/accept")
+        assert resp.status_code == 409
+        assert "invalid_transition" in str(resp.json()["detail"])
+
+    def test_accept_404_nonexistent(self, client):
+        """Accept on nonexistent suggestion -> 404."""
+        resp = client.post("/api/suggestions/00000000-0000-0000-0000-000000000000/accept")
+        assert resp.status_code == 404
+
+    def test_accept_already_accepted_409(self, client):
+        """Accept on already accepted suggestion -> 409."""
+        app_row = _create_app_in_state("sent")
+        suggestion = create_email_suggestion(
+            application_id=app_row["id"],
+            mailbox_from="hr@example.com",
+            subject="Test",
+            snippet="Test",
+            classification="interview_invite",
+            state_proposal="interviewing",
+            received_at=datetime.now(timezone.utc),
+        )
+        client.post(f"/api/suggestions/{suggestion['id']}/accept")
+        resp = client.post(f"/api/suggestions/{suggestion['id']}/accept")
+        assert resp.status_code == 409
+
+
+# ========================================================================
+# Dismiss suggestion (2 tests)
+# ========================================================================
+
+class TestDismissSuggestion:
+    def test_dismiss_marks_as_dismissed(self, client):
+        """POST /suggestions/{id}/dismiss marks as dismissed."""
+        suggestion = create_email_suggestion(
+            application_id=None,
+            mailbox_from="hr@example.com",
+            subject="Test",
+            snippet="Test",
+            classification="question",
+            state_proposal=None,
+            received_at=datetime.now(timezone.utc),
+        )
+
+        resp = client.post(f"/api/suggestions/{suggestion['id']}/dismiss")
+        assert resp.status_code == 200
+        assert resp.json()["status"] == "dismissed"
+
+    def test_dismiss_404_nonexistent(self, client):
+        """Dismiss nonexistent -> 404."""
+        resp = client.post("/api/suggestions/00000000-0000-0000-0000-000000000000/dismiss")
+        assert resp.status_code == 404
+
+
+# ========================================================================
+# GET /suggestions (2 tests)
+# ========================================================================
+
+class TestGetSuggestions:
+    def test_get_suggestions_returns_pending(self, client):
+        """GET /suggestions returns only pending suggestions."""
+        create_email_suggestion(
+            application_id=None,
+            mailbox_from="a@example.com",
+            subject="Subject A",
+            snippet="Snippet A",
+            classification="interview_invite",
+            state_proposal="interviewing",
+            received_at=datetime.now(timezone.utc),
+        )
+        create_email_suggestion(
+            application_id=None,
+            mailbox_from="b@example.com",
+            subject="Subject B",
+            snippet="Snippet B",
+            classification="rejection",
+            state_proposal=None,
+            received_at=datetime.now(timezone.utc),
+        )
+
+        resp = client.get("/api/suggestions")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert len(data) == 2
+        assert all(s["status"] == "pending" for s in data)
+
+    def test_get_suggestions_empty(self, client):
+        """GET /suggestions returns empty list when no suggestions."""
+        resp = client.get("/api/suggestions")
+        assert resp.status_code == 200
+        assert resp.json() == []
+
+
+# ========================================================================
+# Notification channels (6 tests)
+# ========================================================================
+
+class TestLogChannel:
+    def test_log_channel_writes_delivered_true(self):
+        """LogChannel writes notification_log with delivered=true."""
+        reset_channels()
+        ch = LogChannel()
+        result = ch.send("daily_digest", "Test digest", {"count": 5})
+        assert result is True
+
+        logs = list_notification_log(limit=10)
+        assert len(logs) >= 1
+        latest = logs[0]
+        assert latest["channel"] == "log"
+        assert latest["kind"] == "daily_digest"
+        assert latest["delivered"] is True
+        assert latest["error"] is None
+
+    def test_send_notification_log_channel(self):
+        """send_notification via LogChannel creates a log entry."""
+        reset_channels()
+        set_channels([LogChannel()])
+        send_notification("email_suggestion", "Interview invite from TechCorp", {"id": "test"})
+        logs = list_notification_log(limit=10)
+        assert len(logs) >= 1
+        assert logs[0]["kind"] == "email_suggestion"
+        assert logs[0]["delivered"] is True
+        reset_channels()
+
+
+class TestWebhookChannel:
+    def test_webhook_success_2xx(self, monkeypatch):
+        """WebhookChannel with 2xx response writes delivered=true."""
+        class FakeResponse:
+            status_code = 200
+            text = "OK"
+
+        class FakeClient:
+            @staticmethod
+            def post(url, json=None, timeout=None):
+                assert url == "https://hook.example.com/notify"
+                return FakeResponse()
+
+        import httpx
+        monkeypatch.setattr(httpx, "post", FakeClient.post)
+
+        reset_channels()
+        ch = WebhookChannel(url="https://hook.example.com/notify")
+        result = ch.send("daily_digest", "Digest", {"count": 3})
+        assert result is True
+
+        logs = list_notification_log(limit=10)
+        latest = logs[0]
+        assert latest["channel"] == "webhook"
+        assert latest["delivered"] is True
+        assert latest["error"] is None
+        reset_channels()
+
+    def test_webhook_failure_non_2xx(self, monkeypatch):
+        """WebhookChannel with non-2xx writes delivered=false with error."""
+        class FakeResponse:
+            status_code = 500
+            text = "Internal Server Error"
+
+        class FakeClient:
+            @staticmethod
+            def post(url, json=None, timeout=None):
+                return FakeResponse()
+
+        import httpx
+        monkeypatch.setattr(httpx, "post", FakeClient.post)
+
+        reset_channels()
+        ch = WebhookChannel(url="https://hook.example.com/notify")
+        result = ch.send("daily_digest", "Digest", {"count": 3})
+        assert result is False
+
+        logs = list_notification_log(limit=10)
+        latest = logs[0]
+        assert latest["channel"] == "webhook"
+        assert latest["delivered"] is False
+        assert latest["error"] is not None
+        assert "500" in latest["error"]
+        reset_channels()
+
+    def test_webhook_exception_writes_error(self, monkeypatch):
+        """WebhookChannel with connection exception writes delivered=false."""
+        def raise_exc(url, json=None, timeout=None):
+            raise ConnectionError("Connection refused")
+
+        import httpx
+        monkeypatch.setattr(httpx, "post", raise_exc)
+
+        reset_channels()
+        ch = WebhookChannel(url="https://hook.example.com/notify")
+        result = ch.send("daily_digest", "Digest", {"count": 1})
+        assert result is False
+
+        logs = list_notification_log(limit=10)
+        latest = logs[0]
+        assert latest["delivered"] is False
+        assert "Connection refused" in (latest["error"] or "")
+        reset_channels()
+
+    def test_webhook_no_url_writes_error(self):
+        """WebhookChannel with no URL writes delivered=false with config error."""
+        reset_channels()
+        ch = WebhookChannel(url="")
+        result = ch.send("test", "test", {})
+        assert result is False
+
+        logs = list_notification_log(limit=10)
+        latest = logs[0]
+        assert latest["delivered"] is False
+        assert "not configured" in (latest["error"] or "")
+        reset_channels()
+
+
+# ========================================================================
+# Notification log endpoint (2 tests)
+# ========================================================================
+
+class TestNotificationLogEndpoint:
+    def test_get_notifications_log(self, client):
+        """GET /notifications/log returns entries."""
+        reset_channels()
+        set_channels([LogChannel()])
+        send_notification("daily_digest", "Test", {"count": 1})
+        reset_channels()
+
+        resp = client.get("/api/notifications/log")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert len(data) >= 1
+        assert "channel" in data[0]
+        assert "kind" in data[0]
+        assert "delivered" in data[0]
+
+    def test_get_notifications_log_empty(self, client):
+        """GET /notifications/log returns empty when no entries."""
+        resp = client.get("/api/notifications/log")
+        assert resp.status_code == 200
+        assert resp.json() == []
+
+
+# ========================================================================
+# Digest payload shape (2 tests)
+# ========================================================================
+
+class TestDigestPayload:
+    def test_daily_digest_notification_text(self):
+        """Daily digest notification contains expected text fields."""
+        reset_channels()
+        set_channels([LogChannel()])
+        send_notification("daily_digest", "Daily Digest\nScored applications: 5", {
+            "digest_count": 5,
+            "nudge_count": 2,
+            "pending_approvals": 1,
+        })
+        logs = list_notification_log(limit=10)
+        entry = logs[0]
+        assert entry["kind"] == "daily_digest"
+        payload = entry["payload"]
+        assert "text" in payload
+        assert "Daily Digest" in payload["text"]
+        assert payload.get("digest_count") == 5
+        reset_channels()
+
+    def test_daily_digest_payload_has_counts(self):
+        """Digest payload includes digest_count, nudge_count, pending_approvals."""
+        reset_channels()
+        set_channels([LogChannel()])
+        send_notification("daily_digest", "text", {
+            "digest_count": 3,
+            "nudge_count": 1,
+            "pending_approvals": 0,
+        })
+        logs = list_notification_log(limit=10)
+        payload = logs[0]["payload"]
+        assert payload["digest_count"] == 3
+        assert payload["nudge_count"] == 1
+        assert payload["pending_approvals"] == 0
+        reset_channels()
+
+
+# ========================================================================
+# FakeImap end-to-end poll (3 tests)
+# ========================================================================
+
+class TestFakeImapPoll:
+    def test_poll_creates_suggestion_for_matching_email(self):
+        """FakeImap poll creates a suggestion when email matches an application."""
+        app_row = _create_app_in_state("sent", company="TechCorp", url="https://techcorp.com/jobs/1")
+
+        raw_email = _make_email(
+            "hr@techcorp.com",
+            "Interview at TechCorp",
+            "Please come for an interview next Tuesday.",
+        )
+        fake = FakeImap(messages=[(b"1", raw_email)])
+        created = poll_inbox(fake)
+        assert len(created) == 1
+        assert created[0]["classification"] == "interview_invite"
+        assert created[0]["mailbox_from"] == "hr@techcorp.com"
+        assert created[0]["application_id"] == app_row["id"]
+
+    def test_poll_skips_noise_emails(self):
+        """FakeImap poll skips noise classification (no suggestion created)."""
+        _create_app_in_state("sent", company="TechCorp")
+
+        # Mock email_classify to return noise
+        import app.llm as llm_mod
+        original = llm_mod.MOCK_OUTPUTS.get("email_classify", {}).copy()
+        try:
+            llm_mod.MOCK_OUTPUTS["email_classify"] = {
+                "classification": "noise",
+                "state_proposal": None,
+                "reason": "spam",
+            }
+            raw_email = _make_email(
+                "newsletter@spam.com",
+                "Buy our product",
+                "Special offer just for you!",
+            )
+            fake = FakeImap(messages=[(b"1", raw_email)])
+            created = poll_inbox(fake)
+            assert len(created) == 0
+        finally:
+            llm_mod.MOCK_OUTPUTS["email_classify"] = original
+
+    def test_poll_dedupe_skips_same_from_subject_day(self):
+        """FakeImap poll dedupes same from+subject+day."""
+        raw_email = _make_email(
+            "hr@example.com",
+            "Same Subject",
+            "Same body content.",
+        )
+        # First poll
+        fake1 = FakeImap(messages=[(b"1", raw_email)])
+        created1 = poll_inbox(fake1)
+        assert len(created1) == 1
+
+        # Second poll with same message -> dedupe
+        fake2 = FakeImap(messages=[(b"1", raw_email)])
+        created2 = poll_inbox(fake2)
+        assert len(created2) == 0  # deduped
+
+
+# ========================================================================
+# Email parsing helpers (2 tests)
+# ========================================================================
+
+class TestEmailParsing:
+    def test_extract_sender_domain(self):
+        """Extract domain from From header."""
+        assert _extract_sender_domain("John Doe <hr@techcorp.com>") == "techcorp.com"
+        assert _extract_sender_domain("noreply@example.org") == "example.org"
+        assert _extract_sender_domain("") == ""
+
+    def test_build_snippet_truncates(self):
+        """Snippet is truncated to max_len."""
+        long_body = "A" * 500
+        snippet = _build_snippet(long_body, max_len=50)
+        assert len(snippet) <= 53  # 50 + "..."
+        assert snippet.endswith("...")
+
+    def test_build_snippet_short_body(self):
+        """Short body is not truncated."""
+        snippet = _build_snippet("Hello", max_len=300)
+        assert snippet == "Hello"