app/validator/engine.py

import asyncio
import re
import time
import uuid
from datetime import datetime, timezone
from typing import Any, Optional
import structlog

from ..models import (
    ValidationResult,
    ValidationStatus,
    ScoringResult,
    CatchAllResult,
    DNSResult,
    DisposableResult,
    SMTPResult,
    SyntaxResult,
)
from ..config import settings
from .syntax import validate_syntax
from .dns_check import check_dns
from .disposable import check_disposable, domain_in_blocklist, domain_is_known
from .mailbox_provider import MailboxProvider, resolve_provider
from .i18n import EAI_REASON, eai_gate
from ..ml.risk_scorer import AIRiskScorer, ai_scorer, label_for_score, score_many_by_domain
from ..ml.runtime import get_model_async, predict_async
from .mx_intel import mx_is_trusted
from .retry_queue import is_greylist_result
from .scoring import calculate_score

logger = structlog.get_logger()


def _mask_email_for_log(email: str) -> str:
    local_part, separator, domain = email.rpartition("@")
    if not separator or not domain:
        return "[invalid-email]"
    return f"{local_part[:1]}***@{domain}"


def _build_checks_run(
    syntax: SyntaxResult | None,
    dns: DNSResult | None,
    disposable: DisposableResult | None,
    catch_all: CatchAllResult | None,
    smtp: SMTPResult | None,
    scoring: ScoringResult | None,
) -> list[str]:
    """Which of the 7-layer taxonomy keys actually executed for this result.

    Mirrors the frontend fallback in src/lib/verifier.ts classify() so a layer
    the engine did not run is never reported as a pass. DNS without an error
    claims both `dns` and `mx`; a DNS error claims only `dns`.
    """
    checks: list[str] = []
    if syntax is not None:
        checks.append("syntax")
    if dns is not None:
        if dns.error:
            checks.append("dns")
        else:
            checks.extend(["dns", "mx"])
    if disposable is not None:
        checks.append("disposable")
    if catch_all is not None:
        checks.append("catch_all")
    if smtp is not None:
        checks.append("smtp")
    if scoring is not None:
        checks.append("scoring")
    return checks


def _build_reason_code(
    dns: DNSResult | None,
    disposable: DisposableResult | None,
    catch_all: CatchAllResult | None,
    smtp: SMTPResult | None,
    *,
    early: str | None = None,
) -> str | None:
    """Machine cause for the verdict. Ordered most-specific/actionable first.

    disposable outranks everything (frontend classifies it as Disposable). A
    definitive SMTP rejection (verified False + code) outranks the domain-level
    catch-all flag: reporting "catch_all_detected" beside status=invalid is
    incoherent — the explicit "mailbox does not exist" is the actionable fact.
    Catch-all then outranks a merely inconclusive probe.
    """
    if early:
        return early
    if disposable is not None and disposable.is_disposable:
        return "disposable"
    if dns is not None and dns.a_fallback and not dns.has_mx_records:
        return "a_fallback_only"
    if smtp is not None and smtp.verified is False and smtp.smtp_code:
        if smtp.smtp_code >= 500:
            return "rcpt_5xx"
        if smtp.smtp_code >= 400:
            return "rcpt_4xx"
        if smtp.smtp_code >= 200:
            return "rcpt_2xx"
    if catch_all is not None and catch_all.is_catch_all:
        return "catch_all_detected"
    if smtp is not None and smtp.verified is None and smtp.error:
        return (
            "smtp_timeout"
            if re.search(r"timed?\s*out", smtp.error, re.IGNORECASE)
            else "smtp_unavailable"
        )
    return None


def _deterministic_verdict(
    disposable: DisposableResult | None,
    catch_all: CatchAllResult | None,
    smtp: SMTPResult | None,
) -> str:
    """Pre-AI verdict of the deterministic gauntlet (spec Phase 5).

    Order matters: an explicit mailbox rejection outranks the domain-level
    catch-all flag (mirrors _build_reason_code), and a catch-all NEVER reads
    "valid" even when the probe was accepted — it only proves the domain
    accepts everything.
    """
    if disposable is not None and disposable.is_disposable:
        return "disposable"
    if (
        smtp is not None
        and smtp.verified is False
        and smtp.smtp_code is not None
        and smtp.smtp_code >= 500
    ):
        return "invalid"
    if catch_all is not None and catch_all.is_catch_all:
        return "catch_all"
    if smtp is not None and smtp.verified is True:
        return "valid"
    return "unknown"


def _smtp_class(smtp: SMTPResult | None) -> str:
    """Coarse SMTP response class for the AI feature string (no PII)."""
    if smtp is None:
        return "skipped"
    if smtp.verified is True:
        return "accept"
    if smtp.verified is False:
        if smtp.smtp_code and smtp.smtp_code >= 500:
            return "5xx"
        if smtp.smtp_code and smtp.smtp_code >= 400:
            return "4xx"
        return "reject"
    err = (smtp.error or "").lower()
    if "worker" in err or "unreachable" in err:
        return "transport_error"
    return "unavailable"


def _ambiguous(verdict: str) -> bool:
    """Only these two reach the AI scorer (spec gate steps 6 and 8)."""
    return verdict in {"catch_all", "unknown"}


class EmailValidationEngine:
    """
    Master 7-Layer Email Validation Orchestrator.

    Layers:
      1. Syntax / RFC 5321 check
      2. DNS domain existence
      3. MX record verification
      4. Disposable / Role-based / Free provider detection
      5. Catch-all domain detection
      6. SMTP mailbox verification
      7. Deterministic heuristic deliverability scoring
    """

    def __init__(
        self,
        mailbox: MailboxProvider | None = None,
        cache=None,
        retry_queue=None,
        scorer: AIRiskScorer | None = None,
    ):
        self.use_proxies = settings.use_proxies
        # Layers 5 (catch-all) and 6 (SMTP) run through the autonomous mailbox
        # layer: the built-in SMTP engine on this host, or the remote SMTP
        # probe worker when SMTP_WORKER_URL is configured (spec §3.3).
        self.mailbox = mailbox if mailbox is not None else resolve_provider()
        # DNS/MX and definitive mailbox verdicts are cached across runs — the
        # same domain (and often the same address) recurs constantly in bulk
        # lists. Transient failures are never cached; see validation_cache.py.
        if cache is None:
            from .validation_cache import validation_cache as cache

        self.cache = cache
        # Greylisted (4xx) addresses get a bounded second chance. Without this a
        # valid mailbox on a greylisting domain stays permanently "risky".
        if retry_queue is None:
            from .retry_queue import retry_queue as retry_queue

        self.retry_queue = retry_queue
        # AI risk scorer (spec §3.4): pattern-based classification for the
        # ambiguous rows only; circuit breaker + heuristic fallback inside.
        self.scorer = scorer if scorer is not None else ai_scorer

    async def validate(self, email: str, deep: bool = True) -> ValidationResult:
        start_time = time.monotonic()
        run_id = uuid.uuid4().hex[:12]

        log = logger.bind(email=_mask_email_for_log(email), run_id=run_id)
        log.info("validation_started")

        # ---- Layer 1: Syntax -----------------------------------------
        syntax = validate_syntax(email)

        if not syntax.passed:
            elapsed = (time.monotonic() - start_time) * 1000
            scoring = calculate_score(syntax, None, None, None, None)
            log.warning("syntax_failed", error=syntax.error)
            return ValidationResult(
                email=email,
                status=ValidationStatus.INVALID,
                syntax=syntax,
                scoring=scoring,
                processing_time_ms=round(elapsed, 2),
                validated_at=datetime.now(timezone.utc).isoformat(),
                checks_run=_build_checks_run(syntax, None, None, None, None, scoring),
                reason_code=_build_reason_code(None, None, None, None),
                run_id=run_id,
                validator_version=settings.app_version,
                deterministic_verdict="invalid",
            )

        domain = syntax.domain
        normalized_email = syntax.normalized_email or email

        # ---- Layers 2 & 3: DNS + MX ---------------------------------
        dns_result = await check_dns(domain, settings.dns_timeout)

        if dns_result.error_code in {"timeout", "resolver_error"}:
            elapsed = (time.monotonic() - start_time) * 1000
            scoring = calculate_score(syntax, dns_result, None, None, None)
            log.warning(
                "dns_inconclusive", domain=domain, error_code=dns_result.error_code
            )
            return ValidationResult(
                email=normalized_email,
                status=self._determine_status(scoring),
                syntax=syntax,
                dns=dns_result,
                scoring=scoring,
                processing_time_ms=round(elapsed, 2),
                validated_at=datetime.now(timezone.utc).isoformat(),
                checks_run=_build_checks_run(syntax, dns_result, None, None, None, scoring),
                reason_code=_build_reason_code(dns_result, None, None, None, early="dns_timeout"),
                run_id=run_id,
                validator_version=settings.app_version,
                deterministic_verdict="unknown",
            )

        if (
            not dns_result.domain_exists
            or dns_result.error_code == "nxdomain"
            or (not dns_result.has_mx_records and not dns_result.a_fallback)
        ):
            elapsed = (time.monotonic() - start_time) * 1000
            scoring = calculate_score(syntax, dns_result, None, None, None)
            log.warning("dns_failed", domain=domain)
            return ValidationResult(
                email=normalized_email,
                status=ValidationStatus.INVALID,
                syntax=syntax,
                dns=dns_result,
                scoring=scoring,
                processing_time_ms=round(elapsed, 2),
                validated_at=datetime.now(timezone.utc).isoformat(),
                checks_run=_build_checks_run(syntax, dns_result, None, None, None, scoring),
                reason_code=_build_reason_code(
                    dns_result, None, None, None, early="nxdomain"
                ),
                run_id=run_id,
                validator_version=settings.app_version,
                deterministic_verdict="invalid",
            )

        # ---- Layer 4: Disposable Detection ---------------------------
        disposable = check_disposable(normalized_email, domain)

        # ML advisory (single path): unknown domains only, fail-open. Mirrors
        # the batch pipeline so both surfaces agree on the same address.
        # Skipped when the MX runs on a trusted enterprise platform: the model
        # learned 'unfamiliar domain string = disposable' from an imbalanced
        # dataset and flags legitimate small businesses on Google Workspace /
        # M365 — the deterministic MX intelligence is the better signal there.
        ml_model = await get_model_async()
        mx_host = (
            dns_result.mx_records[0]["host"]
            if dns_result.mx_records
            else None
        )
        if (
            ml_model is not None
            and not domain_is_known(domain)
            and deep
            and not mx_is_trusted(mx_host)
        ):
            # Off the event loop: CPU inference inline would stall every other
            # concurrent request (measured 42.6s cold start before this).
            preds = await predict_async(ml_model, [f"john.smith [SEP] {domain}"])
            if preds:
                disposable = disposable.model_copy(
                    update={"ml_label": preds[0][0], "ml_confidence": preds[0][1]}
                )

        if not deep:
            # Fast path: skip SMTP for quick validation
            elapsed = (time.monotonic() - start_time) * 1000
            scoring = calculate_score(
                syntax,
                dns_result,
                disposable,
                None,
                None,
                mailbox_verification_skipped=True,
            )
            status = self._determine_status(scoring)
            return ValidationResult(
                email=normalized_email,
                status=status,
                syntax=syntax,
                dns=dns_result,
                disposable=disposable,
                scoring=scoring,
                processing_time_ms=round(elapsed, 2),
                validated_at=datetime.now(timezone.utc).isoformat(),
                checks_run=_build_checks_run(
                    syntax, dns_result, disposable, None, None, scoring
                ),
                reason_code=_build_reason_code(dns_result, disposable, None, None),
                run_id=run_id,
                validator_version=settings.app_version,
                # Fast path: the mailbox question was never asked, so there is
                # no deterministic mailbox verdict to report.
                deterministic_verdict=None,
            )

        # ---- Layers 5 & 6: Catch-All + SMTP --------------------------
        mx_host = dns_result.mx_records[0]["host"]

        # EAI/SMTPUTF8 gate (spec §7.2): a non-ASCII local part can only be
        # delivered when the receiving MX advertises SMTPUTF8. Without an SMTP
        # dialogue the engine refuses the per-address probe deterministically
        # instead of faking a verdict. The domain-level catch-all probe (a
        # random ASCII local part) still runs — it is independent of the
        # address's own local part.
        eai_reason = eai_gate(syntax.local_part or "")

        catch_all_task = asyncio.create_task(
            self.mailbox.catch_all(domain, mx_host)
        )
        if eai_reason:
            smtp_result = SMTPResult(
                verified=None,
                error=(
                    f"{EAI_REASON}: mailbox probe requires SMTPUTF8 support "
                    "(deterministic refusal)"
                ),
            )
            try:
                catch_all_raw = await catch_all_task
            except Exception as exc:
                # Same contract as the gather path: a failed probe is UNKNOWN.
                catch_all_raw = exc
        else:
            smtp_task = asyncio.create_task(
                self.mailbox.verify(normalized_email, mx_host)
            )
            catch_all_raw, smtp_result = await asyncio.gather(
                catch_all_task, smtp_task, return_exceptions=True
            )

        # Handle catch-all result. An exception means the probe could not run
        # (e.g. port-25 egress blocked) — that is UNKNOWN, not "not catch-all".
        if isinstance(catch_all_raw, Exception):
            catch_all = CatchAllResult(
                is_catch_all=None, confidence=0.0, error=str(catch_all_raw)
            )
        else:
            is_catch_all, confidence = catch_all_raw
            catch_all = CatchAllResult(is_catch_all=is_catch_all, confidence=confidence)

        # Handle SMTP result
        if isinstance(smtp_result, Exception):
            smtp_result = SMTPResult(verified=None, error=str(smtp_result))

        # ---- Layer 7: Scoring ----------------------------------------
        scoring = calculate_score(
            syntax, dns_result, disposable, catch_all, smtp_result
        )

        status = self._determine_status(scoring)
        elapsed = (time.monotonic() - start_time) * 1000

        # ---- Normalized verdict (spec Phase 5): deterministic first, then AI --
        deterministic = _deterministic_verdict(disposable, catch_all, smtp_result)
        ai_score: float | None = None
        ai_label: str | None = None
        fallback_used = False
        if deep and _ambiguous(deterministic):
            features = AIRiskScorer.build_features(
                domain=domain,
                mx_host=mx_host,
                smtp_class=_smtp_class(smtp_result),
                catch_all=catch_all.is_catch_all if catch_all else None,
                local_part=syntax.local_part or "",
                dns_inconsistent=dns_result.dns_inconsistent,
            )
            scored = await self.scorer.score(features)
            ai_score, ai_label, fallback_used = (
                scored.score,
                scored.label,
                scored.fallback_used,
            )
            if scored.label == "high_risk":
                scoring.warnings.append("ai_high_risk")
        if dns_result.dns_inconsistent:
            scoring.warnings.append("dns_inconsistent: resolvers disagree on MX")

        reason = _build_reason_code(dns_result, disposable, catch_all, smtp_result)
        if eai_reason:
            reason = EAI_REASON

        log.info(
            "validation_complete",
            status=status,
            score=scoring.score,
            time_ms=round(elapsed, 2),
        )

        return ValidationResult(
            email=normalized_email,
            status=status,
            syntax=syntax,
            dns=dns_result,
            disposable=disposable,
            catch_all=catch_all,
            smtp=smtp_result,
            scoring=scoring,
            processing_time_ms=round(elapsed, 2),
            validated_at=datetime.now(timezone.utc).isoformat(),
            checks_run=_build_checks_run(
                syntax, dns_result, disposable, catch_all, smtp_result, scoring
            ),
            reason_code=reason,
            run_id=run_id,
            validator_version=settings.app_version,
            deterministic_verdict=deterministic,
            ai_risk_score=ai_score,
            ai_label=ai_label,
            fallback_used=fallback_used,
        )

    async def validate_batch(
        self, emails: list[str], deep: bool = True
    ) -> list[ValidationResult]:
        """Validate many addresses in one pass, reusing per-domain work.

        Bulk throughput comes from doing DNS/MX **once per domain** and handing
        every (email, mx_host) pair to the mailbox provider in a **single** call,
        so the SMTP engine can reuse one session per receiving host instead of a
        fresh TCP handshake per address. Providers that cannot batch fall back to
        the per-address path, so behaviour is never worse than before.

        Results are returned in the same order as `emails`.
        """
        if not emails:
            return []

        run_id = uuid.uuid4().hex[:12]
        out: list[ValidationResult | None] = [None] * len(emails)

        def _stamp(**kw) -> ValidationResult:
            return ValidationResult(
                processing_time_ms=0.0,
                validated_at=datetime.now(timezone.utc).isoformat(),
                run_id=run_id,
                validator_version=settings.app_version,
                **kw,
            )

        # ---- Layer 1: syntax (per address) ---------------------------
        parsed: list[tuple[str, SyntaxResult]] = []
        domain_index: dict[str, list[int]] = {}
        for idx, raw in enumerate(emails):
            syntax = validate_syntax(raw)
            parsed.append((raw, syntax))
            if not syntax.passed:
                scoring = calculate_score(syntax, None, None, None, None)
                out[idx] = _stamp(
                    email=raw,
                    status=ValidationStatus.INVALID,
                    syntax=syntax,
                    scoring=scoring,
                    checks_run=_build_checks_run(syntax, None, None, None, None, scoring),
                    reason_code=_build_reason_code(None, None, None, None),
                )
                continue
            domain_index.setdefault(syntax.domain, []).append(idx)

        # ---- Layers 2 & 3: DNS + MX once per domain (cached) ---------
        domains = list(domain_index.keys())
        domain_dns: dict[str, DNSResult] = {}
        misses: list[str] = []
        for domain in domains:
            cached_dns = await self.cache.get_dns(domain)
            if cached_dns is not None:
                domain_dns[domain] = cached_dns
            else:
                misses.append(domain)

        if misses:
            resolved = await asyncio.gather(
                *[check_dns(d, settings.dns_timeout) for d in misses],
                return_exceptions=True,
            )
            for domain, dns in zip(misses, resolved):
                result = (
                    dns
                    if isinstance(dns, DNSResult)
                    else DNSResult(
                        domain_exists=False,
                        has_mx_records=False,
                        error=str(dns),
                        error_code="resolver_error",
                    )
                )
                domain_dns[domain] = result
                # put_dns ignores transient failures internally.
                await self.cache.put_dns(domain, result)

        pending: list[tuple[int, str, SyntaxResult, str, DNSResult]] = []
        for domain, idxs in domain_index.items():
            dns = domain_dns[domain]
            inconclusive = dns.error_code in {"timeout", "resolver_error"}
            dead = (
                not dns.domain_exists
                or dns.error_code == "nxdomain"
                or (not dns.has_mx_records and not dns.a_fallback)
            )
            for idx in idxs:
                raw, syntax = parsed[idx]
                normalized = syntax.normalized_email or raw
                if inconclusive:
                    scoring = calculate_score(syntax, dns, None, None, None)
                    out[idx] = _stamp(
                        email=normalized,
                        status=self._determine_status(scoring),
                        syntax=syntax,
                        dns=dns,
                        scoring=scoring,
                        checks_run=_build_checks_run(
                            syntax, dns, None, None, None, scoring
                        ),
                        reason_code=_build_reason_code(
                            dns, None, None, None, early="dns_timeout"
                        ),
                        deterministic_verdict="unknown",
                    )
                elif dead:
                    scoring = calculate_score(syntax, dns, None, None, None)
                    out[idx] = _stamp(
                        email=normalized,
                        status=ValidationStatus.INVALID,
                        syntax=syntax,
                        dns=dns,
                        scoring=scoring,
                        checks_run=_build_checks_run(
                            syntax, dns, None, None, None, scoring
                        ),
                        reason_code=_build_reason_code(
                            dns, None, None, None, early="nxdomain"
                        ),
                        deterministic_verdict="invalid",
                    )
                else:
                    pending.append((idx, normalized, syntax, domain, dns))

        # ---- Layer 4 fast path (no mailbox probe) --------------------
        if pending and not deep:
            for idx, normalized, syntax, domain, dns in pending:
                disposable = check_disposable(normalized, domain)
                scoring = calculate_score(
                    syntax, dns, disposable, None, None,
                    mailbox_verification_skipped=True,
                )
                out[idx] = _stamp(
                    email=normalized,
                    status=self._determine_status(scoring),
                    syntax=syntax,
                    dns=dns,
                    disposable=disposable,
                    scoring=scoring,
                    checks_run=_build_checks_run(
                        syntax, dns, disposable, None, None, scoring
                    ),
                    reason_code=_build_reason_code(dns, disposable, None, None),
                    deterministic_verdict=None,
                )
            pending = []

        # ---- Layers 5 & 6: ONE batched mailbox call ------------------
        cached_verdicts: dict[str, tuple[SMTPResult, Optional[bool]]] = {}
        catch_all_map: dict[str, Optional[bool]] = {}
        if pending:
            # Serve anything already answered (definitive verdicts only) so we
            # never re-probe an address we already know the answer to.
            to_probe: list[tuple[int, str, SyntaxResult, str, DNSResult]] = []
            for item in pending:
                _idx, normalized, _syntax, _domain, _dns = item
                hit = await self.cache.get_verdict(normalized, deep)
                if hit is None:
                    to_probe.append(item)
                else:
                    cached_verdicts[normalized] = hit
                    if hit[1] is not None:
                        catch_all_map[item[3]] = hit[1]

            smtp_map: dict[str, SMTPResult] = {}
            pairs: list[tuple[str, str]] = []
            for _idx, normalized, _syntax, _domain, _dns in to_probe:
                if eai_gate(_syntax.local_part or ""):
                    # Deterministic EAI refusal (spec §7.2): the probe requires
                    # an SMTPUTF8 advertisement we cannot observe, so refuse
                    # instead of guessing. Cache/check semantics unchanged.
                    smtp_map[normalized] = SMTPResult(
                        verified=None,
                        error=(
                            f"{EAI_REASON}: mailbox probe requires SMTPUTF8 "
                            "(deterministic refusal)"
                        ),
                    )
                    continue
                if _dns.mx_records:
                    pairs.append((normalized, _dns.mx_records[0]["host"]))
                else:
                    smtp_map[normalized] = SMTPResult(
                        verified=None, error="no MX host available for probe"
                    )
            if pairs:
                try:
                    if hasattr(self.mailbox, "verify_many"):
                        smtp_map, caught = await self.mailbox.verify_many(pairs)
                        for domain, value in (caught or {}).items():
                            catch_all_map.setdefault(domain, value)
                    else:
                        # Provider cannot batch: per-address fallback.
                        for email, mx in pairs:
                            smtp_map[email] = await self.mailbox.verify(email, mx)
                        for domain in {e.rsplit("@", 1)[-1].lower() for e, _ in pairs}:
                            mx = next(
                                m for e, m in pairs
                                if e.rsplit("@", 1)[-1].lower() == domain
                            )
                            value, _conf = await self.mailbox.catch_all(domain, mx)
                            catch_all_map.setdefault(domain, value)
                except Exception as exc:  # one bad stage must not abort the batch
                    smtp_map = {
                        e: SMTPResult(verified=None, error=f"mailbox_stage_error: {exc}")
                        for e, _ in pairs
                    }

            # ---- ML advisory signal (unknown domains, ONE batched call) ----
            # Runs only where the curated lists have no answer - the exact gap
            # a static blocklist cannot close. Any failure here is silently
            # ignored: the deterministic verdict stands.
            # Domains on trusted enterprise MX (Google Workspace / M365 /
            # gateways) are excluded: disposable domains essentially never run
            # there, and the model's 'unfamiliar string' bias would otherwise
            # flag legitimate businesses as disposable.
            ml_map: dict[str, tuple[str, float]] = {}
            ml_model = await get_model_async()
            if ml_model is not None:
                unknown_domains = sorted(
                    {
                        domain
                        for _, _, _, domain, dns in pending
                        if not domain_is_known(domain)
                        and dns.mx_records
                        and not mx_is_trusted(dns.mx_records[0]["host"])
                    }
                )
                if unknown_domains:
                    # One batched call, off the event loop (CPU inference inline
                    # stalls concurrent requests), and fail-open by construction:
                    # predict_async never raises.
                    preds = await predict_async(
                        ml_model, [f"john.smith [SEP] {d}" for d in unknown_domains]
                    )
                    if preds:
                        ml_map = dict(zip(unknown_domains, preds))

            # ---- Layer 7: score each row -----------------------------
            ai_cache: dict[str, Any] = {}  # feature string -> AIScore (dedupe)
            for idx, normalized, syntax, domain, dns in pending:
                cached = cached_verdicts.get(normalized)
                if cached is not None:
                    smtp, ca_cached = cached
                else:
                    smtp = smtp_map.get(normalized) or SMTPResult(
                        verified=None, error="engine returned no result"
                    )
                    ca_cached = None
                    # Persist only definitive answers (verified verdict or a
                    # confirmed catch-all); transient failures are never cached.
                    await self.cache.put_verdict(
                        normalized, deep, smtp, catch_all_map.get(domain)
                    )

                ca_value = (
                    ca_cached if ca_cached is not None else catch_all_map.get(domain)
                )
                catch_all = CatchAllResult(
                    is_catch_all=ca_value,
                    confidence=0.9 if ca_value is not None else 0.0,
                )
                disposable = check_disposable(normalized, domain)
                ml = ml_map.get(domain)
                if ml is not None:
                    disposable = disposable.model_copy(
                        update={"ml_label": ml[0], "ml_confidence": ml[1]}
                    )

                # A 4xx means "come back later", not a verdict. Schedule one
                # bounded retry so a real mailbox on a greylisting domain is not
                # stranded as permanently risky. The queue never raises.
                if cached is None:
                    try:
                        if is_greylist_result(smtp):
                            await self.retry_queue.enqueue(
                                normalized, dns.mx_records[0]["host"], smtp.smtp_code
                            )
                        elif smtp.verified is not None:
                            # Definitive answer — cancel any pending retry.
                            await self.retry_queue.resolve_email(normalized)
                    except Exception:
                        pass

                scoring = calculate_score(
                    syntax, dns, disposable, catch_all, smtp
                )

                # ---- Normalized verdict (spec Phase 5) -------------------
                deterministic = _deterministic_verdict(disposable, catch_all, smtp)
                ai_score: float | None = None
                ai_label: str | None = None
                fallback_used = False
                if deep and _ambiguous(deterministic):
                    feats = AIRiskScorer.build_features(
                        domain=domain,
                        mx_host=dns.mx_records[0]["host"] if dns.mx_records else None,
                        smtp_class=_smtp_class(smtp),
                        catch_all=ca_value,
                        local_part=syntax.local_part or "",
                        dns_inconsistent=dns.dns_inconsistent,
                    )
                    if feats not in ai_cache:
                        ai_cache[feats] = await self.scorer.score(feats)
                    scored = ai_cache[feats]
                    ai_score, ai_label, fallback_used = (
                        scored.score,
                        scored.label,
                        scored.fallback_used,
                    )
                    if scored.label == "high_risk":
                        scoring.warnings.append("ai_high_risk")
                if dns.dns_inconsistent:
                    scoring.warnings.append("dns_inconsistent: resolvers disagree on MX")
                reason = _build_reason_code(dns, disposable, catch_all, smtp)
                if (syntax.local_part or "") and eai_gate(syntax.local_part or ""):
                    reason = EAI_REASON

                out[idx] = _stamp(
                    email=normalized,
                    status=self._determine_status(scoring),
                    syntax=syntax,
                    dns=dns,
                    disposable=disposable,
                    catch_all=catch_all,
                    smtp=smtp,
                    scoring=scoring,
                    checks_run=_build_checks_run(
                        syntax, dns, disposable, catch_all, smtp, scoring
                    ),
                    reason_code=reason,
                    deterministic_verdict=deterministic,
                    ai_risk_score=ai_score,
                    ai_label=ai_label,
                    fallback_used=fallback_used,
                    cache_hit=cached is not None,
                )

        # Guarantee one result per input, in order.
        for idx, result in enumerate(out):
            if result is None:
                raw = emails[idx]
                syntax = parsed[idx][1] if idx < len(parsed) else validate_syntax(raw)
                scoring = calculate_score(syntax, None, None, None, None)
                out[idx] = _stamp(
                    email=raw,
                    status=ValidationStatus.UNKNOWN,
                    syntax=syntax,
                    scoring=scoring,
                    checks_run=_build_checks_run(
                        syntax, None, None, None, None, scoring
                    ),
                    reason_code="engine_missing_result",
                    deterministic_verdict="unknown",
                )
        return [r for r in out if r is not None]

    def _determine_status(self, scoring: ScoringResult) -> ValidationStatus:
        if scoring.score >= 75:
            return ValidationStatus.VALID
        elif scoring.score >= 40:
            return ValidationStatus.RISKY
        else:
            return ValidationStatus.INVALID


# Singleton engine
engine = EmailValidationEngine()