add double opt-in email verification for quote requests

Guest quote submissions now require email verification before the lead
goes live. The verification click also creates a user account and logs
them in. Logged-in users submitting with their own email skip verification.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

CHANGELOG.md

@@ -6,6 +6,65 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Added
+- **Double opt-in email verification for quote requests** — guest quote
+  submissions now require email verification before the lead goes live;
+  verification click also creates a user account and logs them in
+  automatically (GDPR-friendly consent trail)
+- `GET /leads/verify` route — validates token, activates lead
+  (`pending_verification` → `new`, sets `verified_at`), logs user in, sends
+  admin notification and welcome email
+- `send_quote_verification` worker task — branded verification email with
+  project details and "Verify & Activate Quote" CTA button (DEBUG mode
+  prints link to console)
+- `quote_verify_sent.html` template — "Check your email" page shown after
+  guest quote submission
+- Migration 0006 — adds `verified_at TEXT` column to `lead_requests`
+- 9 new tests in `TestQuoteVerification` class covering the full
+  verification flow, expired tokens, duplicate verification, and user
+  creation
+
+### Changed
+- **Inline CTA full copy** — mobile/narrow-screen inline quote CTA now matches
+  sidebar: "Next Step" label, full title, description, 4 checkmark benefits,
+  "Get Supplier Quotes" button, and "Takes ~2 minutes" hint
+- **Signup bar simplified** — removed `×` close button from guest signup bar;
+  now a non-dismissable nudge (still only shown on results tabs via JS)
+- **Investment tab narrower** — CAPEX tab content constrained to 800px max-width
+  so 3-column card grid, table, and chart don't stretch across full 1100px on
+  wide screens
+
+### Changed
+- **Quote form → standalone 9-step HTMX wizard** — extracted "Get Quotes" from
+  planner Step 5 into a standalone multi-step wizard at `/leads/quote` using
+  server-rendered HTMX partials; each step validates server-side and swaps via
+  `hx-post`/`hx-get` with OOB dot progress updates; accumulated state passed
+  forward as hidden JSON field (no JS state management)
+- **Planner reduced to 4 steps** — removed embedded quote form (Step 5) from
+  planner wizard; Step 4 "Get Quotes →" now navigates to `/leads/quote` with
+  pre-filled params (venue, courts, glass, lighting, country, budget)
+- **Planner sidebar CTA** — "Get Supplier Quotes" button now links to standalone
+  quote wizard instead of scrolling to embedded Step 5; sidebar now visible on
+  all tabs including assumptions (was results-only)
+- **Sticky wizard nav** — planner preview bar (CAPEX/CF/IRR) and back/next
+  buttons now stick to the bottom of the viewport so users don't have to scroll
+  to navigate between steps
+- **Mobile quote CTA** — inline "Get Quotes" card shown below main content on
+  screens narrower than 1400px (where the fixed sidebar is hidden)
+- **Step 4 → "Show Results"** — final planner wizard step now says "Show Results"
+  instead of "Get Quotes" since quote flow is a separate standalone wizard
+- **Removed "2-5 suppliers" cap language** — replaced specific supplier count
+  promises with "matched suppliers" across landing page, supplier FAQ, planner
+  sidebar, and quote form privacy box
+
+### Removed
+- Inline quote form from planner (Step 5 HTML, `#wizSuccess`, hidden inputs)
+- `populateWizAutoFill()`, `submitQuote()`, `COUNTRY_NAMES` from planner.js
+- `__PADELNOMICS_QUOTE_URL__` JS variable from planner template
+- Step 5 scoped CSS (~155 lines): `#wizQuoteForm`, `.wiz-autofill-summary`,
+  `.wiz-input`, `.wiz-privacy-box`, `.consent-group`, `.wiz-success`,
+  `.wiz-signup-nudge`, `.wiz-checkbox-label`
+
 ### Added
 - **Supplier tier system** — Migration 0005 adds `tier` (free/growth/pro),
   `logo_url`, `is_verified`, `highlight`, `sticky_until`, `sticky_country`

padelnomics/src/padelnomics/leads/routes.py

@@ -2,12 +2,22 @@
 Leads domain: capture interest in court suppliers and financing.
 """
 import json
+import secrets
 from datetime import datetime
 from pathlib import Path
 
-from quart import Blueprint, flash, g, jsonify, redirect, render_template, request, url_for
+from quart import Blueprint, flash, g, jsonify, redirect, render_template, request, session, url_for
 
-from ..auth.routes import login_required
+from ..auth.routes import (
+    create_auth_token,
+    create_user,
+    get_user_by_email,
+    get_user_by_id,
+    get_valid_token,
+    login_required,
+    mark_token_used,
+    update_user,
+)
 from ..core import config, csrf_protect, execute, fetch_one, send_email
 
 bp = Blueprint(
@@ -150,16 +160,96 @@ async def financing():
     return await render_template("financing.html", prefill=prefill)
 
 
+QUOTE_STEPS = [
+    {"n": 1, "title": "Your Project", "required": ["facility_type"]},
+    {"n": 2, "title": "Location", "required": ["country"]},
+    {"n": 3, "title": "Build Context", "required": []},
+    {"n": 4, "title": "Project Phase", "required": []},
+    {"n": 5, "title": "Timeline", "required": ["timeline"]},
+    {"n": 6, "title": "Financing", "required": []},
+    {"n": 7, "title": "About You", "required": ["stakeholder_type"]},
+    {"n": 8, "title": "Services Needed", "required": []},
+    {"n": 9, "title": "Contact Details", "required": ["contact_name", "contact_email"]},
+]
+
+
+def _parse_accumulated(form_or_args):
+    """Parse accumulated JSON from form data or query args."""
+    raw = form_or_args.get("_accumulated", "{}")
+    try:
+        return json.loads(raw)
+    except (json.JSONDecodeError, TypeError):
+        return {}
+
+
+@bp.route("/quote/step/<int:step>", methods=["GET", "POST"])
+@csrf_protect
+async def quote_step(step):
+    """HTMX endpoint — validate current step and return next step partial."""
+    if step < 1 or step > len(QUOTE_STEPS):
+        return "Invalid step", 400
+
+    if request.method == "POST":
+        form = await request.form
+        accumulated = _parse_accumulated(form)
+        # Merge current step's fields into accumulated
+        for k, v in form.items():
+            if k.startswith("_") or k == "csrf_token":
+                continue
+            if k == "services_needed":
+                accumulated.setdefault("services_needed", [])
+                if v not in accumulated["services_needed"]:
+                    accumulated["services_needed"].append(v)
+            else:
+                accumulated[k] = v
+        # Handle services_needed as getlist for checkboxes
+        services = form.getlist("services_needed")
+        if services:
+            accumulated["services_needed"] = services
+
+        # Validate required fields for current step
+        step_def = QUOTE_STEPS[step - 1]
+        errors = []
+        for field in step_def["required"]:
+            val = accumulated.get(field, "")
+            if isinstance(val, str) and not val.strip():
+                errors.append(field)
+            elif not val:
+                errors.append(field)
+        if errors:
+            return await render_template(
+                f"partials/quote_step_{step}.html",
+                data=accumulated, step=step, steps=QUOTE_STEPS,
+                errors=errors,
+            )
+        # Return next step
+        next_step = step + 1
+        if next_step > len(QUOTE_STEPS):
+            next_step = len(QUOTE_STEPS)
+        return await render_template(
+            f"partials/quote_step_{next_step}.html",
+            data=accumulated, step=next_step, steps=QUOTE_STEPS,
+            errors=[],
+        )
+
+    # GET — render requested step (for back navigation / dot clicks)
+    accumulated = _parse_accumulated(request.args)
+    return await render_template(
+        f"partials/quote_step_{step}.html",
+        data=accumulated, step=step, steps=QUOTE_STEPS,
+        errors=[],
+    )
+
+
 @bp.route("/quote", methods=["GET", "POST"])
 @csrf_protect
 async def quote_request():
-    """3-step lead qualification flow. No login required — guests provide contact info."""
+    """Multi-step quote wizard. No login required — guests provide contact info."""
     if request.method == "POST":
         is_json = request.content_type and "application/json" in request.content_type
 
         if is_json:
             form = await request.get_json()
-            # Normalize: get_json returns a dict, wrap list access
             services = form.get("services_needed", [])
             if isinstance(services, str):
                 services = [services]
@@ -190,9 +280,16 @@ async def quote_request():
         services_json = json.dumps(services) if services else None
 
         user_id = g.user["id"] if g.user else None
-        contact_email = form.get("contact_email", "")
+        contact_email = form.get("contact_email", "").strip().lower()
 
-        await execute(
+        # Logged-in user with matching email → skip verification
+        is_verified_user = (
+            g.user is not None
+            and g.user["email"].lower() == contact_email
+        )
+        status = "new" if is_verified_user else "pending_verification"
+
+        lead_id = await execute(
             """INSERT INTO lead_requests
             (user_id, lead_type, court_count, budget_estimate,
              facility_type, glass_type, lighting_type, build_context,
@@ -202,7 +299,7 @@ async def quote_request():
              contact_name, contact_email, contact_phone, contact_company,
              stakeholder_type,
              heat_score, status, created_at)
-            VALUES (?, 'quote', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'new', ?)""",
+            VALUES (?, 'quote', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
             (
                 user_id,
                 form.get("court_count", 0),
@@ -227,43 +324,165 @@ async def quote_request():
                 form.get("contact_company", ""),
                 form.get("stakeholder_type", ""),
                 heat,
+                status,
                 datetime.utcnow().isoformat(),
             ),
         )
 
-        # Notify admin
+        if is_verified_user:
-        await send_email(
+            # Existing flow: notify admin immediately
-            config.ADMIN_EMAIL,
+            await send_email(
-            f"[{heat.upper()}] New quote request from {contact_email}",
+                config.ADMIN_EMAIL,
-            f"<p><b>Heat:</b> {heat}<br>"
+                f"[{heat.upper()}] New quote request from {contact_email}",
-            f"<b>Contact:</b> {form.get('contact_name')} &lt;{contact_email}&gt;<br>"
+                f"<p><b>Heat:</b> {heat}<br>"
-            f"<b>Stakeholder:</b> {form.get('stakeholder_type')}<br>"
+                f"<b>Contact:</b> {form.get('contact_name')} &lt;{contact_email}&gt;<br>"
-            f"<b>Facility:</b> {form.get('facility_type')} / {form.get('court_count')} courts<br>"
+                f"<b>Stakeholder:</b> {form.get('stakeholder_type')}<br>"
-            f"<b>Glass:</b> {form.get('glass_type')} | <b>Lighting:</b> {form.get('lighting_type')}<br>"
+                f"<b>Facility:</b> {form.get('facility_type')} / {form.get('court_count')} courts<br>"
-            f"<b>Phase:</b> {form.get('location_status')} | <b>Timeline:</b> {form.get('timeline')}<br>"
+                f"<b>Glass:</b> {form.get('glass_type')} | <b>Lighting:</b> {form.get('lighting_type')}<br>"
-            f"<b>Financing:</b> {form.get('financing_status')} | <b>Budget:</b> {form.get('budget_estimate')}<br>"
+                f"<b>Phase:</b> {form.get('location_status')} | <b>Timeline:</b> {form.get('timeline')}<br>"
-            f"<b>City:</b> {form.get('city')} | <b>Country:</b> {form.get('country')}</p>",
+                f"<b>Financing:</b> {form.get('financing_status')} | <b>Budget:</b> {form.get('budget_estimate')}<br>"
-        )
+                f"<b>City:</b> {form.get('city')} | <b>Country:</b> {form.get('country')}</p>",
+            )
+
+            if is_json:
+                return jsonify({"ok": True, "heat": heat})
+
+            return await render_template(
+                "quote_submitted.html",
+                heat=heat,
+                court_count=form.get("court_count", ""),
+                facility_type=form.get("facility_type", ""),
+                country=form.get("country", ""),
+                contact_email=contact_email,
+            )
+
+        # --- Verification needed ---
+        # Get-or-create user for contact_email
+        existing_user = await get_user_by_email(contact_email)
+        if not existing_user:
+            new_user_id = await create_user(contact_email)
+        else:
+            new_user_id = existing_user["id"]
+
+        # Link lead to user if guest submission
+        if user_id is None:
+            await execute(
+                "UPDATE lead_requests SET user_id = ? WHERE id = ?",
+                (new_user_id, lead_id),
+            )
+
+        token = secrets.token_urlsafe(32)
+        await create_auth_token(new_user_id, token, minutes=60)
+
+        from ..worker import enqueue
+        await enqueue("send_quote_verification", {
+            "email": contact_email,
+            "token": token,
+            "lead_id": lead_id,
+            "contact_name": form.get("contact_name", ""),
+            "facility_type": form.get("facility_type", ""),
+            "court_count": form.get("court_count", ""),
+            "country": form.get("country", ""),
+        })
 
         if is_json:
-            return jsonify({"ok": True, "heat": heat})
+            return jsonify({"ok": True, "pending_verification": True})
 
         return await render_template(
-            "quote_submitted.html",
+            "quote_verify_sent.html",
-            heat=heat,
-            court_count=form.get("court_count", ""),
-            facility_type=form.get("facility_type", ""),
-            country=form.get("country", ""),
             contact_email=contact_email,
         )
 
-    # Pre-fill from query params (planner passes calculator state)
+    # GET — render wizard shell with starting step
-    prefill = {
+    data = {}
-        "facility_type": request.args.get("venue", ""),
+    start_step = 1
-        "court_count": request.args.get("courts", ""),
+    venue = request.args.get("venue", "")
-        "glass_type": request.args.get("glass", ""),
+    if venue:
-        "lighting_type": request.args.get("lighting", ""),
+        data = {
-        "budget": request.args.get("budget", ""),
+            "facility_type": venue,
-        "country": request.args.get("country", ""),
+            "court_count": request.args.get("courts", ""),
-    }
+            "glass_type": request.args.get("glass", ""),
-    return await render_template("quote_request.html", prefill=prefill)
+            "lighting_type": request.args.get("lighting", ""),
+            "country": request.args.get("country", ""),
+            "budget_estimate": request.args.get("budget", ""),
+        }
+        start_step = 2  # skip project step, already filled
+    return await render_template(
+        "quote_request.html",
+        data=data, step=start_step, steps=QUOTE_STEPS,
+    )
+
+
+@bp.route("/verify")
+async def verify_quote():
+    """Verify email from quote submission — activates lead and logs user in."""
+    token_str = request.args.get("token")
+    lead_id = request.args.get("lead")
+
+    if not token_str or not lead_id:
+        await flash("Invalid verification link.", "error")
+        return redirect(url_for("leads.quote_request"))
+
+    # Validate token
+    token_data = await get_valid_token(token_str)
+    if not token_data:
+        await flash("This link has expired or already been used. Please submit a new quote request.", "error")
+        return redirect(url_for("leads.quote_request"))
+
+    # Validate lead exists and is pending
+    lead = await fetch_one(
+        "SELECT * FROM lead_requests WHERE id = ? AND status = 'pending_verification'",
+        (lead_id,),
+    )
+    if not lead:
+        await flash("This quote has already been verified or does not exist.", "error")
+        return redirect(url_for("leads.quote_request"))
+
+    # Mark token used
+    await mark_token_used(token_data["id"])
+
+    # Activate lead
+    now = datetime.utcnow().isoformat()
+    await execute(
+        "UPDATE lead_requests SET status = 'new', verified_at = ? WHERE id = ?",
+        (now, lead_id),
+    )
+
+    # Set user name from contact_name if not already set
+    user = await get_user_by_id(token_data["user_id"])
+    if user and not user.get("name"):
+        await update_user(token_data["user_id"], name=lead["contact_name"])
+
+    # Log user in
+    session.permanent = True
+    session["user_id"] = token_data["user_id"]
+    await update_user(token_data["user_id"], last_login_at=now)
+
+    # Send admin notification (deferred from submission)
+    heat = lead["heat_score"] or "cool"
+    contact_email = lead["contact_email"]
+    await send_email(
+        config.ADMIN_EMAIL,
+        f"[{heat.upper()}] New quote request from {contact_email}",
+        f"<p><b>Heat:</b> {heat}<br>"
+        f"<b>Contact:</b> {lead['contact_name']} &lt;{contact_email}&gt;<br>"
+        f"<b>Stakeholder:</b> {lead['stakeholder_type']}<br>"
+        f"<b>Facility:</b> {lead['facility_type']} / {lead['court_count']} courts<br>"
+        f"<b>Glass:</b> {lead['glass_type']} | <b>Lighting:</b> {lead['lighting_type']}<br>"
+        f"<b>Phase:</b> {lead['location_status']} | <b>Timeline:</b> {lead['timeline']}<br>"
+        f"<b>Financing:</b> {lead['financing_status']} | <b>Budget:</b> {lead['budget_estimate']}<br>"
+        f"<b>City:</b> {lead['location']} | <b>Country:</b> {lead['country']}</p>",
+    )
+
+    # Send welcome email
+    from ..worker import enqueue
+    await enqueue("send_welcome", {"email": contact_email})
+
+    return await render_template(
+        "quote_submitted.html",
+        heat=heat,
+        court_count=lead["court_count"] or "",
+        facility_type=lead["facility_type"] or "",
+        country=lead["country"] or "",
+        contact_email=contact_email,
+    )

padelnomics/src/padelnomics/leads/templates/quote_submitted.html

@@ -93,7 +93,7 @@
 
       {% if contact_email %}
       <div class="email-box">
-        &#128231; A confirmation has been sent to <strong>{{ contact_email }}</strong>. Check your inbox (and spam folder).
+        &#128231; Your email <strong>{{ contact_email }}</strong> has been verified.
       </div>
       {% endif %}
 

padelnomics/src/padelnomics/leads/templates/quote_verify_sent.html

@@ -0,0 +1,38 @@
+{% extends "base.html" %}
+{% block title %}Check Your Email - {{ config.APP_NAME }}{% endblock %}
+
+{% block content %}
+<main class="container-page py-12">
+  <div class="card max-w-sm mx-auto mt-8 text-center">
+    <div style="font-size: 2.5rem; margin-bottom: 1rem;">&#9993;</div>
+
+    <h1 class="text-2xl mb-4">Check your email</h1>
+
+    <p class="text-slate-dark">We've sent a verification link to:</p>
+    <p class="font-semibold text-navy my-2">{{ contact_email }}</p>
+
+    <p class="text-slate text-sm" style="margin-top: 1rem;">
+      Click the link in the email to verify your address and activate your quote request.
+      This will also create your {{ config.APP_NAME }} account and log you in automatically.
+    </p>
+
+    <p class="text-slate text-sm" style="margin-top: 0.5rem;">
+      The link expires in 60 minutes.
+    </p>
+
+    <hr>
+
+    <details class="text-left">
+      <summary class="cursor-pointer text-sm font-medium text-navy">Didn't receive the email?</summary>
+      <ul class="list-disc pl-6 mt-2 space-y-1 text-sm text-slate-dark">
+        <li>Check your spam folder</li>
+        <li>Make sure <strong>{{ contact_email }}</strong> is correct</li>
+        <li>Wait a minute — delivery can take a moment</li>
+      </ul>
+      <p class="text-sm text-slate mt-3">
+        Wrong email? <a href="{{ url_for('leads.quote_request') }}">Submit a new request</a>.
+      </p>
+    </details>
+  </div>
+</main>
+{% endblock %}

padelnomics/src/padelnomics/migrations/schema.sql

@@ -154,7 +154,8 @@ CREATE TABLE IF NOT EXISTS lead_requests (
     contact_phone TEXT,
     contact_company TEXT,
     stakeholder_type TEXT,
-    heat_score TEXT DEFAULT 'cool'
+    heat_score TEXT DEFAULT 'cool',
+    verified_at TEXT
 );
 
 CREATE INDEX IF NOT EXISTS idx_leads_status ON lead_requests(status);

padelnomics/src/padelnomics/migrations/versions/0006_add_verified_at_to_leads.py

@@ -0,0 +1,7 @@
+"""Add verified_at column to lead_requests for double opt-in tracking."""
+
+
+def up(conn):
+    cols = {r[1] for r in conn.execute("PRAGMA table_info(lead_requests)").fetchall()}
+    if "verified_at" not in cols:
+        conn.execute("ALTER TABLE lead_requests ADD COLUMN verified_at TEXT")

padelnomics/src/padelnomics/worker.py

@@ -169,6 +169,52 @@ async def handle_send_magic_link(payload: dict) -> None:
     )
 
 
+@task("send_quote_verification")
+async def handle_send_quote_verification(payload: dict) -> None:
+    """Send verification email for quote request."""
+    link = (
+        f"{config.BASE_URL}/leads/verify"
+        f"?token={payload['token']}&lead={payload['lead_id']}"
+    )
+
+    if config.DEBUG:
+        print(f"\n{'='*60}")
+        print(f"  QUOTE VERIFICATION for {payload['email']}")
+        print(f"  {link}")
+        print(f"{'='*60}\n")
+
+    first_name = payload.get("contact_name", "").split()[0] if payload.get("contact_name") else "there"
+    project_desc = ""
+    parts = []
+    if payload.get("court_count"):
+        parts.append(f"{payload['court_count']}-court")
+    if payload.get("facility_type"):
+        parts.append(payload["facility_type"])
+    if payload.get("country"):
+        parts.append(f"in {payload['country']}")
+    if parts:
+        project_desc = f" for your {' '.join(parts)} project"
+
+    body = (
+        f'<h2 style="margin:0 0 16px;color:#0F172A;font-size:20px;">Verify your email to get supplier quotes</h2>'
+        f"<p>Hi {first_name},</p>"
+        f"<p>Thanks for requesting quotes{project_desc}. "
+        f"Click the button below to verify your email and activate your quote request. "
+        f"This will also create your {config.APP_NAME} account so you can track your project.</p>"
+        f"{_email_button(link, 'Verify & Activate Quote')}"
+        f'<p style="font-size:13px;color:#94A3B8;">This link expires in 60 minutes.</p>'
+        f'<p style="font-size:13px;color:#94A3B8;">If the button doesn\'t work, copy and paste this URL into your browser:</p>'
+        f'<p style="font-size:13px;color:#94A3B8;word-break:break-all;">{link}</p>'
+        f'<p style="font-size:13px;color:#94A3B8;">If you didn\'t request this, you can safely ignore this email.</p>'
+    )
+
+    await send_email(
+        to=payload["email"],
+        subject=f"Verify your email to get supplier quotes",
+        html=_email_wrap(body),
+    )
+
+
 @task("send_welcome")
 async def handle_send_welcome(payload: dict) -> None:
     """Send welcome email to new user."""

padelnomics/tests/test_phase0.py

@@ -253,17 +253,46 @@ class TestHeatScore:
 
 class TestQuoteRequest:
     async def test_quote_form_loads(self, client):
-        """GET /leads/quote returns 200 with form."""
+        """GET /leads/quote returns 200 with wizard shell."""
         resp = await client.get("/leads/quote")
         assert resp.status_code == 200
 
     async def test_quote_prefill_from_params(self, client):
-        """Query params pre-fill the form."""
+        """Query params pre-fill the form and start on step 2."""
         resp = await client.get("/leads/quote?venue=outdoor&courts=6")
         assert resp.status_code == 200
 
+    async def test_quote_step_endpoint(self, client):
+        """GET /leads/quote/step/1 returns 200 partial."""
+        resp = await client.get("/leads/quote/step/1")
+        assert resp.status_code == 200
+
+    async def test_quote_step_invalid(self, client):
+        """GET /leads/quote/step/0 returns 400."""
+        resp = await client.get("/leads/quote/step/0")
+        assert resp.status_code == 400
+
+    async def test_quote_step_post_advances(self, client):
+        """POST to step 1 with valid data returns step 2."""
+        await client.get("/leads/quote")
+        async with client.session_transaction() as sess:
+            csrf = sess.get("csrf_token", "")
+
+        resp = await client.post(
+            "/leads/quote/step/1",
+            form={
+                "_accumulated": "{}",
+                "facility_type": "indoor",
+                "court_count": "4",
+                "csrf_token": csrf,
+            },
+        )
+        assert resp.status_code == 200
+        html = (await resp.data).decode()
+        assert "Location" in html
+
     async def test_quote_submit_creates_lead(self, client, db):
-        """POST /leads/quote creates a lead_requests row."""
+        """POST /leads/quote creates a lead_requests row as pending_verification."""
         # Get CSRF token first
         await client.get("/leads/quote")
         async with client.session_transaction() as sess:
@@ -294,13 +323,14 @@ class TestQuoteRequest:
             rows = await cur.fetchall()
         assert len(rows) == 1
         row = dict(rows[0])
+        assert row["status"] == "pending_verification"
         assert row["heat_score"] in ("hot", "warm", "cool")
         assert row["contact_email"] == "test@example.com"
         assert row["facility_type"] == "indoor"
         assert row["stakeholder_type"] == "entrepreneur"
 
     async def test_quote_submit_without_login(self, client, db):
-        """Guests can submit quotes (user_id is null)."""
+        """Guests get a user created and linked; lead is pending_verification."""
         await client.get("/leads/quote")
         async with client.session_transaction() as sess:
             csrf = sess.get("csrf_token", "")
@@ -321,14 +351,15 @@ class TestQuoteRequest:
         assert resp.status_code == 200
 
         async with db.execute(
-            "SELECT user_id FROM lead_requests WHERE contact_email = 'guest@example.com'"
+            "SELECT user_id, status FROM lead_requests WHERE contact_email = 'guest@example.com'"
         ) as cur:
             row = await cur.fetchone()
         assert row is not None
-        assert row[0] is None  # user_id should be NULL for guests
+        assert row[0] is not None  # user_id linked via get-or-create
+        assert row[1] == "pending_verification"
 
     async def test_quote_submit_with_login(self, auth_client, db, test_user):
-        """Logged-in user gets user_id set on lead."""
+        """Logged-in user with matching email skips verification (status='new')."""
         await auth_client.get("/leads/quote")
         async with auth_client.session_transaction() as sess:
             csrf = sess.get("csrf_token", "")
@@ -342,18 +373,19 @@ class TestQuoteRequest:
                 "timeline": "asap",
                 "stakeholder_type": "entrepreneur",
                 "contact_name": "Auth User",
-                "contact_email": "auth@example.com",
+                "contact_email": "test@example.com",  # matches test_user email
                 "csrf_token": csrf,
             },
         )
         assert resp.status_code == 200
 
         async with db.execute(
-            "SELECT user_id FROM lead_requests WHERE contact_email = 'auth@example.com'"
+            "SELECT user_id, status FROM lead_requests WHERE contact_email = 'test@example.com'"
         ) as cur:
             row = await cur.fetchone()
         assert row is not None
         assert row[0] == test_user["id"]
+        assert row[1] == "new"
 
     async def test_venue_search_build_context(self, client, db):
         """Build context 'venue_search' is stored correctly."""
@@ -413,7 +445,7 @@ class TestQuoteRequest:
         assert row[0] == "tennis_club"
 
     async def test_submitted_page_has_context(self, client):
-        """Quote submitted page includes project context."""
+        """Guest quote submission shows 'check your email' verify page."""
         await client.get("/leads/quote")
         async with client.session_transaction() as sess:
             csrf = sess.get("csrf_token", "")
@@ -433,9 +465,8 @@ class TestQuoteRequest:
         )
         assert resp.status_code == 200
         html = (await resp.data).decode()
-        assert "matched" in html.lower()
+        assert "check your email" in html.lower()
-        assert "6-court" in html
+        assert "ctx@example.com" in html
-        assert "DE" in html
 
     async def test_quote_validation_rejects_missing_fields(self, client):
         """POST /leads/quote returns 422 JSON when mandatory fields missing."""
@@ -459,6 +490,173 @@ class TestQuoteRequest:
         assert len(data["errors"]) >= 3  # country, timeline, stakeholder_type + name + email
 
 
+# ════════════════════════════════════════════════════════════
+# Quote verification (double opt-in)
+# ════════════════════════════════════════════════════════════
+
+class TestQuoteVerification:
+    """Double opt-in email verification for quote requests."""
+
+    QUOTE_FORM = {
+        "facility_type": "indoor",
+        "court_count": "4",
+        "country": "DE",
+        "timeline": "3-6mo",
+        "stakeholder_type": "entrepreneur",
+        "contact_name": "Verify Test",
+        "contact_email": "verify@example.com",
+    }
+
+    async def _submit_guest_quote(self, client, db, email="verify@example.com"):
+        """Helper: submit a quote as a guest, return (lead_id, token)."""
+        await client.get("/leads/quote")
+        async with client.session_transaction() as sess:
+            csrf = sess.get("csrf_token", "")
+        form = {**self.QUOTE_FORM, "contact_email": email, "csrf_token": csrf}
+        await client.post("/leads/quote", form=form)
+
+        async with db.execute(
+            "SELECT id FROM lead_requests WHERE contact_email = ?", (email,)
+        ) as cur:
+            lead_id = (await cur.fetchone())[0]
+
+        async with db.execute(
+            "SELECT token FROM auth_tokens ORDER BY id DESC LIMIT 1"
+        ) as cur:
+            token = (await cur.fetchone())[0]
+
+        return lead_id, token
+
+    async def test_guest_quote_creates_pending_lead(self, client, db):
+        """Guest quote creates lead with status='pending_verification'."""
+        lead_id, _ = await self._submit_guest_quote(client, db)
+
+        async with db.execute(
+            "SELECT status FROM lead_requests WHERE id = ?", (lead_id,)
+        ) as cur:
+            row = await cur.fetchone()
+        assert row[0] == "pending_verification"
+
+    async def test_logged_in_same_email_skips_verification(self, auth_client, db, test_user):
+        """Logged-in user with matching email gets status='new' and 'matched' page."""
+        await auth_client.get("/leads/quote")
+        async with auth_client.session_transaction() as sess:
+            csrf = sess.get("csrf_token", "")
+
+        resp = await auth_client.post(
+            "/leads/quote",
+            form={
+                **self.QUOTE_FORM,
+                "contact_email": "test@example.com",  # matches test_user
+                "csrf_token": csrf,
+            },
+        )
+        assert resp.status_code == 200
+        html = (await resp.data).decode()
+        assert "matched" in html.lower()
+
+        async with db.execute(
+            "SELECT status FROM lead_requests WHERE contact_email = 'test@example.com'"
+        ) as cur:
+            row = await cur.fetchone()
+        assert row[0] == "new"
+
+    async def test_logged_in_different_email_needs_verification(self, auth_client, db, test_user):
+        """Logged-in user with different email still needs verification."""
+        await auth_client.get("/leads/quote")
+        async with auth_client.session_transaction() as sess:
+            csrf = sess.get("csrf_token", "")
+
+        resp = await auth_client.post(
+            "/leads/quote",
+            form={
+                **self.QUOTE_FORM,
+                "contact_email": "other@example.com",  # different from test_user
+                "csrf_token": csrf,
+            },
+        )
+        assert resp.status_code == 200
+        html = (await resp.data).decode()
+        assert "check your email" in html.lower()
+
+        async with db.execute(
+            "SELECT status FROM lead_requests WHERE contact_email = 'other@example.com'"
+        ) as cur:
+            row = await cur.fetchone()
+        assert row[0] == "pending_verification"
+
+    async def test_verify_link_activates_lead(self, client, db):
+        """GET /leads/verify with valid token sets status='new' and verified_at."""
+        lead_id, token = await self._submit_guest_quote(client, db)
+
+        resp = await client.get(f"/leads/verify?token={token}&lead={lead_id}")
+        assert resp.status_code == 200
+
+        async with db.execute(
+            "SELECT status, verified_at FROM lead_requests WHERE id = ?", (lead_id,)
+        ) as cur:
+            row = await cur.fetchone()
+        assert row[0] == "new"
+        assert row[1] is not None  # verified_at timestamp set
+
+    async def test_verify_sets_session(self, client, db):
+        """Verification link logs the user in (sets session user_id)."""
+        lead_id, token = await self._submit_guest_quote(client, db)
+
+        await client.get(f"/leads/verify?token={token}&lead={lead_id}")
+
+        async with client.session_transaction() as sess:
+            assert "user_id" in sess
+
+    async def test_verify_expired_token(self, client, db):
+        """Expired/used token redirects with error."""
+        lead_id, token = await self._submit_guest_quote(client, db)
+
+        # Expire the token
+        await db.execute("UPDATE auth_tokens SET expires_at = '2000-01-01T00:00:00'")
+        await db.commit()
+
+        resp = await client.get(
+            f"/leads/verify?token={token}&lead={lead_id}",
+            follow_redirects=False,
+        )
+        assert resp.status_code == 302
+
+    async def test_verify_already_verified_lead(self, client, db):
+        """Attempting to verify an already-activated lead shows error."""
+        lead_id, token = await self._submit_guest_quote(client, db)
+
+        # Manually activate the lead
+        await db.execute(
+            "UPDATE lead_requests SET status = 'new' WHERE id = ?", (lead_id,)
+        )
+        await db.commit()
+
+        resp = await client.get(
+            f"/leads/verify?token={token}&lead={lead_id}",
+            follow_redirects=False,
+        )
+        assert resp.status_code == 302
+
+    async def test_verify_missing_params(self, client, db):
+        """Missing token or lead params redirects."""
+        resp = await client.get("/leads/verify", follow_redirects=False)
+        assert resp.status_code == 302
+
+        resp = await client.get("/leads/verify?token=abc", follow_redirects=False)
+        assert resp.status_code == 302
+
+    async def test_guest_quote_creates_user(self, client, db):
+        """Guest quote submission creates a user row for the contact email."""
+        await self._submit_guest_quote(client, db, email="newuser@example.com")
+
+        async with db.execute(
+            "SELECT id FROM users WHERE email = 'newuser@example.com'"
+        ) as cur:
+            row = await cur.fetchone()
+        assert row is not None
+
+
 # ════════════════════════════════════════════════════════════
 # Migration / schema
 # ════════════════════════════════════════════════════════════
@@ -476,7 +674,7 @@ class TestSchema:
             "contact_phone", "contact_company",
             "wants_financing_help", "decision_process",
             "previous_supplier_contact", "services_needed",
-            "additional_info", "stakeholder_type",
+            "additional_info", "stakeholder_type", "verified_at",
         ):
             assert expected in cols, f"Missing column: {expected}"