feat(alerts): add Pulse AI double opt-in alert subscriptions system

- Add 10 alert criteria types with RPC-based evaluation and 24h rolling cooldown
- Implement pluggable email service with Gmail SMTP and unsubscribe links
- Add ALERT_SUBSCRIPTION query classification and isolated 2-tool resolution
- Add confirmation, unsubscribe, and internal evaluate API routes
- Add background evaluator, digest sender, and cleanup workers
- Add comprehensive test suite with 45 unit tests and 44 simulated tool tests

config.py

@@ -37,3 +37,19 @@ CONTACT_WHATSAPP = os.environ.get("CONTACT_WHATSAPP", "1234567890")
 
 MARKET_CACHE_TTL_SECONDS = int(os.environ.get("MARKET_CACHE_TTL_SECONDS", "300"))
 GEOCODE_COUNTRY_FILTER = os.environ.get("GEOCODE_COUNTRY_FILTER", "")
+
+EMAIL_SMTP_HOST = os.environ.get("EMAIL_SMTP_HOST", "smtp.gmail.com")
+EMAIL_SMTP_PORT = int(os.environ.get("EMAIL_SMTP_PORT", "587"))
+EMAIL_FROM_ADDRESS = os.environ.get("EMAIL_FROM_ADDRESS", "jouledynamicscto@gmail.com")
+EMAIL_SMTP_PASSWORD = os.environ.get("EMAIL_SMTP_PASSWORD", "")
+ALERT_BASE_URL = os.environ.get(
+    "ALERT_BASE_URL",
+    os.environ.get("CONFIRMATION_BASE_URL", "https://johnalbarkaibrahim-sentimentscope.hf.space"),
+)
+SUPABASE_SERVICE_ROLE_KEY = os.environ.get(
+    "SUPABASE_SERVICE_ROLE_KEY", os.environ.get("SUPABASE_KEY", "")
+)
+INTERNAL_API_KEY = os.environ.get("INTERNAL_API_KEY", "")
+ALERT_COOLDOWN_HOURS = int(os.environ.get("ALERT_COOLDOWN_HOURS", "24"))
+MAX_SUBSCRIPTIONS_PER_EMAIL = int(os.environ.get("MAX_SUBSCRIPTIONS_PER_EMAIL", "5"))
+CONFIRMATION_EXPIRY_HOURS = int(os.environ.get("CONFIRMATION_EXPIRY_HOURS", "48"))

misc/alert_subscriptions_migration.sql

@@ -0,0 +1,27 @@
+-- Pulse AI Alert Subscriptions — Table + Indexes
+-- Apply this in the Supabase SQL Editor (Dashboard → SQL Editor → New Query)
+
+CREATE TABLE alert_subscriptions (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    email TEXT NOT NULL,
+    criteria_type TEXT NOT NULL CHECK (criteria_type IN (
+        'spike', 'rate_change', 'price_threshold', 'availability_change',
+        'new_listing', 'tracking_removed', 'anomaly', 'volatility',
+        'trend_reversal', 'digest'
+    )),
+    criteria JSONB NOT NULL,
+    raw_request_text TEXT,
+    confirmed BOOLEAN NOT NULL DEFAULT false,
+    confirmation_token TEXT NOT NULL UNIQUE,
+    unsubscribe_token TEXT NOT NULL UNIQUE,
+    status TEXT NOT NULL DEFAULT 'pending',
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    confirmed_at TIMESTAMPTZ,
+    last_checked_at TIMESTAMPTZ,
+    last_fired_at TIMESTAMPTZ
+);
+
+CREATE INDEX idx_alert_subs_status ON alert_subscriptions(status) WHERE status = 'active';
+CREATE INDEX idx_alert_subs_email ON alert_subscriptions(email);
+CREATE INDEX idx_alert_subs_token ON alert_subscriptions(confirmation_token);
+CREATE INDEX idx_alert_subs_unsub ON alert_subscriptions(unsubscribe_token);

misc/pulse-alert-subscriptions-spec.md

@@ -0,0 +1,161 @@
+# Pulse AI — Alert Subscriptions: Full Technical Specification
+
+## Architecture Overview
+
+```
+User: "let me know when property_a in Miami spikes again"
+        ↓
+ROUTER: classifies as ALERT_SUBSCRIPTION (fast, cheap — exactly 7 classifications)
+        ↓
+AGENT LOOP (_resolve_active_tools):
+  - Injects exactly 2 tools: create_alert_subscription + suggest_actions
+  - Prompt: _PROMPT_BASE + _PROMPT_ALERT + _PROMPT_RESPONSE_RULES
+        ↓
+SYNTHESIS:
+  - Missing details? → invokes suggest_actions with concrete chips
+  - Details complete? → invokes create_alert_subscription (PENDING, unconfirmed)
+        ↓
+CONFIRMATION EMAIL:
+  - System sends double opt-in email with single-use verification token
+  - Link: https://pulse.jouledynamics.com/alerts/confirm?token=<token>
+        ↓
+CONFIRMATION ENDPOINT (GET /api/v1/alerts/confirm):
+  - Service role validates token, sets confirmed = true, status = 'active'
+  - Pulse agent itself CANNOT confirm — double opt-in strictly enforced
+        ↓
+EVALUATION WORKERS (Post-scrape / Scheduled / Webhook):
+  - Alert Evaluator: evaluates 9 event-driven criteria types against fresh data via existing Supabase RPCs
+  - Digest Sender: evaluates scheduled digest subscriptions (daily/weekly)
+  - Cleanup Job: expires and purges unconfirmed pending subscriptions > 48h
+        ↓
+COOLDOWN ENFORCEMENT & DELIVERY:
+  - 24-hour rolling cooldown prevents duplicate fires for persistent conditions
+  - HTML email with unsubscribe link (GET /api/v1/alerts/unsubscribe?token=<token>)
+```
+
+---
+
+## 1. Supported Alert Criteria Types (10 Types)
+
+The alert system supports 10 distinct criteria types spanning event-driven, threshold, and scheduled monitoring:
+
+| `criteria_type` | Description | Target / Required Criteria | Evaluator RPC Used | Default Cooldown |
+|---|---|---|---|---|
+| `spike` | Price or rate spikes exceeding threshold | `{"market": "Miami", "threshold_pct": 25, "direction": "both"}` | `get_spike_alerts` | 24 hours |
+| `rate_change` | Material rate changes over window | `{"market": "Abuja", "days": 7, "min_change_pct": 10}` | `get_market_rate_changes` | 24 hours |
+| `price_threshold` | Property price crosses numeric limit | `{"property_search": "...", "operator": "below", "value": 150000}` | `get_property_snapshot` | 24 hours |
+| `availability_change` | Property or market occupancy change | `{"property_search": "..."}` or `{"market": "Miami"}` | `get_property_snapshot` / `get_availability_rate` | 24 hours |
+| `new_listing` | New property added to tracking | `{"market": "Lagos"}` | `get_recently_changed_tracking` | 24 hours |
+| `tracking_removed` | Property delisted or untracked | `{"market": "Lagos"}` or `{"property_search": "..."}` | `get_recently_changed_tracking` | 24 hours |
+| `anomaly` | Statistical rate deviation | `{"property_search": "...", "days": 30}` | `get_rate_anomaly_report` | 24 hours |
+| `volatility` | High rate fluctuation in market | `{"market": "Abuja", "days": 14, "limit": 5}` | `get_most_volatile_properties` | 24 hours |
+| `trend_reversal` | Market trend direction shift | `{"market": "Miami", "days": 14}` | `get_market_trend` | 24 hours |
+| `digest` | Periodic market recap | `{"market": "Miami", "frequency": "weekly"}` (`"daily"` or `"weekly"`) | `get_real_estate_kpis` + `get_market_averages` | Cron-driven |
+
+---
+
+## 2. Router Integration (`services/query_router.py`)
+
+`ALERT_SUBSCRIPTION` is the 7th valid classification alongside:
+- `OUT_OF_SCOPE`
+- `PATH_A` (live market intelligence)
+- `PATH_B` (general real estate / UI concepts)
+- `BOTH`
+- `GREETING`
+- `COMMERCIAL_HANDOFF`
+- `ALERT_SUBSCRIPTION`
+
+Prompt instructions guide the router to detect explicit future-monitoring intent ("alert me when...", "notify me if...", "send me a weekly digest of...").
+
+---
+
+## 3. Tool Definition & Token Budget (`services/alert_tool_schema.py`)
+
+To preserve strict token economy:
+- Exactly 2 tools are exposed during `ALERT_SUBSCRIPTION` turns: `create_alert_subscription` and `suggest_actions`.
+- No heavy SQL or market analysis tools are loaded into context during subscription creation.
+
+```json
+{
+  "type": "function",
+  "function": {
+    "name": "create_alert_subscription",
+    "description": "Create a PENDING alert subscription. Never call this until you have the condition type, target (market or property), and email address.",
+    "parameters": {
+      "type": "object",
+      "properties": {
+        "email": {"type": "string"},
+        "criteria_type": {
+          "type": "string",
+          "enum": [
+            "spike", "rate_change", "price_threshold", "availability_change",
+            "new_listing", "tracking_removed", "anomaly", "volatility",
+            "trend_reversal", "digest"
+          ]
+        },
+        "criteria": {"type": "object"},
+        "raw_request_text": {"type": "string"}
+      },
+      "required": ["email", "criteria_type", "criteria", "raw_request_text"]
+    }
+  }
+}
+```
+
+---
+
+## 4. Prompt Extension (`_PROMPT_ALERT` in `services/prompts.py`)
+
+When `classification == "ALERT_SUBSCRIPTION"`, the system prompt injects `_PROMPT_ALERT`:
+- Demands 3 essential fields before subscription creation: WHAT condition, WHICH market/property, and EMAIL ADDRESS.
+- Enforces conversational follow-up via `suggest_actions` if any field is missing.
+- Explains that the subscription will be created in `PENDING` state and requires confirmation via email.
+- Enforces anti-spam and privacy constraints (never store personal data beyond email).
+
+---
+
+## 5. Security Architecture & RLS Boundary
+
+1. **Table Schema (`misc/alert_subscriptions_migration.sql`):**
+   - Columns: `id`, `email`, `criteria_type`, `criteria`, `raw_request_text`, `confirmed`, `confirmation_token`, `unsubscribe_token`, `status`, `created_at`, `confirmed_at`, `last_checked_at`, `last_fired_at`.
+   - Double opt-in default: `confirmed BOOLEAN NOT NULL DEFAULT false`, `status TEXT NOT NULL DEFAULT 'pending'`.
+2. **Access Control:**
+   - Frontend / Public agent role: `INSERT` only (cannot `SELECT`, `UPDATE`, or `DELETE`).
+   - Service Role key (`SUPABASE_SERVICE_ROLE_KEY`): Used exclusively by backend API routes and workers to confirm, update `last_fired_at`, and unsubscribe.
+3. **Caps:** Maximum 5 active subscriptions per email address to prevent abuse. Friendly error message returned if exceeded.
+
+---
+
+## 6. Endpoints (`routes/v1/alerts.py`)
+
+- `GET /api/v1/alerts/confirm?token=<token>`: Validates confirmation token, activates subscription (`confirmed = true, status = 'active'`), returns branded HTML response.
+- `GET /api/v1/alerts/unsubscribe?token=<token>`: Validates unsubscribe token, cancels subscription (`status = 'unsubscribed'`), returns branded confirmation.
+- `POST /api/v1/internal/evaluate-alerts`: Protected internal endpoint called by scrapers or webhooks with `X-Internal-Secret` header to trigger evaluation runs.
+
+---
+
+## 7. Workers & Automation (`workers/`)
+
+1. **`alert_evaluator.py`**:
+   - Dispatches across the 9 event-driven criteria types.
+   - Reuses existing audited RPCs (`get_spike_alerts`, `get_property_snapshot`, `get_rate_anomaly_report`, etc.).
+   - Enforces 24-hour rolling cooldown (`ALERT_COOLDOWN_HOURS = 24`).
+2. **`digest_sender.py`**:
+   - Queries active `digest` subscriptions.
+   - Compiles KPIs and average metrics via `get_real_estate_kpis` and `get_market_averages`.
+   - Sends daily or weekly digests based on elapsed time since `last_fired_at`.
+3. **`alert_cleanup.py`**:
+   - Deletes unconfirmed subscriptions older than `CONFIRMATION_EXPIRY_HOURS = 48`.
+
+---
+
+## 8. Pluggable Email Service (`services/email_service.py`)
+
+- Currently backed by Gmail SMTP via `SMTP_USER` and `SMTP_PASSWORD`.
+- Configured via environment variables with fallback support:
+  - `SMTP_HOST` (default: `smtp.gmail.com`)
+  - `SMTP_PORT` (default: `587`)
+  - `SMTP_USER` (default: `jouledynamicscto@gmail.com`)
+  - `FROM_EMAIL` (default: `jouledynamicscto@gmail.com`)
+  - `CONFIRMATION_BASE_URL` (default: `https://pulse.jouledynamics.com`)
+- Ready to swap provider (e.g. Resend, Zoho) purely via `.env` configuration without modifying business logic.

routes/v1/__init__.py

@@ -6,8 +6,11 @@ from fastapi import APIRouter
 from routes.v1.real_estate import router as real_estate_router
 from routes.v1.rag import router as rag_router
 from routes.v1.health import router as health_router
+from routes.v1.alerts import router as alerts_router
 
 router = APIRouter()
 router.include_router(real_estate_router)
 router.include_router(rag_router)
 router.include_router(health_router)
+router.include_router(alerts_router)
+

routes/v1/alerts.py

@@ -0,0 +1,92 @@
+"""Alert subscription management endpoints (confirmation, unsubscribe, worker trigger)."""
+
+from fastapi import APIRouter, Request, status
+from fastapi.responses import HTMLResponse, JSONResponse
+
+from config import INTERNAL_API_KEY
+from services.alert_service import confirm_subscription, unsubscribe
+from services.observability import setup_logger
+
+logger = setup_logger(__name__)
+
+router = APIRouter(
+    prefix="/api/v1",
+    tags=["Alert Subscriptions"],
+)
+
+_SUCCESS_HTML = """
+<html><body style="font-family: Arial, sans-serif; max-width: 600px; margin: 80px auto; text-align: center;">
+<h2 style="color: #1a1a2e;">{title}</h2>
+<p>{message}</p>
+<p style="color: #999; font-size: 12px; margin-top: 40px;">Pulse AI — Real Estate Intelligence by Joule Dynamics</p>
+</body></html>
+"""
+
+_ERROR_HTML = """
+<html><body style="font-family: Arial, sans-serif; max-width: 600px; margin: 80px auto; text-align: center;">
+<h2 style="color: #c0392b;">{title}</h2>
+<p>{message}</p>
+<p style="color: #999; font-size: 12px; margin-top: 40px;">Pulse AI — Real Estate Intelligence by Joule Dynamics</p>
+</body></html>
+"""
+
+
+@router.get("/alerts/confirm", response_class=HTMLResponse)
+async def confirm_alert(token: str):
+    """Confirms a pending alert subscription via token link click."""
+    if not token:
+        return HTMLResponse(
+            _ERROR_HTML.format(title="Invalid Link", message="No confirmation token provided."),
+            status_code=status.HTTP_400_BAD_REQUEST,
+        )
+
+    result = await confirm_subscription(token)
+
+    if result["status"] in ("active", "already_active"):
+        return HTMLResponse(
+            _SUCCESS_HTML.format(title="✅ Subscription Confirmed!", message=result["message"]),
+            status_code=status.HTTP_200_OK,
+        )
+
+    return HTMLResponse(
+        _ERROR_HTML.format(title="Confirmation Failed", message=result["message"]),
+        status_code=status.HTTP_400_BAD_REQUEST,
+    )
+
+
+@router.get("/alerts/unsubscribe", response_class=HTMLResponse)
+async def unsubscribe_alert(token: str):
+    """Unsubscribes from an active alert subscription."""
+    if not token:
+        return HTMLResponse(
+            _ERROR_HTML.format(title="Invalid Link", message="No unsubscribe token provided."),
+            status_code=status.HTTP_400_BAD_REQUEST,
+        )
+
+    result = await unsubscribe(token)
+
+    if result["status"] in ("unsubscribed", "already_unsubscribed"):
+        return HTMLResponse(
+            _SUCCESS_HTML.format(title="Unsubscribed", message=result["message"]),
+            status_code=status.HTTP_200_OK,
+        )
+
+    return HTMLResponse(
+        _ERROR_HTML.format(title="Unsubscribe Failed", message=result["message"]),
+        status_code=status.HTTP_400_BAD_REQUEST,
+    )
+
+
+@router.post("/internal/evaluate-alerts")
+async def trigger_alert_evaluation(request: Request):
+    """Internal endpoint: evaluates all active subscriptions. Secured by X-Internal-Key header."""
+    provided_key = request.headers.get("X-Internal-Key", "")
+    if not INTERNAL_API_KEY or provided_key != INTERNAL_API_KEY:
+        return JSONResponse(
+            status_code=status.HTTP_403_FORBIDDEN,
+            content={"error": "Forbidden"},
+        )
+
+    from workers.alert_evaluator import evaluate_all_active_subscriptions
+    result = await evaluate_all_active_subscriptions()
+    return result

services/agent_loop.py

@@ -372,6 +372,10 @@ def _resolve_active_tools(classification: str, user_query: str, messages: list[d
     if classification == "PATH_B":
         return [SUGGEST_ACTIONS_TOOL]
 
+    if classification == "ALERT_SUBSCRIPTION":
+        from services.alert_tool_schema import CREATE_ALERT_SUBSCRIPTION_TOOL
+        return [CREATE_ALERT_SUBSCRIPTION_TOOL, SUGGEST_ACTIONS_TOOL]
+
     return None
 
 

services/alert_service.py

@@ -0,0 +1,177 @@
+"""Alert subscription lifecycle: create (pending), confirm, unsubscribe."""
+
+import secrets
+from datetime import datetime, timezone
+
+from supabase import create_client
+
+from config import (
+    MAX_SUBSCRIPTIONS_PER_EMAIL,
+    SUPABASE_SERVICE_ROLE_KEY,
+    SUPABASE_URL,
+)
+from services.email_service import (
+    build_criteria_summary,
+    send_confirmation_email,
+    send_subscription_active_email,
+)
+from services.observability import setup_logger
+
+logger = setup_logger(__name__)
+
+_service_client = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)
+
+
+async def create_pending_subscription(
+    email: str, criteria_type: str, criteria: dict, raw_text: str
+) -> dict:
+    """INSERT pending subscription and send confirmation email.
+
+    Enforces per-email subscription cap. Returns a natural language
+    rejection message if the cap is exceeded so the LLM can relay it.
+    """
+    if not email or not criteria_type:
+        return {
+            "status": "error",
+            "message": "Email address and alert type are both required to create a subscription.",
+        }
+
+    active_count = await _count_active_subscriptions(email)
+    if active_count >= MAX_SUBSCRIPTIONS_PER_EMAIL:
+        return {
+            "status": "limit_reached",
+            "message": (
+                f"You already have {active_count} active alert subscriptions on this email address. "
+                f"The maximum is {MAX_SUBSCRIPTIONS_PER_EMAIL}. "
+                "To add a new one, you'd need to unsubscribe from an existing alert first. "
+                "Check your email for unsubscribe links in any previous alert emails."
+            ),
+        }
+
+    confirmation_token = secrets.token_urlsafe(32)
+    unsubscribe_token = secrets.token_urlsafe(32)
+
+    try:
+        _service_client.table("alert_subscriptions").insert({
+            "email": email,
+            "criteria_type": criteria_type,
+            "criteria": criteria,
+            "raw_request_text": raw_text,
+            "confirmation_token": confirmation_token,
+            "unsubscribe_token": unsubscribe_token,
+            "confirmed": False,
+            "status": "pending",
+        }).execute()
+    except Exception as exc:
+        logger.error(f"[alert_service] insert failed: {exc}")
+        return {"status": "error", "message": "Failed to create subscription. Please try again."}
+
+    summary = build_criteria_summary(criteria_type, criteria)
+    await send_confirmation_email(email, confirmation_token, summary)
+
+    logger.info(f"[alert_service] pending subscription created for {email}: {criteria_type}")
+    return {
+        "status": "pending",
+        "message": (
+            f"Alert subscription created! A confirmation email has been sent to {email}. "
+            "Please click the link in that email to activate the alert — "
+            "it won't start monitoring until you confirm."
+        ),
+    }
+
+
+async def confirm_subscription(token: str) -> dict:
+    """Confirm a pending subscription via confirmation token.
+
+    Sets confirmed=true, status='active', sends subscription-active email.
+    """
+    try:
+        result = (
+            _service_client.table("alert_subscriptions")
+            .select("id, email, criteria_type, criteria, unsubscribe_token, status")
+            .eq("confirmation_token", token)
+            .execute()
+        )
+    except Exception as exc:
+        logger.error(f"[alert_service] confirm lookup failed: {exc}")
+        return {"status": "error", "message": "Unable to process confirmation."}
+
+    if not result.data:
+        return {"status": "error", "message": "Invalid or expired confirmation link."}
+
+    sub = result.data[0]
+    if sub["status"] == "active":
+        return {"status": "already_active", "message": "This subscription is already active."}
+
+    if sub["status"] != "pending":
+        return {"status": "error", "message": "This subscription cannot be confirmed."}
+
+    try:
+        _service_client.table("alert_subscriptions").update({
+            "confirmed": True,
+            "status": "active",
+            "confirmed_at": datetime.now(timezone.utc).isoformat(),
+            "confirmation_token": None,
+        }).eq("id", sub["id"]).execute()
+    except Exception as exc:
+        logger.error(f"[alert_service] confirm update failed: {exc}")
+        return {"status": "error", "message": "Unable to activate subscription."}
+
+    await send_subscription_active_email(
+        to=sub["email"],
+        criteria_type=sub["criteria_type"],
+        criteria=sub["criteria"],
+        unsubscribe_token=sub["unsubscribe_token"],
+    )
+
+    logger.info(f"[alert_service] subscription confirmed: {sub['id']}")
+    return {"status": "active", "message": "Your alert subscription is now active!"}
+
+
+async def unsubscribe(token: str) -> dict:
+    """Unsubscribe from an alert subscription via unsubscribe token."""
+    try:
+        result = (
+            _service_client.table("alert_subscriptions")
+            .select("id, status")
+            .eq("unsubscribe_token", token)
+            .execute()
+        )
+    except Exception as exc:
+        logger.error(f"[alert_service] unsubscribe lookup failed: {exc}")
+        return {"status": "error", "message": "Unable to process unsubscribe request."}
+
+    if not result.data:
+        return {"status": "error", "message": "Invalid unsubscribe link."}
+
+    sub = result.data[0]
+    if sub["status"] == "unsubscribed":
+        return {"status": "already_unsubscribed", "message": "You've already unsubscribed from this alert."}
+
+    try:
+        _service_client.table("alert_subscriptions").update({
+            "status": "unsubscribed",
+            "unsubscribe_token": None,
+        }).eq("id", sub["id"]).execute()
+    except Exception as exc:
+        logger.error(f"[alert_service] unsubscribe update failed: {exc}")
+        return {"status": "error", "message": "Unable to process unsubscribe request."}
+
+    logger.info(f"[alert_service] unsubscribed: {sub['id']}")
+    return {"status": "unsubscribed", "message": "You've been unsubscribed. You won't receive any more alerts for this subscription."}
+
+
+async def _count_active_subscriptions(email: str) -> int:
+    """Count active + pending subscriptions for rate-limiting."""
+    try:
+        result = (
+            _service_client.table("alert_subscriptions")
+            .select("id", count="exact")
+            .eq("email", email)
+            .in_("status", ["active", "pending"])
+            .execute()
+        )
+        return result.count or 0
+    except Exception as exc:
+        logger.error(f"[alert_service] count failed: {exc}")
+        return 0

services/alert_tool_schema.py

@@ -0,0 +1,51 @@
+"""Tool schema for the create_alert_subscription LLM tool."""
+
+CREATE_ALERT_SUBSCRIPTION_TOOL = {
+    "type": "function",
+    "function": {
+        "name": "create_alert_subscription",
+        "description": (
+            "Create a PENDING alert subscription. Never call until you have: "
+            "condition type, target (market or property), and email address. "
+            "Does not activate the alert — confirmation happens via email link."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "email": {
+                    "type": "string",
+                    "description": "User's email address for receiving alerts",
+                },
+                "criteria_type": {
+                    "type": "string",
+                    "enum": [
+                        "spike", "rate_change", "price_threshold",
+                        "availability_change", "new_listing", "tracking_removed",
+                        "anomaly", "volatility", "trend_reversal", "digest",
+                    ],
+                    "description": "Type of condition to monitor",
+                },
+                "criteria": {
+                    "type": "object",
+                    "description": (
+                        "Type-specific parameters. Examples: "
+                        "spike: {market, threshold_pct}; "
+                        "rate_change: {market, direction: both|up|down}; "
+                        "price_threshold: {property_search, operator: above|below, value}; "
+                        "availability_change: {property_search, watch_for: available|unavailable|any}; "
+                        "new_listing/tracking_removed: {market}; "
+                        "anomaly: {property_search, deviation_threshold}; "
+                        "volatility: {market, top_n}; "
+                        "trend_reversal: {market, watch_for: rising|falling}; "
+                        "digest: {market, frequency: daily|weekly}"
+                    ),
+                },
+                "raw_request_text": {
+                    "type": "string",
+                    "description": "Original user request verbatim",
+                },
+            },
+            "required": ["email", "criteria_type", "criteria", "raw_request_text"],
+        },
+    },
+}

services/email_service.py

@@ -0,0 +1,197 @@
+"""Pluggable email dispatch interface with Gmail SMTP implementation."""
+
+import asyncio
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+
+from config import (
+    ALERT_BASE_URL,
+    EMAIL_FROM_ADDRESS,
+    EMAIL_SMTP_HOST,
+    EMAIL_SMTP_PASSWORD,
+    EMAIL_SMTP_PORT,
+)
+from services.observability import setup_logger
+
+logger = setup_logger(__name__)
+
+_CRITERIA_LABELS = {
+    "spike": "Price spike",
+    "rate_change": "Rate change",
+    "price_threshold": "Price threshold",
+    "availability_change": "Availability change",
+    "new_listing": "New listing tracked",
+    "tracking_removed": "Property delisted",
+    "anomaly": "Pricing anomaly",
+    "volatility": "Volatility alert",
+    "trend_reversal": "Market trend reversal",
+    "digest": "Market digest",
+}
+
+_FREQUENCY_DESCRIPTIONS = {
+    "spike": "Checked 4× daily after each data scrape. Max 1 alert per 24 hours.",
+    "rate_change": "Checked 4× daily after each data scrape. Max 1 alert per 24 hours.",
+    "price_threshold": "Checked 4× daily after each data scrape. Max 1 alert per 24 hours.",
+    "availability_change": "Checked 4× daily after each data scrape. Max 1 alert per 24 hours.",
+    "new_listing": "Checked 4× daily after each data scrape. Max 1 alert per 24 hours.",
+    "tracking_removed": "Checked 4× daily after each data scrape. Max 1 alert per 24 hours.",
+    "anomaly": "Checked 4× daily after each data scrape. Max 1 alert per 24 hours.",
+    "volatility": "Checked 4× daily after each data scrape. Max 1 alert per 24 hours.",
+    "trend_reversal": "Checked 4× daily after each data scrape. Max 1 alert per 24 hours.",
+    "digest": "Sent on your chosen schedule (daily or weekly).",
+}
+
+
+def build_criteria_summary(criteria_type: str, criteria: dict) -> str:
+    """Convert structured criteria into a human-readable summary string."""
+    label = _CRITERIA_LABELS.get(criteria_type, criteria_type)
+    market = criteria.get("market", "")
+    prop = criteria.get("property_search", "")
+
+    if criteria_type == "spike":
+        threshold = criteria.get("threshold_pct", 25)
+        target = f" in {market}" if market else ""
+        return f"{label} ≥{threshold}%{target}"
+
+    if criteria_type == "price_threshold":
+        op = criteria.get("operator", "below")
+        val = criteria.get("value", "?")
+        return f"{label}: {prop} {op} {val}/night"
+
+    if criteria_type in ("availability_change",):
+        watch = criteria.get("watch_for", "any change")
+        return f"{label}: {prop} becomes {watch}"
+
+    if criteria_type == "digest":
+        freq = criteria.get("frequency", "weekly")
+        return f"{freq.capitalize()} {label} for {market}"
+
+    if criteria_type == "trend_reversal":
+        watch = criteria.get("watch_for", "any direction")
+        return f"{label}: {market} rates start {watch}"
+
+    if market:
+        return f"{label} in {market}"
+
+    if prop:
+        return f"{label} for {prop}"
+
+    return label
+
+
+def build_frequency_description(criteria_type: str, criteria: dict) -> str:
+    """Describe expected alert frequency based on criteria type."""
+    if criteria_type == "digest":
+        freq = criteria.get("frequency", "weekly")
+        return f"Sent {freq}."
+    return _FREQUENCY_DESCRIPTIONS.get(criteria_type, "Checked periodically.")
+
+
+async def send_email(to: str, subject: str, body_html: str) -> bool:
+    """Send email via configured SMTP provider. Returns True on success."""
+    loop = asyncio.get_event_loop()
+    try:
+        await loop.run_in_executor(None, _send_smtp, to, subject, body_html)
+        logger.info(f"[email] sent to={to} subject={subject[:50]}")
+        return True
+    except Exception as exc:
+        logger.error(f"[email] failed to={to}: {exc}")
+        return False
+
+
+def _send_smtp(to: str, subject: str, body_html: str) -> None:
+    """Blocking SMTP send — runs in executor."""
+    msg = MIMEMultipart("alternative")
+    msg["From"] = EMAIL_FROM_ADDRESS
+    msg["To"] = to
+    msg["Subject"] = subject
+    msg.attach(MIMEText(body_html, "html"))
+
+    with smtplib.SMTP(EMAIL_SMTP_HOST, EMAIL_SMTP_PORT) as server:
+        server.starttls()
+        server.login(EMAIL_FROM_ADDRESS, EMAIL_SMTP_PASSWORD)
+        server.sendmail(EMAIL_FROM_ADDRESS, to, msg.as_string())
+
+
+async def send_confirmation_email(to: str, token: str, criteria_summary: str) -> bool:
+    """Send double-opt-in confirmation email with token link."""
+    confirm_url = f"{ALERT_BASE_URL}/api/v1/alerts/confirm?token={token}"
+    subject = "Confirm Your Pulse AI Alert Subscription"
+    body = f"""
+    <div style="font-family: Arial, sans-serif; max-width: 600px; margin: auto; padding: 20px;">
+        <h2 style="color: #1a1a2e;">Confirm Your Alert Subscription</h2>
+        <p>You requested a Pulse AI alert for:</p>
+        <div style="background: #f0f4ff; border-left: 4px solid #4361ee; padding: 12px 16px; margin: 16px 0; border-radius: 4px;">
+            <strong>{criteria_summary}</strong>
+        </div>
+        <p>Click the button below to activate this alert:</p>
+        <a href="{confirm_url}" style="display: inline-block; background: #4361ee; color: white; padding: 12px 24px; text-decoration: none; border-radius: 6px; margin: 16px 0;">
+            Confirm Subscription
+        </a>
+        <p style="color: #666; font-size: 13px;">This link expires in 48 hours. If you didn't request this, ignore this email.</p>
+        <hr style="border: none; border-top: 1px solid #eee; margin: 24px 0;">
+        <p style="color: #999; font-size: 12px;">Pulse AI — Real Estate Intelligence by Joule Dynamics</p>
+    </div>
+    """
+    return await send_email(to, subject, body)
+
+
+async def send_subscription_active_email(
+    to: str,
+    criteria_type: str,
+    criteria: dict,
+    unsubscribe_token: str,
+) -> bool:
+    """Send 'subscription active' email with alert template preview and frequency."""
+    summary = build_criteria_summary(criteria_type, criteria)
+    frequency = build_frequency_description(criteria_type, criteria)
+    unsub_url = f"{ALERT_BASE_URL}/api/v1/alerts/unsubscribe?token={unsubscribe_token}"
+    subject = "✅ Your Pulse AI Alert Is Now Active"
+    body = f"""
+    <div style="font-family: Arial, sans-serif; max-width: 600px; margin: auto; padding: 20px;">
+        <h2 style="color: #1a1a2e;">Your Alert Is Active!</h2>
+        <p>Your Pulse AI alert subscription has been confirmed and is now live.</p>
+
+        <h3 style="color: #4361ee;">What You Subscribed To</h3>
+        <div style="background: #f0f4ff; border-left: 4px solid #4361ee; padding: 12px 16px; margin: 8px 0; border-radius: 4px;">
+            <strong>{summary}</strong>
+        </div>
+
+        <h3 style="color: #4361ee;">How Often You'll Hear From Us</h3>
+        <p>{frequency}</p>
+
+        <h3 style="color: #4361ee;">What Alert Emails Look Like</h3>
+        <div style="background: #fff8e1; border: 1px solid #ffe082; padding: 16px; margin: 8px 0; border-radius: 4px;">
+            <p style="font-size: 13px; color: #666;"><em>Sample alert preview:</em></p>
+            <p><strong>🔔 {summary}</strong></p>
+            <p>Example: "2 properties triggered this alert — Property A rate changed from $120 to $155/night (+29.2%), Property B..."</p>
+            <p style="font-size: 12px;"><a href="#">View in Dashboard</a> · <a href="#">Unsubscribe</a></p>
+        </div>
+
+        <hr style="border: none; border-top: 1px solid #eee; margin: 24px 0;">
+        <p style="color: #999; font-size: 12px;">
+            <a href="{unsub_url}" style="color: #999;">Unsubscribe from this alert</a> ·
+            Pulse AI — Real Estate Intelligence by Joule Dynamics
+        </p>
+    </div>
+    """
+    return await send_email(to, subject, body)
+
+
+async def send_alert_email(
+    to: str, subject: str, alert_body: str, unsubscribe_token: str
+) -> bool:
+    """Send triggered alert email with mandatory unsubscribe link."""
+    unsub_url = f"{ALERT_BASE_URL}/api/v1/alerts/unsubscribe?token={unsubscribe_token}"
+    body = f"""
+    <div style="font-family: Arial, sans-serif; max-width: 600px; margin: auto; padding: 20px;">
+        {alert_body}
+        <hr style="border: none; border-top: 1px solid #eee; margin: 24px 0;">
+        <p style="color: #999; font-size: 12px;">
+            <a href="{unsub_url}" style="color: #999;">Unsubscribe from this alert</a> ·
+            Pulse AI — Real Estate Intelligence by Joule Dynamics
+        </p>
+    </div>
+    """
+    return await send_email(to, subject, body)

services/prompts.py

@@ -3,7 +3,7 @@ Prompt definitions and dynamic prompt composer for the Joule Dynamics Real Estat
 """
 
 ROUTER_PROMPT = """You are the classification router for the Joule Dynamics Real Estate Intelligence Layer.
-Analyze the user query and classify it into EXACTLY ONE of six classifications:
+Analyze the user query and classify it into EXACTLY ONE of seven classifications:
 
 1. "OUT_OF_SCOPE": Query asks about topics completely unrelated to real estate, data monitoring, or the current conversation (e.g. sports, general coding, cooking, recipes). IMPORTANT: Questions asking who you are ("who are you?"), what model you are, or asking to summarize/recap what was discussed in the current chat ("what have we discussed so far?", "summarize our chat") are ALWAYS IN-SCOPE and must NEVER be classified as OUT_OF_SCOPE.
 2. "PATH_A": Query asks a live-data question (prices, spikes, availability, market averages, KPIs, specific listing rates).
@@ -11,10 +11,11 @@ Analyze the user query and classify it into EXACTLY ONE of six classifications:
 4. "BOTH": Query requires BOTH explaining a dashboard UI/methodology concept AND fetching live data metrics via tools.
 5. "GREETING": User is saying hello, thanking the assistant, asking "who are you?", or making casual conversation.
 6. "COMMERCIAL_HANDOFF": Query asks about getting started, hiring Joule Dynamics, custom builds, custom dashboards, pricing for software, or requests tracking for their own specific portfolio outside the demo scope.
+7. "ALERT_SUBSCRIPTION": User wants to be notified about a future event — price spikes, rate changes, availability changes, new listings, pricing anomalies, volatility, trend reversals, or recurring market digests. Keywords: "alert me", "notify me", "let me know when", "tell me when", "tell me whenever", "subscribe", "watch this", "track this for me". This is DISTINCT from PATH_A (which asks about current/historical data, not future notifications).
 
 Respond ONLY with valid JSON matching this schema:
 {
-    "classification": "OUT_OF_SCOPE" | "PATH_A" | "PATH_B" | "BOTH" | "GREETING" | "COMMERCIAL_HANDOFF",
+    "classification": "OUT_OF_SCOPE" | "PATH_A" | "PATH_B" | "BOTH" | "GREETING" | "COMMERCIAL_HANDOFF" | "ALERT_SUBSCRIPTION",
     "tool_categories": ["MARKET" | "ANOMALY" | "PROPERTY" | "GEO"],
     "reason": "1-sentence justification"
 }
@@ -24,7 +25,7 @@ INSTRUCTIONS FOR tool_categories:
   * "ANOMALY": For spike alerts, rate deviations, price crashes, volatile listings.
   * "PROPERTY": For specific listing details, comparisons, availability, price changes.
   * "GEO": For addresses, neighborhoods, coordinates, nearby listings, distances.
-- For "PATH_B", "GREETING", "COMMERCIAL_HANDOFF", or "OUT_OF_SCOPE", set "tool_categories": [].
+- For "PATH_B", "GREETING", "COMMERCIAL_HANDOFF", "ALERT_SUBSCRIPTION", or "OUT_OF_SCOPE", set "tool_categories": [].
 """
 
 _PROMPT_CORE = """\
@@ -95,6 +96,37 @@ When a user asks about custom builds, deploying this system for their business,
 """
 
 
+_PROMPT_ALERT = """
+ALERT SUBSCRIPTIONS: When a user wants to be notified about a future event,
+determine if you have enough structured detail to create a subscription.
+
+REQUIRED before calling create_alert_subscription:
+1. WHAT condition — one of the 10 supported types:
+   - spike: Sharp price jumps ≥ threshold % in a market
+   - rate_change: Any rate increase or decrease in a market
+   - price_threshold: Property price crosses above/below a specific value
+   - availability_change: Property becomes available or unavailable
+   - new_listing: New property added to tracking in a market
+   - tracking_removed: Property removed from active tracking
+   - anomaly: Anomalous pricing deviating from historical baseline
+   - volatility: Property enters the most volatile rankings
+   - trend_reversal: Market rate direction changes (rising to falling or vice versa)
+   - digest: Periodic market summary (daily or weekly)
+2. WHICH market or property (depending on the type)
+3. EMAIL ADDRESS to send alerts to
+
+If ANY of these is missing, ask conversationally and invoke suggest_actions
+with concrete options to help the user decide (e.g. market names, alert type
+choices like "Any spike ≥25%" vs "A specific price threshold").
+
+If all details are present, call create_alert_subscription. This creates a
+PENDING subscription only — tell the user a confirmation email is on its way
+and the alert won't be active until they confirm it via the email link.
+
+NEVER ask for or store any personal information beyond an email address.
+"""
+
+
 def build_system_prompt(classification: str, markets: list | None = None) -> str:
     """
     Compose the system prompt from core + relevant extension blocks only.
@@ -118,4 +150,7 @@ def build_system_prompt(classification: str, markets: list | None = None) -> str
     if classification == "COMMERCIAL_HANDOFF":
         prompt += _PROMPT_COMMERCIAL
 
+    if classification == "ALERT_SUBSCRIPTION":
+        prompt += _PROMPT_ALERT
+
     return prompt

services/query_router.py

@@ -17,7 +17,8 @@ from services.prompts import ROUTER_PROMPT
 logger = setup_logger(__name__)
 
 _VALID_CLASSIFICATIONS = frozenset({
-    "OUT_OF_SCOPE", "PATH_A", "PATH_B", "BOTH", "GREETING", "COMMERCIAL_HANDOFF"
+    "OUT_OF_SCOPE", "PATH_A", "PATH_B", "BOTH", "GREETING", "COMMERCIAL_HANDOFF",
+    "ALERT_SUBSCRIPTION",
 })
 
 

services/tool_executor.py

@@ -86,11 +86,23 @@ async def _data_export_handler(func_args: dict) -> dict:
     return await upload_document_to_appwrite(content, "md")
 
 
+async def _alert_subscription_handler(func_args: dict) -> dict:
+    """Forwards alert subscription creation to alert_service."""
+    from services.alert_service import create_pending_subscription
+    return await create_pending_subscription(
+        email=func_args.get("email", ""),
+        criteria_type=func_args.get("criteria_type", ""),
+        criteria=func_args.get("criteria", {}),
+        raw_text=func_args.get("raw_request_text", ""),
+    )
+
+
 _LOCAL_HANDLERS = {
     "suggest_actions": _suggest_actions_handler,
     "generate_contact_buttons": _contact_buttons_handler,
     "geocode_address": _geocode_handler,
     "generate_data_export": _data_export_handler,
+    "create_alert_subscription": _alert_subscription_handler,
 }
 
 

tests/test_alert_system.py

@@ -0,0 +1,332 @@
+"""Simulated tests for the Pulse AI Alert Subscription system.
+
+All tests use mocks — no LLM, Supabase, or SMTP calls.
+Tests cover: router classification, tool resolution, prompt composition,
+tool executor dispatch, subscription cap enforcement, cooldown logic,
+evaluator dispatch, and criteria summary builder.
+"""
+
+import asyncio
+from datetime import datetime, timezone, timedelta
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture
+def event_loop():
+    """Provide a fresh event loop for each test."""
+    loop = asyncio.new_event_loop()
+    yield loop
+    loop.close()
+
+
+class TestRouterClassification:
+    """Verify ALERT_SUBSCRIPTION is a valid classification."""
+
+    def test_alert_subscription_in_valid_classifications(self):
+        from services.query_router import _VALID_CLASSIFICATIONS
+        assert "ALERT_SUBSCRIPTION" in _VALID_CLASSIFICATIONS
+
+    def test_existing_classifications_preserved(self):
+        from services.query_router import _VALID_CLASSIFICATIONS
+        for cls in ("OUT_OF_SCOPE", "PATH_A", "PATH_B", "BOTH", "GREETING", "COMMERCIAL_HANDOFF"):
+            assert cls in _VALID_CLASSIFICATIONS
+
+    def test_total_classification_count(self):
+        from services.query_router import _VALID_CLASSIFICATIONS
+        assert len(_VALID_CLASSIFICATIONS) == 7
+
+
+class TestToolResolution:
+    """Verify _resolve_active_tools returns exactly 2 tools for ALERT_SUBSCRIPTION."""
+
+    def test_alert_subscription_returns_two_tools(self):
+        from services.agent_loop import _resolve_active_tools
+        tools = _resolve_active_tools("ALERT_SUBSCRIPTION", "alert me", [], [])
+        assert tools is not None
+        assert len(tools) == 2
+        tool_names = {t["function"]["name"] for t in tools}
+        assert tool_names == {"create_alert_subscription", "suggest_actions"}
+
+    def test_path_a_unchanged(self):
+        from services.agent_loop import _resolve_active_tools
+        with patch("services.agent_loop.discover_tools", return_value=[{"function": {"name": "mock"}}]):
+            tools = _resolve_active_tools("PATH_A", "show me spikes", [], ["ANOMALY"])
+            assert tools is not None
+
+    def test_path_b_unchanged(self):
+        from services.agent_loop import _resolve_active_tools
+        tools = _resolve_active_tools("PATH_B", "what does the sparkline mean", [], [])
+        assert tools is not None
+        assert len(tools) == 1
+        assert tools[0]["function"]["name"] == "suggest_actions"
+
+    def test_commercial_unchanged(self):
+        from services.agent_loop import _resolve_active_tools
+        from services.tools import COMMERCIAL_TOOLS
+        tools = _resolve_active_tools("COMMERCIAL_HANDOFF", "how much", [], [])
+        assert tools == COMMERCIAL_TOOLS
+
+
+class TestPromptComposition:
+    """Verify build_system_prompt includes _PROMPT_ALERT for ALERT_SUBSCRIPTION."""
+
+    def test_alert_prompt_included(self):
+        from services.prompts import build_system_prompt
+        prompt = build_system_prompt("ALERT_SUBSCRIPTION", markets=["Miami", "Abuja"])
+        assert "ALERT SUBSCRIPTIONS" in prompt
+        assert "create_alert_subscription" in prompt
+
+    def test_alert_prompt_excludes_data_prompts(self):
+        from services.prompts import build_system_prompt
+        prompt = build_system_prompt("ALERT_SUBSCRIPTION", markets=["Miami"])
+        assert "ADVISORY & STRATEGY" not in prompt
+        assert "DASHBOARD UI" not in prompt
+        assert "COMMERCIAL HANDOFFS" not in prompt
+
+    def test_path_a_prompt_excludes_alert(self):
+        from services.prompts import build_system_prompt
+        prompt = build_system_prompt("PATH_A", markets=["Miami"])
+        assert "ALERT SUBSCRIPTIONS" not in prompt
+        assert "ADVISORY & STRATEGY" in prompt
+
+    def test_all_ten_criteria_types_in_prompt(self):
+        from services.prompts import build_system_prompt
+        prompt = build_system_prompt("ALERT_SUBSCRIPTION", markets=["Miami"])
+        for ctype in ("spike", "rate_change", "price_threshold", "availability_change",
+                       "new_listing", "tracking_removed", "anomaly", "volatility",
+                       "trend_reversal", "digest"):
+            assert ctype in prompt, f"Missing criteria_type '{ctype}' in _PROMPT_ALERT"
+
+    def test_router_prompt_contains_alert_classification(self):
+        from services.prompts import ROUTER_PROMPT
+        assert "ALERT_SUBSCRIPTION" in ROUTER_PROMPT
+        assert "seven classifications" in ROUTER_PROMPT
+
+
+class TestToolExecutorDispatch:
+    """Verify create_alert_subscription routes to the handler."""
+
+    def test_handler_registered(self):
+        from services.tool_executor import _LOCAL_HANDLERS
+        assert "create_alert_subscription" in _LOCAL_HANDLERS
+
+    @pytest.mark.asyncio
+    async def test_handler_calls_alert_service(self):
+        mock_result = {"status": "pending", "message": "Confirmation email sent"}
+        with patch("services.alert_service.create_pending_subscription", new_callable=AsyncMock, return_value=mock_result):
+            from services.tool_executor import execute_tool_by_name
+            result = await execute_tool_by_name("create_alert_subscription", {
+                "email": "test@example.com",
+                "criteria_type": "spike",
+                "criteria": {"market": "Miami", "threshold_pct": 25},
+                "raw_request_text": "Alert me on Miami spikes",
+            })
+            assert result["status"] == "pending"
+
+    def test_existing_handlers_preserved(self):
+        from services.tool_executor import _LOCAL_HANDLERS
+        for handler in ("suggest_actions", "generate_contact_buttons", "geocode_address", "generate_data_export"):
+            assert handler in _LOCAL_HANDLERS
+
+
+class TestAlertToolSchema:
+    """Verify the tool schema structure."""
+
+    def test_schema_has_required_fields(self):
+        from services.alert_tool_schema import CREATE_ALERT_SUBSCRIPTION_TOOL
+        func = CREATE_ALERT_SUBSCRIPTION_TOOL["function"]
+        assert func["name"] == "create_alert_subscription"
+        params = func["parameters"]["properties"]
+        assert "email" in params
+        assert "criteria_type" in params
+        assert "criteria" in params
+        assert "raw_request_text" in params
+
+    def test_criteria_type_enum_has_ten_values(self):
+        from services.alert_tool_schema import CREATE_ALERT_SUBSCRIPTION_TOOL
+        enum_vals = CREATE_ALERT_SUBSCRIPTION_TOOL["function"]["parameters"]["properties"]["criteria_type"]["enum"]
+        assert len(enum_vals) == 10
+        expected = {"spike", "rate_change", "price_threshold", "availability_change",
+                    "new_listing", "tracking_removed", "anomaly", "volatility",
+                    "trend_reversal", "digest"}
+        assert set(enum_vals) == expected
+
+    def test_required_fields(self):
+        from services.alert_tool_schema import CREATE_ALERT_SUBSCRIPTION_TOOL
+        required = CREATE_ALERT_SUBSCRIPTION_TOOL["function"]["parameters"]["required"]
+        assert set(required) == {"email", "criteria_type", "criteria", "raw_request_text"}
+
+
+class TestCriteriaSummaryBuilder:
+    """Verify human-readable criteria summary generation."""
+
+    def test_spike_summary(self):
+        from services.email_service import build_criteria_summary
+        result = build_criteria_summary("spike", {"market": "Miami", "threshold_pct": 25})
+        assert "spike" in result.lower() or "25%" in result
+        assert "Miami" in result
+
+    def test_price_threshold_summary(self):
+        from services.email_service import build_criteria_summary
+        result = build_criteria_summary("price_threshold", {
+            "property_search": "Cozy Studio", "operator": "below", "value": 40000,
+        })
+        assert "Cozy Studio" in result
+        assert "below" in result
+
+    def test_digest_summary(self):
+        from services.email_service import build_criteria_summary
+        result = build_criteria_summary("digest", {"market": "Lagos", "frequency": "weekly"})
+        assert "Lagos" in result
+        assert "Weekly" in result or "weekly" in result.lower()
+
+    def test_trend_reversal_summary(self):
+        from services.email_service import build_criteria_summary
+        result = build_criteria_summary("trend_reversal", {"market": "Abuja", "watch_for": "falling"})
+        assert "Abuja" in result
+        assert "falling" in result
+
+    def test_availability_change_summary(self):
+        from services.email_service import build_criteria_summary
+        result = build_criteria_summary("availability_change", {
+            "property_search": "Beach House", "watch_for": "unavailable",
+        })
+        assert "Beach House" in result
+
+    def test_market_only_types(self):
+        from services.email_service import build_criteria_summary
+        for ctype in ("rate_change", "new_listing", "tracking_removed", "volatility"):
+            result = build_criteria_summary(ctype, {"market": "Miami"})
+            assert "Miami" in result
+
+
+class TestCooldownLogic:
+    """Verify cooldown enforcement in the alert evaluator."""
+
+    def test_no_last_fired_not_in_cooldown(self):
+        from workers.alert_evaluator import _is_in_cooldown
+        assert not _is_in_cooldown({"last_fired_at": None})
+
+    def test_recently_fired_in_cooldown(self):
+        from workers.alert_evaluator import _is_in_cooldown
+        recent = (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat()
+        assert _is_in_cooldown({"last_fired_at": recent}, cooldown_hours=24)
+
+    def test_old_fired_not_in_cooldown(self):
+        from workers.alert_evaluator import _is_in_cooldown
+        old = (datetime.now(timezone.utc) - timedelta(hours=25)).isoformat()
+        assert not _is_in_cooldown({"last_fired_at": old}, cooldown_hours=24)
+
+    def test_exactly_at_boundary(self):
+        from workers.alert_evaluator import _is_in_cooldown
+        boundary = (datetime.now(timezone.utc) - timedelta(hours=24, seconds=1)).isoformat()
+        assert not _is_in_cooldown({"last_fired_at": boundary}, cooldown_hours=24)
+
+
+class TestEvaluatorDispatch:
+    """Verify all 9 event-driven criteria types have evaluator functions."""
+
+    def test_all_evaluators_registered(self):
+        from workers.alert_evaluator import _EVALUATORS
+        expected = {"spike", "rate_change", "price_threshold", "availability_change",
+                    "new_listing", "tracking_removed", "anomaly", "volatility", "trend_reversal"}
+        assert set(_EVALUATORS.keys()) == expected
+
+    def test_digest_not_in_event_evaluators(self):
+        from workers.alert_evaluator import _EVALUATORS
+        assert "digest" not in _EVALUATORS
+
+    def test_evaluators_are_coroutines(self):
+        from workers.alert_evaluator import _EVALUATORS
+        import asyncio
+        for name, fn in _EVALUATORS.items():
+            assert asyncio.iscoroutinefunction(fn), f"Evaluator {name} is not async"
+
+
+class TestDigestSender:
+    """Verify digest schedule logic."""
+
+    def test_never_fired_is_due(self):
+        from workers.digest_sender import _is_due
+        assert _is_due(None, "weekly")
+        assert _is_due(None, "daily")
+
+    def test_daily_not_due_recently(self):
+        from workers.digest_sender import _is_due
+        recent = (datetime.now(timezone.utc) - timedelta(hours=12)).isoformat()
+        assert not _is_due(recent, "daily")
+
+    def test_daily_due_after_23h(self):
+        from workers.digest_sender import _is_due
+        old = (datetime.now(timezone.utc) - timedelta(hours=24)).isoformat()
+        assert _is_due(old, "daily")
+
+    def test_weekly_not_due_recently(self):
+        from workers.digest_sender import _is_due
+        recent = (datetime.now(timezone.utc) - timedelta(days=3)).isoformat()
+        assert not _is_due(recent, "weekly")
+
+    def test_weekly_due_after_6_days(self):
+        from workers.digest_sender import _is_due
+        old = (datetime.now(timezone.utc) - timedelta(days=7)).isoformat()
+        assert _is_due(old, "weekly")
+
+
+class TestFrequencyDescription:
+    """Verify frequency description generation."""
+
+    def test_event_driven_types(self):
+        from services.email_service import build_frequency_description
+        for ctype in ("spike", "rate_change", "price_threshold"):
+            desc = build_frequency_description(ctype, {})
+            assert "4×" in desc or "4x" in desc.lower()
+
+    def test_digest_type(self):
+        from services.email_service import build_frequency_description
+        desc = build_frequency_description("digest", {"frequency": "weekly"})
+        assert "weekly" in desc.lower()
+
+
+class TestConfigConstants:
+    """Verify new config constants are accessible."""
+
+    def test_email_config_exists(self):
+        from config import EMAIL_SMTP_HOST, EMAIL_SMTP_PORT, EMAIL_FROM_ADDRESS
+        assert EMAIL_SMTP_HOST == "smtp.gmail.com"
+        assert EMAIL_SMTP_PORT == 587
+        assert "jouledynamicscto" in EMAIL_FROM_ADDRESS
+
+    def test_alert_config_exists(self):
+        from config import ALERT_COOLDOWN_HOURS, MAX_SUBSCRIPTIONS_PER_EMAIL, CONFIRMATION_EXPIRY_HOURS
+        assert ALERT_COOLDOWN_HOURS == 24
+        assert MAX_SUBSCRIPTIONS_PER_EMAIL == 5
+        assert CONFIRMATION_EXPIRY_HOURS == 48
+
+    def test_service_role_key_fallback(self):
+        from config import SUPABASE_SERVICE_ROLE_KEY, SUPABASE_KEY
+        assert SUPABASE_SERVICE_ROLE_KEY, "SUPABASE_SERVICE_ROLE_KEY should not be empty"
+
+
+class TestRoutesRegistered:
+    """Verify alert routes are registered in the v1 router."""
+
+    def test_alerts_router_imported(self):
+        from routes.v1.alerts import router
+        assert router is not None
+
+    def test_confirm_endpoint_exists(self):
+        from routes.v1.alerts import router
+        paths = [route.path for route in router.routes]
+        assert "/api/v1/alerts/confirm" in paths
+
+    def test_unsubscribe_endpoint_exists(self):
+        from routes.v1.alerts import router
+        paths = [route.path for route in router.routes]
+        assert "/api/v1/alerts/unsubscribe" in paths
+
+    def test_internal_evaluate_endpoint_exists(self):
+        from routes.v1.alerts import router
+        paths = [route.path for route in router.routes]
+        assert "/api/v1/internal/evaluate-alerts" in paths

tests/test_all_tools_simulated.py

@@ -91,7 +91,7 @@ def _assert_stream_events(events: list, tool_name: str):
     assert tool_call_events[0]["tool"] == tool_name, f"Expected {tool_name}, got {tool_call_events[0]['tool']}"
 
 
-async def test_tool_non_streaming(tool_name: str, tool_args: dict) -> dict:
+async def _run_tool_non_streaming(tool_name: str, tool_args: dict) -> dict:
     """Tests a single tool execution within run_agent_loop with mocked LLM calls."""
     call_count = 0
 
@@ -132,7 +132,7 @@ async def test_tool_non_streaming(tool_name: str, tool_args: dict) -> dict:
         }
 
 
-async def test_tool_streaming(tool_name: str, tool_args: dict) -> dict:
+async def _run_tool_streaming(tool_name: str, tool_args: dict) -> dict:
     """Tests a single tool execution within run_agent_loop_streaming with mocked LLM calls."""
     call_count = 0
 
@@ -176,7 +176,7 @@ async def _run_single_case(idx: int, tool_name: str, tool_args: dict) -> tuple[i
     failures = []
 
     try:
-        res_ns = await test_tool_non_streaming(tool_name, tool_args)
+        res_ns = await _run_tool_non_streaming(tool_name, tool_args)
         print(f"[{idx:02d}/22] [SYNC]  PASS: {tool_name.ljust(32)} -> {res_ns['preview']}")
         passes += 1
     except Exception as exc:
@@ -184,7 +184,7 @@ async def _run_single_case(idx: int, tool_name: str, tool_args: dict) -> tuple[i
         failures.append((tool_name, "non-streaming", str(exc)))
 
     try:
-        res_s = await test_tool_streaming(tool_name, tool_args)
+        res_s = await _run_tool_streaming(tool_name, tool_args)
         print(f"[{idx:02d}/22] [STREAM] PASS: {tool_name.ljust(32)} -> {res_s['events_count']} events yielded")
         passes += 1
     except Exception as exc:

workers/__init__.py

@@ -0,0 +1 @@
+"""Workers package for background alert evaluation and maintenance jobs."""

workers/alert_cleanup.py

@@ -0,0 +1,38 @@
+"""Cleanup job: deletes unconfirmed subscriptions older than the expiry window."""
+
+from datetime import datetime, timezone, timedelta
+
+from config import CONFIRMATION_EXPIRY_HOURS, SUPABASE_SERVICE_ROLE_KEY, SUPABASE_URL
+from services.observability import setup_logger
+
+logger = setup_logger(__name__)
+
+
+def _get_service_client():
+    """Lazy-import Supabase service client to avoid circular imports at module level."""
+    from supabase import create_client
+    return create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)
+
+
+async def cleanup_stale_subscriptions() -> int:
+    """Delete pending subscriptions where created_at < now() - expiry window.
+
+    Returns count of deleted subscriptions.
+    """
+    client = _get_service_client()
+    cutoff = (datetime.now(timezone.utc) - timedelta(hours=CONFIRMATION_EXPIRY_HOURS)).isoformat()
+
+    try:
+        result = (
+            client.table("alert_subscriptions")
+            .delete()
+            .eq("status", "pending")
+            .lt("created_at", cutoff)
+            .execute()
+        )
+        count = len(result.data) if result.data else 0
+        logger.info(f"[cleanup] deleted {count} stale pending subscriptions")
+        return count
+    except Exception as exc:
+        logger.error(f"[cleanup] failed: {exc}")
+        return 0

workers/alert_evaluator.py

@@ -0,0 +1,326 @@
+"""Background worker: evaluates active alert subscriptions against fresh data.
+
+Reuses existing Supabase RPCs via execute_tool_rpc() — zero duplicated query logic.
+Each evaluator function is a small unit (CC <= 4) dispatched by criteria_type.
+"""
+
+from datetime import datetime, timezone, timedelta
+
+from config import ALERT_COOLDOWN_HOURS, SUPABASE_SERVICE_ROLE_KEY, SUPABASE_URL
+from services.email_service import build_criteria_summary, send_alert_email
+from services.observability import setup_logger
+from services.supabase_service import execute_tool_rpc
+
+logger = setup_logger(__name__)
+
+
+def _get_service_client():
+    """Lazy-import Supabase service client to avoid circular imports at module level."""
+    from supabase import create_client
+    return create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)
+
+
+def _is_in_cooldown(sub: dict, cooldown_hours: int = ALERT_COOLDOWN_HOURS) -> bool:
+    """Check if subscription is in cooldown window."""
+    last_fired = sub.get("last_fired_at")
+    if not last_fired:
+        return False
+    fired_dt = datetime.fromisoformat(last_fired.replace("Z", "+00:00"))
+    return datetime.now(timezone.utc) - fired_dt < timedelta(hours=cooldown_hours)
+
+
+async def _evaluate_spike(criteria: dict) -> tuple[bool, str]:
+    """Check spike condition using get_spike_alerts RPC."""
+    market = criteria.get("market")
+    threshold = criteria.get("threshold_pct", 25)
+    result = await execute_tool_rpc("get_spike_alerts", {
+        "threshold_param": threshold,
+        "p_market": market,
+        "p_limit": 5,
+        "days_param": 1,
+    })
+    data = result.get("data", [])
+    if not data:
+        return False, ""
+    names = [f"{r.get('property_name', '?')} ({r.get('pct_change', '?')}%)" for r in data[:5]]
+    body = f"<h3>🔔 Price Spike Alert — {market or 'All Markets'}</h3>"
+    body += f"<p>{len(data)} properties spiked ≥{threshold}%:</p><ul>"
+    body += "".join(f"<li>{n}</li>" for n in names) + "</ul>"
+    return True, body
+
+
+async def _evaluate_rate_change(criteria: dict) -> tuple[bool, str]:
+    """Check rate change condition using get_market_rate_changes RPC."""
+    market = criteria.get("market")
+    direction = criteria.get("direction", "both")
+    result = await execute_tool_rpc("get_market_rate_changes", {
+        "p_market": market,
+        "p_days": 1,
+        "p_limit": 5,
+    })
+    data = result.get("data", result)
+    total = data.get("total_changes", 0) if isinstance(data, dict) else 0
+    if total == 0:
+        return False, ""
+
+    if direction == "up" and not data.get("increases", 0):
+        return False, ""
+    if direction == "down" and not data.get("decreases", 0):
+        return False, ""
+
+    body = f"<h3>🔔 Rate Change Alert — {market or 'All Markets'}</h3>"
+    body += f"<p>{total} properties had rate changes in the last scrape cycle.</p>"
+    examples = data.get("examples", [])
+    if examples:
+        body += "<ul>"
+        body += "".join(f"<li>{e.get('name', '?')}: {e.get('change', '?')}</li>" for e in examples[:5])
+        body += "</ul>"
+    return True, body
+
+
+async def _evaluate_price_threshold(criteria: dict) -> tuple[bool, str]:
+    """Check price threshold using get_property_snapshot RPC."""
+    prop = criteria.get("property_search", "")
+    operator = criteria.get("operator", "below")
+    value = criteria.get("value", 0)
+    result = await execute_tool_rpc("get_property_snapshot", {"p_property_search": prop})
+    data = result.get("data", result)
+
+    if isinstance(data, list) and data:
+        data = data[0]
+    rate = data.get("nightly_rate") or data.get("current_rate")
+    if rate is None:
+        return False, ""
+
+    rate = float(rate)
+    condition_met = (operator == "below" and rate < value) or (operator == "above" and rate > value)
+    if not condition_met:
+        return False, ""
+
+    name = data.get("property_name", prop)
+    body = f"<h3>🔔 Price Threshold Alert</h3>"
+    body += f"<p><strong>{name}</strong> is now at <strong>{rate}/night</strong> "
+    body += f"({operator} your threshold of {value}).</p>"
+    return True, body
+
+
+async def _evaluate_availability(criteria: dict) -> tuple[bool, str]:
+    """Check availability change using get_property_snapshot RPC."""
+    prop = criteria.get("property_search", "")
+    watch_for = criteria.get("watch_for", "any")
+    result = await execute_tool_rpc("get_property_snapshot", {"p_property_search": prop})
+    data = result.get("data", result)
+
+    if isinstance(data, list) and data:
+        data = data[0]
+    is_available = data.get("is_available")
+    if is_available is None:
+        return False, ""
+
+    if watch_for == "available" and not is_available:
+        return False, ""
+    if watch_for == "unavailable" and is_available:
+        return False, ""
+
+    status_text = "available" if is_available else "unavailable"
+    name = data.get("property_name", prop)
+    body = f"<h3>🔔 Availability Alert</h3>"
+    body += f"<p><strong>{name}</strong> is now <strong>{status_text}</strong>.</p>"
+    return True, body
+
+
+async def _evaluate_new_listing(criteria: dict) -> tuple[bool, str]:
+    """Check for new listings using get_recently_changed_tracking RPC."""
+    market = criteria.get("market")
+    result = await execute_tool_rpc("get_recently_changed_tracking", {"p_days": 1})
+    data = result.get("data", [])
+    added = [r for r in data if r.get("change_type") == "added"]
+    if market:
+        added = [r for r in added if market.lower() in (r.get("market", "") or "").lower()]
+    if not added:
+        return False, ""
+
+    body = f"<h3>🔔 New Listing Alert — {market or 'All Markets'}</h3>"
+    body += f"<p>{len(added)} new properties added to tracking:</p><ul>"
+    body += "".join(f"<li>{r.get('property_name', '?')}</li>" for r in added[:5]) + "</ul>"
+    return True, body
+
+
+async def _evaluate_tracking_removed(criteria: dict) -> tuple[bool, str]:
+    """Check for removed listings using get_recently_changed_tracking RPC."""
+    market = criteria.get("market")
+    result = await execute_tool_rpc("get_recently_changed_tracking", {"p_days": 1})
+    data = result.get("data", [])
+    removed = [r for r in data if r.get("change_type") == "removed"]
+    if market:
+        removed = [r for r in removed if market.lower() in (r.get("market", "") or "").lower()]
+    if not removed:
+        return False, ""
+
+    body = f"<h3>🔔 Tracking Removed Alert — {market or 'All Markets'}</h3>"
+    body += f"<p>{len(removed)} properties removed from tracking:</p><ul>"
+    body += "".join(f"<li>{r.get('property_name', '?')}</li>" for r in removed[:5]) + "</ul>"
+    return True, body
+
+
+async def _evaluate_anomaly(criteria: dict) -> tuple[bool, str]:
+    """Check for pricing anomalies using get_rate_anomaly_report RPC."""
+    prop = criteria.get("property_search", "")
+    threshold = criteria.get("deviation_threshold", 25)
+    result = await execute_tool_rpc("get_rate_anomaly_report", {
+        "p_property_search": prop,
+        "p_days": 7,
+        "p_deviation_threshold": threshold,
+    })
+    data = result.get("data", result)
+    if isinstance(data, list) and data:
+        data = data[0]
+    anomaly_count = data.get("anomaly_count", 0) if isinstance(data, dict) else 0
+    if not anomaly_count:
+        return False, ""
+
+    name = data.get("property_name", prop)
+    body = f"<h3>🔔 Pricing Anomaly Alert</h3>"
+    body += f"<p><strong>{name}</strong> has {anomaly_count} anomalous pricing events "
+    body += f"(≥{threshold}% deviation from baseline).</p>"
+    return True, body
+
+
+async def _evaluate_volatility(criteria: dict) -> tuple[bool, str]:
+    """Check volatility ranking using get_most_volatile_properties RPC."""
+    market = criteria.get("market")
+    top_n = criteria.get("top_n", 5)
+    result = await execute_tool_rpc("get_most_volatile_properties", {
+        "p_market": market,
+        "p_limit": top_n,
+        "p_days": 7,
+    })
+    data = result.get("data", [])
+    if not data:
+        return False, ""
+
+    body = f"<h3>🔔 Volatility Alert — {market or 'All Markets'}</h3>"
+    body += f"<p>Top {len(data)} most volatile properties:</p><ul>"
+    for r in data[:top_n]:
+        name = r.get("property_name", "?")
+        changes = r.get("change_count", r.get("rate_change_count", "?"))
+        body += f"<li>{name} — {changes} rate changes</li>"
+    body += "</ul>"
+    return True, body
+
+
+async def _evaluate_trend_reversal(criteria: dict) -> tuple[bool, str]:
+    """Check market trend direction using get_market_trend RPC."""
+    market = criteria.get("market")
+    watch_for = criteria.get("watch_for", "falling")
+    result = await execute_tool_rpc("get_market_trend", {
+        "p_market": market,
+        "p_days": 14,
+    })
+    data = result.get("data", [])
+    if not data or len(data) < 7:
+        return False, ""
+
+    recent_half = data[len(data) // 2:]
+    rates = [float(d.get("avg_rate", 0)) for d in recent_half if d.get("avg_rate")]
+    if len(rates) < 2:
+        return False, ""
+
+    slope = rates[-1] - rates[0]
+    direction = "falling" if slope < 0 else "rising"
+    if direction != watch_for:
+        return False, ""
+
+    body = f"<h3>🔔 Trend Reversal Alert — {market}</h3>"
+    body += f"<p>Market rates are now <strong>{direction}</strong> — "
+    body += f"average moved from {rates[0]:.0f} to {rates[-1]:.0f} over the last 7 days.</p>"
+    return True, body
+
+
+_EVALUATORS = {
+    "spike": _evaluate_spike,
+    "rate_change": _evaluate_rate_change,
+    "price_threshold": _evaluate_price_threshold,
+    "availability_change": _evaluate_availability,
+    "new_listing": _evaluate_new_listing,
+    "tracking_removed": _evaluate_tracking_removed,
+    "anomaly": _evaluate_anomaly,
+    "volatility": _evaluate_volatility,
+    "trend_reversal": _evaluate_trend_reversal,
+}
+
+
+async def _evaluate_subscription(sub: dict) -> bool:
+    """Evaluate a single subscription against live data. Returns True if alert was fired."""
+    criteria_type = sub["criteria_type"]
+    if criteria_type == "digest":
+        return False
+
+    evaluator = _EVALUATORS.get(criteria_type)
+    if not evaluator:
+        logger.warning(f"[evaluator] unknown criteria_type: {criteria_type}")
+        return False
+
+    if _is_in_cooldown(sub):
+        logger.info(f"[evaluator] sub {sub['id']} in cooldown, skipping")
+        return False
+
+    condition_met, alert_body = await evaluator(sub["criteria"])
+    if not condition_met:
+        return False
+
+    summary = build_criteria_summary(criteria_type, sub["criteria"])
+    subject = f"🔔 {summary}"
+    await send_alert_email(sub["email"], subject, alert_body, sub["unsubscribe_token"])
+
+    client = _get_service_client()
+    client.table("alert_subscriptions").update({
+        "last_fired_at": datetime.now(timezone.utc).isoformat(),
+        "last_checked_at": datetime.now(timezone.utc).isoformat(),
+    }).eq("id", sub["id"]).execute()
+
+    logger.info(f"[evaluator] fired alert for sub {sub['id']}: {criteria_type}")
+    return True
+
+
+async def evaluate_all_active_subscriptions() -> dict:
+    """Fetch all active subscriptions, evaluate each, fire if condition met."""
+    client = _get_service_client()
+    try:
+        result = (
+            client.table("alert_subscriptions")
+            .select("id, email, criteria_type, criteria, unsubscribe_token, last_fired_at")
+            .eq("status", "active")
+            .eq("confirmed", True)
+            .execute()
+        )
+    except Exception as exc:
+        logger.error(f"[evaluator] failed to fetch subscriptions: {exc}")
+        return {"status": "error", "message": str(exc)}
+
+    subs = result.data or []
+    logger.info(f"[evaluator] evaluating {len(subs)} active subscriptions")
+
+    fired_count = 0
+    checked_count = 0
+    now_iso = datetime.now(timezone.utc).isoformat()
+
+    for sub in subs:
+        checked_count += 1
+        try:
+            fired = await _evaluate_subscription(sub)
+            if fired:
+                fired_count += 1
+            else:
+                client.table("alert_subscriptions").update({
+                    "last_checked_at": now_iso,
+                }).eq("id", sub["id"]).execute()
+        except Exception as exc:
+            logger.error(f"[evaluator] error evaluating sub {sub['id']}: {exc}")
+
+    logger.info(f"[evaluator] done: checked={checked_count}, fired={fired_count}")
+    return {
+        "status": "success",
+        "checked": checked_count,
+        "fired": fired_count,
+    }

workers/digest_sender.py

@@ -0,0 +1,118 @@
+"""Background worker: sends periodic digest emails for digest-type subscriptions."""
+
+from datetime import datetime, timezone
+
+from config import SUPABASE_SERVICE_ROLE_KEY, SUPABASE_URL
+from services.email_service import build_criteria_summary, send_alert_email
+from services.observability import setup_logger
+from services.supabase_service import execute_tool_rpc
+
+logger = setup_logger(__name__)
+
+
+def _get_service_client():
+    """Lazy-import Supabase service client to avoid circular imports at module level."""
+    from supabase import create_client
+    return create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)
+
+
+async def _compile_market_digest(market: str) -> str:
+    """Compile a market digest summary using existing RPCs."""
+    snapshot = await execute_tool_rpc("get_market_snapshot", {
+        "p_market": market,
+        "p_start_date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
+        "p_end_date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
+    })
+    averages = await execute_tool_rpc("get_market_averages", {"market_param": market})
+    spikes = await execute_tool_rpc("get_spike_alerts", {
+        "p_market": market,
+        "threshold_param": 25,
+        "days_param": 7,
+        "p_limit": 3,
+    })
+
+    body = f"<h3>📊 Market Digest — {market}</h3>"
+    body += "<h4>Snapshot</h4>"
+
+    snap_data = snapshot.get("data", snapshot)
+    if isinstance(snap_data, list) and snap_data:
+        snap_data = snap_data[0]
+    if isinstance(snap_data, dict):
+        body += f"<p>Active Properties: {snap_data.get('active_count', '?')}, "
+        body += f"Avg Rate: {snap_data.get('avg_rate', '?')}/night, "
+        body += f"Availability: {snap_data.get('availability_pct', '?')}%</p>"
+
+    avg_data = averages.get("data", averages)
+    if isinstance(avg_data, list) and avg_data:
+        body += "<h4>Market Averages</h4><ul>"
+        for item in avg_data[:5]:
+            body += f"<li>{item.get('property_name', '?')}: {item.get('avg_rate', '?')}/night</li>"
+        body += "</ul>"
+
+    spike_data = spikes.get("data", [])
+    if spike_data:
+        body += f"<h4>Notable Spikes (≥25%)</h4><ul>"
+        for s in spike_data[:3]:
+            body += f"<li>{s.get('property_name', '?')}: {s.get('pct_change', '?')}%</li>"
+        body += "</ul>"
+    else:
+        body += "<p>No significant spikes in the last 7 days.</p>"
+
+    return body
+
+
+async def send_pending_digests() -> dict:
+    """Fetch digest subscriptions due for send, compile market summary, send email."""
+    client = _get_service_client()
+    try:
+        result = (
+            client.table("alert_subscriptions")
+            .select("id, email, criteria, unsubscribe_token, last_fired_at")
+            .eq("status", "active")
+            .eq("confirmed", True)
+            .eq("criteria_type", "digest")
+            .execute()
+        )
+    except Exception as exc:
+        logger.error(f"[digest] failed to fetch digest subs: {exc}")
+        return {"status": "error", "message": str(exc)}
+
+    subs = result.data or []
+    sent_count = 0
+
+    for sub in subs:
+        criteria = sub["criteria"]
+        frequency = criteria.get("frequency", "weekly")
+
+        if not _is_due(sub.get("last_fired_at"), frequency):
+            continue
+
+        market = criteria.get("market", "All Markets")
+        digest_body = await _compile_market_digest(market)
+        summary = build_criteria_summary("digest", criteria)
+        subject = f"📊 {summary}"
+
+        await send_alert_email(sub["email"], subject, digest_body, sub["unsubscribe_token"])
+
+        client.table("alert_subscriptions").update({
+            "last_fired_at": datetime.now(timezone.utc).isoformat(),
+            "last_checked_at": datetime.now(timezone.utc).isoformat(),
+        }).eq("id", sub["id"]).execute()
+
+        sent_count += 1
+        logger.info(f"[digest] sent digest for sub {sub['id']}")
+
+    return {"status": "success", "sent": sent_count}
+
+
+def _is_due(last_fired_at: str | None, frequency: str) -> bool:
+    """Check if a digest subscription is due for sending."""
+    if not last_fired_at:
+        return True
+
+    fired_dt = datetime.fromisoformat(last_fired_at.replace("Z", "+00:00"))
+    elapsed = datetime.now(timezone.utc) - fired_dt
+
+    if frequency == "daily":
+        return elapsed.total_seconds() >= 23 * 3600
+    return elapsed.days >= 6