Initial commit: Howl M365 email management daemon

Full implementation of the Howl email triage system: - Microsoft Graph API integration with MSAL auth (client-credentials and delegated modes) - Claude LLM classification via tool use for structured output - PostgreSQL database with customers, vendors, whitelist, and email_log tables - Alembic migration for full schema - APScheduler daemon with graceful shutdown - Typer CLI (run, dry-run, status commands) - Business rule classifier with overrides (whitelist protection, low-confidence fallback) - Action executor (move to folders, flag, escalate with webhook) - 35 passing unit tests - README, SETUP, and NEXT_STEPS documentation Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-01 16:02:24 -04:00 · 2026-04-01 16:02:24 -04:00 · 3bfda9e585
commit 3bfda9e585
39 changed files with 3656 additions and 0 deletions
--- a/.env.example
+++ b/.env.example
@ -0,0 +1,60 @@
+# =============================================================
+# Howl - M365 Email Management Daemon
+# Copy this file to .env and fill in your values.
+# =============================================================
+
+# --- Microsoft Azure / Graph ---
+AZURE_TENANT_ID=your-tenant-id
+AZURE_CLIENT_ID=your-app-client-id
+AZURE_CLIENT_SECRET=your-app-client-secret   # Required for client_credentials mode only
+
+# Auth mode: client_credentials (shared/app-only) or delegated (personal mailbox)
+GRAPH_AUTH_MODE=client_credentials
+
+# Path to MSAL token cache (delegated mode only — persists refresh token across restarts)
+MSAL_TOKEN_CACHE_PATH=.msal_cache.bin
+
+# Mailbox to monitor (UPN or shared mailbox address)
+GRAPH_MAILBOX=shared@yourcompany.com
+
+# Poll interval in seconds (default: 5 minutes)
+GRAPH_POLL_INTERVAL_SECONDS=300
+
+# Number of messages to fetch per poll cycle
+GRAPH_BATCH_SIZE=50
+
+# Max characters of email body sent to LLM (truncated beyond this)
+GRAPH_MAX_BODY_CHARS=4000
+
+# --- Anthropic ---
+ANTHROPIC_API_KEY=sk-ant-your-key-here
+ANTHROPIC_MODEL=claude-sonnet-4-6
+ANTHROPIC_MAX_TOKENS=1024
+
+# Confidence threshold: below this value → inbox_keep + requires_human_review
+LLM_CONFIDENCE_THRESHOLD=0.60
+
+# --- PostgreSQL ---
+DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/howl
+DATABASE_POOL_SIZE=5
+DATABASE_POOL_MAX_OVERFLOW=10
+
+# --- Processing ---
+# Set to true to analyze emails but take no actions (safe for testing)
+DRY_RUN=false
+MAX_RETRY_COUNT=3
+
+# Optional: Teams or Slack incoming webhook URL for escalation notifications
+NOTIFICATION_WEBHOOK_URL=
+
+# --- Mail Folder Names (Graph display names) ---
+FOLDER_CUSTOMERS=Customers
+FOLDER_VENDORS=Vendors
+FOLDER_REVIEW=Needs Review
+FOLDER_SPAM=Junk Email
+FOLDER_ESCALATE=Escalate
+
+# --- Logging ---
+LOG_LEVEL=INFO
+# json (structured, for production) or text (human-readable, for development)
+LOG_FORMAT=json
--- a/.gitignore
+++ b/.gitignore
@ -0,0 +1,11 @@
+.env
+.msal_cache.bin
+__pycache__/
+*.pyc
+*.egg-info/
+dist/
+.venv/
+venv/
+.pytest_cache/
+.coverage
+htmlcov/
--- a/NEXT_STEPS.md
+++ b/NEXT_STEPS.md
@ -0,0 +1,90 @@
+# Next Steps & Roadmap
+
+## Immediate (before going live)
+
+### 1. Azure app registration
+Follow `SETUP.md` to register an app in Azure AD and grant the required Graph API permissions. This is a prerequisite for any real email processing.
+
+### 2. Seed your database
+Populate `customers`, `customer_emails`, `vendors`, `vendor_emails`, and `whitelist` with your real contact data before running the daemon. The more complete this data is, the better the LLM's classifications will be.
+
+### 3. Run a dry-run validation
+```bash
+DRY_RUN=true howl dry-run --limit 20
+```
+Review the `email_log` table. Check that:
+- Known customers/vendors are being matched correctly
+- LLM classifications feel accurate for your email types
+- Confidence scores make sense (low scores = good candidates for whitelist/notes improvements)
+
+### 4. Tune confidence threshold
+The default `LLM_CONFIDENCE_THRESHOLD=0.60` sends uncertain emails to `Needs Review`. Monitor the `Needs Review` folder for the first week and adjust:
+- Too many emails going there → lower the threshold or add more DB records
+- Wrong actions being taken → raise the threshold
+
+---
+
+## Short-term improvements
+
+### Admin UI for the email log
+A simple read-only web view (Flask or FastAPI) to browse `email_log` entries, see LLM reasoning, and mark records as reviewed. Currently you need `psql` or `howl status` to inspect results.
+
+### CSV / bulk import for contacts
+A `howl import-customers customers.csv` and `howl import-vendors vendors.csv` CLI command to populate the database from spreadsheet exports. Would accept columns like `name, company, email, notes`.
+
+### Human review workflow
+A way to mark emails in the `Needs Review` folder as approved/rejected after human inspection, writing back to `email_log.action_overridden`. This data could seed a feedback loop for improving prompts.
+
+### Multiple mailbox support
+Currently `GRAPH_MAILBOX` is a single address. Supporting a list of mailboxes (e.g., `info@`, `support@`, `billing@`) would let one daemon instance manage several inboxes, each potentially with different folder structures.
+
+### Auto-reply drafting
+For emails classified as `customer_inquiry`, have Claude draft a suggested reply and save it as a draft in M365 (Graph supports creating draft messages). A human reviews and sends — no risk of auto-sending.
+
+---
+
+## Medium-term improvements
+
+### Domain-based routing rules
+Beyond exact email address matching, support routing rules like:
+- All `@acme.com` email → Customers folder regardless of LLM classification
+- All `@noreply.*` addresses → skip LLM, move directly to Vendors
+
+### Feedback loop / fine-tuning data collection
+Export `email_log` rows where `action_overridden=true` as training/evaluation data. Use these to improve prompts or evaluate whether a switch in model or threshold would reduce overrides.
+
+### Attachment handling
+Currently only the email body is analysed. Adding attachment type awareness (e.g., flagging `.pdf` attachments from vendors as likely invoices) would improve invoice detection accuracy.
+
+### Metrics and alerting
+Export processing counts, failure rates, and LLM token usage to a metrics system (Prometheus, Datadog, etc.). Alert on elevated failure rates or processing backlogs.
+
+### Scheduled summaries
+A daily digest email or Teams/Slack message summarising: emails processed, actions taken, items needing review. Useful for staying aware of daemon activity without querying the database.
+
+---
+
+## Database considerations
+
+### Adding indexes for reporting queries
+As `email_log` grows, queries by date range or `sender_type` will benefit from additional indexes:
+```sql
+CREATE INDEX idx_email_log_type_date ON email_log (sender_type, received_at DESC);
+CREATE INDEX idx_email_log_classification ON email_log (llm_classification);
+```
+
+### Archiving old records
+After 90 days, `email_log` rows for completed messages can be moved to a cold archive table or exported to object storage to keep the active table fast.
+
+### Contact data enrichment
+Add fields to `customers` and `vendors` for priority tier, account manager, and SLA level. Pass these to the LLM prompt so high-priority customers get escalated rather than just moved to the folder.
+
+---
+
+## Security considerations
+
+- **Rotate the Azure client secret** every 6–12 months and update `.env`
+- **Restrict `.env` and `.msal_cache.bin` permissions**: `chmod 600 .env .msal_cache.bin`
+- **Do not log email bodies** — the current implementation only logs metadata; ensure `LOG_LEVEL=DEBUG` is not used in production as it may expose content
+- **Review whitelist entries regularly** — expired entries are automatically ignored, but stale permanent entries should be audited
+- **Monitor for phishing mis-classifications** — review the `Junk Email` folder periodically to check that legitimate email isn't being over-aggressively filtered
--- a/README.md
+++ b/README.md
@ -0,0 +1,125 @@
+# Howl
+
+An M365 email management daemon that uses the Microsoft Graph API and Claude AI to automatically classify and route incoming email based on your customer, vendor, and whitelist database.
+
+## How it works
+
+```
+Inbox (M365)
+    │
+    ▼
+Graph API polls for unread mail
+    │
+    ▼
+PostgreSQL lookup — is the sender a customer, vendor, or trusted contact?
+    │
+    ▼
+Claude (LLM) classifies the email using sender context + body content
+    │
+    ▼
+Business rules applied (whitelist always wins, low-confidence → hold for review)
+    │
+    ▼
+Action taken: move to folder / flag / escalate with notification
+    │
+    ▼
+email_log table records every decision for audit and retry
+```
+
+## Features
+
+- Supports both **shared mailboxes** (app-only, no user login) and **personal mailboxes** (delegated device-code auth)
+- **No emails are ever deleted** — only moved, flagged, or escalated
+- **Dry-run mode** — analyze and log without touching any email
+- **Idempotent** — safe to restart; already-processed messages are skipped
+- **Automatic retry** — failed messages are retried on the next poll cycle
+- **Full audit log** — every classification, confidence score, and action is stored in PostgreSQL
+
+## Folder routing
+
+| Sender type | LLM classification | Action |
+|---|---|---|
+| Customer | Inquiry / correspondence | Move to `Customers` folder |
+| Vendor | Invoice / notification | Move to `Vendors` folder |
+| Whitelist | Trusted contact | Mark read, stay in Inbox |
+| Any | Urgent / escalation | Move to `Escalate` + webhook notification |
+| Any | Uncertain (low confidence) | Move to `Needs Review` |
+| Unknown | Spam / newsletter | Move to `Junk Email` |
+
+## Quick start
+
+```bash
+# 1. Create and activate a virtual environment
+python3 -m venv .venv && source .venv/bin/activate
+
+# 2. Install dependencies
+pip install -e ".[dev]"
+
+# 3. Configure environment
+cp .env.example .env
+# Edit .env with your Azure, Anthropic, and PostgreSQL credentials
+
+# 4. Run database migrations
+alembic upgrade head
+
+# 5. Test with dry-run (no email mutations)
+howl dry-run --limit 10
+
+# 6. Start the daemon
+howl run
+```
+
+## CLI commands
+
+| Command | Description |
+|---|---|
+| `howl run` | Start the daemon (blocking, polls on schedule) |
+| `howl dry-run [--limit N]` | Classify up to N emails, log results, take no action |
+| `howl status [--limit N]` | Display recent email_log entries in a table |
+
+## Project structure
+
+```
+howl/
+├── howl/
+│   ├── config.py           # All configuration (pydantic-settings + .env)
+│   ├── main.py             # CLI entry point (Typer)
+│   ├── logging_config.py   # structlog setup (JSON or text)
+│   ├── db/
+│   │   ├── engine.py       # Async SQLAlchemy engine
+│   │   ├── models.py       # ORM models
+│   │   └── queries.py      # lookup_sender(), email log helpers
+│   ├── graph/
+│   │   ├── auth.py         # MSAL token provider (client-credentials or delegated)
+│   │   └── client.py       # Graph API: list, read, move, flag messages
+│   ├── llm/
+│   │   ├── client.py       # Anthropic SDK wrapper with retry
+│   │   ├── prompts.py      # Prompt builder (system prompt + per-email user turn)
+│   │   └── schemas.py      # EmailClassification model + tool definition
+│   ├── pipeline/
+│   │   ├── processor.py    # EmailProcessor — orchestrates the full pipeline
+│   │   ├── classifier.py   # Business rule overrides on LLM output
+│   │   └── actions.py      # ActionExecutor — Graph mutations + webhook
+│   └── daemon/
+│       └── scheduler.py    # APScheduler poll loop + graceful shutdown
+├── migrations/             # Alembic database migrations
+├── tests/                  # Unit and integration tests
+├── .env.example            # Configuration template
+├── SETUP.md                # Detailed setup and configuration guide
+└── NEXT_STEPS.md           # Roadmap and future improvements
+```
+
+## Running tests
+
+```bash
+pytest tests/ -q                          # All tests (no external services needed)
+pytest tests/ -q -m "not integration"    # Unit tests only (default)
+pytest tests/ -q -m integration          # Requires real M365 + Anthropic credentials
+```
+
+## Requirements
+
+- Python 3.11+
+- PostgreSQL 14+
+- Microsoft Azure app registration with `Mail.Read` + `Mail.ReadWrite` permissions
+- Anthropic API key
--- a/SETUP.md
+++ b/SETUP.md
@ -0,0 +1,275 @@
+# Setup Guide
+
+## Prerequisites
+
+- Python 3.11+
+- PostgreSQL 14+ database (local or hosted)
+- Microsoft 365 account with an Azure AD app registration
+- Anthropic API key
+
+---
+
+## 1. Azure AD App Registration
+
+### For a shared mailbox (recommended for production)
+
+This uses **application permissions** — the daemon runs without any user logged in.
+
+1. Go to [portal.azure.com](https://portal.azure.com) → **Azure Active Directory** → **App registrations** → **New registration**
+2. Name it something like `howl-email-daemon`; leave the redirect URI blank
+3. After creation, note the **Application (client) ID** and **Directory (tenant) ID**
+4. Go to **Certificates & secrets** → **New client secret** → copy the value immediately
+5. Go to **API permissions** → **Add a permission** → **Microsoft Graph** → **Application permissions**
+   - Add `Mail.Read`
+   - Add `Mail.ReadWrite`
+6. Click **Grant admin consent** (requires a Global Admin)
+
+Set in `.env`:
+```
+GRAPH_AUTH_MODE=client_credentials
+AZURE_TENANT_ID=<Directory (tenant) ID>
+AZURE_CLIENT_ID=<Application (client) ID>
+AZURE_CLIENT_SECRET=<client secret value>
+GRAPH_MAILBOX=shared@yourcompany.com
+```
+
+### For a personal mailbox
+
+This uses **delegated permissions** — the daemon acts on behalf of a user.
+
+1. Follow steps 1–4 above
+2. Go to **API permissions** → **Add a permission** → **Microsoft Graph** → **Delegated permissions**
+   - Add `Mail.Read`
+   - Add `Mail.ReadWrite`
+3. Under **Authentication** → **Advanced settings**, enable **Allow public client flows**
+
+Set in `.env`:
+```
+GRAPH_AUTH_MODE=delegated
+AZURE_TENANT_ID=<Directory (tenant) ID>
+AZURE_CLIENT_ID=<Application (client) ID>
+# No AZURE_CLIENT_SECRET needed for delegated mode
+GRAPH_MAILBOX=yourname@yourcompany.com
+MSAL_TOKEN_CACHE_PATH=.msal_cache.bin
+```
+
+On first `howl run`, you'll be shown a device code URL and prompted to log in. The token is cached to `MSAL_TOKEN_CACHE_PATH` and refreshes automatically.
+
+---
+
+## 2. PostgreSQL Database
+
+Create a database for Howl:
+
+```sql
+CREATE DATABASE howl;
+CREATE USER howl_user WITH PASSWORD 'your-password';
+GRANT ALL PRIVILEGES ON DATABASE howl TO howl_user;
+```
+
+Set in `.env`:
+```
+DATABASE_URL=postgresql+asyncpg://howl_user:your-password@localhost:5432/howl
+```
+
+Run migrations to create all tables:
+
+```bash
+alembic upgrade head
+```
+
+---
+
+## 3. Environment Configuration
+
+Copy the template and fill in your values:
+
+```bash
+cp .env.example .env
+```
+
+### Required settings
+
+| Variable | Description |
+|---|---|
+| `AZURE_TENANT_ID` | Azure AD directory (tenant) ID |
+| `AZURE_CLIENT_ID` | App registration client ID |
+| `AZURE_CLIENT_SECRET` | Client secret (client_credentials mode only) |
+| `GRAPH_MAILBOX` | Email address of the mailbox to monitor |
+| `ANTHROPIC_API_KEY` | Your Anthropic API key |
+| `DATABASE_URL` | PostgreSQL connection string (asyncpg driver) |
+
+### Key optional settings
+
+| Variable | Default | Description |
+|---|---|---|
+| `GRAPH_AUTH_MODE` | `client_credentials` | `client_credentials` or `delegated` |
+| `GRAPH_POLL_INTERVAL_SECONDS` | `300` | How often to check for new mail (seconds) |
+| `GRAPH_BATCH_SIZE` | `50` | Messages fetched per poll cycle |
+| `GRAPH_MAX_BODY_CHARS` | `4000` | Email body characters sent to LLM (truncated beyond this) |
+| `ANTHROPIC_MODEL` | `claude-sonnet-4-6` | Claude model to use |
+| `LLM_CONFIDENCE_THRESHOLD` | `0.60` | Below this, email goes to Needs Review |
+| `DRY_RUN` | `false` | Set `true` to classify without touching email |
+| `MAX_RETRY_COUNT` | `3` | Retry attempts for failed messages |
+| `NOTIFICATION_WEBHOOK_URL` | _(empty)_ | Teams or Slack webhook for escalation alerts |
+
+### Mail folder names
+
+These are the display names Howl will create/find in the monitored mailbox:
+
+| Variable | Default |
+|---|---|
+| `FOLDER_CUSTOMERS` | `Customers` |
+| `FOLDER_VENDORS` | `Vendors` |
+| `FOLDER_REVIEW` | `Needs Review` |
+| `FOLDER_SPAM` | `Junk Email` |
+| `FOLDER_ESCALATE` | `Escalate` |
+
+Folders are created automatically on first use if they don't already exist.
+
+---
+
+## 4. Populate the Database
+
+Add your contacts before running the daemon. Howl matches sender addresses against these tables to provide context to the LLM.
+
+### Adding customers
+
+```sql
+-- Insert a customer
+INSERT INTO customers (name, company, notes)
+VALUES ('Alice Johnson', 'Acme Corp', 'Key account — priority response required');
+
+-- Attach email addresses
+INSERT INTO customer_emails (customer_id, email_address, is_primary)
+VALUES (
+    (SELECT id FROM customers WHERE name = 'Alice Johnson'),
+    'alice@acme.com',
+    TRUE
+);
+```
+
+### Adding vendors
+
+```sql
+INSERT INTO vendors (name, company, service_category)
+VALUES ('SupplyPro', 'SupplyPro Inc', 'supplier');
+
+INSERT INTO vendor_emails (vendor_id, email_address, is_primary)
+VALUES (
+    (SELECT id FROM vendors WHERE name = 'SupplyPro'),
+    'billing@supplypro.com',
+    TRUE
+);
+```
+
+### Adding whitelist entries
+
+```sql
+-- Exact email address
+INSERT INTO whitelist (email_address, description, added_by)
+VALUES ('audit@partnerfirm.com', 'External audit partner', 'admin');
+
+-- Entire domain (matches any @microsoft.com sender)
+INSERT INTO whitelist (domain, description, added_by)
+VALUES ('microsoft.com', 'Microsoft services', 'admin');
+
+-- Temporary entry (expires in 30 days)
+INSERT INTO whitelist (email_address, description, expires_at, added_by)
+VALUES ('temp@contractor.com', 'Temp contractor access', NOW() + INTERVAL '30 days', 'admin');
+```
+
+---
+
+## 5. Running Howl
+
+### Test first with dry-run
+
+```bash
+# Process up to 10 emails, classify them, but take no action
+howl dry-run --limit 10
+```
+
+Check results:
+```bash
+# See what classifications were made
+howl status --limit 20
+
+# Or query directly
+psql -d howl -c "SELECT sender_address, llm_classification, final_action, status FROM email_log ORDER BY created_at DESC LIMIT 10;"
+```
+
+### Start the daemon
+
+```bash
+howl run
+```
+
+The daemon polls every `GRAPH_POLL_INTERVAL_SECONDS` (default: 5 minutes). Stop it with `Ctrl+C` or `SIGTERM` — it finishes the current batch before exiting.
+
+### Running as a systemd service (Linux)
+
+Create `/etc/systemd/system/howl.service`:
+
+```ini
+[Unit]
+Description=Howl Email Management Daemon
+After=network.target postgresql.service
+
+[Service]
+Type=simple
+User=howl
+WorkingDirectory=/opt/projects/howl
+EnvironmentFile=/opt/projects/howl/.env
+ExecStart=/opt/projects/howl/.venv/bin/howl run
+Restart=on-failure
+RestartSec=10
+
+[Install]
+WantedBy=multi-user.target
+```
+
+```bash
+sudo systemctl enable howl
+sudo systemctl start howl
+sudo journalctl -u howl -f   # Follow logs
+```
+
+---
+
+## 6. Webhook Notifications (optional)
+
+Set `NOTIFICATION_WEBHOOK_URL` to receive alerts when an email is escalated.
+
+### Microsoft Teams
+
+1. In Teams, go to the channel → **...** → **Connectors** → **Incoming Webhook**
+2. Create and copy the webhook URL
+3. Set `NOTIFICATION_WEBHOOK_URL=<webhook URL>`
+
+### Slack
+
+1. In Slack, go to [api.slack.com/apps](https://api.slack.com/apps) → create an app → **Incoming Webhooks**
+2. Activate and add to a channel, copy the URL
+3. Set `NOTIFICATION_WEBHOOK_URL=<webhook URL>`
+
+---
+
+## Troubleshooting
+
+**Authentication error (client_credentials)**
+- Verify the client secret hasn't expired in the Azure portal
+- Ensure admin consent was granted for the application permissions
+
+**401 Unauthorized from Graph**
+- For delegated mode: delete `.msal_cache.bin` and re-authenticate with `howl run`
+- Check that the app registration has the correct permissions for the mailbox
+
+**Messages not being processed**
+- Check `email_log` for `status = 'failed'` rows and read `error_message`
+- Ensure the mailbox address in `GRAPH_MAILBOX` exactly matches the M365 UPN or shared mailbox address
+
+**Low LLM accuracy**
+- Lower `LLM_CONFIDENCE_THRESHOLD` to move more edge cases to `Needs Review`
+- Add `notes` to customer/vendor records — these are included in the LLM context
+- Inspect `llm_raw_response` in `email_log` for Claude's reasoning
--- a/alembic.ini
+++ b/alembic.ini
@ -0,0 +1,38 @@
+[alembic]
+script_location = migrations
+prepend_sys_path = .
+sqlalchemy.url = %(DATABASE_URL)s
+
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+qualname =
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S
--- a/howl/init.py
+++ b/howl/init.py
--- a/howl/config.py
+++ b/howl/config.py
@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import Field, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
+
+    # Azure / Graph
+    azure_tenant_id: str
+    azure_client_id: str
+    azure_client_secret: str = ""
+    graph_auth_mode: Literal["client_credentials", "delegated"] = "client_credentials"
+    msal_token_cache_path: str = ".msal_cache.bin"
+    graph_mailbox: str
+    graph_poll_interval_seconds: int = 300
+    graph_batch_size: int = 50
+    graph_max_body_chars: int = 4000
+
+    # Anthropic
+    anthropic_api_key: str
+    anthropic_model: str = "claude-sonnet-4-6"
+    anthropic_max_tokens: int = 1024
+    llm_confidence_threshold: float = 0.60
+
+    # PostgreSQL
+    database_url: str
+    database_pool_size: int = 5
+    database_pool_max_overflow: int = 10
+
+    # Processing
+    dry_run: bool = False
+    max_retry_count: int = 3
+    notification_webhook_url: str = ""
+
+    # Folder names
+    folder_customers: str = "Customers"
+    folder_vendors: str = "Vendors"
+    folder_review: str = "Needs Review"
+    folder_spam: str = "Junk Email"
+    folder_escalate: str = "Escalate"
+
+    # Logging
+    log_level: str = "INFO"
+    log_format: Literal["json", "text"] = "json"
+
+    @field_validator("graph_auth_mode")
+    @classmethod
+    def validate_client_secret(cls, v: str, info) -> str:
+        # Deferred: full cross-field validation happens at runtime in auth.py
+        return v
+
+    @field_validator("llm_confidence_threshold")
+    @classmethod
+    def validate_confidence(cls, v: float) -> float:
+        if not 0.0 <= v <= 1.0:
+            raise ValueError("llm_confidence_threshold must be between 0.0 and 1.0")
+        return v
+
+
+def get_settings() -> Settings:
+    return Settings()  # type: ignore[call-arg]
--- a/howl/daemon/init.py
+++ b/howl/daemon/init.py
--- a/howl/daemon/scheduler.py
+++ b/howl/daemon/scheduler.py
@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import asyncio
+import signal
+
+import structlog
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+
+from howl.config import Settings
+from howl.pipeline.processor import EmailProcessor
+
+log = structlog.get_logger(__name__)
+
+
+async def start_daemon(processor: EmailProcessor, settings: Settings) -> None:
+    """
+    Start the APScheduler poll loop. Blocks until SIGTERM or SIGINT is received.
+    The scheduler is stopped gracefully after the current job completes.
+    """
+    loop = asyncio.get_running_loop()
+    stop_event = asyncio.Event()
+
+    def _handle_signal():
+        log.info("shutdown_signal_received")
+        stop_event.set()
+
+    loop.add_signal_handler(signal.SIGTERM, _handle_signal)
+    loop.add_signal_handler(signal.SIGINT, _handle_signal)
+
+    scheduler = AsyncIOScheduler()
+
+    async def _poll_job():
+        log.info("poll_started", mailbox=settings.graph_mailbox)
+        try:
+            count = await processor.poll_once()
+            log.info("poll_completed", processed=count)
+        except Exception as exc:
+            log.exception("poll_error", error=str(exc))
+
+    scheduler.add_job(
+        _poll_job,
+        trigger="interval",
+        seconds=settings.graph_poll_interval_seconds,
+        id="email_poll",
+        next_run_time=__import__("datetime").datetime.now(),  # Run immediately on start
+        max_instances=1,
+        coalesce=True,
+    )
+
+    scheduler.start()
+    log.info(
+        "daemon_started",
+        mailbox=settings.graph_mailbox,
+        interval_seconds=settings.graph_poll_interval_seconds,
+        dry_run=settings.dry_run,
+    )
+
+    await stop_event.wait()
+
+    log.info("daemon_stopping")
+    scheduler.shutdown(wait=True)
+    log.info("daemon_stopped")
--- a/howl/db/init.py
+++ b/howl/db/init.py
--- a/howl/db/engine.py
+++ b/howl/db/engine.py
@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import AsyncGenerator
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from howl.config import Settings
+
+_engine = None
+_session_factory = None
+
+
+def init_engine(settings: Settings) -> None:
+    global _engine, _session_factory
+    _engine = create_async_engine(
+        settings.database_url,
+        pool_size=settings.database_pool_size,
+        max_overflow=settings.database_pool_max_overflow,
+        pool_pre_ping=True,
+        echo=False,
+    )
+    _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
+
+
+@asynccontextmanager
+async def async_session() -> AsyncGenerator[AsyncSession, None]:
+    if _session_factory is None:
+        raise RuntimeError("Database engine not initialised — call init_engine() first")
+    async with _session_factory() as session:
+        try:
+            yield session
+            await session.commit()
+        except Exception:
+            await session.rollback()
+            raise
+
+
+async def dispose_engine() -> None:
+    if _engine is not None:
+        await _engine.dispose()
--- a/howl/db/models.py
+++ b/howl/db/models.py
@ -0,0 +1,209 @@
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import (
+    Boolean,
+    CheckConstraint,
+    DateTime,
+    Enum,
+    Index,
+    Integer,
+    Numeric,
+    String,
+    Text,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.dialects.postgresql import JSONB, UUID
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+# ---------------------------------------------------------------------------
+# Customers
+# ---------------------------------------------------------------------------
+
+
+class Customer(Base):
+    __tablename__ = "customers"
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    name: Mapped[str] = mapped_column(Text, nullable=False)
+    company: Mapped[Optional[str]] = mapped_column(Text)
+    phone: Mapped[Optional[str]] = mapped_column(Text)
+    notes: Mapped[Optional[str]] = mapped_column(Text)
+    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
+    )
+
+    emails: Mapped[list[CustomerEmail]] = relationship(
+        "CustomerEmail", back_populates="customer", cascade="all, delete-orphan"
+    )
+
+
+class CustomerEmail(Base):
+    __tablename__ = "customer_emails"
+    __table_args__ = (
+        UniqueConstraint("email_address", name="uq_customer_email"),
+        Index("idx_customer_emails_address", func.lower(String()), postgresql_ops={}),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    customer_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        nullable=False,
+    )
+    email_address: Mapped[str] = mapped_column(Text, nullable=False)
+    label: Mapped[Optional[str]] = mapped_column(Text)
+    is_primary: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+
+    customer: Mapped[Customer] = relationship("Customer", back_populates="emails")
+
+
+# ---------------------------------------------------------------------------
+# Vendors
+# ---------------------------------------------------------------------------
+
+
+class Vendor(Base):
+    __tablename__ = "vendors"
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    name: Mapped[str] = mapped_column(Text, nullable=False)
+    company: Mapped[Optional[str]] = mapped_column(Text)
+    service_category: Mapped[Optional[str]] = mapped_column(Text)
+    phone: Mapped[Optional[str]] = mapped_column(Text)
+    notes: Mapped[Optional[str]] = mapped_column(Text)
+    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
+    )
+
+    emails: Mapped[list[VendorEmail]] = relationship(
+        "VendorEmail", back_populates="vendor", cascade="all, delete-orphan"
+    )
+
+
+class VendorEmail(Base):
+    __tablename__ = "vendor_emails"
+    __table_args__ = (UniqueConstraint("email_address", name="uq_vendor_email"),)
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    vendor_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False)
+    email_address: Mapped[str] = mapped_column(Text, nullable=False)
+    label: Mapped[Optional[str]] = mapped_column(Text)
+    is_primary: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+
+    vendor: Mapped[Vendor] = relationship("Vendor", back_populates="emails")
+
+
+# ---------------------------------------------------------------------------
+# Whitelist
+# ---------------------------------------------------------------------------
+
+
+class Whitelist(Base):
+    __tablename__ = "whitelist"
+    __table_args__ = (
+        CheckConstraint(
+            "email_address IS NOT NULL OR domain IS NOT NULL",
+            name="chk_whitelist_has_target",
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    email_address: Mapped[Optional[str]] = mapped_column(Text)
+    domain: Mapped[Optional[str]] = mapped_column(Text)
+    description: Mapped[Optional[str]] = mapped_column(Text)
+    added_by: Mapped[Optional[str]] = mapped_column(Text)
+    expires_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
+    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+
+
+# ---------------------------------------------------------------------------
+# Email Log
+# ---------------------------------------------------------------------------
+
+SENDER_TYPE_ENUM = Enum(
+    "customer", "vendor", "whitelist", "unknown", name="sender_type", create_type=False
+)
+EMAIL_ACTION_ENUM = Enum(
+    "inbox_keep",
+    "flag_follow_up",
+    "move_customer",
+    "move_vendor",
+    "move_whitelist",
+    "move_spam",
+    "move_review",
+    "escalate",
+    name="email_action",
+    create_type=False,
+)
+PROCESSING_STATUS_ENUM = Enum(
+    "pending", "processing", "completed", "failed", "skipped",
+    name="processing_status",
+    create_type=False,
+)
+
+
+class EmailLog(Base):
+    __tablename__ = "email_log"
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+
+    # Graph identifiers
+    graph_message_id: Mapped[str] = mapped_column(Text, nullable=False, unique=True)
+    graph_conversation_id: Mapped[Optional[str]] = mapped_column(Text)
+    mailbox: Mapped[str] = mapped_column(Text, nullable=False)
+
+    # Envelope
+    sender_address: Mapped[str] = mapped_column(Text, nullable=False)
+    sender_name: Mapped[Optional[str]] = mapped_column(Text)
+    subject: Mapped[Optional[str]] = mapped_column(Text)
+    received_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
+    has_attachments: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+
+    # DB lookup
+    sender_type: Mapped[str] = mapped_column(SENDER_TYPE_ENUM, nullable=False, default="unknown")
+    matched_entity_id: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True))
+    matched_entity_table: Mapped[Optional[str]] = mapped_column(Text)
+
+    # LLM
+    llm_model: Mapped[Optional[str]] = mapped_column(Text)
+    llm_input_tokens: Mapped[Optional[int]] = mapped_column(Integer)
+    llm_output_tokens: Mapped[Optional[int]] = mapped_column(Integer)
+    llm_raw_response: Mapped[Optional[dict]] = mapped_column(JSONB)
+    llm_classification: Mapped[Optional[str]] = mapped_column(Text)
+    llm_confidence: Mapped[Optional[float]] = mapped_column(Numeric(4, 3))
+    llm_reasoning: Mapped[Optional[str]] = mapped_column(Text)
+    llm_suggested_action: Mapped[Optional[str]] = mapped_column(EMAIL_ACTION_ENUM)
+
+    # Decision
+    final_action: Mapped[str] = mapped_column(EMAIL_ACTION_ENUM, nullable=False, default="inbox_keep")
+    action_overridden: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    action_executed_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
+    action_error: Mapped[Optional[str]] = mapped_column(Text)
+
+    # Pipeline metadata
+    status: Mapped[str] = mapped_column(PROCESSING_STATUS_ENUM, nullable=False, default="pending")
+    processing_started_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
+    processing_completed_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
+    error_message: Mapped[Optional[str]] = mapped_column(Text)
+    retry_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
+    )
--- a/howl/db/queries.py
+++ b/howl/db/queries.py
@ -0,0 +1,166 @@
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Optional
+
+import structlog
+from sqlalchemy import select, text, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from howl.db.models import EmailLog
+
+log = structlog.get_logger(__name__)
+
+
+@dataclass
+class SenderMatch:
+    sender_type: str  # 'customer' | 'vendor' | 'whitelist' | 'unknown'
+    entity_id: Optional[uuid.UUID] = None
+    entity_table: Optional[str] = None
+    entity_name: Optional[str] = None
+    notes: Optional[str] = None
+
+
+async def lookup_sender(session: AsyncSession, email_address: str) -> SenderMatch:
+    """
+    Look up a sender email address across customers, vendors, and whitelist in a
+    single query. Returns the first match found, or a SenderMatch with
+    sender_type='unknown' if no match.
+    """
+    addr_lower = email_address.lower()
+    domain = addr_lower.split("@")[-1] if "@" in addr_lower else ""
+
+    # Single UNION query across all three tables
+    sql = text("""
+        SELECT 'customer' AS sender_type,
+               c.id        AS entity_id,
+               'customers' AS entity_table,
+               c.name      AS entity_name,
+               c.notes     AS notes
+        FROM customer_emails ce
+        JOIN customers c ON c.id = ce.customer_id
+        WHERE LOWER(ce.email_address) = :addr
+          AND c.is_active = TRUE
+
+        UNION ALL
+
+        SELECT 'vendor'  AS sender_type,
+               v.id      AS entity_id,
+               'vendors' AS entity_table,
+               v.name    AS entity_name,
+               v.notes   AS notes
+        FROM vendor_emails ve
+        JOIN vendors v ON v.id = ve.vendor_id
+        WHERE LOWER(ve.email_address) = :addr
+          AND v.is_active = TRUE
+
+        UNION ALL
+
+        SELECT 'whitelist' AS sender_type,
+               w.id        AS entity_id,
+               'whitelist' AS entity_table,
+               w.description AS entity_name,
+               NULL        AS notes
+        FROM whitelist w
+        WHERE w.is_active = TRUE
+          AND (w.expires_at IS NULL OR w.expires_at > NOW())
+          AND (
+              LOWER(w.email_address) = :addr
+              OR LOWER(w.domain) = :domain
+          )
+
+        LIMIT 1
+    """)
+
+    result = await session.execute(sql, {"addr": addr_lower, "domain": domain})
+    row = result.mappings().first()
+
+    if row is None:
+        return SenderMatch(sender_type="unknown")
+
+    return SenderMatch(
+        sender_type=row["sender_type"],
+        entity_id=row["entity_id"],
+        entity_table=row["entity_table"],
+        entity_name=row["entity_name"],
+        notes=row["notes"],
+    )
+
+
+async def is_already_processed(session: AsyncSession, graph_message_id: str) -> bool:
+    """Return True if this Graph message ID has already been logged (idempotency guard)."""
+    result = await session.execute(
+        select(EmailLog.id).where(EmailLog.graph_message_id == graph_message_id).limit(1)
+    )
+    return result.scalar() is not None
+
+
+async def insert_email_log(
+    session: AsyncSession,
+    *,
+    graph_message_id: str,
+    graph_conversation_id: Optional[str],
+    mailbox: str,
+    sender_address: str,
+    sender_name: Optional[str],
+    subject: Optional[str],
+    received_at: Optional[datetime],
+    has_attachments: bool,
+) -> EmailLog:
+    """Create an EmailLog row in 'processing' status and flush it."""
+    log_entry = EmailLog(
+        graph_message_id=graph_message_id,
+        graph_conversation_id=graph_conversation_id,
+        mailbox=mailbox,
+        sender_address=sender_address,
+        sender_name=sender_name,
+        subject=subject,
+        received_at=received_at,
+        has_attachments=has_attachments,
+        status="processing",
+        processing_started_at=datetime.now(timezone.utc),
+    )
+    session.add(log_entry)
+    await session.flush()
+    return log_entry
+
+
+async def update_email_log(
+    session: AsyncSession,
+    log_id: uuid.UUID,
+    **kwargs,
+) -> None:
+    """Update fields on an EmailLog row by ID."""
+    if not kwargs:
+        return
+    await session.execute(
+        update(EmailLog)
+        .where(EmailLog.id == log_id)
+        .values(**kwargs, updated_at=datetime.now(timezone.utc))
+    )
+
+
+async def get_failed_emails(
+    session: AsyncSession, max_retries: int, limit: int = 100
+) -> list[EmailLog]:
+    """Return failed emails that haven't exceeded the retry limit."""
+    result = await session.execute(
+        select(EmailLog)
+        .where(
+            EmailLog.status == "failed",
+            EmailLog.retry_count < max_retries,
+        )
+        .order_by(EmailLog.created_at)
+        .limit(limit)
+    )
+    return list(result.scalars().all())
+
+
+async def get_recent_logs(session: AsyncSession, limit: int = 20) -> list[EmailLog]:
+    """Return the most recent email log entries for status display."""
+    result = await session.execute(
+        select(EmailLog).order_by(EmailLog.created_at.desc()).limit(limit)
+    )
+    return list(result.scalars().all())
--- a/howl/graph/init.py
+++ b/howl/graph/init.py
--- a/howl/graph/auth.py
+++ b/howl/graph/auth.py
@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import json
+import os
+from typing import Callable
+
+import msal
+import structlog
+
+from howl.config import Settings
+
+log = structlog.get_logger(__name__)
+
+# Graph API scope for app-only access
+_GRAPH_SCOPE = ["https://graph.microsoft.com/.default"]
+
+
+def _load_token_cache(cache_path: str) -> msal.SerializableTokenCache:
+    cache = msal.SerializableTokenCache()
+    if os.path.exists(cache_path):
+        with open(cache_path) as f:
+            cache.deserialize(f.read())
+    return cache
+
+
+def _save_token_cache(cache: msal.SerializableTokenCache, cache_path: str) -> None:
+    if cache.has_state_changed:
+        with open(cache_path, "w") as f:
+            f.write(cache.serialize())
+
+
+def build_token_provider(settings: Settings) -> Callable[[], str]:
+    """
+    Return a callable that returns a valid bearer token string.
+    The callable handles silent refresh automatically.
+    Supports both client_credentials and delegated auth modes.
+    """
+    if settings.graph_auth_mode == "client_credentials":
+        return _build_client_credentials_provider(settings)
+    else:
+        return _build_delegated_provider(settings)
+
+
+def _build_client_credentials_provider(settings: Settings) -> Callable[[], str]:
+    if not settings.azure_client_secret:
+        raise ValueError(
+            "AZURE_CLIENT_SECRET is required for client_credentials auth mode"
+        )
+
+    app = msal.ConfidentialClientApplication(
+        client_id=settings.azure_client_id,
+        client_credential=settings.azure_client_secret,
+        authority=f"https://login.microsoftonline.com/{settings.azure_tenant_id}",
+    )
+
+    def get_token() -> str:
+        result = app.acquire_token_silent(scopes=_GRAPH_SCOPE, account=None)
+        if not result:
+            result = app.acquire_token_for_client(scopes=_GRAPH_SCOPE)
+
+        if "access_token" not in result:
+            error = result.get("error_description", result.get("error", "Unknown error"))
+            raise RuntimeError(f"MSAL token acquisition failed: {error}")
+
+        log.debug("token_acquired", auth_mode="client_credentials")
+        return result["access_token"]
+
+    return get_token
+
+
+def _build_delegated_provider(settings: Settings) -> Callable[[], str]:
+    cache = _load_token_cache(settings.msal_token_cache_path)
+
+    app = msal.PublicClientApplication(
+        client_id=settings.azure_client_id,
+        authority=f"https://login.microsoftonline.com/{settings.azure_tenant_id}",
+        token_cache=cache,
+    )
+
+    def get_token() -> str:
+        accounts = app.get_accounts()
+        result = None
+
+        if accounts:
+            result = app.acquire_token_silent(scopes=_GRAPH_SCOPE, account=accounts[0])
+
+        if not result:
+            # Device code flow for first-time / expired auth
+            flow = app.initiate_device_flow(scopes=_GRAPH_SCOPE)
+            if "user_code" not in flow:
+                raise RuntimeError(f"Device flow initiation failed: {flow}")
+
+            # Print instructions — this is intentionally printed to stdout for user action
+            print(f"\n{flow['message']}\n")
+            result = app.acquire_token_by_device_flow(flow)
+
+        _save_token_cache(cache, settings.msal_token_cache_path)
+
+        if "access_token" not in result:
+            error = result.get("error_description", result.get("error", "Unknown error"))
+            raise RuntimeError(f"MSAL token acquisition failed: {error}")
+
+        log.debug("token_acquired", auth_mode="delegated")
+        return result["access_token"]
+
+    return get_token
--- a/howl/graph/client.py
+++ b/howl/graph/client.py
@ -0,0 +1,238 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Callable, Optional
+
+import httpx
+import structlog
+from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential
+
+log = structlog.get_logger(__name__)
+
+GRAPH_BASE = "https://graph.microsoft.com/v1.0"
+
+
+@dataclass
+class Message:
+    id: str
+    conversation_id: Optional[str]
+    subject: Optional[str]
+    sender_address: str
+    sender_name: Optional[str]
+    received_at: Optional[datetime]
+    has_attachments: bool
+    importance: str
+    body_preview: Optional[str] = None
+
+
+def _is_retryable(exc: BaseException) -> bool:
+    if isinstance(exc, httpx.HTTPStatusError):
+        return exc.response.status_code in (429, 500, 502, 503, 504)
+    return isinstance(exc, (httpx.TimeoutException, httpx.NetworkError))
+
+
+def _retry_wait(retry_state) -> float:
+    """Respect Retry-After header when present."""
+    exc = retry_state.outcome.exception()
+    if isinstance(exc, httpx.HTTPStatusError):
+        retry_after = exc.response.headers.get("Retry-After")
+        if retry_after:
+            try:
+                return float(retry_after)
+            except ValueError:
+                pass
+    # Exponential backoff: 2, 4, 8 seconds
+    return min(2 ** retry_state.attempt_number, 30)
+
+
+class GraphClient:
+    def __init__(self, token_provider: Callable[[], str]) -> None:
+        self._token_provider = token_provider
+        self._http = httpx.Client(timeout=30)
+        self._folder_cache: dict[str, str] = {}  # mailbox:name → folder_id
+
+    def _headers(self) -> dict[str, str]:
+        return {
+            "Authorization": f"Bearer {self._token_provider()}",
+            "Content-Type": "application/json",
+        }
+
+    def _get(self, url: str, params: dict | None = None) -> dict:
+        response = self._http.get(url, headers=self._headers(), params=params)
+        response.raise_for_status()
+        return response.json()
+
+    def _post(self, url: str, json: dict) -> dict:
+        response = self._http.post(url, headers=self._headers(), json=json)
+        response.raise_for_status()
+        return response.json()
+
+    def _patch(self, url: str, json: dict) -> None:
+        response = self._http.patch(url, headers=self._headers(), json=json)
+        response.raise_for_status()
+
+    # ------------------------------------------------------------------
+    # Messages
+    # ------------------------------------------------------------------
+
+    @retry(
+        retry=retry_if_exception(_is_retryable),
+        stop=stop_after_attempt(4),
+        wait=wait_exponential(multiplier=1, min=2, max=30),
+        reraise=True,
+    )
+    def list_unread_messages(
+        self,
+        mailbox: str,
+        folder: str = "Inbox",
+        top: int = 50,
+        skip_token: Optional[str] = None,
+    ) -> tuple[list[Message], Optional[str]]:
+        """
+        Return a batch of unread messages and the next skip token (or None if done).
+        """
+        url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders/{folder}/messages"
+        params: dict = {
+            "$filter": "isRead eq false",
+            "$top": str(top),
+            "$select": (
+                "id,conversationId,subject,sender,receivedDateTime,"
+                "hasAttachments,importance,bodyPreview"
+            ),
+            "$orderby": "receivedDateTime asc",
+        }
+        if skip_token:
+            params["$skipToken"] = skip_token
+
+        data = self._get(url, params)
+        messages = [_parse_message(m) for m in data.get("value", [])]
+
+        next_link = data.get("@odata.nextLink")
+        next_skip = _extract_skip_token(next_link) if next_link else None
+
+        log.debug("listed_messages", mailbox=mailbox, count=len(messages))
+        return messages, next_skip
+
+    @retry(
+        retry=retry_if_exception(_is_retryable),
+        stop=stop_after_attempt(4),
+        wait=wait_exponential(multiplier=1, min=2, max=30),
+        reraise=True,
+    )
+    def get_message_body(self, mailbox: str, message_id: str) -> str:
+        """Fetch the full message body as plain text (HTML stripped if needed)."""
+        url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}"
+        data = self._get(url, params={"$select": "body"})
+        content_type = data.get("body", {}).get("contentType", "text")
+        body = data.get("body", {}).get("content", "")
+
+        if content_type == "html":
+            body = _strip_html(body)
+
+        return body
+
+    @retry(
+        retry=retry_if_exception(_is_retryable),
+        stop=stop_after_attempt(4),
+        wait=wait_exponential(multiplier=1, min=2, max=30),
+        reraise=True,
+    )
+    def move_message(self, mailbox: str, message_id: str, destination_folder_id: str) -> None:
+        url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}/move"
+        self._post(url, {"destinationId": destination_folder_id})
+        log.debug("message_moved", mailbox=mailbox, message_id=message_id, folder=destination_folder_id)
+
+    @retry(
+        retry=retry_if_exception(_is_retryable),
+        stop=stop_after_attempt(4),
+        wait=wait_exponential(multiplier=1, min=2, max=30),
+        reraise=True,
+    )
+    def flag_message(self, mailbox: str, message_id: str) -> None:
+        url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}"
+        self._patch(url, {"flag": {"flagStatus": "flagged"}})
+        log.debug("message_flagged", mailbox=mailbox, message_id=message_id)
+
+    @retry(
+        retry=retry_if_exception(_is_retryable),
+        stop=stop_after_attempt(4),
+        wait=wait_exponential(multiplier=1, min=2, max=30),
+        reraise=True,
+    )
+    def mark_as_read(self, mailbox: str, message_id: str) -> None:
+        url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}"
+        self._patch(url, {"isRead": True})
+
+    # ------------------------------------------------------------------
+    # Folders
+    # ------------------------------------------------------------------
+
+    def get_or_create_folder(self, mailbox: str, folder_name: str) -> str:
+        """
+        Return the folder ID for a given display name, creating it if it doesn't exist.
+        Results are cached in-memory.
+        """
+        cache_key = f"{mailbox}:{folder_name}"
+        if cache_key in self._folder_cache:
+            return self._folder_cache[cache_key]
+
+        folder_id = self._find_folder(mailbox, folder_name)
+        if folder_id is None:
+            folder_id = self._create_folder(mailbox, folder_name)
+
+        self._folder_cache[cache_key] = folder_id
+        return folder_id
+
+    def _find_folder(self, mailbox: str, folder_name: str) -> Optional[str]:
+        url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders"
+        data = self._get(url, params={"$filter": f"displayName eq '{folder_name}'", "$top": "1"})
+        folders = data.get("value", [])
+        return folders[0]["id"] if folders else None
+
+    def _create_folder(self, mailbox: str, folder_name: str) -> str:
+        url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders"
+        data = self._post(url, {"displayName": folder_name})
+        log.info("folder_created", mailbox=mailbox, folder_name=folder_name)
+        return data["id"]
+
+    def close(self) -> None:
+        self._http.close()
+
+
+# ------------------------------------------------------------------
+# Helpers
+# ------------------------------------------------------------------
+
+def _parse_message(raw: dict) -> Message:
+    sender = raw.get("sender", {}).get("emailAddress", {})
+    received_str = raw.get("receivedDateTime")
+    received_at = datetime.fromisoformat(received_str.replace("Z", "+00:00")) if received_str else None
+
+    return Message(
+        id=raw["id"],
+        conversation_id=raw.get("conversationId"),
+        subject=raw.get("subject"),
+        sender_address=sender.get("address", ""),
+        sender_name=sender.get("name"),
+        received_at=received_at,
+        has_attachments=raw.get("hasAttachments", False),
+        importance=raw.get("importance", "normal"),
+        body_preview=raw.get("bodyPreview"),
+    )
+
+
+def _extract_skip_token(next_link: str) -> Optional[str]:
+    match = re.search(r"\$skipToken=([^&]+)", next_link)
+    return match.group(1) if match else None
+
+
+def _strip_html(html: str) -> str:
+    try:
+        from bs4 import BeautifulSoup
+        soup = BeautifulSoup(html, "html.parser")
+        return soup.get_text(separator="\n")
+    except ImportError:
+        # Fallback: crude tag stripping
+        return re.sub(r"<[^>]+>", "", html)
--- a/howl/llm/init.py
+++ b/howl/llm/init.py
--- a/howl/llm/client.py
+++ b/howl/llm/client.py
@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import structlog
+from anthropic import Anthropic
+from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential
+
+from howl.config import Settings
+from howl.db.queries import SenderMatch
+from howl.graph.client import Message
+from howl.llm.prompts import SYSTEM_PROMPT, build_prompt
+from howl.llm.schemas import CLASSIFY_EMAIL_TOOL, EmailClassification
+
+log = structlog.get_logger(__name__)
+
+
+class LLMClient:
+    def __init__(self, settings: Settings) -> None:
+        self._client = Anthropic(api_key=settings.anthropic_api_key)
+        self._model = settings.anthropic_model
+        self._max_tokens = settings.anthropic_max_tokens
+        self._max_body_chars = settings.graph_max_body_chars
+
+    @retry(
+        retry=retry_if_exception_type(Exception),
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=2, max=15),
+        reraise=True,
+    )
+    def classify(
+        self,
+        message: Message,
+        body: str,
+        sender_match: SenderMatch,
+    ) -> tuple[EmailClassification, dict]:
+        """
+        Send the email to Claude for classification via tool use.
+
+        Returns:
+            (EmailClassification, raw_response_dict)
+        """
+        user_prompt = build_prompt(message, body, sender_match, self._max_body_chars)
+
+        response = self._client.messages.create(
+            model=self._model,
+            max_tokens=self._max_tokens,
+            system=SYSTEM_PROMPT,
+            tools=[CLASSIFY_EMAIL_TOOL],
+            tool_choice={"type": "tool", "name": "classify_email"},
+            messages=[{"role": "user", "content": user_prompt}],
+        )
+
+        # Extract the tool use block
+        tool_use_block = next(
+            (block for block in response.content if block.type == "tool_use"),
+            None,
+        )
+
+        if tool_use_block is None:
+            raise ValueError("Claude did not call the classify_email tool")
+
+        raw_input = tool_use_block.input
+        classification = EmailClassification.model_validate(raw_input)
+
+        raw_response = {
+            "model": response.model,
+            "stop_reason": response.stop_reason,
+            "usage": {
+                "input_tokens": response.usage.input_tokens,
+                "output_tokens": response.usage.output_tokens,
+            },
+            "tool_input": raw_input,
+        }
+
+        log.debug(
+            "email_classified",
+            classification=classification.classification,
+            action=classification.action,
+            confidence=classification.confidence,
+            model=response.model,
+        )
+
+        return classification, raw_response
--- a/howl/llm/prompts.py
+++ b/howl/llm/prompts.py
@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from howl.db.queries import SenderMatch
+from howl.graph.client import Message
+
+SYSTEM_PROMPT = """\
+You are an email triage assistant. Your job is to classify incoming emails and \
+recommend the appropriate action for each one.
+
+You will be given:
+- The sender's email address and display name
+- Whether the sender is a known customer, vendor, trusted/whitelisted contact, or unknown
+- If known: the contact's name and any notes on file
+- The email subject, receipt timestamp, and body (which may be truncated)
+
+Your goal is to help manage the inbox efficiently and accurately. Use the \
+classify_email tool to return your structured analysis.
+
+Classification guidelines:
+- customer_inquiry: A customer asking a question, requesting support, or following up
+- vendor_invoice: An invoice, bill, or payment request from a vendor
+- vendor_notification: Order confirmations, shipping notifications, account updates from vendors
+- newsletter: Marketing emails, product announcements, digests (even from known contacts)
+- spam: Unsolicited promotional or bulk email from unknown senders
+- phishing: Emails attempting to deceive the recipient (fake invoices, credential requests, etc.)
+- support_request: A request for help or technical support
+- urgent_action_required: Time-sensitive email requiring immediate attention
+- unknown: Cannot determine the nature of the email
+
+Action guidelines:
+- move_customer: Use for customer emails that are inquiries, requests, or correspondence
+- move_vendor: Use for vendor invoices, notifications, and correspondence
+- move_whitelist: Use for trusted senders that are neither customers nor vendors
+- flag_follow_up: Use when the email needs a response but is not urgent
+- move_spam: Use only for high-confidence spam or newsletters from unknown senders
+- move_review: Use when uncertain or when the email needs human judgement
+- inbox_keep: Keep in inbox (use sparingly — prefer a more specific action)
+- escalate: Use for urgent, time-critical, or potentially problematic emails
+
+Be conservative: when uncertain, recommend move_review and set requires_human_review=true \
+rather than taking an irreversible action. Never recommend moving email from a known \
+customer or vendor to spam regardless of the content.\
+"""
+
+
+def build_prompt(
+    message: Message,
+    body: str,
+    sender_match: SenderMatch,
+    max_body_chars: int = 4000,
+) -> str:
+    """Build the user-turn prompt string for a single email."""
+    lines: list[str] = []
+
+    lines.append(f"Sender: {message.sender_name or '(no name)'} <{message.sender_address}>")
+    lines.append(f"Sender Type: {sender_match.sender_type}")
+
+    if sender_match.entity_name:
+        lines.append(f"Known as: {sender_match.entity_name}")
+    if sender_match.notes:
+        lines.append(f"Notes: {sender_match.notes}")
+
+    lines.append(f"Subject: {message.subject or '(no subject)'}")
+    lines.append(f"Received: {message.received_at.isoformat() if message.received_at else 'unknown'}")
+    lines.append(f"Has Attachments: {'yes' if message.has_attachments else 'no'}")
+    lines.append(f"Importance: {message.importance}")
+
+    truncated = len(body) > max_body_chars
+    display_body = body[:max_body_chars] if truncated else body
+
+    lines.append("")
+    lines.append("Body" + (" (truncated)" if truncated else "") + ":")
+    lines.append("---")
+    lines.append(display_body)
+    lines.append("---")
+
+    return "\n".join(lines)
--- a/howl/llm/schemas.py
+++ b/howl/llm/schemas.py
@ -0,0 +1,122 @@
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field, field_validator
+
+
+EmailActionLiteral = Literal[
+    "inbox_keep",
+    "flag_follow_up",
+    "move_customer",
+    "move_vendor",
+    "move_whitelist",
+    "move_spam",
+    "move_review",
+    "escalate",
+]
+
+ClassificationLiteral = Literal[
+    "customer_inquiry",
+    "vendor_invoice",
+    "vendor_notification",
+    "newsletter",
+    "spam",
+    "phishing",
+    "support_request",
+    "urgent_action_required",
+    "unknown",
+]
+
+
+class EmailClassification(BaseModel):
+    classification: ClassificationLiteral
+    confidence: float = Field(..., ge=0.0, le=1.0)
+    action: EmailActionLiteral
+    reasoning: str
+    priority: Literal["low", "normal", "high", "urgent"]
+    requires_human_review: bool
+    tags: list[str] = Field(default_factory=list)
+
+    @field_validator("confidence")
+    @classmethod
+    def round_confidence(cls, v: float) -> float:
+        return round(v, 3)
+
+
+# Tool definition passed to Claude
+CLASSIFY_EMAIL_TOOL = {
+    "name": "classify_email",
+    "description": (
+        "Classify an incoming email and recommend the appropriate action. "
+        "Always call this tool with your analysis."
+    ),
+    "input_schema": {
+        "type": "object",
+        "properties": {
+            "classification": {
+                "type": "string",
+                "enum": [
+                    "customer_inquiry",
+                    "vendor_invoice",
+                    "vendor_notification",
+                    "newsletter",
+                    "spam",
+                    "phishing",
+                    "support_request",
+                    "urgent_action_required",
+                    "unknown",
+                ],
+                "description": "The category of this email.",
+            },
+            "confidence": {
+                "type": "number",
+                "minimum": 0.0,
+                "maximum": 1.0,
+                "description": "Your confidence in the classification (0.0 = uncertain, 1.0 = certain).",
+            },
+            "action": {
+                "type": "string",
+                "enum": [
+                    "inbox_keep",
+                    "flag_follow_up",
+                    "move_customer",
+                    "move_vendor",
+                    "move_whitelist",
+                    "move_spam",
+                    "move_review",
+                    "escalate",
+                ],
+                "description": "Recommended action to take on this email.",
+            },
+            "reasoning": {
+                "type": "string",
+                "description": "Brief explanation of why you chose this classification and action.",
+            },
+            "priority": {
+                "type": "string",
+                "enum": ["low", "normal", "high", "urgent"],
+                "description": "The priority level of this email.",
+            },
+            "requires_human_review": {
+                "type": "boolean",
+                "description": (
+                    "Set to true if a human should review this email before or after action is taken."
+                ),
+            },
+            "tags": {
+                "type": "array",
+                "items": {"type": "string"},
+                "description": "Optional free-form tags for further categorisation.",
+            },
+        },
+        "required": [
+            "classification",
+            "confidence",
+            "action",
+            "reasoning",
+            "priority",
+            "requires_human_review",
+        ],
+    },
+}
--- a/howl/logging_config.py
+++ b/howl/logging_config.py
@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import logging
+import sys
+
+import structlog
+
+
+def configure_logging(log_level: str = "INFO", log_format: str = "json") -> None:
+    level = getattr(logging, log_level.upper(), logging.INFO)
+
+    shared_processors: list = [
+        structlog.contextvars.merge_contextvars,
+        structlog.stdlib.add_logger_name,
+        structlog.stdlib.add_log_level,
+        structlog.processors.TimeStamper(fmt="iso"),
+        structlog.processors.StackInfoRenderer(),
+    ]
+
+    if log_format == "json":
+        renderer = structlog.processors.JSONRenderer()
+    else:
+        renderer = structlog.dev.ConsoleRenderer()
+
+    structlog.configure(
+        processors=shared_processors + [
+            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
+        ],
+        logger_factory=structlog.stdlib.LoggerFactory(),
+        wrapper_class=structlog.stdlib.BoundLogger,
+        cache_logger_on_first_use=True,
+    )
+
+    formatter = structlog.stdlib.ProcessorFormatter(
+        foreign_pre_chain=shared_processors,
+        processors=[
+            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
+            renderer,
+        ],
+    )
+
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setFormatter(formatter)
+
+    root_logger = logging.getLogger()
+    root_logger.handlers = [handler]
+    root_logger.setLevel(level)
+
+    # Quiet noisy libraries
+    logging.getLogger("httpx").setLevel(logging.WARNING)
+    logging.getLogger("httpcore").setLevel(logging.WARNING)
+    logging.getLogger("apscheduler").setLevel(logging.WARNING)
--- a/howl/main.py
+++ b/howl/main.py
@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Optional
+
+import structlog
+import typer
+from rich.console import Console
+from rich.table import Table
+
+app = typer.Typer(name="howl", help="M365 email management daemon")
+console = Console()
+log = structlog.get_logger(__name__)
+
+
+def _bootstrap() -> tuple:
+    """Load settings, configure logging, initialise DB and Graph clients."""
+    from howl.config import get_settings
+    from howl.db.engine import init_engine
+    from howl.graph.auth import build_token_provider
+    from howl.graph.client import GraphClient
+    from howl.llm.client import LLMClient
+    from howl.logging_config import configure_logging
+    from howl.pipeline.processor import EmailProcessor
+
+    settings = get_settings()
+    configure_logging(settings.log_level, settings.log_format)
+    init_engine(settings)
+
+    token_provider = build_token_provider(settings)
+    graph = GraphClient(token_provider)
+    llm = LLMClient(settings)
+    processor = EmailProcessor(graph, llm, settings)
+
+    return settings, graph, processor
+
+
+@app.command()
+def run() -> None:
+    """Start the email management daemon (blocking)."""
+    settings, graph, processor = _bootstrap()
+
+    if settings.dry_run:
+        console.print("[yellow]DRY RUN mode enabled — no email actions will be taken[/yellow]")
+
+    from howl.daemon.scheduler import start_daemon
+
+    asyncio.run(start_daemon(processor, settings))
+
+
+@app.command(name="dry-run")
+def dry_run(
+    limit: int = typer.Option(10, "--limit", "-n", help="Max emails to process"),
+) -> None:
+    """
+    Process emails in dry-run mode (no mutations). Overrides DRY_RUN env var.
+    """
+    import os
+    os.environ["DRY_RUN"] = "true"
+
+    settings, graph, processor = _bootstrap()
+
+    console.print(f"[cyan]Dry run: processing up to {limit} emails from {settings.graph_mailbox}[/cyan]")
+
+    async def _run():
+        mailbox = settings.graph_mailbox
+        messages, _ = graph.list_unread_messages(mailbox, top=limit)
+        for message in messages[:limit]:
+            await processor.process_message(message)
+
+    asyncio.run(_run())
+    console.print("[green]Dry run complete. Check the email_log table for results.[/green]")
+
+
+@app.command()
+def status(
+    limit: int = typer.Option(20, "--limit", "-n", help="Number of recent log entries to show"),
+) -> None:
+    """Show recent email processing activity."""
+    settings, _, _ = _bootstrap()
+
+    async def _run():
+        from howl.db import engine as db_engine
+        from howl.db.queries import get_recent_logs
+
+        async with db_engine.async_session() as session:
+            logs = await get_recent_logs(session, limit=limit)
+
+        table = Table(title=f"Recent Email Log (last {limit})")
+        table.add_column("Received", style="dim", width=20)
+        table.add_column("Sender", width=30)
+        table.add_column("Subject", width=35)
+        table.add_column("Type", width=10)
+        table.add_column("Action", width=16)
+        table.add_column("Status", width=10)
+
+        for entry in logs:
+            received = entry.received_at.strftime("%Y-%m-%d %H:%M") if entry.received_at else "-"
+            sender = entry.sender_address[:28] + ".." if len(entry.sender_address) > 30 else entry.sender_address
+            subject = (entry.subject or "")[:33] + ".." if len(entry.subject or "") > 35 else (entry.subject or "-")
+            status_style = {
+                "completed": "green",
+                "failed": "red",
+                "processing": "yellow",
+                "pending": "dim",
+                "skipped": "dim",
+            }.get(entry.status, "")
+
+            table.add_row(
+                received,
+                sender,
+                subject,
+                entry.sender_type,
+                entry.final_action,
+                f"[{status_style}]{entry.status}[/{status_style}]",
+            )
+
+        console.print(table)
+
+    asyncio.run(_run())
+
+
+if __name__ == "__main__":
+    app()
--- a/howl/pipeline/init.py
+++ b/howl/pipeline/init.py
--- a/howl/pipeline/actions.py
+++ b/howl/pipeline/actions.py
@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import httpx
+import structlog
+
+from howl.config import Settings
+from howl.db.models import EmailLog
+from howl.graph.client import GraphClient
+
+log = structlog.get_logger(__name__)
+
+
+class ActionExecutor:
+    def __init__(self, graph: GraphClient, settings: Settings) -> None:
+        self._graph = graph
+        self._settings = settings
+
+    def execute(self, log_entry: EmailLog, final_action: str) -> None:
+        """
+        Execute the final action on the email via Graph API.
+        Raises on failure; caller is responsible for logging the error.
+        """
+        mailbox = log_entry.mailbox
+        message_id = log_entry.graph_message_id
+
+        if self._settings.dry_run:
+            log.info("dry_run_action", action=final_action, message_id=message_id)
+            return
+
+        if final_action == "move_customer":
+            folder_id = self._graph.get_or_create_folder(mailbox, self._settings.folder_customers)
+            self._graph.move_message(mailbox, message_id, folder_id)
+
+        elif final_action == "move_vendor":
+            folder_id = self._graph.get_or_create_folder(mailbox, self._settings.folder_vendors)
+            self._graph.move_message(mailbox, message_id, folder_id)
+
+        elif final_action == "move_whitelist":
+            # Whitelisted contacts can go to a generic "Trusted" folder or stay in inbox
+            # Here we keep them in inbox and just flag them as read
+            self._graph.mark_as_read(mailbox, message_id)
+
+        elif final_action == "move_spam":
+            folder_id = self._graph.get_or_create_folder(mailbox, self._settings.folder_spam)
+            self._graph.move_message(mailbox, message_id, folder_id)
+
+        elif final_action == "move_review":
+            folder_id = self._graph.get_or_create_folder(mailbox, self._settings.folder_review)
+            self._graph.move_message(mailbox, message_id, folder_id)
+
+        elif final_action == "flag_follow_up":
+            self._graph.flag_message(mailbox, message_id)
+
+        elif final_action == "escalate":
+            folder_id = self._graph.get_or_create_folder(mailbox, self._settings.folder_escalate)
+            self._graph.move_message(mailbox, message_id, folder_id)
+            self._graph.flag_message(mailbox, message_id)
+            self._send_notification(log_entry)
+
+        elif final_action == "inbox_keep":
+            # No move — just mark as read after the pipeline completes
+            pass
+
+        else:
+            log.warning("unknown_action", action=final_action, message_id=message_id)
+
+        log.info(
+            "action_executed",
+            action=final_action,
+            message_id=message_id,
+            mailbox=mailbox,
+            dry_run=False,
+        )
+
+    def _send_notification(self, log_entry: EmailLog) -> None:
+        """POST a notification to the configured webhook URL (Teams / Slack)."""
+        url = self._settings.notification_webhook_url
+        if not url:
+            return
+
+        payload = {
+            "text": (
+                f"**Escalated email** from {log_entry.sender_name or log_entry.sender_address}\n"
+                f"Subject: {log_entry.subject or '(no subject)'}\n"
+                f"Mailbox: {log_entry.mailbox}"
+            )
+        }
+
+        try:
+            response = httpx.post(url, json=payload, timeout=10)
+            response.raise_for_status()
+            log.info("notification_sent", webhook_url=url)
+        except Exception as exc:
+            # Notification failure is non-fatal
+            log.warning("notification_failed", error=str(exc))
--- a/howl/pipeline/classifier.py
+++ b/howl/pipeline/classifier.py
@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import structlog
+
+from howl.config import Settings
+from howl.db.queries import SenderMatch
+from howl.llm.schemas import EmailClassification
+
+log = structlog.get_logger(__name__)
+
+
+def decide(
+    sender_match: SenderMatch,
+    classification: EmailClassification,
+    settings: Settings,
+) -> tuple[str, bool]:
+    """
+    Apply business rule overrides on top of the raw LLM classification.
+
+    Returns:
+        (final_action, action_overridden)
+    """
+    llm_action = classification.action
+    final_action = llm_action
+    overridden = False
+
+    # Rule 1: Whitelist senders are never moved to spam/junk.
+    if sender_match.sender_type == "whitelist" and llm_action in ("move_spam",):
+        final_action = "inbox_keep"
+        overridden = True
+        log.info(
+            "classifier_override",
+            rule="whitelist_protects_from_spam",
+            original_action=llm_action,
+            final_action=final_action,
+        )
+
+    # Rule 2: Known customers are never spammed — move to review instead.
+    elif sender_match.sender_type == "customer" and llm_action == "move_spam":
+        final_action = "move_review"
+        overridden = True
+        log.info(
+            "classifier_override",
+            rule="customer_never_to_spam",
+            original_action=llm_action,
+            final_action=final_action,
+        )
+
+    # Rule 3: Known vendors are never spammed — move to review instead.
+    elif sender_match.sender_type == "vendor" and llm_action == "move_spam":
+        final_action = "move_review"
+        overridden = True
+        log.info(
+            "classifier_override",
+            rule="vendor_never_to_spam",
+            original_action=llm_action,
+            final_action=final_action,
+        )
+
+    # Rule 4: LLM explicitly flagged for human review — move to review folder.
+    if classification.requires_human_review and final_action not in (
+        "move_review",
+        "escalate",
+        "flag_follow_up",
+    ):
+        final_action = "move_review"
+        overridden = True
+        log.info(
+            "classifier_override",
+            rule="requires_human_review",
+            original_action=llm_action,
+            final_action=final_action,
+        )
+
+    # Rule 5: Low-confidence classification — fall back to inbox_keep.
+    if (
+        classification.confidence < settings.llm_confidence_threshold
+        and final_action not in ("move_review", "escalate")
+    ):
+        final_action = "inbox_keep"
+        overridden = True
+        log.info(
+            "classifier_override",
+            rule="low_confidence_fallback",
+            confidence=classification.confidence,
+            threshold=settings.llm_confidence_threshold,
+            original_action=llm_action,
+            final_action=final_action,
+        )
+
+    return final_action, overridden
--- a/howl/pipeline/processor.py
+++ b/howl/pipeline/processor.py
@ -0,0 +1,229 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import structlog
+
+from howl.config import Settings
+from howl.db import engine as db_engine
+from howl.db import queries
+from howl.graph.client import GraphClient, Message
+from howl.llm.client import LLMClient
+from howl.pipeline import actions as action_module
+from howl.pipeline import classifier
+
+log = structlog.get_logger(__name__)
+
+
+class EmailProcessor:
+    def __init__(
+        self,
+        graph: GraphClient,
+        llm: LLMClient,
+        settings: Settings,
+    ) -> None:
+        self._graph = graph
+        self._llm = llm
+        self._settings = settings
+        self._executor = action_module.ActionExecutor(graph, settings)
+
+    async def process_message(self, message: Message) -> None:
+        """
+        Run a single email through the full pipeline:
+        idempotency check → DB lookup → LLM classify → decide → act → log.
+        """
+        log_ctx = structlog.contextvars.bound_contextvars(
+            message_id=message.id,
+            sender=message.sender_address,
+        )
+
+        async with db_engine.async_session() as session:
+            # Idempotency guard
+            if await queries.is_already_processed(session, message.id):
+                log.debug("message_already_processed", message_id=message.id)
+                return
+
+            # Create log record immediately for crash recovery
+            log_entry = await queries.insert_email_log(
+                session,
+                graph_message_id=message.id,
+                graph_conversation_id=message.conversation_id,
+                mailbox=self._settings.graph_mailbox,
+                sender_address=message.sender_address,
+                sender_name=message.sender_name,
+                subject=message.subject,
+                received_at=message.received_at,
+                has_attachments=message.has_attachments,
+            )
+
+        try:
+            # DB sender lookup
+            async with db_engine.async_session() as session:
+                sender_match = await queries.lookup_sender(session, message.sender_address)
+
+            log.info(
+                "sender_matched",
+                sender_type=sender_match.sender_type,
+                entity=sender_match.entity_name,
+                message_id=message.id,
+            )
+
+            # Fetch full email body
+            body = self._graph.get_message_body(self._settings.graph_mailbox, message.id)
+
+            # LLM classification
+            classification, raw_response = self._llm.classify(message, body, sender_match)
+
+            # Business rule override
+            final_action, overridden = classifier.decide(
+                sender_match, classification, self._settings
+            )
+
+            # Update log with LLM results
+            async with db_engine.async_session() as session:
+                await queries.update_email_log(
+                    session,
+                    log_entry.id,
+                    sender_type=sender_match.sender_type,
+                    matched_entity_id=sender_match.entity_id,
+                    matched_entity_table=sender_match.entity_table,
+                    llm_model=raw_response.get("model"),
+                    llm_input_tokens=raw_response.get("usage", {}).get("input_tokens"),
+                    llm_output_tokens=raw_response.get("usage", {}).get("output_tokens"),
+                    llm_raw_response=raw_response,
+                    llm_classification=classification.classification,
+                    llm_confidence=classification.confidence,
+                    llm_reasoning=classification.reasoning,
+                    llm_suggested_action=classification.action,
+                    final_action=final_action,
+                    action_overridden=overridden,
+                )
+
+            # Execute action
+            self._executor.execute(log_entry, final_action)
+
+            # Mark as read (no-op in dry_run — action executor already skipped)
+            if not self._settings.dry_run:
+                self._graph.mark_as_read(self._settings.graph_mailbox, message.id)
+
+            # Mark log as completed
+            async with db_engine.async_session() as session:
+                await queries.update_email_log(
+                    session,
+                    log_entry.id,
+                    status="completed",
+                    action_executed_at=datetime.now(timezone.utc),
+                    processing_completed_at=datetime.now(timezone.utc),
+                )
+
+            log.info(
+                "message_processed",
+                message_id=message.id,
+                action=final_action,
+                classification=classification.classification,
+                confidence=classification.confidence,
+            )
+
+        except Exception as exc:
+            log.exception("message_processing_failed", message_id=message.id, error=str(exc))
+            async with db_engine.async_session() as session:
+                await queries.update_email_log(
+                    session,
+                    log_entry.id,
+                    status="failed",
+                    error_message=str(exc),
+                    processing_completed_at=datetime.now(timezone.utc),
+                    retry_count=log_entry.retry_count + 1,
+                )
+
+    async def poll_once(self) -> int:
+        """
+        Fetch one batch of unread messages and process each one.
+        Returns the number of messages processed.
+        """
+        mailbox = self._settings.graph_mailbox
+        batch_size = self._settings.graph_batch_size
+        processed = 0
+        skip_token = None
+
+        while True:
+            messages, skip_token = self._graph.list_unread_messages(
+                mailbox, top=batch_size, skip_token=skip_token
+            )
+            for message in messages:
+                await self.process_message(message)
+                processed += 1
+
+            if not skip_token:
+                break
+
+        # Retry previously failed messages
+        async with db_engine.async_session() as session:
+            failed = await queries.get_failed_emails(
+                session, self._settings.max_retry_count
+            )
+
+        for log_entry in failed:
+            # Re-construct a minimal Message for retry
+            retry_msg = Message(
+                id=log_entry.graph_message_id,
+                conversation_id=log_entry.graph_conversation_id,
+                subject=log_entry.subject,
+                sender_address=log_entry.sender_address,
+                sender_name=log_entry.sender_name,
+                received_at=log_entry.received_at,
+                has_attachments=log_entry.has_attachments,
+                importance="normal",
+            )
+            # Reset to pending so it gets re-processed
+            async with db_engine.async_session() as session:
+                await queries.update_email_log(session, log_entry.id, status="pending")
+
+            # Remove from log so idempotency guard allows reprocessing
+            # (we update status to pending above then process_message checks is_already_processed
+            #  which checks for any row — so we need a different path)
+            await self._retry_failed(log_entry, retry_msg)
+
+        return processed
+
+    async def _retry_failed(self, log_entry, message: Message) -> None:
+        """Re-run pipeline for a failed message, updating existing log row."""
+        try:
+            sender_match_result = None
+            async with db_engine.async_session() as session:
+                sender_match_result = await queries.lookup_sender(session, message.sender_address)
+                await queries.update_email_log(
+                    session, log_entry.id,
+                    status="processing",
+                    processing_started_at=datetime.now(timezone.utc),
+                )
+
+            body = self._graph.get_message_body(self._settings.graph_mailbox, message.id)
+            classification, raw_response = self._llm.classify(message, body, sender_match_result)
+            final_action, overridden = classifier.decide(sender_match_result, classification, self._settings)
+            self._executor.execute(log_entry, final_action)
+
+            if not self._settings.dry_run:
+                self._graph.mark_as_read(self._settings.graph_mailbox, message.id)
+
+            async with db_engine.async_session() as session:
+                await queries.update_email_log(
+                    session, log_entry.id,
+                    status="completed",
+                    final_action=final_action,
+                    action_overridden=overridden,
+                    action_executed_at=datetime.now(timezone.utc),
+                    processing_completed_at=datetime.now(timezone.utc),
+                    error_message=None,
+                )
+
+        except Exception as exc:
+            log.exception("retry_failed", message_id=message.id, error=str(exc))
+            async with db_engine.async_session() as session:
+                await queries.update_email_log(
+                    session, log_entry.id,
+                    status="failed",
+                    error_message=str(exc),
+                    retry_count=log_entry.retry_count + 1,
+                    processing_completed_at=datetime.now(timezone.utc),
+                )
--- a/migrations/env.py
+++ b/migrations/env.py
@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import os
+from logging.config import fileConfig
+
+from alembic import context
+from sqlalchemy import engine_from_config, pool
+
+from howl.db.models import Base
+
+config = context.config
+
+# Allow DATABASE_URL env var to override alembic.ini
+database_url = os.environ.get("DATABASE_URL", "")
+if database_url:
+    # Alembic uses psycopg (sync) driver for migrations
+    sync_url = database_url.replace("postgresql+asyncpg://", "postgresql+psycopg://")
+    config.set_main_option("sqlalchemy.url", sync_url)
+
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+target_metadata = Base.metadata
+
+
+def run_migrations_offline() -> None:
+    url = config.get_main_option("sqlalchemy.url")
+    context.configure(
+        url=url,
+        target_metadata=target_metadata,
+        literal_binds=True,
+        dialect_opts={"paramstyle": "named"},
+    )
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online() -> None:
+    connectable = engine_from_config(
+        config.get_section(config.config_ini_section, {}),
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
+    with connectable.connect() as connection:
+        context.configure(connection=connection, target_metadata=target_metadata)
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()
--- a/migrations/script.py.mako
+++ b/migrations/script.py.mako
@ -0,0 +1,28 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+
+"""
+from __future__ import annotations
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+# revision identifiers, used by Alembic.
+revision: str = ${repr(up_revision)}
+down_revision: Union[str, None] = ${repr(down_revision)}
+branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
+depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}
--- a/migrations/versions/0001_initial_schema.py
+++ b/migrations/versions/0001_initial_schema.py
@ -0,0 +1,200 @@
+"""Initial schema
+
+Revision ID: 0001
+Revises:
+Create Date: 2026-04-01
+"""
+from __future__ import annotations
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "0001"
+down_revision: Union[str, None] = None
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Enum types
+    sender_type = postgresql.ENUM(
+        "customer", "vendor", "whitelist", "unknown",
+        name="sender_type", create_type=True,
+    )
+    email_action = postgresql.ENUM(
+        "inbox_keep", "flag_follow_up",
+        "move_customer", "move_vendor", "move_whitelist",
+        "move_spam", "move_review", "escalate",
+        name="email_action", create_type=True,
+    )
+    processing_status = postgresql.ENUM(
+        "pending", "processing", "completed", "failed", "skipped",
+        name="processing_status", create_type=True,
+    )
+    sender_type.create(op.get_bind(), checkfirst=True)
+    email_action.create(op.get_bind(), checkfirst=True)
+    processing_status.create(op.get_bind(), checkfirst=True)
+
+    # updated_at trigger function
+    op.execute("""
+        CREATE OR REPLACE FUNCTION set_updated_at()
+        RETURNS TRIGGER AS $$
+        BEGIN
+            NEW.updated_at = NOW();
+            RETURN NEW;
+        END;
+        $$ LANGUAGE plpgsql;
+    """)
+
+    # customers
+    op.create_table(
+        "customers",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
+        sa.Column("name", sa.Text, nullable=False),
+        sa.Column("company", sa.Text),
+        sa.Column("phone", sa.Text),
+        sa.Column("notes", sa.Text),
+        sa.Column("is_active", sa.Boolean, nullable=False, server_default="TRUE"),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")),
+    )
+    op.execute("""
+        CREATE TRIGGER trg_customers_updated_at
+        BEFORE UPDATE ON customers
+        FOR EACH ROW EXECUTE FUNCTION set_updated_at();
+    """)
+
+    # customer_emails
+    op.create_table(
+        "customer_emails",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
+        sa.Column("customer_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("customers.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("email_address", sa.Text, nullable=False),
+        sa.Column("label", sa.Text),
+        sa.Column("is_primary", sa.Boolean, nullable=False, server_default="FALSE"),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")),
+        sa.UniqueConstraint("email_address", name="uq_customer_email"),
+    )
+    op.create_index("idx_customer_emails_address", "customer_emails", [sa.text("LOWER(email_address)")])
+
+    # vendors
+    op.create_table(
+        "vendors",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
+        sa.Column("name", sa.Text, nullable=False),
+        sa.Column("company", sa.Text),
+        sa.Column("service_category", sa.Text),
+        sa.Column("phone", sa.Text),
+        sa.Column("notes", sa.Text),
+        sa.Column("is_active", sa.Boolean, nullable=False, server_default="TRUE"),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")),
+    )
+    op.execute("""
+        CREATE TRIGGER trg_vendors_updated_at
+        BEFORE UPDATE ON vendors
+        FOR EACH ROW EXECUTE FUNCTION set_updated_at();
+    """)
+
+    # vendor_emails
+    op.create_table(
+        "vendor_emails",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
+        sa.Column("vendor_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("vendors.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("email_address", sa.Text, nullable=False),
+        sa.Column("label", sa.Text),
+        sa.Column("is_primary", sa.Boolean, nullable=False, server_default="FALSE"),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")),
+        sa.UniqueConstraint("email_address", name="uq_vendor_email"),
+    )
+    op.create_index("idx_vendor_emails_address", "vendor_emails", [sa.text("LOWER(email_address)")])
+
+    # whitelist
+    op.create_table(
+        "whitelist",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
+        sa.Column("email_address", sa.Text),
+        sa.Column("domain", sa.Text),
+        sa.Column("description", sa.Text),
+        sa.Column("added_by", sa.Text),
+        sa.Column("expires_at", sa.DateTime(timezone=True)),
+        sa.Column("is_active", sa.Boolean, nullable=False, server_default="TRUE"),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")),
+        sa.CheckConstraint("email_address IS NOT NULL OR domain IS NOT NULL", name="chk_whitelist_has_target"),
+    )
+    op.create_index(
+        "idx_whitelist_email", "whitelist",
+        [sa.text("LOWER(email_address)")],
+        postgresql_where=sa.text("email_address IS NOT NULL"),
+    )
+    op.create_index(
+        "idx_whitelist_domain", "whitelist",
+        [sa.text("LOWER(domain)")],
+        postgresql_where=sa.text("domain IS NOT NULL"),
+    )
+
+    # email_log
+    op.create_table(
+        "email_log",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
+        sa.Column("graph_message_id", sa.Text, nullable=False, unique=True),
+        sa.Column("graph_conversation_id", sa.Text),
+        sa.Column("mailbox", sa.Text, nullable=False),
+        sa.Column("sender_address", sa.Text, nullable=False),
+        sa.Column("sender_name", sa.Text),
+        sa.Column("subject", sa.Text),
+        sa.Column("received_at", sa.DateTime(timezone=True)),
+        sa.Column("has_attachments", sa.Boolean, nullable=False, server_default="FALSE"),
+        sa.Column("sender_type", sa.Enum("customer", "vendor", "whitelist", "unknown", name="sender_type", create_type=False), nullable=False, server_default="unknown"),
+        sa.Column("matched_entity_id", postgresql.UUID(as_uuid=True)),
+        sa.Column("matched_entity_table", sa.Text),
+        sa.Column("llm_model", sa.Text),
+        sa.Column("llm_input_tokens", sa.Integer),
+        sa.Column("llm_output_tokens", sa.Integer),
+        sa.Column("llm_raw_response", postgresql.JSONB),
+        sa.Column("llm_classification", sa.Text),
+        sa.Column("llm_confidence", sa.Numeric(4, 3)),
+        sa.Column("llm_reasoning", sa.Text),
+        sa.Column("llm_suggested_action", sa.Enum("inbox_keep", "flag_follow_up", "move_customer", "move_vendor", "move_whitelist", "move_spam", "move_review", "escalate", name="email_action", create_type=False)),
+        sa.Column("final_action", sa.Enum("inbox_keep", "flag_follow_up", "move_customer", "move_vendor", "move_whitelist", "move_spam", "move_review", "escalate", name="email_action", create_type=False), nullable=False, server_default="inbox_keep"),
+        sa.Column("action_overridden", sa.Boolean, nullable=False, server_default="FALSE"),
+        sa.Column("action_executed_at", sa.DateTime(timezone=True)),
+        sa.Column("action_error", sa.Text),
+        sa.Column("status", sa.Enum("pending", "processing", "completed", "failed", "skipped", name="processing_status", create_type=False), nullable=False, server_default="pending"),
+        sa.Column("processing_started_at", sa.DateTime(timezone=True)),
+        sa.Column("processing_completed_at", sa.DateTime(timezone=True)),
+        sa.Column("error_message", sa.Text),
+        sa.Column("retry_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")),
+    )
+    op.create_index("idx_email_log_sender", "email_log", [sa.text("LOWER(sender_address)")])
+    op.create_index(
+        "idx_email_log_status", "email_log", ["status"],
+        postgresql_where=sa.text("status IN ('pending', 'processing')"),
+    )
+    op.create_index("idx_email_log_received", "email_log", [sa.text("received_at DESC")])
+    op.execute("""
+        CREATE TRIGGER trg_email_log_updated_at
+        BEFORE UPDATE ON email_log
+        FOR EACH ROW EXECUTE FUNCTION set_updated_at();
+    """)
+
+
+def downgrade() -> None:
+    op.execute("DROP TRIGGER IF EXISTS trg_email_log_updated_at ON email_log")
+    op.drop_table("email_log")
+    op.execute("DROP TRIGGER IF EXISTS trg_vendors_updated_at ON vendors")
+    op.drop_table("vendor_emails")
+    op.drop_table("vendors")
+    op.execute("DROP TRIGGER IF EXISTS trg_customers_updated_at ON customers")
+    op.drop_table("customer_emails")
+    op.drop_table("customers")
+    op.drop_table("whitelist")
+    op.execute("DROP TYPE IF EXISTS sender_type")
+    op.execute("DROP TYPE IF EXISTS email_action")
+    op.execute("DROP TYPE IF EXISTS processing_status")
+    op.execute("DROP FUNCTION IF EXISTS set_updated_at")
--- a/pyproject.toml
+++ b/pyproject.toml
@ -0,0 +1,57 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "howl"
+version = "0.1.0"
+description = "M365 email management daemon with LLM-powered filtering"
+requires-python = ">=3.11"
+dependencies = [
+    # Microsoft Graph + auth
+    "msal>=1.31.0",
+    "msgraph-sdk>=1.9.0",
+    "httpx>=0.27.0",
+    # Database
+    "sqlalchemy>=2.0.0",
+    "alembic>=1.13.0",
+    "asyncpg>=0.30.0",
+    "psycopg[binary]>=3.2.0",
+    # LLM
+    "anthropic>=0.40.0",
+    # Config & validation
+    "pydantic>=2.9.0",
+    "pydantic-settings>=2.6.0",
+    # CLI
+    "typer>=0.13.0",
+    "rich>=13.0.0",
+    # Scheduling
+    "apscheduler>=3.10.0",
+    # Logging
+    "structlog>=24.0.0",
+    # Retry
+    "tenacity>=9.0.0",
+    # HTML stripping for email bodies
+    "beautifulsoup4>=4.12.0",
+]
+
+[project.scripts]
+howl = "howl.main:app"
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0.0",
+    "pytest-asyncio>=0.24.0",
+    "respx>=0.21.0",
+    "pytest-cov>=5.0.0",
+]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]
+markers = [
+    "integration: marks tests that require real external services (deselect with '-m not integration')",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["howl"]
--- a/tests/init.py
+++ b/tests/init.py
--- a/tests/conftest.py
+++ b/tests/conftest.py
@ -0,0 +1,157 @@
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Generator
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from howl.config import Settings
+from howl.db.queries import SenderMatch
+from howl.graph.client import GraphClient, Message
+from howl.llm.schemas import EmailClassification
+
+FIXTURES_DIR = Path(__file__).parent / "fixtures"
+
+
+# ---------------------------------------------------------------------------
+# Settings fixture (no real credentials needed)
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def settings() -> Settings:
+    return Settings(
+        azure_tenant_id="test-tenant",
+        azure_client_id="test-client-id",
+        azure_client_secret="test-secret",
+        graph_auth_mode="client_credentials",
+        graph_mailbox="test@example.com",
+        anthropic_api_key="test-anthropic-key",
+        database_url="postgresql+asyncpg://test:test@localhost:5432/howl_test",
+        dry_run=True,
+        log_level="WARNING",
+        log_format="text",
+    )
+
+
+# ---------------------------------------------------------------------------
+# Sample email messages
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def sample_emails() -> list[dict]:
+    with open(FIXTURES_DIR / "sample_emails.json") as f:
+        return json.load(f)
+
+
+@pytest.fixture
+def llm_responses() -> dict:
+    with open(FIXTURES_DIR / "llm_responses.json") as f:
+        return json.load(f)
+
+
+def make_message(raw: dict) -> Message:
+    sender = raw.get("sender", {}).get("emailAddress", {})
+    received_str = raw.get("receivedDateTime")
+    received_at = datetime.fromisoformat(received_str.replace("Z", "+00:00")) if received_str else None
+    return Message(
+        id=raw["id"],
+        conversation_id=raw.get("conversationId"),
+        subject=raw.get("subject"),
+        sender_address=sender.get("address", ""),
+        sender_name=sender.get("name"),
+        received_at=received_at,
+        has_attachments=raw.get("hasAttachments", False),
+        importance=raw.get("importance", "normal"),
+        body_preview=raw.get("bodyPreview"),
+    )
+
+
+@pytest.fixture
+def customer_message(sample_emails) -> Message:
+    return make_message(next(e for e in sample_emails if e["id"] == "msg-001-customer-inquiry"))
+
+
+@pytest.fixture
+def spam_message(sample_emails) -> Message:
+    return make_message(next(e for e in sample_emails if e["id"] == "msg-004-spam"))
+
+
+@pytest.fixture
+def urgent_message(sample_emails) -> Message:
+    return make_message(next(e for e in sample_emails if e["id"] == "msg-007-urgent"))
+
+
+# ---------------------------------------------------------------------------
+# Mock Graph client
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def mock_graph(sample_emails) -> MagicMock:
+    graph = MagicMock(spec=GraphClient)
+    raw_messages = [make_message(e) for e in sample_emails]
+    graph.list_unread_messages.return_value = (raw_messages[:3], None)
+    graph.get_message_body.return_value = "This is the email body text."
+    graph.get_or_create_folder.return_value = str(uuid.uuid4())
+    graph.move_message.return_value = None
+    graph.flag_message.return_value = None
+    graph.mark_as_read.return_value = None
+    return graph
+
+
+# ---------------------------------------------------------------------------
+# Mock LLM client
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def mock_llm_classification(llm_responses) -> EmailClassification:
+    return EmailClassification.model_validate(llm_responses["customer_inquiry"])
+
+
+@pytest.fixture
+def mock_llm(mock_llm_classification) -> MagicMock:
+    from howl.llm.client import LLMClient
+    llm = MagicMock(spec=LLMClient)
+    llm.classify.return_value = (
+        mock_llm_classification,
+        {
+            "model": "claude-sonnet-4-6",
+            "stop_reason": "tool_use",
+            "usage": {"input_tokens": 350, "output_tokens": 80},
+            "tool_input": mock_llm_classification.model_dump(),
+        },
+    )
+    return llm
+
+
+# ---------------------------------------------------------------------------
+# Sender matches
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def customer_match() -> SenderMatch:
+    return SenderMatch(
+        sender_type="customer",
+        entity_id=uuid.uuid4(),
+        entity_table="customers",
+        entity_name="Acme Corp",
+        notes="Key account - handle with priority",
+    )
+
+
+@pytest.fixture
+def unknown_match() -> SenderMatch:
+    return SenderMatch(sender_type="unknown")
+
+
+@pytest.fixture
+def whitelist_match() -> SenderMatch:
+    return SenderMatch(
+        sender_type="whitelist",
+        entity_id=uuid.uuid4(),
+        entity_table="whitelist",
+        entity_name="Trusted partner domain",
+    )
--- a/tests/fixtures/llm_responses.json
+++ b/tests/fixtures/llm_responses.json
@ -0,0 +1,65 @@
+{
+  "customer_inquiry": {
+    "classification": "customer_inquiry",
+    "confidence": 0.95,
+    "action": "move_customer",
+    "reasoning": "This is a known customer following up on an order. Should be routed to the Customers folder for response.",
+    "priority": "normal",
+    "requires_human_review": false,
+    "tags": ["order-followup", "customer-service"]
+  },
+  "vendor_invoice": {
+    "classification": "vendor_invoice",
+    "confidence": 0.97,
+    "action": "move_vendor",
+    "reasoning": "Invoice from a known vendor with a specific invoice number and payment details. Route to Vendors folder.",
+    "priority": "normal",
+    "requires_human_review": true,
+    "tags": ["invoice", "payment-due"]
+  },
+  "spam": {
+    "classification": "spam",
+    "confidence": 0.98,
+    "action": "move_spam",
+    "reasoning": "Classic spam characteristics: gift card offer, urgency language, unknown sender domain.",
+    "priority": "low",
+    "requires_human_review": false,
+    "tags": ["gift-card-scam", "promotional"]
+  },
+  "phishing": {
+    "classification": "phishing",
+    "confidence": 0.96,
+    "action": "move_spam",
+    "reasoning": "Phishing attempt: impersonates Microsoft using a suspicious domain, requests credential verification.",
+    "priority": "high",
+    "requires_human_review": true,
+    "tags": ["phishing", "credential-theft", "impersonation"]
+  },
+  "urgent": {
+    "classification": "urgent_action_required",
+    "confidence": 0.93,
+    "action": "escalate",
+    "reasoning": "Known customer reporting production system outage. Requires immediate human attention.",
+    "priority": "urgent",
+    "requires_human_review": true,
+    "tags": ["production-outage", "customer-escalation"]
+  },
+  "low_confidence": {
+    "classification": "unknown",
+    "confidence": 0.45,
+    "action": "move_review",
+    "reasoning": "Cannot determine the nature of this email with sufficient confidence. Human review recommended.",
+    "priority": "normal",
+    "requires_human_review": true,
+    "tags": []
+  },
+  "newsletter": {
+    "classification": "newsletter",
+    "confidence": 0.92,
+    "action": "move_spam",
+    "reasoning": "Unsolicited newsletter from an unknown sender. Contains unsubscribe link.",
+    "priority": "low",
+    "requires_human_review": false,
+    "tags": ["newsletter", "marketing"]
+  }
+}
--- a/tests/fixtures/sample_emails.json
+++ b/tests/fixtures/sample_emails.json
@ -0,0 +1,142 @@
+[
+  {
+    "id": "msg-001-customer-inquiry",
+    "conversationId": "conv-001",
+    "subject": "Question about my order #45231",
+    "sender": {"emailAddress": {"address": "alice@acme.com", "name": "Alice Johnson"}},
+    "receivedDateTime": "2026-04-01T09:15:00Z",
+    "hasAttachments": false,
+    "importance": "normal",
+    "bodyPreview": "Hi, I wanted to check on the status of my order...",
+    "body": {
+      "contentType": "text",
+      "content": "Hi,\n\nI wanted to check on the status of my order #45231 that was placed on March 28th. I haven't received any shipping notification yet and was hoping you could provide an update.\n\nThank you,\nAlice Johnson\nAcme Corp"
+    }
+  },
+  {
+    "id": "msg-002-vendor-invoice",
+    "conversationId": "conv-002",
+    "subject": "Invoice INV-2026-0342 from SupplyPro",
+    "sender": {"emailAddress": {"address": "billing@supplypro.com", "name": "SupplyPro Billing"}},
+    "receivedDateTime": "2026-04-01T10:30:00Z",
+    "hasAttachments": true,
+    "importance": "normal",
+    "bodyPreview": "Please find attached Invoice INV-2026-0342...",
+    "body": {
+      "contentType": "text",
+      "content": "Dear Customer,\n\nPlease find attached Invoice INV-2026-0342 for $4,250.00 due April 30, 2026.\n\nPayment terms: Net 30\nBank transfer details are included in the attached PDF.\n\nRegards,\nSupplyPro Billing Team"
+    }
+  },
+  {
+    "id": "msg-003-whitelist-notification",
+    "conversationId": "conv-003",
+    "subject": "Quarterly audit reminder - action required",
+    "sender": {"emailAddress": {"address": "audit@partnerfirm.com", "name": "Partner Firm Audit"}},
+    "receivedDateTime": "2026-04-01T11:00:00Z",
+    "hasAttachments": false,
+    "importance": "high",
+    "bodyPreview": "This is a reminder about the upcoming quarterly audit...",
+    "body": {
+      "contentType": "text",
+      "content": "Hello,\n\nThis is a reminder that the Q1 2026 quarterly audit is scheduled for April 15th. Please ensure all financial records from January through March are compiled and available for review.\n\nPlease confirm receipt of this notice.\n\nBest regards,\nPartner Firm Audit Team"
+    }
+  },
+  {
+    "id": "msg-004-spam",
+    "conversationId": "conv-004",
+    "subject": "You've been selected! Claim your $500 gift card NOW",
+    "sender": {"emailAddress": {"address": "promo@random-offers.xyz", "name": "Special Rewards"}},
+    "receivedDateTime": "2026-04-01T08:00:00Z",
+    "hasAttachments": false,
+    "importance": "normal",
+    "bodyPreview": "Congratulations! You have been specially selected...",
+    "body": {
+      "contentType": "text",
+      "content": "CONGRATULATIONS!! You have been specially selected to receive a $500 gift card. Click here to claim your reward within 24 hours. Limited time offer! Act NOW!!!"
+    }
+  },
+  {
+    "id": "msg-005-phishing",
+    "conversationId": "conv-005",
+    "subject": "Urgent: Your account requires verification",
+    "sender": {"emailAddress": {"address": "security@microsoft-support-helpdesk.com", "name": "Microsoft Security"}},
+    "receivedDateTime": "2026-04-01T07:45:00Z",
+    "hasAttachments": false,
+    "importance": "high",
+    "bodyPreview": "Your account has been flagged for suspicious activity...",
+    "body": {
+      "contentType": "text",
+      "content": "Dear User,\n\nYour Microsoft account has been flagged for suspicious activity. You must verify your identity within 24 hours or your account will be suspended.\n\nClick here to verify: http://microsoft-verify.malicious-site.com/verify\n\nMicrosoft Security Team"
+    }
+  },
+  {
+    "id": "msg-006-newsletter",
+    "conversationId": "conv-006",
+    "subject": "This week in SaaS: Top 10 tools for 2026",
+    "sender": {"emailAddress": {"address": "newsletter@techdigest.io", "name": "Tech Digest Weekly"}},
+    "receivedDateTime": "2026-04-01T06:00:00Z",
+    "hasAttachments": false,
+    "importance": "low",
+    "bodyPreview": "Welcome to this week's edition of Tech Digest...",
+    "body": {
+      "contentType": "text",
+      "content": "Welcome to Tech Digest Weekly!\n\nThis week we cover:\n- Top 10 SaaS tools for productivity in 2026\n- AI coding assistants reviewed\n- Cloud cost optimization tips\n\nRead more at techdigest.io\n\nUnsubscribe | Manage preferences"
+    }
+  },
+  {
+    "id": "msg-007-urgent",
+    "conversationId": "conv-007",
+    "subject": "URGENT: Production system down - need immediate help",
+    "sender": {"emailAddress": {"address": "bob@bigclient.com", "name": "Bob Smith"}},
+    "receivedDateTime": "2026-04-01T14:22:00Z",
+    "hasAttachments": false,
+    "importance": "high",
+    "bodyPreview": "Our production system went down 20 minutes ago...",
+    "body": {
+      "contentType": "text",
+      "content": "Hi Team,\n\nOur production system went down about 20 minutes ago and we're losing orders. We believe it's related to the API changes deployed this morning. We need immediate assistance.\n\nPlease call me ASAP at 555-0199.\n\nBob Smith\nVP Engineering, BigClient Inc"
+    }
+  },
+  {
+    "id": "msg-008-vendor-notification",
+    "conversationId": "conv-008",
+    "subject": "Your subscription renewal confirmation - CloudTools Pro",
+    "sender": {"emailAddress": {"address": "noreply@cloudtools.io", "name": "CloudTools"}},
+    "receivedDateTime": "2026-04-01T12:00:00Z",
+    "hasAttachments": false,
+    "importance": "normal",
+    "bodyPreview": "Your CloudTools Pro subscription has been renewed...",
+    "body": {
+      "contentType": "text",
+      "content": "Thank you for renewing your CloudTools Pro subscription.\n\nPlan: Pro Annual\nRenewal date: April 1, 2026\nNext renewal: April 1, 2027\nAmount charged: $1,199.00\n\nManage your subscription at app.cloudtools.io/billing"
+    }
+  },
+  {
+    "id": "msg-009-unknown-legit",
+    "conversationId": "conv-009",
+    "subject": "Partnership inquiry - integration opportunity",
+    "sender": {"emailAddress": {"address": "partnerships@newcompany.com", "name": "New Company Partnerships"}},
+    "receivedDateTime": "2026-04-01T13:15:00Z",
+    "hasAttachments": false,
+    "importance": "normal",
+    "bodyPreview": "Hello, I'm reaching out regarding a potential partnership...",
+    "body": {
+      "contentType": "text",
+      "content": "Hello,\n\nI'm reaching out regarding a potential partnership opportunity. Our company provides complementary services to yours and we believe there could be significant mutual benefit in exploring an integration.\n\nWould you be open to a 30-minute call this week?\n\nBest,\nJen Williams\nHead of Partnerships, New Company"
+    }
+  },
+  {
+    "id": "msg-010-domain-whitelist",
+    "conversationId": "conv-010",
+    "subject": "Updated compliance documentation",
+    "sender": {"emailAddress": {"address": "compliance@microsoft.com", "name": "Microsoft Compliance"}},
+    "receivedDateTime": "2026-04-01T15:00:00Z",
+    "hasAttachments": true,
+    "importance": "normal",
+    "bodyPreview": "Please review the attached updated compliance documentation...",
+    "body": {
+      "contentType": "text",
+      "content": "Dear Customer,\n\nPlease review the attached updated Microsoft compliance documentation relevant to your enterprise subscription. These updates reflect changes in data processing terms effective May 1, 2026.\n\nMicrosoft Compliance Team"
+    }
+  }
+]
--- a/tests/test_classifier.py
+++ b/tests/test_classifier.py
@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import pytest
+
+from howl.db.queries import SenderMatch
+from howl.llm.schemas import EmailClassification
+from howl.pipeline import classifier
+
+
+def make_classification(**kwargs) -> EmailClassification:
+    defaults = {
+        "classification": "customer_inquiry",
+        "confidence": 0.90,
+        "action": "move_customer",
+        "reasoning": "Test classification",
+        "priority": "normal",
+        "requires_human_review": False,
+        "tags": [],
+    }
+    defaults.update(kwargs)
+    return EmailClassification.model_validate(defaults)
+
+
+def test_whitelist_sender_not_moved_to_spam(settings, whitelist_match):
+    classification = make_classification(action="move_spam", confidence=0.85)
+    final_action, overridden = classifier.decide(whitelist_match, classification, settings)
+    assert final_action == "inbox_keep"
+    assert overridden is True
+
+
+def test_customer_sender_not_moved_to_spam(settings, customer_match):
+    classification = make_classification(action="move_spam", confidence=0.85)
+    final_action, overridden = classifier.decide(customer_match, classification, settings)
+    assert final_action == "move_review"
+    assert overridden is True
+
+
+def test_requires_human_review_forces_move_review(settings, unknown_match):
+    classification = make_classification(
+        action="inbox_keep",
+        confidence=0.75,
+        requires_human_review=True,
+    )
+    final_action, overridden = classifier.decide(unknown_match, classification, settings)
+    assert final_action == "move_review"
+    assert overridden is True
+
+
+def test_low_confidence_falls_back_to_inbox_keep(settings, unknown_match):
+    # Settings has llm_confidence_threshold=0.60, this is 0.45
+    classification = make_classification(
+        action="move_spam",
+        confidence=0.45,
+        requires_human_review=False,
+    )
+    final_action, overridden = classifier.decide(unknown_match, classification, settings)
+    assert final_action == "inbox_keep"
+    assert overridden is True
+
+
+def test_high_confidence_customer_passes_through(settings, customer_match):
+    classification = make_classification(
+        action="move_customer",
+        confidence=0.95,
+        requires_human_review=False,
+    )
+    final_action, overridden = classifier.decide(customer_match, classification, settings)
+    assert final_action == "move_customer"
+    assert overridden is False
+
+
+def test_escalate_not_overridden_by_low_confidence(settings, customer_match):
+    """Escalate action should survive even low confidence — it's already going to human review."""
+    classification = make_classification(
+        action="escalate",
+        confidence=0.40,
+        requires_human_review=True,
+    )
+    final_action, overridden = classifier.decide(customer_match, classification, settings)
+    assert final_action == "escalate"
+
+
+def test_flag_follow_up_survives_requires_human_review(settings, unknown_match):
+    """flag_follow_up is an acceptable requires_human_review action — should not be overridden."""
+    classification = make_classification(
+        action="flag_follow_up",
+        confidence=0.80,
+        requires_human_review=True,
+    )
+    final_action, overridden = classifier.decide(unknown_match, classification, settings)
+    assert final_action == "flag_follow_up"
--- a/tests/test_graph_client.py
+++ b/tests/test_graph_client.py
@ -0,0 +1,183 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+import respx
+import httpx
+
+from howl.graph.client import GraphClient, _parse_message, _extract_skip_token, _strip_html
+
+
+GRAPH_BASE = "https://graph.microsoft.com/v1.0"
+
+
+def make_graph_client() -> GraphClient:
+    return GraphClient(token_provider=lambda: "test-bearer-token")
+
+
+# ---------------------------------------------------------------------------
+# Unit tests (no HTTP)
+# ---------------------------------------------------------------------------
+
+def test_parse_message_extracts_fields(sample_emails):
+    raw = sample_emails[0]  # customer inquiry
+    msg = _parse_message(raw)
+    assert msg.id == "msg-001-customer-inquiry"
+    assert msg.sender_address == "alice@acme.com"
+    assert msg.sender_name == "Alice Johnson"
+    assert msg.subject == "Question about my order #45231"
+    assert msg.has_attachments is False
+    assert isinstance(msg.received_at, datetime)
+
+
+def test_parse_message_with_no_sender():
+    raw = {"id": "x", "conversationId": None}
+    msg = _parse_message(raw)
+    assert msg.sender_address == ""
+    assert msg.received_at is None
+
+
+def test_extract_skip_token_from_next_link():
+    link = "https://graph.microsoft.com/v1.0/users/mb/messages?$skipToken=abc123&$top=50"
+    assert _extract_skip_token(link) == "abc123"
+
+
+def test_extract_skip_token_returns_none_when_absent():
+    assert _extract_skip_token("https://graph.microsoft.com/v1.0/users/mb/messages") is None
+
+
+def test_strip_html_removes_tags():
+    html = "<html><body><p>Hello <b>World</b></p></body></html>"
+    text = _strip_html(html)
+    assert "<" not in text
+    assert "Hello" in text
+    assert "World" in text
+
+
+# ---------------------------------------------------------------------------
+# HTTP-level tests with respx
+# ---------------------------------------------------------------------------
+
+@respx.mock
+def test_list_unread_messages_single_page():
+    client = make_graph_client()
+    mailbox = "test@example.com"
+    url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders/Inbox/messages"
+
+    respx.get(url).mock(return_value=httpx.Response(200, json={
+        "value": [
+            {
+                "id": "msg-001",
+                "conversationId": "conv-001",
+                "subject": "Test",
+                "sender": {"emailAddress": {"address": "a@b.com", "name": "A B"}},
+                "receivedDateTime": "2026-04-01T10:00:00Z",
+                "hasAttachments": False,
+                "importance": "normal",
+            }
+        ]
+    }))
+
+    messages, next_skip = client.list_unread_messages(mailbox, top=50)
+    assert len(messages) == 1
+    assert messages[0].id == "msg-001"
+    assert next_skip is None
+
+
+@respx.mock
+def test_list_unread_messages_pagination():
+    client = make_graph_client()
+    mailbox = "test@example.com"
+    url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders/Inbox/messages"
+
+    respx.get(url).mock(return_value=httpx.Response(200, json={
+        "value": [{"id": "msg-001", "conversationId": None, "subject": "S",
+                   "sender": {"emailAddress": {"address": "a@b.com"}},
+                   "receivedDateTime": "2026-04-01T10:00:00Z",
+                   "hasAttachments": False, "importance": "normal"}],
+        "@odata.nextLink": f"{url}?$skipToken=NEXT_TOKEN_ABC"
+    }))
+
+    messages, next_skip = client.list_unread_messages(mailbox, top=1)
+    assert len(messages) == 1
+    assert next_skip == "NEXT_TOKEN_ABC"
+
+
+@respx.mock
+def test_get_message_body_plain_text():
+    client = make_graph_client()
+    mailbox = "test@example.com"
+    message_id = "msg-001"
+    url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}"
+
+    respx.get(url).mock(return_value=httpx.Response(200, json={
+        "body": {"contentType": "text", "content": "Hello, this is the body."}
+    }))
+
+    body = client.get_message_body(mailbox, message_id)
+    assert body == "Hello, this is the body."
+
+
+@respx.mock
+def test_get_message_body_html_stripped():
+    client = make_graph_client()
+    mailbox = "test@example.com"
+    message_id = "msg-002"
+    url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}"
+
+    respx.get(url).mock(return_value=httpx.Response(200, json={
+        "body": {"contentType": "html", "content": "<html><body><p>Hello</p></body></html>"}
+    }))
+
+    body = client.get_message_body(mailbox, message_id)
+    assert "<" not in body
+    assert "Hello" in body
+
+
+@respx.mock
+def test_get_or_create_folder_creates_when_not_found():
+    client = make_graph_client()
+    mailbox = "test@example.com"
+    list_url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders"
+    create_url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders"
+
+    # First call: list returns empty
+    respx.get(list_url).mock(return_value=httpx.Response(200, json={"value": []}))
+    # Second call: create returns new folder
+    respx.post(create_url).mock(return_value=httpx.Response(201, json={
+        "id": "folder-id-123",
+        "displayName": "Customers"
+    }))
+
+    folder_id = client.get_or_create_folder(mailbox, "Customers")
+    assert folder_id == "folder-id-123"
+
+    # Second call should use cache, no additional HTTP requests
+    folder_id_cached = client.get_or_create_folder(mailbox, "Customers")
+    assert folder_id_cached == folder_id
+
+
+@respx.mock
+def test_move_message():
+    client = make_graph_client()
+    mailbox = "test@example.com"
+    message_id = "msg-001"
+    folder_id = "folder-abc"
+    url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}/move"
+
+    respx.post(url).mock(return_value=httpx.Response(201, json={"id": "msg-001-moved"}))
+    client.move_message(mailbox, message_id, folder_id)  # Should not raise
+
+
+@respx.mock
+def test_raises_on_401():
+    client = make_graph_client()
+    mailbox = "test@example.com"
+    url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders/Inbox/messages"
+
+    respx.get(url).mock(return_value=httpx.Response(401, json={"error": {"code": "Unauthorized"}}))
+
+    with pytest.raises(httpx.HTTPStatusError):
+        client.list_unread_messages(mailbox, top=10)
--- a/tests/test_llm_client.py
+++ b/tests/test_llm_client.py
@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from howl.llm.schemas import CLASSIFY_EMAIL_TOOL, EmailClassification
+from howl.llm.prompts import build_prompt, SYSTEM_PROMPT
+
+FIXTURES_DIR = Path(__file__).parent / "fixtures"
+
+
+def test_email_classification_validates_valid_data(llm_responses):
+    cls = EmailClassification.model_validate(llm_responses["customer_inquiry"])
+    assert cls.classification == "customer_inquiry"
+    assert cls.action == "move_customer"
+    assert 0.0 <= cls.confidence <= 1.0
+    assert cls.reasoning
+
+
+def test_email_classification_rejects_invalid_confidence():
+    with pytest.raises(Exception):
+        EmailClassification.model_validate({
+            "classification": "unknown",
+            "confidence": 1.5,  # Out of range
+            "action": "inbox_keep",
+            "reasoning": "test",
+            "priority": "normal",
+            "requires_human_review": False,
+        })
+
+
+def test_email_classification_requires_human_review_for_phishing(llm_responses):
+    cls = EmailClassification.model_validate(llm_responses["phishing"])
+    assert cls.requires_human_review is True
+
+
+def test_email_classification_low_confidence(llm_responses):
+    cls = EmailClassification.model_validate(llm_responses["low_confidence"])
+    assert cls.confidence < 0.60
+    assert cls.requires_human_review is True
+
+
+def test_build_prompt_includes_sender_info(customer_message, customer_match):
+    prompt = build_prompt(customer_message, "Hello, checking on my order.", customer_match)
+    assert customer_message.sender_address in prompt
+    assert "customer" in prompt
+    assert "Acme Corp" in prompt
+    assert "Key account" in prompt
+
+
+def test_build_prompt_truncates_body(customer_message, unknown_match):
+    long_body = "x" * 10000
+    prompt = build_prompt(customer_message, long_body, unknown_match, max_body_chars=100)
+    assert "truncated" in prompt
+    # Body in prompt should be 100 chars, not 10000
+    assert "x" * 101 not in prompt
+
+
+def test_build_prompt_unknown_sender(spam_message, unknown_match):
+    prompt = build_prompt(spam_message, "Claim your prize!", unknown_match)
+    assert "unknown" in prompt
+    # No entity name should appear
+    assert "Known as:" not in prompt
+
+
+def test_classify_email_tool_schema_is_valid():
+    """Tool definition should have all required fields for Anthropic tool use."""
+    assert CLASSIFY_EMAIL_TOOL["name"] == "classify_email"
+    schema = CLASSIFY_EMAIL_TOOL["input_schema"]
+    assert schema["type"] == "object"
+    required = schema["required"]
+    assert "classification" in required
+    assert "confidence" in required
+    assert "action" in required
+    assert "reasoning" in required
+    assert "requires_human_review" in required
+
+
+def test_llm_client_calls_anthropic_and_parses_response(
+    settings, customer_message, customer_match
+):
+    """Test that LLMClient correctly calls the Anthropic SDK and parses tool use output."""
+    from howl.llm.client import LLMClient
+
+    mock_tool_use = MagicMock()
+    mock_tool_use.type = "tool_use"
+    mock_tool_use.input = {
+        "classification": "customer_inquiry",
+        "confidence": 0.92,
+        "action": "move_customer",
+        "reasoning": "Known customer inquiry.",
+        "priority": "normal",
+        "requires_human_review": False,
+        "tags": [],
+    }
+
+    mock_response = MagicMock()
+    mock_response.content = [mock_tool_use]
+    mock_response.model = "claude-sonnet-4-6"
+    mock_response.stop_reason = "tool_use"
+    mock_response.usage.input_tokens = 300
+    mock_response.usage.output_tokens = 75
+
+    with patch("howl.llm.client.Anthropic") as MockAnthropic:
+        MockAnthropic.return_value.messages.create.return_value = mock_response
+        client = LLMClient(settings)
+        classification, raw = client.classify(
+            customer_message, "Hello, checking my order status.", customer_match
+        )
+
+    assert classification.classification == "customer_inquiry"
+    assert classification.action == "move_customer"
+    assert raw["usage"]["input_tokens"] == 300
+
+
+def test_llm_client_raises_if_no_tool_use(settings, customer_message, customer_match):
+    """LLMClient should raise ValueError if Claude doesn't call the tool."""
+    from howl.llm.client import LLMClient
+
+    mock_text_block = MagicMock()
+    mock_text_block.type = "text"
+
+    mock_response = MagicMock()
+    mock_response.content = [mock_text_block]
+
+    with patch("howl.llm.client.Anthropic") as MockAnthropic:
+        MockAnthropic.return_value.messages.create.return_value = mock_response
+        client = LLMClient(settings)
+        with pytest.raises(ValueError, match="classify_email"):
+            client.classify(customer_message, "body", customer_match)
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@ -0,0 +1,187 @@
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+from datetime import datetime, timezone
+
+import pytest
+
+from howl.db.queries import SenderMatch
+from howl.llm.schemas import EmailClassification
+from howl.pipeline import classifier
+from howl.pipeline.actions import ActionExecutor
+from howl.pipeline.processor import EmailProcessor
+
+
+# ---------------------------------------------------------------------------
+# ActionExecutor tests (dry_run mode — no actual HTTP calls)
+# ---------------------------------------------------------------------------
+
+def test_action_executor_dry_run_does_not_call_graph(settings, mock_graph):
+    from howl.db.models import EmailLog
+    assert settings.dry_run is True  # conftest sets dry_run=True
+
+    executor = ActionExecutor(mock_graph, settings)
+
+    log_entry = MagicMock(spec=EmailLog)
+    log_entry.mailbox = "test@example.com"
+    log_entry.graph_message_id = "msg-001"
+    log_entry.sender_name = "Alice"
+    log_entry.sender_address = "alice@test.com"
+    log_entry.subject = "Test"
+
+    executor.execute(log_entry, "move_customer")
+    mock_graph.move_message.assert_not_called()
+    mock_graph.get_or_create_folder.assert_not_called()
+
+
+def test_action_executor_move_customer_live(settings, mock_graph):
+    settings = settings.model_copy(update={"dry_run": False})
+    from howl.db.models import EmailLog
+
+    executor = ActionExecutor(mock_graph, settings)
+    log_entry = MagicMock(spec=EmailLog)
+    log_entry.mailbox = "test@example.com"
+    log_entry.graph_message_id = "msg-001"
+    log_entry.sender_name = "Alice"
+    log_entry.sender_address = "alice@test.com"
+    log_entry.subject = "Test"
+
+    executor.execute(log_entry, "move_customer")
+    mock_graph.get_or_create_folder.assert_called_once_with("test@example.com", settings.folder_customers)
+    mock_graph.move_message.assert_called_once()
+
+
+def test_action_executor_escalate_sends_notification(settings, mock_graph):
+    settings = settings.model_copy(update={
+        "dry_run": False,
+        "notification_webhook_url": "https://hooks.example.com/notify",
+    })
+    from howl.db.models import EmailLog
+
+    executor = ActionExecutor(mock_graph, settings)
+    log_entry = MagicMock(spec=EmailLog)
+    log_entry.mailbox = "test@example.com"
+    log_entry.graph_message_id = "msg-escalate"
+    log_entry.sender_name = "Bob"
+    log_entry.sender_address = "bob@client.com"
+    log_entry.subject = "URGENT"
+
+    with patch("httpx.post") as mock_post:
+        mock_post.return_value = MagicMock(status_code=200, raise_for_status=MagicMock())
+        executor.execute(log_entry, "escalate")
+
+    mock_graph.flag_message.assert_called_once()
+    mock_post.assert_called_once()
+    call_kwargs = mock_post.call_args
+    assert call_kwargs[0][0] == "https://hooks.example.com/notify"
+    assert "text" in call_kwargs[1]["json"]
+    assert call_kwargs[1]["timeout"] == 10
+
+
+# ---------------------------------------------------------------------------
+# Full pipeline integration (all external calls mocked)
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_processor_process_message_happy_path(
+    settings, mock_graph, mock_llm, customer_message, customer_match
+):
+    """Full pipeline: message → DB lookup → LLM → classify → action → log."""
+    processor = EmailProcessor(mock_graph, mock_llm, settings)
+
+    inserted_log = MagicMock()
+    inserted_log.id = "log-uuid-001"
+    inserted_log.retry_count = 0
+
+    with (
+        patch("howl.pipeline.processor.db_engine") as mock_db_engine,
+        patch("howl.pipeline.processor.queries") as mock_queries,
+    ):
+        # Set up async session context manager
+        mock_session = AsyncMock()
+        mock_db_engine.async_session.return_value.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_db_engine.async_session.return_value.__aexit__ = AsyncMock(return_value=False)
+
+        mock_queries.is_already_processed = AsyncMock(return_value=False)
+        mock_queries.insert_email_log = AsyncMock(return_value=inserted_log)
+        mock_queries.lookup_sender = AsyncMock(return_value=customer_match)
+        mock_queries.update_email_log = AsyncMock()
+
+        await processor.process_message(customer_message)
+
+    # Verify DB was consulted for idempotency
+    mock_queries.is_already_processed.assert_called_once_with(mock_session, customer_message.id)
+
+    # Verify email log was created
+    mock_queries.insert_email_log.assert_called_once()
+
+    # Verify sender was looked up
+    mock_queries.lookup_sender.assert_called_once_with(mock_session, customer_message.sender_address)
+
+    # Verify LLM was called
+    mock_llm.classify.assert_called_once()
+
+    # Verify log was updated with completed status
+    final_update_calls = [
+        c for c in mock_queries.update_email_log.call_args_list
+        if c.kwargs.get("status") == "completed" or (len(c.args) > 2 and "completed" in str(c))
+    ]
+    # At minimum update_email_log should have been called
+    assert mock_queries.update_email_log.call_count >= 2
+
+
+@pytest.mark.asyncio
+async def test_processor_skips_already_processed_message(
+    settings, mock_graph, mock_llm, customer_message
+):
+    processor = EmailProcessor(mock_graph, mock_llm, settings)
+
+    with (
+        patch("howl.pipeline.processor.db_engine") as mock_db_engine,
+        patch("howl.pipeline.processor.queries") as mock_queries,
+    ):
+        mock_session = AsyncMock()
+        mock_db_engine.async_session.return_value.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_db_engine.async_session.return_value.__aexit__ = AsyncMock(return_value=False)
+        mock_queries.is_already_processed = AsyncMock(return_value=True)
+
+        await processor.process_message(customer_message)
+
+    # Should not proceed past idempotency check
+    mock_llm.classify.assert_not_called()
+    mock_graph.move_message.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_processor_handles_llm_failure_gracefully(
+    settings, mock_graph, mock_llm, customer_message, customer_match
+):
+    """If LLM throws, the pipeline should catch it and log failure — not crash."""
+    mock_llm.classify.side_effect = RuntimeError("Anthropic API timeout")
+
+    processor = EmailProcessor(mock_graph, mock_llm, settings)
+    inserted_log = MagicMock()
+    inserted_log.id = "log-uuid-002"
+    inserted_log.retry_count = 0
+
+    with (
+        patch("howl.pipeline.processor.db_engine") as mock_db_engine,
+        patch("howl.pipeline.processor.queries") as mock_queries,
+    ):
+        mock_session = AsyncMock()
+        mock_db_engine.async_session.return_value.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_db_engine.async_session.return_value.__aexit__ = AsyncMock(return_value=False)
+        mock_queries.is_already_processed = AsyncMock(return_value=False)
+        mock_queries.insert_email_log = AsyncMock(return_value=inserted_log)
+        mock_queries.lookup_sender = AsyncMock(return_value=customer_match)
+        mock_queries.update_email_log = AsyncMock()
+
+        # Should NOT raise
+        await processor.process_message(customer_message)
+
+    # Should have logged failure
+    failed_updates = [
+        c for c in mock_queries.update_email_log.call_args_list
+        if c.kwargs.get("status") == "failed"
+    ]
+    assert len(failed_updates) >= 1