Spaces:

sitcod3
/

modul-x

Running

Z User

v3.1.0: Part B architecture migration — eliminate Hugging Face inference credit dependency

6c516b6 1 day ago

22.6 kB

from urllib.parse import urlsplit
from typing import List, Optional

from fastapi import APIRouter, HTTPException, Query, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import json
import time
import asyncio
import uuid

from ..models import (
    SingleEmailRequest, BulkEmailRequest,
    ValidationResult, BulkValidationResponse,
    TaskStatusResponse
)
from ..validator.engine import engine
from ..utils.cache import get_cache, set_cache
from ..config import settings
from ..db import repo

# Strong references to in-flight background bulk jobs (the event loop only
# keeps weak ones, so a task can vanish mid-run and its results are lost), and
# a cap so a burst of bulk requests cannot spawn unbounded SMTP work.
_background_tasks: set[asyncio.Task] = set()
_bulk_semaphore = asyncio.Semaphore(settings.bulk_concurrency)

# Celery task handle (module level so the routing is testable/patchable).
# Import failure degrades to the background-task path; Celery itself is only
# required when celery_bulk_enabled=True.
try:
    from ..tasks.celery_tasks import bulk_validate_task  # noqa: E402
except Exception:  # pragma: no cover - celery not installed
    bulk_validate_task = None

router = APIRouter()


async def process_bulk_sync(emails: list, webhook_url: str = None, deep: bool = True) -> dict:
    """
    Process bulk validation synchronously in chunks.
    Used when Celery is unavailable.
    `deep` defaults to True so bulk runs the full 7-layer pipeline (catch-all +
    SMTP) exactly like single verification — the product charges 1 credit/email
    and must not silently degrade to a syntax+DNS-only check.
    Chunks feed the engine's *batched* path: DNS/MX resolves once per domain and
    the mailbox stage takes the whole chunk in one call, so the SMTP engine reuses
    a session per receiving host instead of opening a TCP connection per address.
    """
    results = []
    total = len(emails)
    chunk_size = settings.bulk_sync_chunk_size  # Configurable chunk size

    for i in range(0, total, chunk_size):
        chunk = emails[i:i + chunk_size]
        try:
            chunk_results = await engine.validate_batch(chunk, deep=deep)
        except Exception as exc:  # never lose a chunk to one bad stage
            results.extend(
                {
                    "email": email,
                    "error": str(exc) or "Unknown error",
                    "status": "error",
                }
                for email in chunk
            )
            continue

        # validate_batch guarantees one result per input, in input order.
        if len(chunk_results) == len(chunk):
            pairs = zip(chunk, chunk_results)
        else:
            # Defensive: never lose or mis-attribute a row.
            by_email = {getattr(r, "email", "").lower(): r for r in chunk_results}
            pairs = ((email, by_email.get(email.lower())) for email in chunk)

        for email, result in pairs:
            if isinstance(result, ValidationResult):
                results.append(result.model_dump())
            else:
                results.append(
                    {
                        "email": email,
                        "error": "engine returned no result for this address",
                        "status": "error",
                    }
                )
    
    # Webhook notification if provided
    if webhook_url:
        import httpx
        try:
            async with httpx.AsyncClient(timeout=10) as client:
                await client.post(webhook_url, json={"results": results})
        except Exception:
            pass

    # Run-quality monitor (report item 10): fire the signed system-alert
    # webhook when this run's risky/unknown share crosses the threshold.
    try:
        from ..utils.rate_alerts import maybe_alert_risky_rate

        await maybe_alert_risky_rate(
            results,
            webhook_url=webhook_url,
            job_id="sync",
            threshold=settings.risky_rate_alert_threshold,
        )
    except Exception:
        pass

    return results


# ---- Single Email Validation --------------------------------------
@router.post(
    "/validate/single",
    response_model=ValidationResult,
    summary="Validate a single email address",
    tags=["Validation"]
)
async def validate_single(
    request: SingleEmailRequest,
    deep: bool = Query(default=True, description="Enable SMTP verification (slower but more accurate)")
):
    """
    Full 7-layer email validation:
    Syntax → DNS → MX → Disposable → Catch-All → SMTP → Score
    """
    email = request.email

    # Check cache first
    cache_key = f"val:{email}:{'deep' if deep else 'fast'}"
    cached = await get_cache(cache_key)
    if cached:
        result = ValidationResult(**json.loads(cached))
        return result.model_copy(update={"cache_hit": True})

    # Run full validation
    result = await engine.validate(email, deep=deep)

    # Cache the result
    await set_cache(cache_key, result.model_dump_json(), ttl=settings.redis_cache_ttl)

    # Persist to Supabase (best-effort)
    if settings.enable_db_persistence:
        repo.save_validation_result(result)

    return result


# ---- Bulk Email Validation ----------------------------------------
@router.post(
    "/validate/bulk",
    summary="Validate multiple email addresses (async)",
    tags=["Validation"]
)
async def validate_bulk(request: BulkEmailRequest):
    """
    Async bulk validation. Returns a task_id.
    Poll /validate/bulk/status/{task_id} for results.
    Uses Celery when Redis is available; otherwise falls back to a
    synchronous pass (now supports any batch size).
    """
    print(f"[BULK] Received bulk validation request for {len(request.emails)} emails")
    
    # Only use Celery when the broker is actually reachable — otherwise
    # .delay() would block retrying the Redis connection.
    try:
        from ..utils.cache import redis_available
        use_celery = (settings.celery_bulk_enabled
                      and await redis_available())
        print(f"[BULK] Redis available: {use_celery}")
    except Exception as e:
        print(f"[BULK] Redis check failed: {str(e)}")
        use_celery = False

    if use_celery:
        try:
            task = bulk_validate_task.delay(request.emails, request.webhook_url, request.deep)
            # Record the job in Supabase so it can be tracked across restarts
            if settings.enable_db_persistence:
                repo.create_bulk_job(task.id, request.emails)
            return BulkValidationResponse(
                task_id=task.id,
                status="processing",
                total=len(request.emails),
                message=f"Processing {len(request.emails)} emails. Poll /validate/bulk/status/{task.id}"
            )
        except Exception as exc:
            print(f"[BULK] Celery unavailable ({exc}) — using synchronous fallback.")

    # Synchronous fallback for any size list — but NON-BLOCKING: the request
    # returns immediately with a task_id, and the results are processed in a
    # background asyncio task. The old blocking behavior held the HTTP request
    # for the entire validation (60-112s on a blocked-egress host), which
    # exceeded the frontend's 45s fetch timeout and made the Results page
    # show nothing.
    task_id = uuid.uuid4().hex
    print(f"[BULK] Processing {len(request.emails)} emails in background (task {task_id})")

    # Record the job BEFORE processing: without this row, finish_bulk_job is an
    # UPDATE that matches nothing, and in-flight polls hit the "Redis down"
    # branch and 404 — which the frontend treats as terminal.
    if settings.enable_db_persistence:
        try:
            repo.create_bulk_job(task_id, request.emails)
        except Exception as exc:
            print(f"[BULK] create_bulk_job failed: {exc}")

    async def _run_and_store():
        async with _bulk_semaphore:
            try:
                results = await process_bulk_sync(
                    request.emails, request.webhook_url, deep=request.deep
                )
                if settings.enable_db_persistence:
                    try:
                        repo.finish_bulk_job(task_id, results)
                    except Exception as exc:
                        print(f"[BULK] finish_bulk_job failed: {exc}")
                await set_cache(
                    f"bulk_result:{task_id}",
                    json.dumps({"status": "completed", "total": len(request.emails),
                                "results": results}),
                    ttl=settings.redis_cache_ttl,
                )
                print(f"[BULK] task {task_id} completed: {len(results)} results")
            except Exception as exc:
                print(f"[BULK] task {task_id} failed: {exc}")
                await set_cache(
                    f"bulk_result:{task_id}",
                    json.dumps({"status": "error", "error": str(exc),
                                "results": []}),
                    ttl=settings.redis_cache_ttl,
                )

    # The loop holds only a WEAK reference to tasks: without a strong ref the
    # job can be garbage-collected mid-run and its results silently lost.
    task = asyncio.create_task(_run_and_store())
    _background_tasks.add(task)
    task.add_done_callback(_background_tasks.discard)

    return JSONResponse(content={
        "task_id": task_id,
        "status": "processing",
        "total": len(request.emails),
        "message": f"Processing {len(request.emails)} emails. Poll /validate/bulk/status/{task_id}",
    })


@router.get("/validate/bulk/status/{task_id}", response_model=TaskStatusResponse, tags=["Validation"])
async def get_bulk_status(task_id: str):
    """Poll the status of a bulk validation task."""
    from ..utils.cache import redis_available

    def _row_to_result(row):
        """Best-effort coercion of a stored/fallback row into a ValidationResult."""
        if row is None or not isinstance(row, dict):
            return None
        if "scoring" not in row and "score" in row:
            row = {
                **row,
                "status": row.get("status") or "unknown",
                "scoring": {
                    "score": row.get("score") or 0,
                    "risk_level": row.get("risk_level") or "low",
                },
            }
        try:
            return ValidationResult(**row)
        except Exception:
            return None

    def _normalize_results(raw):
        rows = raw if isinstance(raw, list) else []
        results = [r for r in (_row_to_result(row) for row in rows) if r is not None]
        return results or None  # None when absent → serialized as "results": null

    # Background-processed bulk results (the non-Celery async path) are stored
    # in the shared cache — check it FIRST so polling works for every task id.
    stored = await get_cache(f"bulk_result:{task_id}")
    if stored:
        try:
            data = json.loads(stored)
            return TaskStatusResponse(
                task_id=task_id,
                status=data.get("status", "completed"),
                progress=data.get("total"),
                total=data.get("total"),
                results=_normalize_results(data.get("results")),
            )
        except Exception:
            pass  # corrupt payload -> fall through to the other backends

    if task_id == "sync":
        # Sync-fallback results are persisted next to the "sync" job row; read
        # them back so polling keeps working after the original response.
        if settings.enable_db_persistence:
            job = repo.get_bulk_job(task_id)
            if job:
                return TaskStatusResponse(
                    task_id=task_id,
                    status=job.get("status", "completed") or "completed",
                    progress=job.get("processed"),
                    total=job.get("total"),
                    results=_normalize_results(job.get("results")),
                )
        return TaskStatusResponse(task_id=task_id, status="completed")
    if not await redis_available():
        # Redis down → fall back to Supabase job tracking (never hang on the backend)
        if settings.enable_db_persistence:
            job = repo.get_bulk_job(task_id)
            if job:
                return TaskStatusResponse(
                    task_id=task_id,
                    status=job.get("status", "unknown"),
                    progress=job.get("processed"),
                    total=job.get("total"),
                    results=_normalize_results(job.get("results"))
                )
        # An in-flight background job has no cached result yet. Report
        # "processing" instead of 404 — the frontend treats 404 as terminal and
        # would abandon a job that is still running.
        if any(not t.done() for t in _background_tasks):
            return TaskStatusResponse(task_id=task_id, status="processing")
        raise HTTPException(status_code=404, detail=f"Task not found: {task_id}")

    try:
        from celery.result import AsyncResult
        result = AsyncResult(task_id)
        payload = {
            "task_id": task_id,
            "status": result.status.lower(),
        }
        # Surface live progress while the task is still running. Celery's
        # PROGRESS state stores {"current": n, "total": n} in result.info.
        if result.status == "PROGRESS" and isinstance(result.info, dict):
            payload["progress"] = result.info.get("current")
            payload["total"] = result.info.get("total")
        if result.ready():
            raw = result.result if isinstance(result.result, list) else []
            payload["results"] = _normalize_results(raw)
            # Persist finished job so it survives worker restarts
            if settings.enable_db_persistence:
                repo.finish_bulk_job(task_id, raw)
        return TaskStatusResponse(**payload)
    except Exception as e:
        # Fallback: read job progress from Supabase
        if settings.enable_db_persistence:
            job = repo.get_bulk_job(task_id)
            if job:
                return TaskStatusResponse(
                    task_id=task_id,
                    status=job.get("status", "unknown"),
                    progress=job.get("processed"),
                    total=job.get("total"),
                    results=_normalize_results(job.get("results"))
                )
        raise HTTPException(status_code=404, detail=f"Task not found: {str(e)}")


# ---- Domain Info ---------------------------------------------------
@router.get("/domain/{domain}", summary="Get full DNS/authentication info for a domain", tags=["Domain"])
async def get_domain_info(domain: str):
    """Get MX, SPF, DMARC, DKIM records for a domain."""
    from ..validator.dns_check import check_dns
    result = await check_dns(domain, timeout=settings.dns_timeout)
    if settings.enable_db_persistence:
        repo.save_domain_record(domain, result)
    return result


# ---- Quick Check (Syntax + DNS only, no SMTP) ---------------------
@router.get("/validate/quick/{email}", summary="Quick syntax + DNS check (no SMTP)", tags=["Validation"])
async def quick_validate(email: str):
    """Syntax + DNS only. Fast. No SMTP probe."""
    result = await engine.validate(email, deep=False)
    return result


# ---- Health Check --------------------------------------------------
@router.get("/health", tags=["System"])
async def health():
    db = await repo.db_status()
    
    # Check Redis and Celery status
    redis_status = False
    celery_status = False
    try:
        from ..utils.cache import redis_available
        redis_status = await redis_available()
    except Exception:
        pass
    
    if redis_status:
        try:
            from celery import Celery
            test_celery = Celery(
                "health_check",
                broker=settings.celery_broker_url,
                backend=settings.celery_result_backend
            )
            # Try to inspect workers
            inspect = test_celery.control.inspect(timeout=2)
            celery_status = inspect.stats() is not None
        except Exception:
            pass
    
    # Autonomous-engine worker summary (spec §3): never fatal, best-effort.
    workers = {
        "dns": {"doh_enabled": settings.doh_enabled},
        "smtp": {
            "provider": engine.mailbox.name,
            "worker_configured": bool(settings.smtp_worker_url),
        },
        "ai_scorer": {
            "configured": bool(settings.ai_scorer_model) and settings.ai_scorer_local,
        },
    }

    return {
        "status": "healthy",
        "app": settings.app_name,
        "version": settings.app_version,
        "database": db,
        "redis": {"available": redis_status},
        "celery": {"available": celery_status},
        "workers": workers,
        "timestamp": time.time()
    }


# ---- Database Status ------------------------------------------------
@router.get("/db/status", tags=["System"], response_model=dict)
async def db_status_endpoint():
    """Supabase connectivity + schema readiness."""
    return await repo.db_status()


# ---- ML advisory diagnostics ----------------------------------------
@router.get("/ml/status", tags=["System"])
async def ml_status():
    """Operability of the advisory DistilBERT layer (flag / repo / load state)."""
    from ..config import settings
    from ..ml.runtime import get_model

    model = get_model()
    sample = None
    if model is not None and model.loaded:
        try:
            sample = model.predict(["john.smith [SEP] stripe.com"])
        except Exception as exc:  # pragma: no cover - defensive
            sample = [f"predict raised: {exc}"]
    return {
        "enabled": settings.ml_disposable_enabled,
        "repo": settings.ml_model_repo,
        "loaded": bool(model is not None and model.loaded),
        "last_error": getattr(model, "last_error", None) if model else None,
        "predict_sample": sample,
    }


# ---- History --------------------------------------------------------
@router.get("/history", tags=["Validation"])
async def get_history(limit: int = Query(default=50, ge=1, le=500)):
    """Most recent validations persisted in Supabase (newest first)."""
    return repo.get_validation_history(limit=limit)


# ---- Stats ---------------------------------------------------------
@router.get("/stats", tags=["System"])
async def stats():
    from ..validator.proxy_engine import proxy_pool
    db_stats = repo.get_stats()
    return {
        "proxy_pool": proxy_pool.stats,
        "database": db_stats,
        "version": settings.app_version
    }


# ---- Autonomous engine status (spec §3: four-worker visibility) -----
@router.get("/engine/status", tags=["System"])
async def engine_status():
    """Live state of every autonomous-engine worker (no PII)."""
    from ..validator.egress import check_port25_egress

    egress = await asyncio.to_thread(check_port25_egress)
    scorer_status = engine.scorer.status()
    worker_url = settings.smtp_worker_url
    return {
        "version": settings.app_version,
        "dns": {
            "doh_enabled": settings.doh_enabled,
            "resolvers": ["dns.google", "cloudflare-dns.com"],
            "dual_consistency": settings.doh_enabled,
        },
        "smtp": {
            "provider": engine.mailbox.name,
            "helo_domain": settings.smtp_helo_domain,
            "remote_worker": (
                {"configured": True, "host": urlsplit(worker_url).hostname}
                if worker_url
                else {"configured": False}
            ),
            "egress_port25": egress.as_dict(),
        },
        "ai_scorer": scorer_status,
        "catch_all_ttl_seconds": settings.catch_all_ttl_seconds,
        "dns_cache_ttl_seconds": settings.dns_cache_ttl_seconds,
    }


# ---- Inference burn-rate instrumentation (spec Phase 6) --------------
@router.get("/metrics/inference", tags=["System"])
async def inference_metrics():
    """AI-scorer burn rate against the <5% target (spec Phase 6)."""
    return engine.scorer.metrics.as_dict()


# ---- SMTP probe worker endpoints (spec §3.3) ------------------------
# Any deployment of this image can serve as the port-25 SMTP worker for an
# orchestrator whose own egress is blocked. Guarded by a shared secret
# (X-Worker-Token) when SMTP_WORKER_TOKEN is configured on either side.
async def _require_worker_token(request: Request) -> None:
    expected = settings.smtp_worker_token
    if not expected:
        return
    supplied = request.headers.get("x-worker-token", "")
    if supplied != expected:
        raise HTTPException(status_code=401, detail="worker token invalid")


class WorkerProbeRequest(BaseModel):
    email: Optional[str] = None
    domain: Optional[str] = None
    mx_host: str
    probe: str = "mailbox"  # "mailbox" | "catch_all"


class WorkerBatchRequest(BaseModel):
    pairs: List[List[str]]


@router.post("/smtp/probe", tags=["Worker"])
async def smtp_probe(request: Request, body: WorkerProbeRequest):
    """One mailbox or catch-all probe from THIS host's port-25 egress."""
    await _require_worker_token(request)
    from ..validator.mailbox_provider import AutonomousSmtpProvider

    worker = AutonomousSmtpProvider()
    if body.probe == "catch_all":
        if not body.domain:
            raise HTTPException(status_code=422, detail="domain required for catch_all probe")
        is_ca, confidence = await worker.catch_all(body.domain, body.mx_host)
        return {"is_catch_all": is_ca, "confidence": confidence}
    if not body.email:
        raise HTTPException(status_code=422, detail="email required for mailbox probe")
    result = await worker.verify(body.email, body.mx_host)
    return json.loads(result.model_dump_json())


@router.post("/smtp/probe/batch", tags=["Worker"])
async def smtp_probe_batch(request: Request, body: WorkerBatchRequest):
    """Batched mailbox probes (one session per receiving MX host)."""
    await _require_worker_token(request)
    from ..validator.mailbox_provider import AutonomousSmtpProvider

    worker = AutonomousSmtpProvider()
    pairs = [(row[0], row[1]) for row in body.pairs if len(row) >= 2]
    results, catch_all = await worker.verify_many(pairs)
    return {
        "results": {e: json.loads(r.model_dump_json()) for e, r in results.items()},
        "catch_all": catch_all,
    }

	from urllib.parse import urlsplit
	from typing import List, Optional

	from fastapi import APIRouter, HTTPException, Query, Request
	from fastapi.responses import JSONResponse
	from pydantic import BaseModel
	import json
	import time
	import asyncio
	import uuid

	from ..models import (
	SingleEmailRequest, BulkEmailRequest,
	ValidationResult, BulkValidationResponse,
	TaskStatusResponse
	)
	from ..validator.engine import engine
	from ..utils.cache import get_cache, set_cache
	from ..config import settings
	from ..db import repo

	# Strong references to in-flight background bulk jobs (the event loop only
	# keeps weak ones, so a task can vanish mid-run and its results are lost), and
	# a cap so a burst of bulk requests cannot spawn unbounded SMTP work.
	_background_tasks: set[asyncio.Task] = set()
	_bulk_semaphore = asyncio.Semaphore(settings.bulk_concurrency)

	# Celery task handle (module level so the routing is testable/patchable).
	# Import failure degrades to the background-task path; Celery itself is only
	# required when celery_bulk_enabled=True.
	try:
	from ..tasks.celery_tasks import bulk_validate_task # noqa: E402
	except Exception: # pragma: no cover - celery not installed
	bulk_validate_task = None

	router = APIRouter()


	async def process_bulk_sync(emails: list, webhook_url: str = None, deep: bool = True) -> dict:
	"""
	Process bulk validation synchronously in chunks.
	Used when Celery is unavailable.

	`deep` defaults to True so bulk runs the full 7-layer pipeline (catch-all +
	SMTP) exactly like single verification — the product charges 1 credit/email
	and must not silently degrade to a syntax+DNS-only check.

	Chunks feed the engine's batched path: DNS/MX resolves once per domain and
	the mailbox stage takes the whole chunk in one call, so the SMTP engine reuses
	a session per receiving host instead of opening a TCP connection per address.
	"""
	results = []
	total = len(emails)
	chunk_size = settings.bulk_sync_chunk_size # Configurable chunk size

	for i in range(0, total, chunk_size):
	chunk = emails[i:i + chunk_size]
	try:
	chunk_results = await engine.validate_batch(chunk, deep=deep)
	except Exception as exc: # never lose a chunk to one bad stage
	results.extend(
	{
	"email": email,
	"error": str(exc) or "Unknown error",
	"status": "error",
	}
	for email in chunk
	)
	continue

	# validate_batch guarantees one result per input, in input order.
	if len(chunk_results) == len(chunk):
	pairs = zip(chunk, chunk_results)
	else:
	# Defensive: never lose or mis-attribute a row.
	by_email = {getattr(r, "email", "").lower(): r for r in chunk_results}
	pairs = ((email, by_email.get(email.lower())) for email in chunk)

	for email, result in pairs:
	if isinstance(result, ValidationResult):
	results.append(result.model_dump())
	else:
	results.append(
	{
	"email": email,
	"error": "engine returned no result for this address",
	"status": "error",
	}
	)

	# Webhook notification if provided
	if webhook_url:
	import httpx
	try:
	async with httpx.AsyncClient(timeout=10) as client:
	await client.post(webhook_url, json={"results": results})
	except Exception:
	pass

	# Run-quality monitor (report item 10): fire the signed system-alert
	# webhook when this run's risky/unknown share crosses the threshold.
	try:
	from ..utils.rate_alerts import maybe_alert_risky_rate

	await maybe_alert_risky_rate(
	results,
	webhook_url=webhook_url,
	job_id="sync",
	threshold=settings.risky_rate_alert_threshold,
	)
	except Exception:
	pass

	return results


	# ---- Single Email Validation --------------------------------------
	@router.post(
	"/validate/single",
	response_model=ValidationResult,
	summary="Validate a single email address",
	tags=["Validation"]
	)
	async def validate_single(
	request: SingleEmailRequest,
	deep: bool = Query(default=True, description="Enable SMTP verification (slower but more accurate)")
	):
	"""
	Full 7-layer email validation:
	Syntax → DNS → MX → Disposable → Catch-All → SMTP → Score
	"""
	email = request.email

	# Check cache first
	cache_key = f"val:{email}:{'deep' if deep else 'fast'}"
	cached = await get_cache(cache_key)
	if cached:
	result = ValidationResult(**json.loads(cached))
	return result.model_copy(update={"cache_hit": True})

	# Run full validation
	result = await engine.validate(email, deep=deep)

	# Cache the result
	await set_cache(cache_key, result.model_dump_json(), ttl=settings.redis_cache_ttl)

	# Persist to Supabase (best-effort)
	if settings.enable_db_persistence:
	repo.save_validation_result(result)

	return result


	# ---- Bulk Email Validation ----------------------------------------
	@router.post(
	"/validate/bulk",
	summary="Validate multiple email addresses (async)",
	tags=["Validation"]
	)
	async def validate_bulk(request: BulkEmailRequest):
	"""
	Async bulk validation. Returns a task_id.
	Poll /validate/bulk/status/{task_id} for results.

	Uses Celery when Redis is available; otherwise falls back to a
	synchronous pass (now supports any batch size).
	"""
	print(f"[BULK] Received bulk validation request for {len(request.emails)} emails")

	# Only use Celery when the broker is actually reachable — otherwise
	# .delay() would block retrying the Redis connection.
	try:
	from ..utils.cache import redis_available
	use_celery = (settings.celery_bulk_enabled
	and await redis_available())
	print(f"[BULK] Redis available: {use_celery}")
	except Exception as e:
	print(f"[BULK] Redis check failed: {str(e)}")
	use_celery = False

	if use_celery:
	try:
	task = bulk_validate_task.delay(request.emails, request.webhook_url, request.deep)
	# Record the job in Supabase so it can be tracked across restarts
	if settings.enable_db_persistence:
	repo.create_bulk_job(task.id, request.emails)
	return BulkValidationResponse(
	task_id=task.id,
	status="processing",
	total=len(request.emails),
	message=f"Processing {len(request.emails)} emails. Poll /validate/bulk/status/{task.id}"
	)
	except Exception as exc:
	print(f"[BULK] Celery unavailable ({exc}) — using synchronous fallback.")

	# Synchronous fallback for any size list — but NON-BLOCKING: the request
	# returns immediately with a task_id, and the results are processed in a
	# background asyncio task. The old blocking behavior held the HTTP request
	# for the entire validation (60-112s on a blocked-egress host), which
	# exceeded the frontend's 45s fetch timeout and made the Results page
	# show nothing.
	task_id = uuid.uuid4().hex
	print(f"[BULK] Processing {len(request.emails)} emails in background (task {task_id})")

	# Record the job BEFORE processing: without this row, finish_bulk_job is an
	# UPDATE that matches nothing, and in-flight polls hit the "Redis down"
	# branch and 404 — which the frontend treats as terminal.
	if settings.enable_db_persistence:
	try:
	repo.create_bulk_job(task_id, request.emails)
	except Exception as exc:
	print(f"[BULK] create_bulk_job failed: {exc}")

	async def _run_and_store():
	async with _bulk_semaphore:
	try:
	results = await process_bulk_sync(
	request.emails, request.webhook_url, deep=request.deep
	)
	if settings.enable_db_persistence:
	try:
	repo.finish_bulk_job(task_id, results)
	except Exception as exc:
	print(f"[BULK] finish_bulk_job failed: {exc}")
	await set_cache(
	f"bulk_result:{task_id}",
	json.dumps({"status": "completed", "total": len(request.emails),
	"results": results}),
	ttl=settings.redis_cache_ttl,
	)
	print(f"[BULK] task {task_id} completed: {len(results)} results")
	except Exception as exc:
	print(f"[BULK] task {task_id} failed: {exc}")
	await set_cache(
	f"bulk_result:{task_id}",
	json.dumps({"status": "error", "error": str(exc),
	"results": []}),
	ttl=settings.redis_cache_ttl,
	)

	# The loop holds only a WEAK reference to tasks: without a strong ref the
	# job can be garbage-collected mid-run and its results silently lost.
	task = asyncio.create_task(_run_and_store())
	_background_tasks.add(task)
	task.add_done_callback(_background_tasks.discard)

	return JSONResponse(content={
	"task_id": task_id,
	"status": "processing",
	"total": len(request.emails),
	"message": f"Processing {len(request.emails)} emails. Poll /validate/bulk/status/{task_id}",
	})


	@router.get("/validate/bulk/status/{task_id}", response_model=TaskStatusResponse, tags=["Validation"])
	async def get_bulk_status(task_id: str):
	"""Poll the status of a bulk validation task."""
	from ..utils.cache import redis_available

	def _row_to_result(row):
	"""Best-effort coercion of a stored/fallback row into a ValidationResult."""
	if row is None or not isinstance(row, dict):
	return None
	if "scoring" not in row and "score" in row:
	row = {
	**row,
	"status": row.get("status") or "unknown",
	"scoring": {
	"score": row.get("score") or 0,
	"risk_level": row.get("risk_level") or "low",
	},
	}
	try:
	return ValidationResult(**row)
	except Exception:
	return None

	def _normalize_results(raw):
	rows = raw if isinstance(raw, list) else []
	results = [r for r in (_row_to_result(row) for row in rows) if r is not None]
	return results or None # None when absent → serialized as "results": null

	# Background-processed bulk results (the non-Celery async path) are stored
	# in the shared cache — check it FIRST so polling works for every task id.
	stored = await get_cache(f"bulk_result:{task_id}")
	if stored:
	try:
	data = json.loads(stored)
	return TaskStatusResponse(
	task_id=task_id,
	status=data.get("status", "completed"),
	progress=data.get("total"),
	total=data.get("total"),
	results=_normalize_results(data.get("results")),
	)
	except Exception:
	pass # corrupt payload -> fall through to the other backends

	if task_id == "sync":
	# Sync-fallback results are persisted next to the "sync" job row; read
	# them back so polling keeps working after the original response.
	if settings.enable_db_persistence:
	job = repo.get_bulk_job(task_id)
	if job:
	return TaskStatusResponse(
	task_id=task_id,
	status=job.get("status", "completed") or "completed",
	progress=job.get("processed"),
	total=job.get("total"),
	results=_normalize_results(job.get("results")),
	)
	return TaskStatusResponse(task_id=task_id, status="completed")
	if not await redis_available():
	# Redis down → fall back to Supabase job tracking (never hang on the backend)
	if settings.enable_db_persistence:
	job = repo.get_bulk_job(task_id)
	if job:
	return TaskStatusResponse(
	task_id=task_id,
	status=job.get("status", "unknown"),
	progress=job.get("processed"),
	total=job.get("total"),
	results=_normalize_results(job.get("results"))
	)
	# An in-flight background job has no cached result yet. Report
	# "processing" instead of 404 — the frontend treats 404 as terminal and
	# would abandon a job that is still running.
	if any(not t.done() for t in _background_tasks):
	return TaskStatusResponse(task_id=task_id, status="processing")
	raise HTTPException(status_code=404, detail=f"Task not found: {task_id}")

	try:
	from celery.result import AsyncResult
	result = AsyncResult(task_id)
	payload = {
	"task_id": task_id,
	"status": result.status.lower(),
	}
	# Surface live progress while the task is still running. Celery's
	# PROGRESS state stores {"current": n, "total": n} in result.info.
	if result.status == "PROGRESS" and isinstance(result.info, dict):
	payload["progress"] = result.info.get("current")
	payload["total"] = result.info.get("total")
	if result.ready():
	raw = result.result if isinstance(result.result, list) else []
	payload["results"] = _normalize_results(raw)
	# Persist finished job so it survives worker restarts
	if settings.enable_db_persistence:
	repo.finish_bulk_job(task_id, raw)
	return TaskStatusResponse(**payload)
	except Exception as e:
	# Fallback: read job progress from Supabase
	if settings.enable_db_persistence:
	job = repo.get_bulk_job(task_id)
	if job:
	return TaskStatusResponse(
	task_id=task_id,
	status=job.get("status", "unknown"),
	progress=job.get("processed"),
	total=job.get("total"),
	results=_normalize_results(job.get("results"))
	)
	raise HTTPException(status_code=404, detail=f"Task not found: {str(e)}")


	# ---- Domain Info ---------------------------------------------------
	@router.get("/domain/{domain}", summary="Get full DNS/authentication info for a domain", tags=["Domain"])
	async def get_domain_info(domain: str):
	"""Get MX, SPF, DMARC, DKIM records for a domain."""
	from ..validator.dns_check import check_dns
	result = await check_dns(domain, timeout=settings.dns_timeout)
	if settings.enable_db_persistence:
	repo.save_domain_record(domain, result)
	return result


	# ---- Quick Check (Syntax + DNS only, no SMTP) ---------------------
	@router.get("/validate/quick/{email}", summary="Quick syntax + DNS check (no SMTP)", tags=["Validation"])
	async def quick_validate(email: str):
	"""Syntax + DNS only. Fast. No SMTP probe."""
	result = await engine.validate(email, deep=False)
	return result


	# ---- Health Check --------------------------------------------------
	@router.get("/health", tags=["System"])
	async def health():
	db = await repo.db_status()

	# Check Redis and Celery status
	redis_status = False
	celery_status = False
	try:
	from ..utils.cache import redis_available
	redis_status = await redis_available()
	except Exception:
	pass

	if redis_status:
	try:
	from celery import Celery
	test_celery = Celery(
	"health_check",
	broker=settings.celery_broker_url,
	backend=settings.celery_result_backend
	)
	# Try to inspect workers
	inspect = test_celery.control.inspect(timeout=2)
	celery_status = inspect.stats() is not None
	except Exception:
	pass

	# Autonomous-engine worker summary (spec §3): never fatal, best-effort.
	workers = {
	"dns": {"doh_enabled": settings.doh_enabled},
	"smtp": {
	"provider": engine.mailbox.name,
	"worker_configured": bool(settings.smtp_worker_url),
	},
	"ai_scorer": {
	"configured": bool(settings.ai_scorer_model) and settings.ai_scorer_local,
	},
	}

	return {
	"status": "healthy",
	"app": settings.app_name,
	"version": settings.app_version,
	"database": db,
	"redis": {"available": redis_status},
	"celery": {"available": celery_status},
	"workers": workers,
	"timestamp": time.time()
	}


	# ---- Database Status ------------------------------------------------
	@router.get("/db/status", tags=["System"], response_model=dict)
	async def db_status_endpoint():
	"""Supabase connectivity + schema readiness."""
	return await repo.db_status()


	# ---- ML advisory diagnostics ----------------------------------------
	@router.get("/ml/status", tags=["System"])
	async def ml_status():
	"""Operability of the advisory DistilBERT layer (flag / repo / load state)."""
	from ..config import settings
	from ..ml.runtime import get_model

	model = get_model()
	sample = None
	if model is not None and model.loaded:
	try:
	sample = model.predict(["john.smith [SEP] stripe.com"])
	except Exception as exc: # pragma: no cover - defensive
	sample = [f"predict raised: {exc}"]
	return {
	"enabled": settings.ml_disposable_enabled,
	"repo": settings.ml_model_repo,
	"loaded": bool(model is not None and model.loaded),
	"last_error": getattr(model, "last_error", None) if model else None,
	"predict_sample": sample,
	}


	# ---- History --------------------------------------------------------
	@router.get("/history", tags=["Validation"])
	async def get_history(limit: int = Query(default=50, ge=1, le=500)):
	"""Most recent validations persisted in Supabase (newest first)."""
	return repo.get_validation_history(limit=limit)


	# ---- Stats ---------------------------------------------------------
	@router.get("/stats", tags=["System"])
	async def stats():
	from ..validator.proxy_engine import proxy_pool
	db_stats = repo.get_stats()
	return {
	"proxy_pool": proxy_pool.stats,
	"database": db_stats,
	"version": settings.app_version
	}


	# ---- Autonomous engine status (spec §3: four-worker visibility) -----
	@router.get("/engine/status", tags=["System"])
	async def engine_status():
	"""Live state of every autonomous-engine worker (no PII)."""
	from ..validator.egress import check_port25_egress

	egress = await asyncio.to_thread(check_port25_egress)
	scorer_status = engine.scorer.status()
	worker_url = settings.smtp_worker_url
	return {
	"version": settings.app_version,
	"dns": {
	"doh_enabled": settings.doh_enabled,
	"resolvers": ["dns.google", "cloudflare-dns.com"],
	"dual_consistency": settings.doh_enabled,
	},
	"smtp": {
	"provider": engine.mailbox.name,
	"helo_domain": settings.smtp_helo_domain,
	"remote_worker": (
	{"configured": True, "host": urlsplit(worker_url).hostname}
	if worker_url
	else {"configured": False}
	),
	"egress_port25": egress.as_dict(),
	},
	"ai_scorer": scorer_status,
	"catch_all_ttl_seconds": settings.catch_all_ttl_seconds,
	"dns_cache_ttl_seconds": settings.dns_cache_ttl_seconds,
	}


	# ---- Inference burn-rate instrumentation (spec Phase 6) --------------
	@router.get("/metrics/inference", tags=["System"])
	async def inference_metrics():
	"""AI-scorer burn rate against the <5% target (spec Phase 6)."""
	return engine.scorer.metrics.as_dict()


	# ---- SMTP probe worker endpoints (spec §3.3) ------------------------
	# Any deployment of this image can serve as the port-25 SMTP worker for an
	# orchestrator whose own egress is blocked. Guarded by a shared secret
	# (X-Worker-Token) when SMTP_WORKER_TOKEN is configured on either side.
	async def _require_worker_token(request: Request) -> None:
	expected = settings.smtp_worker_token
	if not expected:
	return
	supplied = request.headers.get("x-worker-token", "")
	if supplied != expected:
	raise HTTPException(status_code=401, detail="worker token invalid")


	class WorkerProbeRequest(BaseModel):
	email: Optional[str] = None
	domain: Optional[str] = None
	mx_host: str
	probe: str = "mailbox" # "mailbox" \| "catch_all"


	class WorkerBatchRequest(BaseModel):
	pairs: List[List[str]]


	@router.post("/smtp/probe", tags=["Worker"])
	async def smtp_probe(request: Request, body: WorkerProbeRequest):
	"""One mailbox or catch-all probe from THIS host's port-25 egress."""
	await _require_worker_token(request)
	from ..validator.mailbox_provider import AutonomousSmtpProvider

	worker = AutonomousSmtpProvider()
	if body.probe == "catch_all":
	if not body.domain:
	raise HTTPException(status_code=422, detail="domain required for catch_all probe")
	is_ca, confidence = await worker.catch_all(body.domain, body.mx_host)
	return {"is_catch_all": is_ca, "confidence": confidence}
	if not body.email:
	raise HTTPException(status_code=422, detail="email required for mailbox probe")
	result = await worker.verify(body.email, body.mx_host)
	return json.loads(result.model_dump_json())


	@router.post("/smtp/probe/batch", tags=["Worker"])
	async def smtp_probe_batch(request: Request, body: WorkerBatchRequest):
	"""Batched mailbox probes (one session per receiving MX host)."""
	await _require_worker_token(request)
	from ..validator.mailbox_provider import AutonomousSmtpProvider

	worker = AutonomousSmtpProvider()
	pairs = [(row[0], row[1]) for row in body.pairs if len(row) >= 2]
	results, catch_all = await worker.verify_many(pairs)
	return {
	"results": {e: json.loads(r.model_dump_json()) for e, r in results.items()},
	"catch_all": catch_all,
	}