diff --git a/apps/api/app/db/migrate.py b/apps/api/app/db/migrate.py index 2e06910..ed642df 100644 --- a/apps/api/app/db/migrate.py +++ b/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 """ diff --git a/apps/api/app/imap_watch.py b/apps/api/app/imap_watch.py new file mode 100644 index 0000000..28f7090 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/apps/api/app/llm.py b/apps/api/app/llm.py index c9c1515..7b9bc84 100644 --- a/apps/api/app/llm.py +++ b/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.", + }, } diff --git a/apps/api/app/main.py b/apps/api/app/main.py index b5857ee..b3ae3df 100644 --- a/apps/api/app/main.py +++ b/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, - } \ No newline at end of file + } + + +# --- 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) \ No newline at end of file diff --git a/apps/api/app/notify.py b/apps/api/app/notify.py new file mode 100644 index 0000000..48c312a --- /dev/null +++ b/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] \ No newline at end of file diff --git a/apps/api/app/scheduler.py b/apps/api/app/scheduler.py index 45de8bc..74748ce 100644 --- a/apps/api/app/scheduler.py +++ b/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: diff --git a/apps/api/app/schemas.py b/apps/api/app/schemas.py index 47b746d..30acad0 100644 --- a/apps/api/app/schemas.py +++ b/apps/api/app/schemas.py @@ -319,4 +319,31 @@ class SeedDemoResponse(BaseModel): profile: str postings: int applications: int - sections: int \ No newline at end of file + 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 \ No newline at end of file diff --git a/apps/api/migrations/003_email_notify.sql b/apps/api/migrations/003_email_notify.sql new file mode 100644 index 0000000..817fc63 --- /dev/null +++ b/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() +); \ No newline at end of file diff --git a/apps/api/tests/conftest.py b/apps/api/tests/conftest.py index 2235f0a..303e767 100644 --- a/apps/api/tests/conftest.py +++ b/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 """ diff --git a/apps/api/tests/test_v11_email_notify.py b/apps/api/tests/test_v11_email_notify.py new file mode 100644 index 0000000..3bce5d5 --- /dev/null +++ b/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 ") == "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" \ No newline at end of file