feat(llm_judge): multi-sample verdicts with median aggregation

docs/TASK_DEFINITION_GUIDE.md

@@ -834,8 +834,13 @@ Have an LLM grade the task against a rubric written in the task YAML. **Continuo
 | `model` | `anthropic.claude-sonnet-4-6` | Judge model id (vendor-prefixed; auto-translated per backend) |
 | `temperature` | `0.0` | Sampling temperature (0.0 = deterministic) |
 | `max_tokens` | `2000` | Maximum tokens in the judge's response |
+| `samples` | `1` | Number of independent judge invocations (1–9). `1` = single call, unchanged behavior. With N > 1 the criterion scores the **median** of N sampled verdicts; rationale/findings come from the sample closest to the median (earliest on ties); per-sample scores land in `details`; token usage sums across samples. Samples run concurrently (wall-clock ≈ one call); judge spend multiplies by N and is not gated by `max_usd`. |
 | `max_file_chars` | `20000` | Per-file (and agent_output) truncation applied before building the prompt |
 
+**Sampling the judge.** Even at `temperature: 0.0`, judge verdicts are not perfectly repeatable — the same artifacts can draw a strict or a lenient reading of the rubric on different calls. Set `samples: 3` (odd values recommended) to invoke the judge three times over the same rendered prompt and score the median verdict, damping a single outlier reading. Because every sample grades identical artifacts, the spread across samples is pure judge variance — unlike experiment-level `replicates`, which re-run the whole agent and mix agent variance into the same number. A sample that returns no verdict degrades to the median of the remaining valid samples (with a note in `details`) rather than scoring the criterion 0.0; only when every sample fails does the single-sample failure behavior apply.
+
+**Cost and latency.** The N samples are dispatched concurrently, so wall-clock stays close to a single judge call regardless of N — `samples` does not eat into `task_timeout` beyond the one-call baseline. Judge **spend** still multiplies by N, and judge usage is deliberately excluded from `run_limits.max_usd` (that cap meters the *agent's* bill): on a dataset-fanned suite, `samples: 9` multiplies ungated judge cost 9× per row — size N with that in mind. `samples` is deliberately `llm_judge`-only: an `agent_judge` verdict costs a full tool-using sub-agent run, so multi-sampling it would multiply a far larger bill for the same variance signal — `agent_judge` does not accept the key (strict YAML validation rejects it at load time).
+
 **Transport selection.** The judge call is routed by the active `API_BACKEND`:
 
 | `API_BACKEND` | Credentials | Judge transport |

src/coder_eval/criteria/llm_judge.py

@@ -1,9 +1,13 @@
 """LLM-as-a-judge success criterion checker."""
 
 import logging
-from typing import TYPE_CHECKING
+from collections.abc import Iterable
+from concurrent.futures import ThreadPoolExecutor
+from statistics import median
+from typing import TYPE_CHECKING, NamedTuple
 
 from coder_eval.criteria.base import BaseCriterion, CheckContext, register_criterion
+from coder_eval.errors import JudgeInfrastructureError
 from coder_eval.evaluation.judge_anthropic import invoke_anthropic_judge
 from coder_eval.evaluation.judge_bedrock import invoke_bedrock_judge
 from coder_eval.evaluation.judge_context import (
@@ -113,53 +117,31 @@ def _check_impl(
 
         scrub_key = reference_code if criterion.include_reference else None
 
-        # Attribute the judge's API call to ``JudgeCriterionResult.token_usage``
-        # from the usage the backend reported in its response.
-        verdict, parse_error, raw_verdict_text, response_usage = _invoke_tool_channel(
+        # One grading path for every N: samples=1 is the one-element case whose
+        # median is its own score and whose summed usage is its own usage.
+        return _grade_with_sampling(
             criterion=criterion,
             route=route,
-            system_msg=_SYSTEM_PROMPT,
             user_msg=user_msg,
+            judge_ctx=judge_ctx,
+            scrub_key=scrub_key,
         )
-        judge_usage = response_usage
-
-        # Sanitize any raw model text we persist to CriterionResult.details. A misbehaving
-        # model could echo the reference back in an unparseable response, so we scrub it.
-        scrubbed = scrub_reference(raw_verdict_text, scrub_key)
-
-        def _maybe_transcript() -> JudgeTranscript | None:
-            if not criterion.capture_transcript:
-                return None
-            return build_judge_transcript(
-                raw_verdict=raw_verdict_text,
-                max_chars=criterion.max_transcript_chars,
-                judge_system_prompt=_SYSTEM_PROMPT,
-                judge_prompt=user_msg,
-                scrub_key=scrub_key,
-            )
 
-        if parse_error is not None:
-            return JudgeCriterionResult(
-                criterion_type=criterion.type,
-                description=criterion.description,
-                score=0.0,
-                details=scrubbed[:500],
-                error=scrub_reference(parse_error, scrub_key),
-                transcript=_maybe_transcript(),
-                token_usage=judge_usage,
-            )
-        assert verdict is not None  # parser contract: verdict is set iff parse_error is None
 
-        details = format_details(verdict.score, verdict.rationale, judge_ctx.missing_files, judge_ctx.degraded_notes)
-        return JudgeCriterionResult(
-            criterion_type=criterion.type,
-            description=criterion.description,
-            score=verdict.score,
-            details=scrub_reference(details, scrub_key),
-            findings=[scrub_reference(f, scrub_key) for f in verdict.findings],
-            transcript=_maybe_transcript(),
-            token_usage=judge_usage,
-        )
+class _SampleOutcome(NamedTuple):
+    """One judge invocation's outcome.
+
+    ``raw_verdict_text`` is the JSON-dumped verdict for the transcript when a
+    verdict was parsed, or a fallback marker when the model failed to call the
+    tool — preserves the "judge transcript carries the structured payload"
+    invariant. ``response_usage`` is the usage the model reported (``None``
+    when the backend surfaced none).
+    """
+
+    verdict: JudgeVerdict | None
+    parse_error: str | None
+    raw_verdict_text: str
+    response_usage: TokenUsage | None
 
 
 def _invoke_tool_channel(
@@ -168,16 +150,8 @@ def _invoke_tool_channel(
     route: "ApiRoute | None",
     system_msg: str,
     user_msg: str,
-) -> tuple[JudgeVerdict | None, str | None, str, TokenUsage | None]:
-    """Dispatch the tool-channel invocation by route.
-
-    Returns ``(verdict, parse_error, raw_verdict_text, response_usage)``.
-    ``raw_verdict_text`` is the JSON-dumped verdict for the transcript when
-    present, or a fallback marker when the model failed to call the tool —
-    preserves the "judge transcript carries the structured payload" invariant.
-    ``response_usage`` is the usage the model reported (``None`` when the
-    backend surfaced none).
-    """
+) -> _SampleOutcome:
+    """Dispatch the tool-channel invocation by route."""
     response_usage: TokenUsage | None
     match route:
         case BedrockRoute():
@@ -206,11 +180,176 @@ def _invoke_tool_channel(
         case _:
             # route is None or an unexpected type — the unconfigured-arm guard in
             # _check_impl handles None before dispatch, so this is defensive only.
-            return None, "llm_judge: no usable API route", "(no route)", None
+            return _SampleOutcome(
+                verdict=None,
+                parse_error="llm_judge: no usable API route",
+                raw_verdict_text="(no route)",
+                response_usage=None,
+            )
 
     if verdict is not None:
-        return verdict, None, verdict.model_dump_json(), response_usage
-    return None, err, f"(no verdict — {err})", response_usage
+        return _SampleOutcome(
+            verdict=verdict,
+            parse_error=None,
+            raw_verdict_text=verdict.model_dump_json(),
+            response_usage=response_usage,
+        )
+    return _SampleOutcome(
+        verdict=None,
+        parse_error=err,
+        raw_verdict_text=f"(no verdict — {err})",
+        response_usage=response_usage,
+    )
+
+
+def _grade_with_sampling(
+    *,
+    criterion: LLMJudgeCriterion,
+    route: "ApiRoute",
+    user_msg: str,
+    judge_ctx: JudgeContext,
+    scrub_key: str | None,
+) -> JudgeCriterionResult:
+    """Invoke the judge ``criterion.samples`` times and score the median verdict.
+
+    The single grading path: ``samples: 1`` is the one-element case (its median
+    is its own score, the summed usage is its own usage), so single-sample and
+    multi-sample results are built by the same code. Samples are independent and
+    grade the SAME rendered prompt, so they are dispatched concurrently — score
+    spread across samples is judge variance by construction, and wall-clock stays
+    close to one judge call. Futures are collected in SUBMISSION order, not
+    completion order, so the representative tie-break (earliest sample), the
+    first-sample diagnostic, and error triage stay deterministic.
+
+    A sample that produces no verdict (a transport failure, or a response with no
+    usable ``submit_verdict`` call) degrades to the median of the remaining valid
+    samples — every swallowed exception is logged at WARNING and named in the
+    degraded note so a systematic first-party bug stays loud. When NO sample
+    produces a verdict, the single-sample failure semantics apply: an
+    infrastructure failure escalates (``JudgeInfrastructureError`` propagates to
+    ``FinalStatus.ERROR`` — judge infra failure is not an agent failure), any
+    other exception reaches ``@handle_criterion_errors``, and all-parse-failures
+    score 0.0 with the first sample's diagnostic.
+    """
+    outcomes: list[_SampleOutcome] = []
+    failure_labels: list[str] = []
+    first_infra: JudgeInfrastructureError | None = None
+    first_unexpected: Exception | None = None
+    with ThreadPoolExecutor(max_workers=criterion.samples) as pool:
+        futures = [
+            pool.submit(
+                _invoke_tool_channel,
+                criterion=criterion,
+                route=route,
+                system_msg=_SYSTEM_PROMPT,
+                user_msg=user_msg,
+            )
+            for _ in range(criterion.samples)
+        ]
+        for future in futures:
+            try:
+                outcomes.append(future.result())
+            except JudgeInfrastructureError as exc:
+                logger.warning("llm_judge sample failed: %s", exc, exc_info=True)
+                failure_labels.append(f"{type(exc).__name__}: {exc}")
+                if first_infra is None:
+                    first_infra = exc
+            except Exception as exc:
+                logger.warning("llm_judge sample failed: %s", exc, exc_info=True)
+                failure_labels.append(f"{type(exc).__name__}: {exc}")
+                if first_unexpected is None:
+                    first_unexpected = exc
+
+    # Cost is real for every sample that returned a response, verdict or not.
+    token_usage = _sum_usage(o.response_usage for o in outcomes)
+    valid: list[tuple[JudgeVerdict, str]] = [(o.verdict, o.raw_verdict_text) for o in outcomes if o.verdict is not None]
+
+    if not valid:
+        if first_infra is not None:
+            raise first_infra
+        if first_unexpected is not None:
+            raise first_unexpected
+        first = outcomes[0]
+        assert first.parse_error is not None  # parser contract: verdict is set iff parse_error is None
+        scrubbed = scrub_reference(first.raw_verdict_text, scrub_key)
+        return JudgeCriterionResult(
+            criterion_type=criterion.type,
+            description=criterion.description,
+            score=0.0,
+            details=scrubbed[:500],
+            error=scrub_reference(first.parse_error, scrub_key),
+            transcript=_build_transcript(
+                criterion=criterion,
+                raw_verdict_text=first.raw_verdict_text,
+                user_msg=user_msg,
+                scrub_key=scrub_key,
+            ),
+            token_usage=token_usage,
+        )
+
+    scores = [verdict.score for verdict, _ in valid]
+    median_score = median(scores)
+    rep_verdict, rep_raw = min(valid, key=lambda pair: abs(pair[0].score - median_score))
+
+    degraded_notes = list(judge_ctx.degraded_notes)
+    failed_samples = criterion.samples - len(valid)
+    if criterion.samples > 1 and failed_samples:
+        note = (
+            f"{failed_samples}/{criterion.samples} judge samples produced no verdict; "
+            f"median over {len(valid)} valid samples"
+        )
+        if failure_labels:
+            note = f"{note} ({'; '.join(failure_labels)})"
+        degraded_notes.append(note)
+
+    details = format_details(median_score, rep_verdict.rationale, judge_ctx.missing_files, degraded_notes)
+    if criterion.samples > 1:
+        rendered_scores = ", ".join(f"{s:.3f}" for s in scores)
+        details = f"{details}\nsample_scores: [{rendered_scores}]"
+    return JudgeCriterionResult(
+        criterion_type=criterion.type,
+        description=criterion.description,
+        score=median_score,
+        details=scrub_reference(details, scrub_key),
+        findings=[scrub_reference(f, scrub_key) for f in rep_verdict.findings],
+        transcript=_build_transcript(
+            criterion=criterion,
+            raw_verdict_text=rep_raw,
+            user_msg=user_msg,
+            scrub_key=scrub_key,
+        ),
+        token_usage=token_usage,
+    )
+
+
+def _sum_usage(usages: Iterable[TokenUsage | None]) -> TokenUsage | None:
+    """Field-wise sum of the reported usages; ``None`` when no sample reported any."""
+    present = [u for u in usages if u is not None]
+    if not present:
+        return None
+    total = present[0]
+    for u in present[1:]:
+        total = total + u
+    return total
+
+
+def _build_transcript(
+    *,
+    criterion: LLMJudgeCriterion,
+    raw_verdict_text: str,
+    user_msg: str,
+    scrub_key: str | None,
+) -> JudgeTranscript | None:
+    """Build the persisted judge transcript when ``capture_transcript`` is on."""
+    if not criterion.capture_transcript:
+        return None
+    return build_judge_transcript(
+        raw_verdict=raw_verdict_text,
+        max_chars=criterion.max_transcript_chars,
+        judge_system_prompt=_SYSTEM_PROMPT,
+        judge_prompt=user_msg,
+        scrub_key=scrub_key,
+    )
 
 
 def _render_user_message(prompt: str, context: JudgeContext) -> str:

src/coder_eval/models/criteria.py

@@ -760,6 +760,28 @@ class LLMJudgeCriterion(BaseSuccessCriterion):
             "(score + rationale + a handful of findings) without runaway."
         ),
     )
+    samples: int = Field(
+        default=1,
+        ge=1,
+        le=9,
+        description=(
+            "Number of independent judge invocations. The default (1) keeps the single-call "
+            "behavior. With N > 1, the judge grades the same rendered prompt N times and the "
+            "criterion scores the MEDIAN of the sampled scores; rationale/findings/transcript "
+            "come from the sample whose score is closest to the median (earliest on ties), so "
+            "the audit trail is a real verdict rather than a synthetic blend. Per-sample scores "
+            "are rendered in ``details`` and token usage is summed across samples. With N >= 2, "
+            "a sample that produces no verdict degrades to the median of the valid samples "
+            "(with a note in ``details``); when every sample fails, single-sample failure "
+            "semantics apply unchanged. Odd N recommended — an even-N median averages the two "
+            "middle scores. Samples run concurrently, so wall-clock stays close to one judge "
+            "call; judge SPEND still multiplies by N and is not gated by ``run_limits.max_usd``. "
+            "Capped at 9 because the median's variance damping flattens well before that while "
+            "cost grows linearly — past 9 you pay for noise reduction you can no longer measure. "
+            "Unrelated to the dataset-level ``sample_per_stratum`` / CLI ``--sample`` (those "
+            "subsample dataset ROWS; this re-samples the judge's verdict on one row)."
+        ),
+    )
     max_file_chars: int = Field(
         default=20_000,
         gt=0,