WA1: migration 003, imap_watch, notify, scheduler jobs, suggestions+notifications endpoints, email_classify mock

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
             """