feat(jobs): server-driven bulk job runner — survives browser tab closure

POST /api/v1/jobs accepts the whole list, validates the JWT + job ownership,
and finalizes the run server-side: per-chunk verifier_checks upserts,
processed_count/valid/risky/bad heartbeats on verifier_jobs, completed/error
finalization and risky+unprocessed refunds via the service-role ledger.

This removes the 'submitting client is gone' failure class (10 of the last 15
failed CSV runs on 2026-10-20/21): the browser no longer orchestrates chunking,
so a closed/throttled tab can no longer orphan a run holding thousands of
credits. classify_engine_result mirrors src/lib/verifier.ts classify() exactly
(NXDOMAIN -> Invalid/bad, dns_timeout/a_fallback_only -> Unknown/risky).

Frontend counterpart: runBulk v2 posts /jobs and polls the DB row only, with a
legacy chunk-loop fallback for engines predating this route (404/405).

Tests: tests/test_verifier_jobs.py (12 cases: classification edges, auth,
ownership, lifecycle finalization, chunk-failure refund path).

app/api/jobs.py

@@ -0,0 +1,346 @@
+"""Server-driven bulk job runner — the fix for the "submitting client is gone"
+failure class (10 of the last 15 failed CSV runs on 2026-10-20/21).
+
+Before this module the BROWSER orchestrated every bulk run: it chunked the
+file, submitted each chunk to /validate/bulk, polled the engine, mirrored rows
+into verifier_checks and finalized the job. If the tab closed, throttled or
+slept mid-run, the job orphaned in 'processing' until the reaper closed it and
+refunded the unprocessed remainder. The engine itself was healthy — its chunk
+task even completed 500/500 while the browser row froze at 0/3000.
+
+Contract (v2):
+  POST /api/v1/jobs  {job_id, emails, deep?, final?}
+    - JWT required (productFetch always attaches the Supabase session token).
+    - verifier_jobs row must exist, belong to the JWT subject and still be
+      open (queued/processing).
+    - A background task validates the list in chunks (bulk_sync_chunk_size),
+      upserts verifier_checks rows after EVERY chunk, heartbeats verifier_jobs
+      (processed_count + counts + updated_at) so the dashboard and the reaper
+      both see live progress, and — on the final batch — flips the job to
+      'completed' and refunds risky rows server-side.
+    - Any chunk-level failure finalizes the job as 'error' and refunds BOTH
+      the risky rows and the unprocessed remainder (same ledger mechanics as
+      the client finalize and the pg_cron reaper; idempotent via
+      refunded_credits).
+  The client polls verifier_jobs directly after accepting 202 — it no longer
+  needs to stay alive for the run to finish.
+"""
+import asyncio
+import json
+import logging
+import time
+from typing import List, Optional
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel
+
+from ..auth import verify_supabase_jwt
+from ..config import settings
+from ..db import repo
+from ..models import ValidationResult
+from ..validator.engine import engine
+
+logger = logging.getLogger("emailvalidator.jobs")
+
+router = APIRouter()
+
+# Strong references so the event loop cannot GC an in-flight runner, and a
+# semaphore so concurrent job submissions cannot spawn unbounded SMTP work.
+_job_tasks: set[asyncio.Task] = set()
+_job_semaphore = asyncio.Semaphore(settings.bulk_concurrency)
+
+GROUPS = ("good", "risky", "bad")
+
+
+class JobRequest(BaseModel):
+    job_id: str
+    emails: List[str]
+    deep: bool = True
+    final: bool = True
+    user_id: Optional[str] = None  # informational; ownership comes from the JWT
+
+
+# ---- classification (mirror of src/lib/verifier.ts classify()) ------------
+
+
+def classify_engine_result(r: ValidationResult) -> dict:
+    """Map an engine ValidationResult to the verifier_checks row shape the
+    dashboard renders. Mirrors the frontend classify() branch order so a row
+    never disagrees with a client-side re-classification:
+      disposable → bad | dead domain → bad | dns inconclusive → risky |
+      valid+smtp → good | engine invalid → bad | catch-all → risky | else risky.
+    """
+    disposable = bool(getattr(r.disposable, "is_disposable", False))
+    catch_all = bool(getattr(r.catch_all, "is_catch_all", False))
+    dns = r.dns
+    status = (r.status.value if hasattr(r.status, "value") else str(r.status)) or "unknown"
+
+    checks_run = r.checks_run or []
+    dns_error = (dns.error or "") if dns else ""
+    dns_failure = dns_error == "dns_timeout" or "timeout" in dns_error.lower()
+    a_fallback_only = bool(dns and dns.domain_exists and not dns.has_mx_records)
+    smtp_verified = bool(r.smtp and r.smtp.verified is True)
+    reason = r.reason_code
+
+    if disposable:
+        out_status, result, grp = "disposable", "Disposable", "bad"
+        reason = reason or "disposable"
+    elif dns is not None and dns.domain_exists is False:
+        # NXDOMAIN is definitive (the stale-build regression that showed a
+        # 0% Bad column on a 3000-row list must never resurrect server-side).
+        out_status, result, grp = "invalid", "Invalid", "bad"
+        reason = reason or "nxdomain"
+    elif dns_failure or a_fallback_only:
+        out_status, result, grp = "unknown", "Unknown", "risky"
+        reason = reason or ("dns_timeout" if dns_failure else "a_fallback_only")
+    elif status == "valid" and smtp_verified:
+        out_status, result, grp = "valid", "Ok", "good"
+    elif status == "invalid":
+        out_status, result, grp = "invalid", "Invalid", "bad"
+    elif catch_all:
+        out_status, result, grp = "catch_all", "Catch-all", "risky"
+        reason = reason or "catch_all_detected"
+    else:
+        out_status, result, grp = "unknown", "Unknown", "risky"
+        reason = reason or "smtp_unavailable"
+
+    score = r.scoring.score if r.scoring is not None else None
+    return {
+        "email": r.email,
+        "status": out_status,
+        "result": result,
+        "grp": grp,
+        "score": score,
+        "reason": reason,
+        "pass_through": None,
+        "checks_run": checks_run,
+        "verified_at": r.validated_at,
+        "run_id": r.run_id,
+        "validator_version": r.validator_version,
+    }
+
+
+# ---- persistence helpers (service-role writes, best-effort like repo) ------
+
+
+async def _write_checks(rows: List[dict]) -> None:
+    if not rows:
+        return
+    try:
+        await asyncio.to_thread(repo.write_verifier_checks, rows)
+    except Exception as exc:  # never lose a run to one write blip
+        logger.warning("job_checks_write_failed: %s", exc)
+
+
+async def _update_job(job_id: str, **fields) -> None:
+    try:
+        await asyncio.to_thread(repo.update_verifier_job, job_id, fields)
+    except Exception as exc:
+        logger.warning("job_update_failed job=%s: %s", job_id, exc)
+
+
+async def _finalize_failed(job_id: str, error: str, counts: dict, processed: int) -> None:
+    """Close a failed job and refund risky + unprocessed server-side.
+
+    Mirrors the ledger mechanics of public.refund_job_risky() (migration
+    20261020000000) minus its auth.uid() guard — the engine writes with the
+    service-role key, under which auth.uid() is null and the SQL function
+    would raise. Idempotency comes from verifier_jobs.refunded_credits.
+    """
+    await _update_job(
+        job_id,
+        status="error",
+        error=error[:500],
+        processed_count=processed,
+        valid_count=counts.get("good", 0),
+        risky_count=counts.get("risky", 0),
+        bad_count=counts.get("bad", 0),
+    )
+    try:
+        job = await asyncio.to_thread(repo.get_verifier_job, job_id)
+        if not job:
+            return
+        total = int(job.get("total_count") or 0)
+        charged = int(job.get("credits_charged") or 0)
+        refunded = int(job.get("refunded_credits") or 0)
+        risky = int(job.get("risky_count") or 0)
+        unprocessed = max(total - int(job.get("processed_count") or 0), 0)
+        due = min(risky + unprocessed, charged) - refunded
+        if due > 0:
+            await asyncio.to_thread(repo.refund_credits, job["user_id"], due,
+                                    "Auto-refund — risky and unprocessed results are not billed")
+            await _update_job(job_id, refunded_credits=refunded + due)
+    except Exception as exc:
+        logger.warning("job_refund_failed job=%s: %s", job_id, exc)
+
+
+# ---- the runner ------------------------------------------------------------
+
+
+async def _run_verifier_job(job_id: str, user_id: str, emails: List[str], deep: bool,
+                            final: bool) -> None:
+    started = time.monotonic()
+    counts = {g: 0 for g in GROUPS}
+    processed = 0
+    # Defensive dedupe (the frontend already deduped; a client bug must not
+    # double-write rows — (job_id, email) upsert would silently dedupe them
+    # while counts drift).
+    seen: set[str] = set()
+    ordered: List[str] = []
+    for e in emails:
+        e = (e or "").strip().lower()
+        if e and "@" in e and e not in seen:
+            seen.add(e)
+            ordered.append(e)
+    emails = ordered
+    total = len(emails)
+    if total == 0:
+        await _finalize_failed(job_id, "No valid addresses reached the engine.", counts, 0)
+        return
+
+    chunk_size = max(int(settings.bulk_sync_chunk_size or 50), 10)
+    async with _job_semaphore:
+        try:
+            await _update_job(job_id, status="processing")
+            for i in range(0, total, chunk_size):
+                chunk = emails[i:i + chunk_size]
+                try:
+                    results = await engine.validate_batch(chunk, deep=deep)
+                except Exception as exc:  # one bad stage must not kill the job
+                    logger.warning("job_chunk_failed job=%s: %s", job_id, exc)
+                    await _finalize_failed(
+                        job_id,
+                        f"Engine chunk failed after {processed}/{total} addresses: {exc}",
+                        counts, processed,
+                    )
+                    return
+                if len(results) != len(chunk):
+                    await _finalize_failed(
+                        job_id,
+                        f"Engine returned {len(results)} of {len(chunk)} results for one chunk",
+                        counts, processed,
+                    )
+                    return
+                rows = [classify_engine_result(r) for r in results]
+                await _write_checks(rows)
+                for row in rows:
+                    counts[row["grp"]] = counts.get(row["grp"], 0) + 1
+                processed += len(chunk)
+                # Heartbeat + live progress in one write; the dashboard polls
+                # this row instead of the engine, and the reaper reads it.
+                await _update_job(
+                    job_id,
+                    processed_count=processed,
+                    valid_count=counts["good"],
+                    risky_count=counts["risky"],
+                    bad_count=counts["bad"],
+                )
+            if final:
+                await _update_job(
+                    job_id,
+                    status="completed",
+                    error=None,
+                    processed_count=processed,
+                    valid_count=counts["good"],
+                    risky_count=counts["risky"],
+                    bad_count=counts["bad"],
+                )
+                # Risky rows are never billed; the completed-job refund covers
+                # exactly the risky_count (unprocessed term is zero by design).
+                try:
+                    job = await asyncio.to_thread(repo.get_verifier_job, job_id)
+                    if job:
+                        charged = int(job.get("credits_charged") or 0)
+                        refunded = int(job.get("refunded_credits") or 0)
+                        due = min(counts["risky"], charged) - refunded
+                        if due > 0:
+                            await asyncio.to_thread(
+                                repo.refund_credits, job["user_id"], due,
+                                "Auto-refund — risky results are not billed")
+                            await _update_job(job_id, refunded_credits=refunded + due)
+                except Exception as exc:
+                    logger.warning("job_risky_refund_failed job=%s: %s", job_id, exc)
+            logger.info("job_batch_done job=%s processed=%s/%s final=%s secs=%.1f",
+                        job_id, processed, total, final, time.monotonic() - started)
+        except Exception as exc:  # absolute safety net
+            logger.exception("job_runner_crashed job=%s", job_id)
+            await _finalize_failed(job_id, f"Runner crashed: {exc}", counts, processed)
+
+
+# ---- routes ----------------------------------------------------------------
+
+
+@router.post("/jobs", tags=["Bulk Jobs"])
+async def create_server_job(request: Request, payload: JobRequest):
+    """Accept a bulk job for server-side processing (202 + DB polling)."""
+    token = request.headers.get("authorization", "")
+    token = token[7:].strip() if token.lower().startswith("bearer ") else ""
+    if not token:
+        raise HTTPException(401, "Missing bearer token — sign in first.")
+    try:
+        claims = await verify_supabase_jwt(token)
+    except Exception as exc:
+        raise HTTPException(401, f"Invalid or expired token: {exc}")
+    user_id = claims.get("sub", "")
+    if not user_id:
+        raise HTTPException(401, "Token has no subject claim.")
+    if not payload.emails:
+        raise HTTPException(400, "emails must not be empty.")
+    if not settings.supabase_configured:
+        raise HTTPException(503, "Server persistence is not configured (SUPABASE_URL/SERVICE key).")
+
+    job = await asyncio.to_thread(repo.get_verifier_job, payload.job_id)
+    if not job:
+        raise HTTPException(404, f"Job {payload.job_id} not found — create it first.")
+    if str(job.get("user_id")) != user_id:
+        # Ownership mismatch: identical response for missing/foreign jobs.
+        raise HTTPException(404, f"Job {payload.job_id} not found for this account.")
+    if job.get("status") not in ("queued", "processing"):
+        raise HTTPException(409, f"Job {payload.job_id} is already {job.get('status')}.")
+    if not payload.emails:
+        raise HTTPException(400, "emails must not be empty.")
+
+    # Cap one submission so a giant file cannot pin the runner forever: the
+    # client splits into final-flagged batches (20k ≈ 600 KB JSON).
+    MAX_PER_CALL = 20000
+    emails = payload.emails[:MAX_PER_CALL]
+
+    task = asyncio.create_task(
+        _run_verifier_job(payload.job_id, user_id, emails, payload.deep, payload.final)
+    )
+    _job_tasks.add(task)
+    task.add_done_callback(_job_tasks.discard)
+
+    return {
+        "accepted": len(emails),
+        "job_id": payload.job_id,
+        "status": "processing",
+        "final": payload.final,
+        "message": "Processing server-side — poll verifier_jobs for progress.",
+    }
+
+
+@router.get("/jobs/{job_id}", tags=["Bulk Jobs"])
+async def job_status(job_id: str, request: Request):
+    """Lightweight engine-side echo of the job row (debug aid; the dashboard
+    polls the DB directly and does not depend on this)."""
+    token = request.headers.get("authorization", "")
+    token = token[7:].strip() if token.lower().startswith("bearer ") else ""
+    if not token:
+        raise HTTPException(401, "Missing bearer token — sign in first.")
+    try:
+        claims = await verify_supabase_jwt(token)
+    except Exception as exc:
+        raise HTTPException(401, f"Invalid or expired token: {exc}")
+    job = await asyncio.to_thread(repo.get_verifier_job, job_id)
+    if not job or str(job.get("user_id")) != claims.get("sub", ""):
+        raise HTTPException(404, f"Job {job_id} not found for this account.")
+    return {
+        "job_id": job_id,
+        "status": job.get("status"),
+        "total": job.get("total_count"),
+        "processed": job.get("processed_count"),
+        "valid": job.get("valid_count"),
+        "risky": job.get("risky_count"),
+        "bad": job.get("bad_count"),
+    }

app/db/repo.py

@@ -220,3 +220,70 @@ def save_domain_record(domain: str, dns_result: Any) -> Optional[Dict[str, Any]]
         "has_dkim": bool(dns_result and dns_result.has_dkim),
     }
     return _upsert("domain_records", payload, on_conflict="domain")
+
+
+# ---------------------------------------------------------------------------
+# Verifier jobs — server-driven bulk runs (app/api/jobs.py)
+# The engine writes the same tables the browser used to write, so a job now
+# finishes even when the submitting tab is gone.
+# ---------------------------------------------------------------------------
+def get_verifier_job(job_id: str) -> Optional[Dict[str, Any]]:
+    rows = _select("verifier_jobs", "*", limit=1, eq={"id": job_id})
+    return rows[0] if rows else None
+
+
+def update_verifier_job(job_id: str, fields: Dict[str, Any]) -> None:
+    """Patch the job row. Callers pass business fields only; updated_at is
+    always touched so the stale-job reaper sees live heartbeats."""
+    payload = {**fields, "updated_at": "now()"}
+    client = _client()
+    if client is None:
+        return
+    try:
+        client.table("verifier_jobs").update(payload).eq("id", job_id).execute()
+    except Exception as exc:
+        logger.warning("supabase_verifier_job_update_failed", job_id=job_id, error=str(exc))
+
+
+def write_verifier_checks(rows: List[Dict[str, Any]]) -> None:
+    """Upsert verifier_checks rows, idempotent on (job_id, email)."""
+    if not rows:
+        return
+    client = _client()
+    if client is None:
+        return
+    try:
+        client.table("verifier_checks").upsert(rows, on_conflict="job_id,email").execute()
+    except Exception as exc:
+        logger.warning("supabase_verifier_checks_upsert_failed", count=len(rows), error=str(exc))
+
+
+def refund_credits(user_id: str, amount: int, description: str) -> None:
+    """Credit balance + ledger row (service-role twin of the SQL refund path).
+    Idempotency is the caller's duty via verifier_jobs.refunded_credits."""
+    if amount <= 0 or not user_id:
+        return
+    client = _client()
+    if client is None:
+        return
+    try:
+        # Atomic increment via an RPC is unavailable server-side without a
+        # dedicated function; PostgREST PATCH with a computed value races, so
+        # read-modify-write under a single-writer assumption (one runner task
+        # per job). The refunded_credits ledger column keeps each refund
+        # one-shot even if a retry replays the same due amount.
+        resp = client.table("user_credits").select("balance").eq("user_id", user_id).limit(1).execute()
+        rows = resp.data or []
+        balance = int(rows[0]["balance"]) if rows else 0
+        client.table("user_credits").update({
+            "balance": balance + int(amount),
+            "last_updated": "now()",
+        }).eq("user_id", user_id).execute()
+        client.table("credit_transactions").insert({
+            "user_id": user_id,
+            "type": "refund",
+            "amount": int(amount),
+            "description": description,
+        }).execute()
+    except Exception as exc:
+        logger.warning("supabase_refund_credits_failed", user_id=user_id, error=str(exc))

app/main.py

@@ -10,6 +10,7 @@ import time
 
 from .config import settings
 from .api.routes import router
+from .api.jobs import router as jobs_router
 from .auth import license_middleware
 from .utils.logger import setup_logging
 from .db import repo
@@ -176,6 +177,9 @@ async def global_exception_handler(request: Request, exc: Exception):
 
 # ---- Include Routes -------------------------------------------------
 app.include_router(router, prefix="/api/v1")
+# Server-driven bulk jobs (POST /api/v1/jobs): the engine finalizes the run
+# with DB heartbeats so a closed browser tab can no longer orphan it.
+app.include_router(jobs_router, prefix="/api/v1")
 
 
 # ---- Health alias for the website productApi probe -------------------

tests/test_verifier_jobs.py

@@ -0,0 +1,278 @@
+"""Server-driven bulk job runner (app/api/jobs.py).
+
+Regression class this locks down (2026-10-20/21 incidents, 10 of the last 15
+failed CSV runs): the browser used to orchestrate chunking, so a closed tab
+orphaned verifier_jobs rows in 'processing' holding thousands of credits. The
+engine now finalizes the run server-side with per-chunk DB writes and
+heartbeats, and refunds risky + unprocessed rows on failure.
+"""
+import asyncio
+import unittest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from fastapi.testclient import TestClient
+
+from app.api import jobs as jobs_mod
+from app.api.jobs import classify_engine_result
+from app.main import app
+from app.models import (
+    CatchAllResult,
+    DNSResult,
+    DisposableResult,
+    ValidationResult,
+    ScoringResult,
+    SMTPResult,
+    SyntaxResult,
+    ValidationStatus,
+)
+
+client = TestClient(app)
+
+
+def _result(email: str, *, status=ValidationStatus.UNKNOWN, domain_exists=True,
+            has_mx=True, dns_error=None, disposable=False, catch_all=False,
+            smtp_verified=None, smtp_code=None, score=70, reason_code=None) -> ValidationResult:
+    dns = DNSResult(
+        domain_exists=domain_exists,
+        has_mx_records=has_mx,
+        error=dns_error,
+        error_code=dns_error,
+    )
+    syntax = SyntaxResult(passed=True, normalized_email=email, domain=email.split("@")[-1])
+    return ValidationResult(
+        email=email,
+        status=status,
+        syntax=syntax,
+        dns=dns,
+        disposable=DisposableResult(is_disposable=disposable, is_free_provider=False,
+                                    is_role_based=False, confidence=0.9) if disposable else None,
+        catch_all=CatchAllResult(is_catch_all=catch_all, confidence=0.9) if catch_all else None,
+        smtp=SMTPResult(verified=smtp_verified, smtp_code=smtp_code) if smtp_code or smtp_verified is not None else None,
+        scoring=ScoringResult(score=score, risk_level="medium", breakdown={}, warnings=[]),
+        processing_time_ms=1.0,
+        validated_at="2026-10-21T00:00:00+00:00",
+        checks_run=["syntax", "dns"],
+        reason_code=reason_code,
+        run_id="testrun",
+        validator_version="3.1.0",
+    )
+
+
+class ClassifyTests(unittest.TestCase):
+    """classify_engine_result must mirror src/lib/verifier.ts classify()."""
+
+    def test_nxdomain_is_invalid_bad(self):
+        # 185/3000 rows in the 2026-10-21 run were dead domains mis-classified
+        # as Unknown/risky by the stale frontend build. Server-side rows must
+        # be Invalid/bad — a domain that does not exist cannot accept mail.
+        row = classify_engine_result(
+            _result("a@dead.test", status=ValidationStatus.INVALID,
+                    domain_exists=False, has_mx=False, dns_error="nxdomain",
+                    reason_code="nxdomain")
+        )
+        self.assertEqual(row["grp"], "bad")
+        self.assertEqual(row["status"], "invalid")
+        self.assertEqual(row["result"], "Invalid")
+
+    def test_smtp_unavailable_is_risky(self):
+        row = classify_engine_result(
+            _result("b@gmail.test", status=ValidationStatus.RISKY,
+                    smtp_verified=None, reason_code="smtp_unavailable")
+        )
+        self.assertEqual(row["grp"], "risky")
+        self.assertEqual(row["result"], "Unknown")
+
+    def test_valid_with_smtp_is_good(self):
+        row = classify_engine_result(
+            _result("c@corp.test", status=ValidationStatus.VALID,
+                    smtp_verified=True, smtp_code=250, score=95)
+        )
+        self.assertEqual(row["grp"], "good")
+        self.assertEqual(row["result"], "Ok")
+
+    def test_disposable_is_bad(self):
+        row = classify_engine_result(_result("x@mailinator.test", disposable=True))
+        self.assertEqual(row["grp"], "bad")
+        self.assertEqual(row["result"], "Disposable")
+
+    def test_catch_all_is_risky(self):
+        row = classify_engine_result(
+            _result("d@ca.test", status=ValidationStatus.RISKY, catch_all=True)
+        )
+        self.assertEqual(row["grp"], "risky")
+        self.assertEqual(row["result"], "Catch-all")
+
+    def test_dns_timeout_is_risky_not_bad(self):
+        # Transient DNS states are genuinely undecidable — golden invariant.
+        row = classify_engine_result(
+            _result("e@corp.test", status=ValidationStatus.UNKNOWN,
+                    dns_error="timeout", reason_code="dns_timeout")
+        )
+        self.assertEqual(row["grp"], "risky")
+        self.assertEqual(row["reason"], "dns_timeout")
+
+
+class JobEndpointTests(unittest.TestCase):
+    """POST /api/v1/jobs auth, ownership and lifecycle."""
+
+    def setUp(self):
+        self.job_row = {
+            "id": "11111111-1111-1111-1111-111111111111",
+            "user_id": "22222222-2222-2222-2222-222222222222",
+            "name": "t.csv",
+            "status": "queued",
+            "total_count": 3,
+            "processed_count": 0,
+            "valid_count": 0, "risky_count": 0, "bad_count": 0,
+            "credits_charged": 3, "refunded_credits": 0,
+        }
+        patcher = patch("app.api.jobs.repo", autospec=True)
+        self.repo_mock = patcher.start()
+        self.addCleanup(patcher.stop)
+        # The route 503s without persistence config; tests run without a real
+        # Supabase, so present the route with a configured project.
+        for field, value in (("supabase_url", "https://test.supabase.co"),
+                             ("supabase_service_role_key", "svc-key-test")):
+            p = patch.object(jobs_mod.settings, field, value)
+            p.start()
+            self.addCleanup(p.stop)
+        self.repo_mock.get_verifier_job.return_value = self.job_row
+        self.repo_mock.update_verifier_job.return_value = None
+        self.repo_mock.write_verifier_checks.return_value = None
+        self.repo_mock.refund_credits.return_value = None
+
+    def _auth(self, sub="22222222-2222-2222-2222-222222222222"):
+        return patch(
+            "app.api.jobs.verify_supabase_jwt",
+            new=AsyncMock(return_value={"sub": sub, "email": "t@t.test"}),
+        )
+
+    def test_401_without_token(self):
+        res = client.post("/api/v1/jobs", json={"job_id": "x", "emails": ["a@b.test"]})
+        self.assertEqual(res.status_code, 401)
+
+    def test_404_foreign_job(self):
+        # Same-shaped id, DIFFERENT owner: the token subject does not match the
+        # row's user_id, and the route must answer 404 (never 403/200).
+        with self._auth(sub="99999999-9999-9999-9999-999999999999"):
+            res = client.post(
+                "/api/v1/jobs",
+                json={"job_id": "11111111-1111-1111-1111-111111111111",
+                      "emails": ["a@b.test"]},
+                headers={"Authorization": "Bearer faketoken"},
+            )
+        self.assertEqual(res.status_code, 404)
+
+    def test_409_when_job_closed(self):
+        self.job_row["status"] = "completed"
+        with self._auth():
+            res = client.post(
+                "/api/v1/jobs",
+                json={"job_id": "11111111-1111-1111-1111-111111111111",
+                      "emails": ["a@b.test"]},
+                headers={"Authorization": "Bearer faketoken"},
+            )
+        self.assertEqual(res.status_code, 409)
+
+    def test_400_empty_emails(self):
+        with self._auth():
+            res = client.post(
+                "/api/v1/jobs",
+                json={"job_id": "11111111-1111-1111-1111-111111111111", "emails": []},
+                headers={"Authorization": "Bearer faketoken"},
+            )
+        self.assertEqual(res.status_code, 400)
+
+    def test_run_finalizes_completed_and_refunds_risky(self):
+        """Happy path: 3 emails, 1 risky (smtp unavailable) → job completed,
+        risky row refunded once (refunded_credits ledger updated).
+        The endpoint wiring is asserted on the POST response; the lifecycle is
+        asserted by driving the runner coroutine to completion on a real loop
+        (TestClient does not guarantee background-task scheduling after the
+        response returns)."""
+
+        async def fake_batch(emails, deep=True):
+            return [
+                _result(emails[0], status=ValidationStatus.VALID, smtp_verified=True,
+                        smtp_code=250, score=95),
+                _result(emails[1], status=ValidationStatus.RISKY, reason_code="smtp_unavailable"),
+                _result(emails[2], status=ValidationStatus.INVALID, domain_exists=False,
+                        has_mx=False, dns_error="nxdomain", reason_code="nxdomain"),
+            ]
+
+        self.job_row["refunded_credits"] = 0
+        emails = ["a@x.test", "b@x.test", "c@x.test"]
+        # The endpoint spawns the runner as a fire-and-forget task; in tests
+        # that task would outlive the patch context and run the REAL engine on
+        # .test domains (all NXDOMAIN). Swallow the spawn — the lifecycle is
+        # driven deterministically below on a real loop.
+        def _no_task(coro):
+            coro.close()
+            return MagicMock()
+        with patch("app.api.jobs.engine.validate_batch", new=AsyncMock(side_effect=fake_batch)), \
+             patch("app.api.jobs.settings.bulk_sync_chunk_size", 2), \
+             patch("app.api.jobs.asyncio.create_task", side_effect=_no_task), \
+             self._auth():
+            res = client.post(
+                "/api/v1/jobs",
+                json={"job_id": "11111111-1111-1111-1111-111111111111",
+                      "emails": emails},
+                headers={"Authorization": "Bearer faketoken"},
+            )
+            self.assertEqual(res.status_code, 200)
+            self.assertEqual(res.json()["accepted"], 3)
+            # Deterministic lifecycle: drive the same runner invocation to
+            # completion INSIDE the patch context (a test-client background
+            # task would outlive it and run the real engine on .test domains).
+            asyncio.run(jobs_mod._run_verifier_job(
+                "11111111-1111-1111-1111-111111111111",
+                "22222222-2222-2222-2222-222222222222", emails, True, True))
+        updates = [c.args[1] for c in self.repo_mock.update_verifier_job.call_args_list
+                   if len(c.args) >= 2 and isinstance(c.args[1], dict)]
+        final = [u for u in updates if u.get("status") == "completed"]
+        self.assertTrue(final, f"runner never completed; updates={updates}")
+        last = final[-1]
+        self.assertEqual(last["bad_count"], 1)   # nxdomain → bad
+        self.assertEqual(last["risky_count"], 1)  # smtp unavailable → risky
+        self.assertEqual(last["valid_count"], 1)
+        # risky refund path executed server-side
+        self.repo_mock.refund_credits.assert_called_once()
+        self.assertEqual(self.repo_mock.refund_credits.call_args[0][1], 1)
+
+    def test_chunk_failure_finalizes_error_and_refunds(self):
+        """Engine crash mid-run → job error + risky + unprocessed refund."""
+
+        async def boom(emails, deep=True):
+            raise RuntimeError("smtp stage exploded")
+
+        self.job_row["total_count"] = 5
+        self.job_row["credits_charged"] = 5
+        def _no_task(coro):
+            coro.close()
+            return MagicMock()
+        with patch("app.api.jobs.engine.validate_batch", new=AsyncMock(side_effect=boom)), \
+             patch("app.api.jobs.asyncio.create_task", side_effect=_no_task), \
+             self._auth():
+            res = client.post(
+                "/api/v1/jobs",
+                json={"job_id": "11111111-1111-1111-1111-111111111111",
+                      "emails": ["a@x.test"] * 5},
+                headers={"Authorization": "Bearer faketoken"},
+            )
+            self.assertEqual(res.status_code, 200)
+            # Deterministic lifecycle inside the patch context.
+            asyncio.run(jobs_mod._run_verifier_job(
+                "11111111-1111-1111-1111-111111111111",
+                "22222222-2222-2222-2222-222222222222", ["a@x.test"] * 5, True, True))
+        updates = [c.args[1] for c in self.repo_mock.update_verifier_job.call_args_list
+                   if len(c.args) >= 2 and isinstance(c.args[1], dict)]
+        err_calls = [u for u in updates if u.get("status") == "error"]
+        self.assertTrue(err_calls, f"runner never finalized the failed job; updates={updates}")
+        self.assertIn("Engine chunk failed", err_calls[-1]["error"])
+        # refund = risky(0) + unprocessed(5) clamped to charged(5)
+        self.repo_mock.refund_credits.assert_called_once()
+        self.assertEqual(self.repo_mock.refund_credits.call_args[0][1], 5)
+
+
+if __name__ == "__main__":
+    unittest.main()