feat(quality): A4 — L2 Pro premium judge (8 dims, gpt-4o, Pro-gated, opt-in)

config.py

@@ -91,6 +91,19 @@ class Config:
     # Hard ceiling on the L1 call (seconds). Anything longer is a skip.
     QUALITY_L1_TIMEOUT_SEC = float(os.getenv("QUALITY_L1_TIMEOUT_SEC", "8.0"))
 
+    # ============== Quality Layer (L2 — Pro tier) ==============
+    # Track A4 of the dev plan — STRONGER LLM judge (8 dimensions).
+    # Gated to Pro+ plans in the route. Default off everywhere.
+    # Cost: ~$0.005–$0.02/job (gpt-4o) or ~$0.001/job (gpt-4o-mini).
+    # Set QUALITY_L2_TIER_GATE=false to allow L2 for free tier too.
+    QUALITY_L2_ENABLED = os.getenv("QUALITY_L2_ENABLED", "false").lower() == "true"
+    QUALITY_L2_LOG_ONLY = os.getenv("QUALITY_L2_LOG_ONLY", "true").lower() == "true"
+    QUALITY_L2_SAMPLE_SIZE = int(os.getenv("QUALITY_L2_SAMPLE_SIZE", "15"))
+    QUALITY_L2_MIN_CHUNKS = int(os.getenv("QUALITY_L2_MIN_CHUNKS", "20"))
+    QUALITY_L2_TIMEOUT_SEC = float(os.getenv("QUALITY_L2_TIMEOUT_SEC", "20.0"))
+    # When true, only Pro+ plans can use L2. Otherwise, all plans can.
+    QUALITY_L2_TIER_GATE = os.getenv("QUALITY_L2_TIER_GATE", "true").lower() == "true"
+
 
     # ============== API Configuration ==============
     API_TITLE = "Document Translation API"

middleware/metrics.py

@@ -70,6 +70,28 @@ quality_l1_judge_cost_usd = Histogram(
     buckets=(0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1),
 )
 
+# ---- L2 Pro premium judge (Track A4) ----
+
+quality_l2_judge_total = Counter(
+    "quality_l2_judge_total",
+    "Total L2 (Pro premium judge, 8-dim) verdicts",
+    ["verdict", "model", "tier"],  # verdict: pass | fail | skip | error
+)
+
+quality_l2_judge_duration_seconds = Histogram(
+    "quality_l2_judge_duration_seconds",
+    "L2 Pro judge call duration in seconds",
+    ["model"],
+    buckets=(0.5, 1, 2, 5, 10, 20, 30, 60),
+)
+
+quality_l2_judge_cost_usd = Histogram(
+    "quality_l2_judge_cost_usd",
+    "L2 Pro judge estimated cost in USD",
+    ["model"],
+    buckets=(0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0),
+)
+
 # ---- Retry metrics ----
 
 translation_retry_total = Counter(
@@ -138,6 +160,29 @@ def record_l1_verdict(
         quality_l1_judge_cost_usd.labels(model=model).observe(cost_usd)
 
 
+def record_l2_verdict(
+    verdict: str,
+    model: str = "unknown",
+    tier: str = "pro",
+    duration_seconds: float = None,
+    cost_usd: float = None,
+):
+    """Record an L2 Pro judge verdict.
+
+    Args:
+        verdict: pass | fail | skip | error
+        model: model name (e.g. gpt-4o)
+        tier: pro | business | enterprise
+        duration_seconds: optional, observed in histogram
+        cost_usd: optional, observed in histogram
+    """
+    quality_l2_judge_total.labels(verdict=verdict, model=model, tier=tier).inc()
+    if duration_seconds is not None:
+        quality_l2_judge_duration_seconds.labels(model=model).observe(duration_seconds)
+    if cost_usd is not None:
+        quality_l2_judge_cost_usd.labels(model=model).observe(cost_usd)
+
+
 def record_translation_retry(reason: str, tier: str = "free"):
     """Record a translation retry triggered by a quality issue.
 

routes/translate_routes.py

@@ -1455,6 +1455,49 @@ async def _run_translation_job(
             except Exception:
                 pass
 
+        # ------------------------------------------------------------------
+        # Quality L2 layer (Track A4 — Pro premium judge)
+        # Stronger LLM (gpt-4o default), 8 dimensions, 15 samples.
+        # Gated to Pro+ plans (configurable via QUALITY_L2_TIER_GATE).
+        # Default OFF everywhere — observation first.
+        # Cost: ~$0.005–$0.02/job (gpt-4o), ~$0.001/job (gpt-4o-mini).
+        # ------------------------------------------------------------------
+        if getattr(config, "QUALITY_L2_ENABLED", False) and quality_samples:
+            try:
+                from services.quality import run_l2_check
+
+                # Tier gate: Pro+ plans only (unless gate is disabled)
+                user_tier = (
+                    _tier_for_quota(current_user.plan) if current_user else "free"
+                )
+                tier_gate_on = getattr(config, "QUALITY_L2_TIER_GATE", True)
+                if not tier_gate_on or user_tier in ("pro", "business", "enterprise"):
+                    translated_chunks_for_l2 = [s["translated"] for s in quality_samples]
+                    l2_result = await run_l2_check(
+                        source_chunks=[""] * len(translated_chunks_for_l2),
+                        translated_chunks=translated_chunks_for_l2,
+                        target_lang=target_lang,
+                        l0_failed_indices=l0_failed_indices,
+                        job_id=job_id,
+                        file_extension=file_extension,
+                        max_samples=getattr(config, "QUALITY_L2_SAMPLE_SIZE", 15),
+                        min_chunks=getattr(config, "QUALITY_L2_MIN_CHUNKS", 20),
+                        log_only=getattr(config, "QUALITY_L2_LOG_ONLY", True),
+                    )
+                else:
+                    # Free/Starter user — skip L2 silently (gated)
+                    logger.info(
+                        "quality_l2_check_skipped",
+                        job_id=job_id,
+                        reason="tier_gated",
+                        tier=user_tier,
+                    )
+            except Exception as l2_err:
+                # L2 must NEVER break a job. Log and continue.
+                logger.warning(
+                    f"Job {job_id}: quality L2 layer failed: {l2_err}"
+                )
+
         if user_id:
             # Determine cost factor based on selected provider and model
             cost_factor = 1

services/quality/l2_judge.py

@@ -0,0 +1,395 @@
+"""
+L2 Pro Premium Judge — stronger model, more dimensions, Pro-tier only.
+
+Why a SEPARATE module from L1?
+  - L1 is fast + cheap (deepseek-chat, 4 dimensions, 5 samples)
+  - L2 is slow + expensive (gpt-4o, 8 dimensions, 15 samples)
+  - Different defaults, different config, different metrics
+  - L2 is gated to the Pro plan; L1 is universal
+
+L2 dimensions (8):
+  1. accurate       — meaning preserved
+  2. fluent         — natural in target language
+  3. correct_lang   — in the target language (not source leakage)
+  4. no_leaks       — no prompt artifacts
+  5. terminology    — domain terms correctly handled
+  6. style          — appropriate register (formal/informal/technical)
+  7. completeness   — no content dropped or added
+  8. formatting     — codes, numbers, units preserved
+
+L1 was a binary pass/fail. L2 returns per-dimension scores (0/1) so the
+caller can decide which dimensions matter for a given job.
+
+DESIGN CONSTRAINTS (same as L1):
+  - 100% API-based. No local models, no GPU.
+  - Async, with a hard timeout.
+  - Defensive: never raises.
+  - Output is structured JSON.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import re
+import time
+from dataclasses import dataclass, field, asdict
+from typing import List, Optional, Tuple, Dict, Any
+
+from core.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+# ---------- Result dataclasses ----------
+
+@dataclass
+class L2DimensionVerdict:
+    """8-dimension verdict for a single chunk."""
+    accurate: bool = False
+    fluent: bool = False
+    correct_lang: bool = False
+    no_leaks: bool = False
+    terminology: bool = False
+    style: bool = False
+    completeness: bool = False
+    formatting: bool = False
+    reason: str = ""
+
+    @property
+    def passed_count(self) -> int:
+        return sum([
+            self.accurate, self.fluent, self.correct_lang, self.no_leaks,
+            self.terminology, self.style, self.completeness, self.formatting,
+        ])
+
+    @property
+    def total(self) -> int:
+        return 8
+
+    @property
+    def score(self) -> float:
+        return self.passed_count / self.total
+
+    @property
+    def passed(self) -> bool:
+        # L2 is conservative: any single fail = chunk fails
+        return self.passed_count == self.total
+
+    def to_log_dict(self) -> dict:
+        return asdict(self)
+
+
+@dataclass
+class L2Result:
+    """Aggregate result of an L2 check on a sample of chunks."""
+    verdict: str                  # "pass", "fail", "skip"
+    chunks_evaluated: int = 0
+    chunks_passed: int = 0
+    chunks_failed: int = 0
+    failure_rate: float = 0.0
+    average_score: float = 0.0    # mean of per-chunk scores (0.0 to 1.0)
+    dimension_pass_rates: Dict[str, float] = field(default_factory=dict)
+    samples: List[dict] = field(default_factory=list)
+    model_used: str = ""
+    elapsed_ms: float = 0.0
+    cost_estimate_usd: float = 0.0
+    error: str = ""
+
+    def to_log_dict(self) -> dict:
+        return asdict(self)
+
+
+# ---------- Prompt template (8 dimensions) ----------
+
+L2_JUDGE_SYSTEM_PROMPT = """You are an expert translation quality evaluator using MQM-inspired criteria.
+
+For each (SOURCE, TRANSLATION) pair, check these 8 criteria (yes/no for each):
+
+1. ACCURATE     — Does the translation preserve the meaning of the source?
+2. FLUENT       — Is the translation natural and grammatical in {target_lang_name}?
+3. CORRECT_LANG — Is the translation actually in {target_lang_name} (ISO: {target_lang})?
+4. NO_LEAKS     — Is the translation free of prompt artifacts, source-language text, or meta-commentary?
+5. TERMINOLOGY  — Are domain-specific terms (technical, legal, medical, etc.) correctly translated?
+6. STYLE        — Is the register/tone appropriate (formal/informal/technical matching the source)?
+7. COMPLETENESS — Is all content present, with nothing added or dropped?
+8. FORMATTING   — Are codes, numbers, dates, and units preserved exactly?
+
+A translation FAILS if ANY criterion is "no". The "reason" must be in English and ≤ 20 words.
+
+Respond with a JSON array, one object per pair, in the same order. NO other text, NO markdown fences:
+[
+  {{"accurate": "yes"|"no", "fluent": "yes"|"no", "correct_lang": "yes"|"no", "no_leaks": "yes"|"no", "terminology": "yes"|"no", "style": "yes"|"no", "completeness": "yes"|"no", "formatting": "yes"|"no", "reason": "short justification"}}
+]
+"""
+
+
+# ---------- LLM client ----------
+
+class L2ProJudge:
+    """
+    Calls a STRONG LLM via the OpenAI-compatible API to judge translation
+    quality across 8 dimensions. Pro-tier only.
+
+    Default model: gpt-4o (strongest general judge we can afford at scale).
+    For sub-$0.01/job cost, we limit to 15 samples per job and 8 dimensions.
+    """
+
+    def __init__(
+        self,
+        api_key: str,
+        base_url: str = "https://api.openai.com/v1",
+        model: str = "gpt-4o",
+        timeout_seconds: float = 20.0,
+        max_retries: int = 1,
+    ):
+        if not api_key:
+            raise ValueError("api_key is required for L2ProJudge")
+        self._api_key = api_key
+        self._base_url = base_url.rstrip("/")
+        self._model = model
+        self._timeout = timeout_seconds
+        self._max_retries = max_retries
+        self._client = None
+
+    def _get_client(self):
+        if self._client is None:
+            try:
+                import openai
+                self._client = openai.AsyncOpenAI(
+                    api_key=self._api_key,
+                    base_url=self._base_url,
+                    timeout=self._timeout,
+                )
+            except Exception as e:
+                logger.warning("l2_judge_client_init_failed", error=str(e))
+                return None
+        return self._client
+
+    async def judge_batch(
+        self,
+        pairs: List[Tuple[str, str]],
+        target_lang: str,
+        target_lang_name: str = "",
+    ) -> L2Result:
+        """
+        Judge a batch of (source, translation) pairs across 8 dimensions.
+
+        Returns an L2Result with verdict="skip" on any internal error.
+        Never raises.
+        """
+        start = time.time()
+        empty = L2Result(verdict="skip", error="not_run")
+
+        if not pairs:
+            return L2Result(verdict="skip", error="empty pairs",
+                            elapsed_ms=round((time.time() - start) * 1000, 2))
+
+        client = self._get_client()
+        if client is None:
+            return L2Result(verdict="skip", error="client unavailable",
+                            elapsed_ms=round((time.time() - start) * 1000, 2))
+
+        user_lines = [f"Target language: {target_lang} ({target_lang_name})\n"]
+        for i, (src, trans) in enumerate(pairs, 1):
+            user_lines.append(f"\n--- Pair {i} ---")
+            user_lines.append(f"SOURCE: {src}")
+            user_lines.append(f"TRANSLATION: {trans}")
+        user_msg = "\n".join(user_lines)
+
+        system_prompt = L2_JUDGE_SYSTEM_PROMPT.format(
+            target_lang=target_lang,
+            target_lang_name=target_lang_name or target_lang,
+        )
+
+        try:
+            response = await asyncio.wait_for(
+                self._call_with_retries(client, system_prompt, user_msg),
+                timeout=self._timeout + 5.0,
+            )
+        except asyncio.TimeoutError:
+            elapsed_ms = round((time.time() - start) * 1000, 2)
+            logger.warning("l2_judge_timeout",
+                           timeout_s=self._timeout, elapsed_ms=elapsed_ms)
+            return L2Result(verdict="skip", error="timeout", elapsed_ms=elapsed_ms)
+        except Exception as e:
+            elapsed_ms = round((time.time() - start) * 1000, 2)
+            logger.warning("l2_judge_error",
+                           error=str(e)[:200], elapsed_ms=elapsed_ms)
+            return L2Result(verdict="skip", error=str(e)[:200],
+                            elapsed_ms=elapsed_ms)
+
+        verdicts = self._parse_response(response, len(pairs))
+
+        if not verdicts:
+            elapsed_ms = round((time.time() - start) * 1000, 2)
+            return L2Result(verdict="skip", error="parse_failed",
+                            elapsed_ms=elapsed_ms)
+
+        # Aggregate
+        passed = sum(1 for v in verdicts if v.passed)
+        failed = len(verdicts) - passed
+        failure_rate = failed / len(verdicts) if verdicts else 0.0
+        average_score = sum(v.score for v in verdicts) / len(verdicts)
+
+        # Per-dimension pass rates
+        dimensions = [
+            "accurate", "fluent", "correct_lang", "no_leaks",
+            "terminology", "style", "completeness", "formatting",
+        ]
+        dim_pass_rates = {}
+        for dim in dimensions:
+            count = sum(1 for v in verdicts if getattr(v, dim))
+            dim_pass_rates[dim] = round(count / len(verdicts), 3) if verdicts else 0.0
+
+        # L2 verdict: strict — any chunk fail = overall fail
+        verdict = "pass" if failed == 0 else "fail"
+
+        elapsed_ms = round((time.time() - start) * 1000, 2)
+        cost_estimate = self._estimate_cost(len(pairs))
+
+        return L2Result(
+            verdict=verdict,
+            chunks_evaluated=len(verdicts),
+            chunks_passed=passed,
+            chunks_failed=failed,
+            failure_rate=round(failure_rate, 3),
+            average_score=round(average_score, 3),
+            dimension_pass_rates=dim_pass_rates,
+            samples=[v.to_log_dict() for v in verdicts],
+            model_used=self._model,
+            elapsed_ms=elapsed_ms,
+            cost_estimate_usd=cost_estimate,
+        )
+
+    async def _call_with_retries(self, client, system_prompt: str, user_msg: str):
+        """Call the LLM with retry on transient errors."""
+        last_exc = None
+        for attempt in range(self._max_retries + 1):
+            try:
+                response = await client.chat.completions.create(
+                    model=self._model,
+                    messages=[
+                        {"role": "system", "content": system_prompt},
+                        {"role": "user", "content": user_msg},
+                    ],
+                    temperature=0.0,
+                    max_tokens=1200,          # larger than L1 (more dimensions)
+                    response_format={"type": "json_object"},
+                )
+                return response
+            except Exception as e:
+                last_exc = e
+                if attempt < self._max_retries:
+                    await asyncio.sleep(0.8)
+        raise last_exc
+
+    def _parse_response(self, response, expected_count: int) -> List[L2DimensionVerdict]:
+        """Parse the LLM response into a list of 8-dimension verdicts."""
+        try:
+            content = response.choices[0].message.content or ""
+        except (AttributeError, IndexError) as e:
+            logger.warning("l2_judge_bad_response", error=str(e))
+            return []
+
+        content = content.strip()
+        if content.startswith("```"):
+            content = re.sub(r"^```(?:json)?\s*\n?", "", content)
+            content = re.sub(r"\n?```\s*$", "", content)
+
+        try:
+            data = json.loads(content)
+        except json.JSONDecodeError as e:
+            logger.warning("l2_judge_json_parse_error",
+                           error=str(e), content_preview=content[:200])
+            return []
+
+        if isinstance(data, dict):
+            items = None
+            for key in ("verdicts", "results", "translations", "data"):
+                if key in data and isinstance(data[key], list):
+                    items = data[key]
+                    break
+            if items is None:
+                for v in data.values():
+                    if isinstance(v, list):
+                        items = v
+                        break
+            if items is None:
+                logger.warning("l2_judge_no_list_in_response")
+                return []
+        elif isinstance(data, list):
+            items = data
+        else:
+            logger.warning("l2_judge_unexpected_response_type",
+                           type_=type(data).__name__)
+            return []
+
+        verdicts: List[L2DimensionVerdict] = []
+        for item in items:
+            try:
+                v = L2DimensionVerdict(
+                    accurate=str(item.get("accurate", "")).lower() == "yes",
+                    fluent=str(item.get("fluent", "")).lower() == "yes",
+                    correct_lang=str(item.get("correct_lang", "")).lower() == "yes",
+                    no_leaks=str(item.get("no_leaks", "")).lower() == "yes",
+                    terminology=str(item.get("terminology", "")).lower() == "yes",
+                    style=str(item.get("style", "")).lower() == "yes",
+                    completeness=str(item.get("completeness", "")).lower() == "yes",
+                    formatting=str(item.get("formatting", "")).lower() == "yes",
+                    reason=str(item.get("reason", ""))[:300],
+                )
+                verdicts.append(v)
+            except Exception as e:
+                logger.warning("l2_judge_item_parse_error",
+                               error=str(e), item=str(item)[:200])
+
+        return verdicts
+
+    def _estimate_cost(self, num_pairs: int) -> float:
+        """Rough USD cost estimate for the call."""
+        # L2 has more dimensions = longer output
+        input_tokens = 250 + (num_pairs * 280)
+        output_tokens = num_pairs * 110
+        model_lower = self._model.lower()
+        # IMPORTANT: check 'mini' BEFORE full 'gpt-4o' because
+        # 'gpt-4o-mini' contains 'gpt-4o'.
+        if "gpt-4o-mini" in model_lower:
+            input_cost = input_tokens / 1_000_000 * 0.15
+            output_cost = output_tokens / 1_000_000 * 0.60
+        elif "gpt-4o" in model_lower:
+            input_cost = input_tokens / 1_000_000 * 2.50
+            output_cost = output_tokens / 1_000_000 * 10.00
+        elif "claude" in model_lower:
+            input_cost = input_tokens / 1_000_000 * 3.00
+            output_cost = output_tokens / 1_000_000 * 15.00
+        else:
+            # Generic conservative
+            input_cost = input_tokens / 1_000_000 * 1.00
+            output_cost = output_tokens / 1_000_000 * 3.00
+        return round(input_cost + output_cost, 6)
+
+
+# ---------- Convenience factory ----------
+
+def make_l2_judge_from_env() -> Optional[L2ProJudge]:
+    """
+    Build an L2ProJudge from environment variables. Returns None if
+    no API key is configured.
+
+    Reads:
+      - L2_JUDGE_API_KEY       (required)
+      - L2_JUDGE_BASE_URL      (default: OpenAI)
+      - L2_JUDGE_MODEL         (default: gpt-4o)
+      - L2_JUDGE_TIMEOUT       (default: 20.0)
+    """
+    import os
+    api_key = os.getenv("L2_JUDGE_API_KEY", "").strip()
+    if not api_key:
+        return None
+    return L2ProJudge(
+        api_key=api_key,
+        base_url=os.getenv("L2_JUDGE_BASE_URL", "https://api.openai.com/v1"),
+        model=os.getenv("L2_JUDGE_MODEL", "gpt-4o"),
+        timeout_seconds=float(os.getenv("L2_JUDGE_TIMEOUT", "20.0")),
+    )

services/quality/pipeline.py

@@ -23,6 +23,7 @@ from core.logging import get_logger
 from .script_detector import evaluate_document, DocumentQualityResult
 from .sampler import sample_chunks_for_l1
 from .llm_judge import L1Result, LLMJudge
+from .l2_judge import L2Result, L2ProJudge
 
 logger = get_logger(__name__)
 
@@ -271,3 +272,155 @@ def make_judge_from_env_safe() -> Optional[LLMJudge]:
     except Exception as e:
         logger.warning("l1_judge_init_failed", error=str(e)[:200])
         return None
+
+
+# ---------- L2 (Pro tier) ----------
+
+async def run_l2_check(
+    source_chunks: List[str],
+    translated_chunks: List[str],
+    target_lang: Optional[str],
+    l0_failed_indices: Optional[Set[int]] = None,
+    job_id: Optional[str] = None,
+    file_extension: Optional[str] = None,
+    max_samples: int = 15,
+    min_chunks: int = 20,
+    judge: Optional[L2ProJudge] = None,
+    log_only: bool = True,
+) -> L2Result:
+    """
+    Run the L2 Pro premium judge (8 dimensions, gpt-4o default).
+
+    Args:
+        source_chunks: Original texts.
+        translated_chunks: Translated texts.
+        target_lang: Target language code (e.g. "fr", "en").
+        l0_failed_indices: Indices that L0 flagged as bad — skipped.
+        job_id: For logging.
+        file_extension: For logging.
+        max_samples: How many chunks to send to the LLM.
+        min_chunks: Skip the check if document has fewer chunks.
+        judge: An L2ProJudge instance. If None, created from env vars.
+        log_only: If True, never propagate the verdict (observation mode).
+                  If False, the caller can decide what to do with the verdict.
+
+    Returns an L2Result. verdict="skip" on any internal error.
+    Never raises — defensive wrapper.
+    """
+    skip = L2Result(verdict="skip", error="not_run")
+
+    if l0_failed_indices is None:
+        l0_failed_indices = set()
+
+    # Sample (reuse the L1 sampler — it's just chunk selection, model-agnostic)
+    sample = sample_chunks_for_l1(
+        source_chunks, translated_chunks, l0_failed_indices,
+        max_samples=max_samples, min_chunks=min_chunks,
+    )
+    if not sample:
+        logger.info(
+            "quality_l2_check_skipped",
+            job_id=job_id,
+            reason="insufficient_chunks_or_all_flagged",
+            chunk_count=len(source_chunks),
+        )
+        _record_l2_metric(verdict="skip", model="none")
+        return skip
+
+    # Get the judge
+    if judge is None:
+        judge = make_l2_judge_from_env_safe()
+
+    if judge is None:
+        logger.info(
+            "quality_l2_check_skipped",
+            job_id=job_id,
+            reason="no_l2_judge_configured",
+        )
+        _record_l2_metric(verdict="skip", model="none")
+        return skip
+
+    # Get the language name for the prompt
+    target_lang_name = _LANG_NAMES.get((target_lang or "").lower(), target_lang or "auto")
+
+    # Call the LLM
+    try:
+        result = await judge.judge_batch(sample, target_lang or "auto", target_lang_name)
+    except Exception as e:
+        logger.warning(
+            "quality_l2_check_failed",
+            job_id=job_id,
+            error=str(e)[:200],
+            error_type=type(e).__name__,
+        )
+        _record_l2_metric(verdict="error", model="unknown")
+        return L2Result(verdict="skip", error=str(e)[:200])
+
+    # Log (always) — caller decides what to do
+    logger.info(
+        "quality_l2_check",
+        job_id=job_id,
+        file_extension=file_extension,
+        target_lang=target_lang,
+        verdict=result.verdict,
+        chunks_evaluated=result.chunks_evaluated,
+        chunks_passed=result.chunks_passed,
+        chunks_failed=result.chunks_failed,
+        failure_rate=result.failure_rate,
+        average_score=result.average_score,
+        dimension_pass_rates=result.dimension_pass_rates,
+        model=result.model_used,
+        elapsed_ms=result.elapsed_ms,
+        cost_estimate_usd=result.cost_estimate_usd,
+        log_only=log_only,
+    )
+
+    # Record Prometheus metric
+    duration_s = None
+    if result.elapsed_ms is not None:
+        try:
+            duration_s = float(result.elapsed_ms) / 1000.0
+        except Exception:
+            duration_s = None
+    _record_l2_metric(
+        verdict=result.verdict or "skip",
+        model=result.model_used or "unknown",
+        duration_seconds=duration_s,
+        cost_usd=result.cost_estimate_usd,
+    )
+    return result
+
+
+def _record_l2_metric(
+    verdict: str,
+    model: str = "unknown",
+    duration_seconds: float = None,
+    cost_usd: float = None,
+) -> None:
+    """Best-effort Prometheus metric emission for L2.
+
+    Never raises.
+    """
+    try:
+        from middleware.metrics import record_l2_verdict
+        record_l2_verdict(
+            verdict=verdict,
+            model=model,
+            duration_seconds=duration_seconds,
+            cost_usd=cost_usd,
+        )
+    except Exception:
+        pass
+
+
+def make_l2_judge_from_env_safe() -> Optional[L2ProJudge]:
+    """Read env vars and build an L2 judge, or return None if not configured.
+
+    Defensive wrapper — a misconfigured L2 environment NEVER breaks a job.
+    """
+    try:
+        from .l2_judge import make_l2_judge_from_env
+        return make_l2_judge_from_env()
+    except Exception as e:
+        logger.warning("l2_judge_init_failed", error=str(e)[:200])
+        return None

tests/services/quality/test_l2_judge.py

@@ -0,0 +1,494 @@
+"""
+Tests for Track A4 — L2 Pro premium judge.
+
+Covers:
+  - L2DimensionVerdict: 8 dimensions + scoring
+  - L2Result: aggregation + dimension pass rates
+  - L2ProJudge: construction, missing api_key, cost estimation
+  - make_l2_judge_from_env: env-var-driven factory
+  - run_l2_check: defensive wrapper
+"""
+import os
+import json
+import pytest
+from unittest.mock import MagicMock, AsyncMock, patch
+
+from services.quality.l2_judge import (
+    L2DimensionVerdict,
+    L2Result,
+    L2ProJudge,
+    make_l2_judge_from_env,
+    L2_JUDGE_SYSTEM_PROMPT,
+)
+
+
+# ============================================================================
+# L2DimensionVerdict
+# ============================================================================
+
+class TestL2DimensionVerdict:
+    def test_all_pass(self):
+        v = L2DimensionVerdict(
+            accurate=True, fluent=True, correct_lang=True, no_leaks=True,
+            terminology=True, style=True, completeness=True, formatting=True,
+        )
+        assert v.passed_count == 8
+        assert v.total == 8
+        assert v.score == 1.0
+        assert v.passed is True
+
+    def test_one_fails(self):
+        v = L2DimensionVerdict(
+            accurate=True, fluent=True, correct_lang=True, no_leaks=True,
+            terminology=False,  # <-- fails
+            style=True, completeness=True, formatting=True,
+        )
+        assert v.passed_count == 7
+        assert v.total == 8
+        assert v.score == pytest.approx(0.875)
+        # L2 is strict: one fail = chunk fails
+        assert v.passed is False
+
+    def test_all_fail(self):
+        v = L2DimensionVerdict()  # all default False
+        assert v.passed_count == 0
+        assert v.score == 0.0
+        assert v.passed is False
+
+    def test_default_construction(self):
+        v = L2DimensionVerdict()
+        assert v.accurate is False
+        assert v.reason == ""
+
+
+# ============================================================================
+# L2Result
+# ============================================================================
+
+class TestL2Result:
+    def test_default_construction(self):
+        r = L2Result(verdict="pass")
+        assert r.verdict == "pass"
+        assert r.chunks_evaluated == 0
+        assert r.dimension_pass_rates == {}
+        assert r.error == ""
+
+    def test_to_log_dict(self):
+        r = L2Result(
+            verdict="pass",
+            chunks_evaluated=10,
+            chunks_passed=8,
+            chunks_failed=2,
+            failure_rate=0.2,
+            average_score=0.85,
+            model_used="gpt-4o",
+            cost_estimate_usd=0.012,
+        )
+        d = r.to_log_dict()
+        assert d["verdict"] == "pass"
+        assert d["chunks_evaluated"] == 10
+        assert d["model_used"] == "gpt-4o"
+        assert d["cost_estimate_usd"] == 0.012
+
+
+# ============================================================================
+# L2ProJudge construction
+# ============================================================================
+
+class TestL2ProJudgeConstruction:
+    def test_requires_api_key(self):
+        with pytest.raises(ValueError, match="api_key is required"):
+            L2ProJudge(api_key="")
+
+    def test_basic_construction(self):
+        judge = L2ProJudge(api_key="sk-test")
+        assert judge._api_key == "sk-test"
+        assert judge._model == "gpt-4o"
+        assert judge._base_url == "https://api.openai.com/v1"
+
+    def test_custom_model(self):
+        judge = L2ProJudge(
+            api_key="sk-test",
+            model="gpt-4o-mini",
+            base_url="https://api.openai.com/v1",
+        )
+        assert judge._model == "gpt-4o-mini"
+
+    def test_strips_trailing_slash_from_base_url(self):
+        judge = L2ProJudge(
+            api_key="sk-test",
+            base_url="https://api.example.com/v1/",
+        )
+        assert judge._base_url == "https://api.example.com/v1"
+
+
+# ============================================================================
+# Cost estimation
+# ============================================================================
+
+class TestL2CostEstimation:
+    def test_gpt4o_cost(self):
+        judge = L2ProJudge(api_key="sk", model="gpt-4o")
+        cost = judge._estimate_cost(15)  # 15 samples default
+        # Should be in the $0.01–$0.05 range for 15 chunks
+        assert 0.001 < cost < 0.10
+
+    def test_gpt4o_mini_cheaper(self):
+        judge_mini = L2ProJudge(api_key="sk", model="gpt-4o-mini")
+        judge_full = L2ProJudge(api_key="sk", model="gpt-4o")
+        cost_mini = judge_mini._estimate_cost(15)
+        cost_full = judge_full._estimate_cost(15)
+        # gpt-4o-mini should be cheaper than gpt-4o
+        assert cost_mini < cost_full
+
+    def test_zero_pairs(self):
+        judge = L2ProJudge(api_key="sk", model="gpt-4o")
+        cost = judge._estimate_cost(0)
+        # Even with 0 pairs, the system prompt has some cost
+        assert cost >= 0
+
+
+# ============================================================================
+# judge_batch — defensive
+# ============================================================================
+
+class TestL2JudgeBatch:
+    @pytest.mark.asyncio
+    async def test_empty_pairs_skips(self):
+        judge = L2ProJudge(api_key="sk")
+        result = await judge.judge_batch([], "fr", "French")
+        assert result.verdict == "skip"
+        assert result.error == "empty pairs"
+
+    @pytest.mark.asyncio
+    async def test_client_unavailable_skips(self):
+        judge = L2ProJudge(api_key="sk")
+        # Simulate a client init failure
+        with patch.object(judge, "_get_client", return_value=None):
+            result = await judge.judge_batch(
+                [("Hello", "Bonjour")], "fr", "French"
+            )
+        assert result.verdict == "skip"
+        assert "unavailable" in result.error or "client" in result.error.lower()
+
+    @pytest.mark.asyncio
+    async def test_successful_judgement(self):
+        """A mock client that returns a well-formed JSON response should
+        produce a passing L2Result."""
+        judge = L2ProJudge(api_key="sk", model="gpt-4o")
+
+        # Mock the client
+        mock_response = MagicMock()
+        mock_response.choices = [MagicMock()]
+        mock_response.choices[0].message.content = json.dumps([
+            {
+                "accurate": "yes", "fluent": "yes", "correct_lang": "yes",
+                "no_leaks": "yes", "terminology": "yes", "style": "yes",
+                "completeness": "yes", "formatting": "yes",
+                "reason": "Perfect translation",
+            }
+        ])
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create = AsyncMock(return_value=mock_response)
+
+        with patch.object(judge, "_get_client", return_value=mock_client):
+            result = await judge.judge_batch(
+                [("Hello", "Bonjour")], "fr", "French"
+            )
+
+        assert result.verdict == "pass"
+        assert result.chunks_evaluated == 1
+        assert result.chunks_passed == 1
+        assert result.chunks_failed == 0
+        assert result.failure_rate == 0.0
+        assert result.average_score == 1.0
+        # All 8 dimensions should have pass rate 1.0
+        for dim in ["accurate", "fluent", "terminology", "style"]:
+            assert result.dimension_pass_rates[dim] == 1.0
+
+    @pytest.mark.asyncio
+    async def test_partial_failure(self):
+        """If one of 8 dimensions fails, the chunk should fail."""
+        judge = L2ProJudge(api_key="sk", model="gpt-4o")
+
+        mock_response = MagicMock()
+        mock_response.choices = [MagicMock()]
+        # fluent = no, others yes
+        mock_response.choices[0].message.content = json.dumps([
+            {
+                "accurate": "yes", "fluent": "no", "correct_lang": "yes",
+                "no_leaks": "yes", "terminology": "yes", "style": "yes",
+                "completeness": "yes", "formatting": "yes",
+                "reason": "Awkward phrasing",
+            }
+        ])
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create = AsyncMock(return_value=mock_response)
+
+        with patch.object(judge, "_get_client", return_value=mock_client):
+            result = await judge.judge_batch(
+                [("Hello", "Bonjour")], "fr", "French"
+            )
+
+        # L2 is strict: one fail = overall fail
+        assert result.verdict == "fail"
+        assert result.chunks_evaluated == 1
+        assert result.chunks_passed == 0
+        assert result.chunks_failed == 1
+        assert result.dimension_pass_rates["fluent"] == 0.0
+        assert result.dimension_pass_rates["accurate"] == 1.0
+
+    @pytest.mark.asyncio
+    async def test_handles_markdown_fences(self):
+        """The judge should strip markdown code fences from responses."""
+        judge = L2ProJudge(api_key="sk", model="gpt-4o")
+
+        mock_response = MagicMock()
+        mock_response.choices = [MagicMock()]
+        mock_response.choices[0].message.content = (
+            "```json\n"
+            + json.dumps([{
+                "accurate": "yes", "fluent": "yes", "correct_lang": "yes",
+                "no_leaks": "yes", "terminology": "yes", "style": "yes",
+                "completeness": "yes", "formatting": "yes",
+                "reason": "ok",
+            }])
+            + "\n```"
+        )
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create = AsyncMock(return_value=mock_response)
+
+        with patch.object(judge, "_get_client", return_value=mock_client):
+            result = await judge.judge_batch(
+                [("Hello", "Bonjour")], "fr", "French"
+            )
+
+        assert result.verdict == "pass"
+        assert result.chunks_evaluated == 1
+
+    @pytest.mark.asyncio
+    async def test_handles_dict_with_list(self):
+        """Some LLMs return {"verdicts": [...]} instead of a raw list."""
+        judge = L2ProJudge(api_key="sk", model="gpt-4o")
+
+        mock_response = MagicMock()
+        mock_response.choices = [MagicMock()]
+        mock_response.choices[0].message.content = json.dumps({
+            "verdicts": [
+                {
+                    "accurate": "yes", "fluent": "yes", "correct_lang": "yes",
+                    "no_leaks": "yes", "terminology": "yes", "style": "yes",
+                    "completeness": "yes", "formatting": "yes",
+                    "reason": "good",
+                }
+            ]
+        })
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create = AsyncMock(return_value=mock_response)
+
+        with patch.object(judge, "_get_client", return_value=mock_client):
+            result = await judge.judge_batch(
+                [("Hello", "Bonjour")], "fr", "French"
+            )
+
+        assert result.verdict == "pass"
+
+    @pytest.mark.asyncio
+    async def test_timeout_returns_skip(self):
+        import asyncio
+        judge = L2ProJudge(api_key="sk", model="gpt-4o", timeout_seconds=0.1)
+
+        # Mock client that raises TimeoutError
+        mock_client = MagicMock()
+        mock_client.chat.completions.create = AsyncMock(
+            side_effect=asyncio.TimeoutError()
+        )
+
+        with patch.object(judge, "_get_client", return_value=mock_client):
+            result = await judge.judge_batch(
+                [("Hello", "Bonjour")], "fr", "French"
+            )
+
+        assert result.verdict == "skip"
+        assert "timeout" in result.error.lower()
+
+    @pytest.mark.asyncio
+    async def test_never_raises(self):
+        """Even on unexpected error, the judge should return a skip, not raise."""
+        judge = L2ProJudge(api_key="sk", model="gpt-4o")
+
+        # Mock client that raises a generic exception
+        mock_client = MagicMock()
+        mock_client.chat.completions.create = AsyncMock(
+            side_effect=RuntimeError("something unexpected")
+        )
+
+        with patch.object(judge, "_get_client", return_value=mock_client):
+            # Should NOT raise
+            result = await judge.judge_batch(
+                [("Hello", "Bonjour")], "fr", "French"
+            )
+        assert result.verdict == "skip"
+
+    @pytest.mark.asyncio
+    async def test_dimension_pass_rates_aggregated(self):
+        """Per-dimension pass rates should aggregate across chunks."""
+        judge = L2ProJudge(api_key="sk", model="gpt-4o")
+
+        mock_response = MagicMock()
+        mock_response.choices = [MagicMock()]
+        # 2 chunks: chunk 1 all-pass, chunk 2 fluent-fail
+        mock_response.choices[0].message.content = json.dumps([
+            {
+                "accurate": "yes", "fluent": "yes", "correct_lang": "yes",
+                "no_leaks": "yes", "terminology": "yes", "style": "yes",
+                "completeness": "yes", "formatting": "yes",
+                "reason": "ok",
+            },
+            {
+                "accurate": "yes", "fluent": "no", "correct_lang": "yes",
+                "no_leaks": "yes", "terminology": "yes", "style": "yes",
+                "completeness": "yes", "formatting": "yes",
+                "reason": "awkward",
+            },
+        ])
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create = AsyncMock(return_value=mock_response)
+
+        with patch.object(judge, "_get_client", return_value=mock_client):
+            result = await judge.judge_batch(
+                [("Hello", "Bonjour"), ("Goodbye", "Au revoir")],
+                "fr", "French"
+            )
+
+        assert result.chunks_evaluated == 2
+        # fluent: 1/2 = 0.5
+        assert result.dimension_pass_rates["fluent"] == 0.5
+        # accurate: 2/2 = 1.0
+        assert result.dimension_pass_rates["accurate"] == 1.0
+
+
+# ============================================================================
+# Factory
+# ============================================================================
+
+class TestL2JudgeFactory:
+    def test_no_api_key_returns_none(self, monkeypatch):
+        monkeypatch.delenv("L2_JUDGE_API_KEY", raising=False)
+        assert make_l2_judge_from_env() is None
+
+    def test_api_key_creates_judge(self, monkeypatch):
+        monkeypatch.setenv("L2_JUDGE_API_KEY", "sk-test")
+        monkeypatch.setenv("L2_JUDGE_MODEL", "gpt-4o")
+        monkeypatch.setenv("L2_JUDGE_BASE_URL", "https://api.openai.com/v1")
+        judge = make_l2_judge_from_env()
+        assert judge is not None
+        assert judge._api_key == "sk-test"
+        assert judge._model == "gpt-4o"
+
+    def test_api_key_with_default_model(self, monkeypatch):
+        monkeypatch.setenv("L2_JUDGE_API_KEY", "sk-test")
+        # Clear other vars to test defaults
+        monkeypatch.delenv("L2_JUDGE_MODEL", raising=False)
+        monkeypatch.delenv("L2_JUDGE_BASE_URL", raising=False)
+        judge = make_l2_judge_from_env()
+        assert judge._model == "gpt-4o"
+        assert judge._base_url == "https://api.openai.com/v1"
+
+
+# ============================================================================
+# Pipeline integration
+# ============================================================================
+
+class TestL2PipelineIntegration:
+    @pytest.mark.asyncio
+    async def test_run_l2_check_no_judge_skips(self):
+        from services.quality.pipeline import run_l2_check
+
+        result = await run_l2_check(
+            source_chunks=["Hello"] * 25,
+            translated_chunks=["Bonjour"] * 25,
+            target_lang="fr",
+            file_extension="docx",
+            judge=None,  # Will try to load from env, but env has no key
+        )
+        # Should return skip, not raise
+        assert result.verdict == "skip"
+
+    @pytest.mark.asyncio
+    async def test_run_l2_check_with_mock_judge(self):
+        from services.quality.pipeline import run_l2_check
+
+        # Build a judge that always returns pass
+        judge = L2ProJudge(api_key="sk")
+
+        mock_response = MagicMock()
+        mock_response.choices = [MagicMock()]
+        mock_response.choices[0].message.content = json.dumps([
+            {
+                "accurate": "yes", "fluent": "yes", "correct_lang": "yes",
+                "no_leaks": "yes", "terminology": "yes", "style": "yes",
+                "completeness": "yes", "formatting": "yes",
+                "reason": "ok",
+            }
+        ] * 5)
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create = AsyncMock(return_value=mock_response)
+        judge._client = mock_client
+
+        # 25 chunks > default min_chunks of 20
+        result = await run_l2_check(
+            source_chunks=["Hello world"] * 25,
+            translated_chunks=["Bonjour le monde"] * 25,
+            target_lang="fr",
+            file_extension="docx",
+            judge=judge,
+            max_samples=5,
+        )
+
+        assert result.verdict == "pass"
+        assert result.chunks_evaluated >= 1
+        assert result.chunks_passed >= 1
+
+    @pytest.mark.asyncio
+    async def test_run_l2_check_too_few_chunks_skips(self):
+        from services.quality.pipeline import run_l2_check
+
+        # 5 chunks < default min_chunks of 20
+        result = await run_l2_check(
+            source_chunks=["a"] * 5,
+            translated_chunks=["b"] * 5,
+            target_lang="fr",
+            min_chunks=20,
+        )
+        # Should skip due to insufficient chunks
+        assert result.verdict == "skip"
+
+
+# ============================================================================
+# 8-dimension coverage
+# ============================================================================
+
+class TestL28DimensionCoverage:
+    """Sanity check: the L2 prompt template actually mentions all 8 dimensions."""
+
+    def test_prompt_has_all_8_dimensions(self):
+        for dim in [
+            "ACCURATE", "FLUENT", "CORRECT_LANG", "NO_LEAKS",
+            "TERMINOLOGY", "STYLE", "COMPLETENESS", "FORMATTING",
+        ]:
+            assert dim in L2_JUDGE_SYSTEM_PROMPT, (
+                f"Dimension {dim!r} missing from L2 prompt template"
+            )
+
+    def test_prompt_has_format_hint(self):
+        # Should tell the model to respond with a JSON array
+        assert "JSON" in L2_JUDGE_SYSTEM_PROMPT
+        assert "yes" in L2_JUDGE_SYSTEM_PROMPT
+        assert "no" in L2_JUDGE_SYSTEM_PROMPT