diff --git a/vero/src/vero/evaluation/engine.py b/vero/src/vero/evaluation/engine.py
index 201f51c..b2f2196 100644
--- a/vero/src/vero/evaluation/engine.py
+++ b/vero/src/vero/evaluation/engine.py
@@ -193,6 +193,7 @@ async def evaluate_admin(
         split: str,
         commit: str,
         sample_ids: list[int] | None = None,
+        model: str | None = None,
     ) -> Experiment:
         """Admin/verifier evaluation: explicit ``task``, no budget, no allowlist.
 
@@ -200,7 +201,23 @@ async def evaluate_admin(
         this scores an arbitrary ``(task, dataset_id, split)`` — including held-out
         tasks/splits the agent never had access to. Used by the verifier to score
         the selected commit on its configured targets.
+
+        ``model`` overrides the executor model for this one eval (rides
+        ``task_params`` so the eval strategy can honor it; the Mode-B
+        HarborRunner does, the Mode-A vero-task path ignores it). Used for
+        transfer targets: scoring the champion under a model it was NOT
+        optimized on.
         """
+        params = self.run_constraints
+        if model is not None:
+            params = params.model_copy(
+                update={
+                    "task_params": {
+                        **(params.task_params or {}),
+                        "harbor_model_override": model,
+                    }
+                }
+            )
         return await self.evaluator.evaluate(
             commit=commit,
             dataset_id=dataset_id,
@@ -208,7 +225,7 @@ async def evaluate_admin(
             task=task,
             sample_ids=sample_ids,
             db=self.db,
-            evaluation_parameters=self.run_constraints,
+            evaluation_parameters=params,
         )
 
     def status(self) -> dict[tuple[str, str], SplitBudget]:
diff --git a/vero/src/vero/harbor/build/compiler.py b/vero/src/vero/harbor/build/compiler.py
index 2021399..dc42431 100644
--- a/vero/src/vero/harbor/build/compiler.py
+++ b/vero/src/vero/harbor/build/compiler.py
@@ -236,6 +236,7 @@ def _serve_config(
                 "split": t.split,
                 "reward_key": t.reward_key,
                 "sample_ids": t.sample_ids,
+                "model": t.model,
             }
             for t in config.targets
         ],
@@ -243,6 +244,7 @@ def _serve_config(
         "submit_enabled": config.submit_enabled,
         "score_baseline": config.score_baseline,
         "k_anonymity_floor": config.k_anonymity_floor,
+        "instruct_exhaust_budget": config.instruct_exhaust_budget,
         "agent_volume": AGENT_VOLUME,
         "admin_volume": ADMIN_VOLUME,
         "admin_token_path": TOKEN_PATH,
@@ -405,6 +407,7 @@ def compile_task(
         and {"sample_ids", "num_samples"}
         <= {f.name for f in dataclasses.fields(EvalRequest)}
         and any(s.access == "viewable" for s in config.splits),
+        exhaust_budget=config.instruct_exhaust_budget,
     )
     _render(jenv, "task.toml.j2", out / "task.toml", **ctx)
     _render(jenv, "instruction.md.j2", out / "instruction.md", **ctx)
diff --git a/vero/src/vero/harbor/build/config.py b/vero/src/vero/harbor/build/config.py
index 8cf6c9a..6893847 100644
--- a/vero/src/vero/harbor/build/config.py
+++ b/vero/src/vero/harbor/build/config.py
@@ -36,6 +36,12 @@ class TargetSpec(BaseModel):
     split: str
     reward_key: str = "reward"
     sample_ids: list[int] | None = None
+    # Executor-model override for this target (transfer probe; Mode B only):
+    # score the selected commit AND the baseline under a model it was not
+    # optimized on, so model-specific couplings (measured live: hardcoded
+    # temperature=0 crashing 72/72 on an executor that rejects it) surface at
+    # finalize instead of one substrate away.
+    model: str | None = None
 
 
 class _BuildConfigBase(BaseModel):
@@ -72,6 +78,13 @@ class _BuildConfigBase(BaseModel):
     # mean_score, so singleton subsets would hand back per-sample labels.
     k_anonymity_floor: int = 5
 
+    # Instruction lever: render the "unspent budget is wasted" persistence
+    # bullet that tells the optimizer to keep spending (re-measure the champion,
+    # try one more variant) instead of stopping early. On by default (current
+    # behavior); off makes stopping-early a choice the agent arrives at itself,
+    # which is the ablation arm for measuring what the exhortation contributes.
+    instruct_exhaust_budget: bool = True
+
     # write-access: paths in the target repo the optimizer may NOT edit
     # (the scorer, by default). Applied as unix perms in main before the agent runs.
     read_only_paths: list[str] = Field(default_factory=list)
diff --git a/vero/src/vero/harbor/build/templates/instruction.md.j2 b/vero/src/vero/harbor/build/templates/instruction.md.j2
index df25032..6447398 100644
--- a/vero/src/vero/harbor/build/templates/instruction.md.j2
+++ b/vero/src/vero/harbor/build/templates/instruction.md.j2
@@ -51,9 +51,11 @@ in rough proportion to its size, so it is the cheap way to triage ideas:
   from a regression. Repeat baseline evals are metered. `vero harbor status`
   shows the baseline sha and whether the free eval is still available.
 {% endif %}
-- Scores are noisy. Unspent budget is wasted: if you finish with evals left, spend
+- Scores are noisy.
+{% if exhaust_budget %}- Unspent budget is wasted: if you finish with evals left, spend
   them re-measuring your best candidate to confirm its score, or trying one more
   variant, rather than stopping early.
+{% endif %}
 - The test split is hidden: you cannot evaluate it, and its labels never reach this
   container. Trying to read it will fail.
 - The scorer is locked. Only the eval sidecar scores.
diff --git a/vero/src/vero/harbor/config.py b/vero/src/vero/harbor/config.py
index dea857f..f6b0b4b 100644
--- a/vero/src/vero/harbor/config.py
+++ b/vero/src/vero/harbor/config.py
@@ -40,6 +40,32 @@ class HarborConfig:
     # without running anything). None keeps the current behavior: the
     # candidate env supplies harbor, and is trusted to.
     harbor_requirement: str | None = None
+    # Bounded within-eval retry for infra-destroyed samples. A sample whose
+    # EVERY attempt died of a transient infrastructure cause (connection,
+    # timeout, rate limit, 5xx) was never measured at all: re-run it after a
+    # backoff instead of booking the outage as a permanent error. Measured
+    # live: a 65-second host DNS blip killed 44 of 72 attempts of one eval
+    # with ConnectionError, and nothing in the record distinguished the blip
+    # from a bad candidate.
+    #
+    # OFF BY DEFAULT, and it must stay off when the candidate is an
+    # adversarial optimizer. The qualifying predicate is built from exception
+    # types raised inside candidate code, and agents are stochastic: a
+    # candidate that raises an allowlisted exception whenever an attempt is
+    # going badly loses nothing on partially-good samples (its fakes
+    # zero-fill like honest failures) but converts every all-bad sample from
+    # a booked 0.0 into a fresh re-roll. That is one-sided selection over
+    # attempt sets, exactly what the zero-fill invariant exists to prevent.
+    # Enable only for trusted-candidate evaluations (frozen agents,
+    # operator-run matrices), where re-measuring an outage is pure signal
+    # recovery. Candidate crashes and exhausted key budgets never retry
+    # regardless (a crash is a result; a spent key cannot recover by
+    # waiting), and recovered samples carry an ``infra_retry`` audit marker.
+    infra_retry_rounds: int = 0
+    # Backoff before retry round N is N times this many seconds. Transient
+    # infra needs time, not immediacy: instant retries burned 6 of 8 run
+    # attempts inside one live DNS blip.
+    infra_retry_delay_s: float = 30.0
     extra_args: list[str] = field(default_factory=list)  # passthrough harbor run flags
 
     def __post_init__(self) -> None:
@@ -50,6 +76,18 @@ def __post_init__(self) -> None:
                 f"aggregate_attempts must be 'best' or 'mean', got "
                 f"{self.aggregate_attempts!r}"
             )
+        if self.infra_retry_rounds < 0:
+            raise ValueError(
+                f"infra_retry_rounds must be >= 0, got {self.infra_retry_rounds}"
+            )
+        # A zero (or negative) delay silently nullifies the backoff, and an
+        # instant retry re-enters the same outage: 6 of 8 run attempts once
+        # burned inside a single live DNS blip for exactly this reason.
+        if self.infra_retry_rounds > 0 and self.infra_retry_delay_s <= 0:
+            raise ValueError(
+                f"infra_retry_delay_s must be > 0 when infra retries are "
+                f"enabled, got {self.infra_retry_delay_s}"
+            )
 
     @property
     def is_registry(self) -> bool:
diff --git a/vero/src/vero/harbor/protocol.py b/vero/src/vero/harbor/protocol.py
index c43f101..156c8e3 100644
--- a/vero/src/vero/harbor/protocol.py
+++ b/vero/src/vero/harbor/protocol.py
@@ -104,7 +104,10 @@ def build_status(
     split_accesses: list[SplitAccess],
     base_commit: str | None = None,
     free_baseline_available: bool = False,
-    k_anonymity_floor: int = 1,
+    # Default matches EvaluationSidecar's enforcement default: a caller that
+    # forgets to pass the floor must not advertise a laxer one than the
+    # sidecar enforces (agents would send sub-floor requests that 400).
+    k_anonymity_floor: int = 5,
 ) -> StatusSummary:
     """Build the agent-facing status from the budget ledger + split tiers.
 
diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py
index f262261..c12d927 100644
--- a/vero/src/vero/harbor/runner.py
+++ b/vero/src/vero/harbor/runner.py
@@ -10,6 +10,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
 from pathlib import Path
@@ -31,6 +32,68 @@
 
 logger = logging.getLogger(__name__)
 
+# Dead-attempt classification. An attempt that died of one of these exception
+# types never got a fair shot at the task: the model endpoint, the network, or
+# the key quota failed before the candidate could be measured. Classification
+# is diagnostic and retry-gating ONLY: every dead attempt still scores 0.0
+# regardless of class, because excusing infra-labeled deaths from the score
+# would hand the candidate a lever (raise ConnectionError on hard tasks and
+# have those attempts dropped). Conservative allowlist: anything unlisted
+# counts against the candidate.
+_INFRA_EXCEPTION_TYPES = frozenset({
+    "APIConnectionError",
+    "APITimeoutError",
+    "ConnectTimeout",
+    "ConnectionError",
+    "InternalServerError",
+    "RateLimitError",
+    "ReadTimeout",
+    "ServiceUnavailableError",
+    "Timeout",
+    "TimeoutError",
+})
+_INFRA_SUFFIX = "[infra]"
+# litellm surfaces a spent key budget as a BadRequestError; only the message
+# distinguishes it from a candidate-caused bad request. Infra, but PERSISTENT:
+# waiting cannot refill a key, so it alarms instead of retrying. (Measured
+# live: a key crossed its spend cap mid-matrix and two cells of BadRequestError
+# zeros were nearly booked as a portability finding.)
+_KEY_BUDGET_MARKER = "budget has been exceeded"
+_KEY_BUDGET_SUFFIX = "[infra:llm-key-budget]"
+
+
+def _dead_attempt_label(exception_info: dict | None) -> str:
+    """Cause label for one dead attempt; infra causes carry a class suffix so
+    every downstream record (metrics, error strings, per-sample output) shows
+    at a glance whether the deaths measure the candidate or the plumbing. An
+    attempt can die with no recorded exception at all (the verifier simply
+    produced no rewards); keep it countable."""
+    info = exception_info or {}
+    exc = info.get("exception_type")
+    if not exc:
+        return "no_rewards_recorded"
+    # The class suffixes below are load-bearing (retry gating, the
+    # n_dead_infra metric, the key-budget alarm) and round-trip through
+    # persisted output, while the exception type name is candidate-authored:
+    # a class literally named "XError[infra]" must not walk in pre-suffixed.
+    # Neutralize brackets before classifying; genuine types contain none.
+    exc = exc.replace("[", "(").replace("]", ")")
+    if exc == "BadRequestError" and _KEY_BUDGET_MARKER in (
+        info.get("exception_message") or ""
+    ).lower():
+        return f"{exc}{_KEY_BUDGET_SUFFIX}"
+    if exc in _INFRA_EXCEPTION_TYPES:
+        return f"{exc}{_INFRA_SUFFIX}"
+    return exc
+
+
+def _is_infra(label: str) -> bool:
+    return label.endswith((_INFRA_SUFFIX, _KEY_BUDGET_SUFFIX))
+
+
+def _is_transient_infra(label: str) -> bool:
+    return label.endswith(_INFRA_SUFFIX)
+
 
 class HarborRunner:
     """Mode-B EvalStrategy: nested `harbor run` + collate -> SampleResults."""
@@ -76,6 +139,151 @@ async def produce_sample_results(
                 str(workspace.project_path), params, [t for _, t in pending], jobs_dir
             )
         self._collate(jobs_dir, pairs, params, ran=[t for _, t in pending])
+        await self._retry_transient_infra(workspace, params, pairs, jobs_dir)
+
+    async def _retry_transient_infra(
+        self,
+        workspace: Workspace,
+        params: EvaluationParameters,
+        pairs: list[tuple[int, str]],
+        jobs_dir: Path,
+    ) -> None:
+        """Bounded re-run of samples that were outaged, not measured.
+
+        Scope is deliberately narrow: only samples whose EVERY attempt died of
+        a transient infra cause qualify. A partially scored sample is a noisy
+        measurement whose infra-dead attempts stay zero-filled; a candidate
+        crash is a result; an exhausted key budget cannot recover by waiting.
+        Measured live: a 65-second host DNS blip killed 44 of 72 attempts of
+        one eval with ConnectionError, and with no pause between attempts the
+        whole burst fit inside the blip, booking a near-zero that nothing in
+        the record distinguished from a bad candidate.
+
+        OFF BY DEFAULT (see HarborConfig.infra_retry_rounds), because against
+        an adversarial candidate this is a re-roll lever: the qualifying
+        predicate is built from exception types raised inside candidate code,
+        so a stochastic candidate that raises an allowlisted exception
+        whenever an attempt is going badly converts its all-bad samples from
+        booked zeros into fresh re-rolls. Enable only for trusted-candidate
+        evaluations. Two record-integrity rules hold either way: each round
+        runs in a fresh SIBLING of the jobs dir and collates from there alone
+        (never inside it: resume-path collations rglob the whole jobs dir, and
+        nested round dirs would pool this round's dead attempts into a later
+        resumed mean), and a recovered sample's booked output carries an
+        ``infra_retry`` audit marker naming the discarded dead attempts, so
+        the re-measurement is never invisible in the durable record.
+        """
+        # Every discarded round per sample, in round order: a sample that
+        # rides several retry rounds before recovering must surface ALL the
+        # rounds it burned, not just the last one before recovery.
+        discard_history: dict[int, list[dict[str, int]]] = {}
+        for round_no in range(1, self.config.infra_retry_rounds + 1):
+            retry: list[tuple[int, str]] = []
+            for sid, t in pairs:
+                dead = self._transient_infra_dead(params, sid)
+                if dead:
+                    retry.append((sid, t))
+                    discard_history.setdefault(sid, []).append(dead)
+            if not retry:
+                return
+            delay = self.config.infra_retry_delay_s * round_no
+            logger.warning(
+                f"{len(retry)} sample(s) lost every attempt to transient infra "
+                f"causes; retry round {round_no}/{self.config.infra_retry_rounds} "
+                f"in {delay:.0f}s."
+            )
+            await asyncio.sleep(delay)
+            # Sibling of the jobs dir, never a subdir (see docstring), with a
+            # globally fresh ordinal: a dir left over from an earlier eval of
+            # the same result_dir must never leak its stale trials into this
+            # round's collation.
+            ordinal = len(list(jobs_dir.parent.glob("jobs-infra-retry-*"))) + 1
+            round_dir = jobs_dir.parent / f"jobs-infra-retry-{ordinal}"
+            await self._run_harbor(
+                str(workspace.project_path), params, [t for _, t in retry], round_dir
+            )
+            # Collate only what the round actually produced: overwriting a
+            # cause-rich infra error with "no Harbor trial result" (because the
+            # retry round itself died early) would degrade the record.
+            produced = set(self._load_trials(round_dir))
+            got = [(sid, t) for sid, t in retry if t in produced]
+            if not got:
+                logger.warning(
+                    f"infra retry round {round_no} produced no trials; keeping "
+                    f"the recorded errors."
+                )
+                continue
+            self._collate(round_dir, got, params, ran=[t for _, t in got])
+            self._mark_recovered(params, got, discard_history, round_no)
+
+    def _mark_recovered(
+        self,
+        params: EvaluationParameters,
+        got: list[tuple[int, str]],
+        discard_history: dict[int, list[dict[str, int]]],
+        round_no: int,
+    ) -> None:
+        """Stamp the audit marker onto samples the retry round recovered.
+
+        A successful retry rebooks the sample from the fresh round alone, so
+        without this the durable record would show a clean measurement with no
+        trace that full rounds of dead attempts were discarded, and a
+        discarded round is exactly the kind of fact an auditor of the scores
+        must be able to see. ``discarded_rounds`` lists every burned round in
+        order, not just the one immediately before recovery."""
+        for sid, _ in got:
+            result = self._existing(params, sid)
+            if result is None or result.is_error():
+                continue  # still failed; the error itself is the audit trail
+            output = result.output if isinstance(result.output, dict) else {}
+            output["infra_retry"] = {
+                "recovered_round": round_no,
+                "discarded_rounds": discard_history.get(sid, []),
+            }
+            result.output = output
+            save_sample_result(
+                get_vero_home_dir() / "sessions",
+                params.session_id,
+                params.result_id,
+                sample_id=sid,
+                result=result,
+            )
+
+    def _transient_infra_dead(
+        self, params: EvaluationParameters, sample_id: int
+    ) -> dict[str, int] | None:
+        """The persisted dead-cause dict when the sample was never measured
+        (it errored AND every dead attempt carries a transient-infra label),
+        else None. Samples with no structured cause record are not retryable:
+        the conservative default is "measured", the same fail-closed direction
+        as zero-filling."""
+        existing = self._existing(params, sample_id)
+        if existing is None or not existing.is_error():
+            return None
+        output = existing.output if isinstance(existing.output, dict) else {}
+        dead = output.get("dead_exception_types") or {}
+        if dead and all(_is_transient_infra(k) for k in dead):
+            return dead
+        return None
+
+    @staticmethod
+    def _alarm_key_budget(task_name: str, dead: dict[str, int]) -> None:
+        """The one infra cause that deserves its own alarm: an exhausted key
+        budget fails every subsequent LLM call identically, so from the first
+        occurrence onward the run is measuring the outage, not the agent, and
+        neither retrying nor waiting fixes it. ERROR level: this is the line
+        an operator greps for after a run full of inexplicable zeros."""
+        n = sum(v for k, v in dead.items() if k.endswith(_KEY_BUDGET_SUFFIX))
+        if n:
+            logger.error(
+                f"Task '{task_name}': {n} attempt(s) report the LLM key's "
+                f"spend budget as exhausted. If real, every call on this key "
+                f"fails the same way until it is refilled or swapped, and "
+                f"scores recorded meanwhile measure the outage, not the "
+                f"candidate. The signature is read from candidate-process "
+                f"exceptions, so corroborate against the key's own spend "
+                f"records before invalidating results."
+            )
 
     # ------------------------------------------------------------------
     # Task selection (host-side; just task names)
@@ -129,8 +337,12 @@ def _build_command(
             "--n-attempts", str(c.n_attempts),
             "--max-retries", str(c.max_retries),
         ]
-        if c.model:
-            cmd += ["-m", c.model]
+        # Per-eval executor override (transfer targets): the verifier scores
+        # the champion under a model it was not optimized on. Rides
+        # task_params so it needs no runner state.
+        model = (params.task_params or {}).get("harbor_model_override") or c.model
+        if model:
+            cmd += ["-m", str(model)]
         for task_name in task_names:
             cmd += ["-i", task_name]
         cmd += ["--jobs-dir", str(jobs_dir), *c.extra_args]
@@ -205,6 +417,9 @@ def _collate(
             self.config.aggregate_attempts == "mean"
             or self.feedback_transcripts
             or self.expose_attempt_detail
+            # The infra retry decides from per-attempt death causes, which are
+            # only recorded when attempts are loaded at collation.
+            or self.config.infra_retry_rounds > 0
         )
         groups = self._trial_groups(jobs_dir) if need_attempts else {}
         for sample_id, task_name in pairs:
@@ -370,10 +585,7 @@ def _out(output: dict) -> dict:
                 else:
                     measured.append(0.0)
                     n_dead += 1
-                    exc = (t.get("exception_info") or {}).get("exception_type")
-                    # An attempt can die without a recorded exception (the
-                    # verifier simply produced no rewards); keep it countable.
-                    key = exc or "no_rewards_recorded"
+                    key = _dead_attempt_label(t.get("exception_info"))
                     dead_types[key] = dead_types.get(key, 0) + 1
             if n_scored:
                 if len(measured) < self.config.n_attempts or n_dead:
@@ -394,6 +606,7 @@ def _out(output: dict) -> dict:
                 if dead_types:
                     # dict, not metrics: metrics are float-valued by contract.
                     mean_output["dead_exception_types"] = dead_types
+                self._alarm_key_budget(task_name, dead_types)
                 return SampleResult(
                     score=mean,
                     feedback=self._failure_feedback(mean, attempts),
@@ -402,6 +615,15 @@ def _out(output: dict) -> dict:
                         "n_attempts": float(len(attempts)),
                         "n_scored": float(n_scored),
                         "n_dead": float(n_dead),
+                        # Zeros with an infra-labeled cause: still counted in
+                        # the mean (see the classification note at module top)
+                        # but split out for the analyst deciding whether a
+                        # cell's zeros measured the plumbing. Labels derive
+                        # from candidate-process exceptions: corroborating
+                        # evidence for invalidating a cell, not proof.
+                        "n_dead_infra": float(
+                            sum(v for k, v in dead_types.items() if _is_infra(k))
+                        ),
                         "n_clean": float(n_clean),
                     },
                     output=_out(mean_output),
@@ -409,11 +631,45 @@ def _out(output: dict) -> dict:
                 )
         rewards = (trial.get("verifier_result") or {}).get("rewards") or {}
         if not rewards:
+            # Name the cause in the error string itself: it is the one field
+            # that flows everywhere (DB, per-sample files, the verifier's
+            # target_errors), and "no verifier rewards" alone cannot separate
+            # a champion that crashes deterministically on this executor from
+            # an infra outage: a distinction that decides whether the cell is
+            # a measurement or a re-run.
+            cause = ""
+            dead: dict[str, int] = {}
+            if attempts:
+                for t in attempts:
+                    if (t.get("verifier_result") or {}).get("rewards"):
+                        continue
+                    key = _dead_attempt_label(t.get("exception_info"))
+                    dead[key] = dead.get(key, 0) + 1
+                if dead:
+                    causes = ", ".join(
+                        f"{k} x{v}" for k, v in sorted(dead.items(), key=lambda i: -i[1])
+                    )
+                    cause = f" (attempts died: {causes})"
+            self._alarm_key_budget(task_name, dead)
+            no_rewards_output = {
+                "task_name": task_name,
+                "trial_name": trial.get("trial_name"),
+            }
+            if dead:
+                # Structured twin of the error string above: the within-eval
+                # infra retry reads this to decide whether the sample was
+                # measured or merely outaged (string parsing would be the
+                # fragile alternative).
+                no_rewards_output["dead_exception_types"] = dead
             return SampleResult(
-                error=f"No verifier rewards for task '{task_name}'.",
-                output=_out(
-                    {"task_name": task_name, "trial_name": trial.get("trial_name")}
-                ),
+                error=f"No verifier rewards for task '{task_name}'.{cause}",
+                # The agent died before scoring. A candidate edit that CRASHES
+                # the agent lands here, and "no verifier rewards" alone gives
+                # the optimizer no way to see its own crash; the transcript
+                # does. Passed as score 0.0: an unscored attempt counts as a
+                # failure everywhere else too.
+                feedback=self._failure_feedback(0.0, attempts),
+                output=_out(no_rewards_output),
                 **common,
             )
         score = self._extract_reward(rewards)
@@ -553,6 +809,12 @@ def _read_transcript_tail(self, trial_dir: Path) -> str | None:
                 data = path.read_bytes()
             except OSError:
                 continue
+            # An empty transcript carries nothing: keep looking (an empty pane
+            # must fall through to the trajectory), and if every candidate is
+            # empty return None so the caller tries the next failed attempt
+            # instead of emitting "" as feedback.
+            if not data:
+                continue
             # errors="replace": a multibyte char straddling the cap boundary is
             # rendered as U+FFFD rather than crashing the collation.
             return data[-self.feedback_max_bytes :].decode("utf-8", errors="replace")
diff --git a/vero/src/vero/harbor/serve.py b/vero/src/vero/harbor/serve.py
index 00eb327..bd42329 100644
--- a/vero/src/vero/harbor/serve.py
+++ b/vero/src/vero/harbor/serve.py
@@ -10,6 +10,7 @@
 
 import json
 import logging
+import shutil
 from pathlib import Path
 from typing import Annotated, Literal
 
@@ -42,6 +43,8 @@ class _TargetCfg(BaseModel):
     split: str
     reward_key: str = "reward"
     sample_ids: list[int] | None = None
+    # Executor-model override for this target (transfer probe; Mode B only).
+    model: str | None = None
 
 
 class _ServeConfigBase(BaseModel):
@@ -82,6 +85,10 @@ class _ServeConfigBase(BaseModel):
     # EvaluationSidecar.k_anonymity_floor for the leak this closes.
     k_anonymity_floor: int = 5
 
+    # Consumed at COMPILE time (the instruction's exhaust-budget bullet);
+    # recorded here so serve.json mirrors build.yaml, like instruct_multifidelity.
+    instruct_exhaust_budget: bool = True
+
     # volumes / token
     agent_volume: str
     admin_volume: str
@@ -155,8 +162,12 @@ def _load_or_build_ledger(
     a sidecar restart would reset all spent budget to full, letting the agent
     regain its full evaluation budget by triggering a restart. On startup we
     reconstruct each SplitBudget and restore its persisted ``remaining_*`` values.
-    Falls back to the configured budgets if the file is missing or unreadable
-    (fail-safe to the configured budget, never to unlimited).
+
+    A MISSING file is a fresh boot: configured budgets. A file that exists but
+    cannot be parsed fails CLOSED: metered budgets restore with zero remaining.
+    The old fallback (configured budgets) refunded the agent everything already
+    spent, so any crash that corrupted the flush minted budget; spend that
+    cannot be read must be treated as fully spent, never as never-happened.
     """
     if persist_path.exists():
         try:
@@ -181,11 +192,28 @@ def _load_or_build_ledger(
             )
             return BudgetLedger(budgets, persist_path=persist_path)
         except (json.JSONDecodeError, KeyError, OSError) as e:
-            logger.warning(
-                "Could not reload persisted ledger %s (%s); using configured budgets.",
+            logger.error(
+                "Persisted ledger %s exists but is unreadable (%s); failing "
+                "CLOSED: metered agent budgets restore as exhausted. Admin and "
+                "finalize are unaffected. The unreadable file is preserved at "
+                "%s; delete ledger.json deliberately to boot fresh.",
                 persist_path,
                 e,
+                persist_path.with_suffix(".corrupt"),
             )
+            try:  # keep the evidence: the next flush overwrites persist_path
+                shutil.copyfile(persist_path, persist_path.with_suffix(".corrupt"))
+            except OSError:
+                pass
+            budgets = []
+            for cfg in budget_cfgs:
+                b = SplitBudget(**cfg)
+                if b.total_sample_budget is not None:
+                    b.remaining_sample_budget = 0
+                if b.total_run_budget is not None:
+                    b.remaining_run_budget = 0
+                budgets.append(b)
+            return BudgetLedger(budgets, persist_path=persist_path)
     return BudgetLedger(
         [SplitBudget(**b) for b in budget_cfgs], persist_path=persist_path
     )
diff --git a/vero/src/vero/harbor/server.py b/vero/src/vero/harbor/server.py
index d29b7a4..4dff350 100644
--- a/vero/src/vero/harbor/server.py
+++ b/vero/src/vero/harbor/server.py
@@ -11,6 +11,7 @@
 
 import json
 import logging
+import re
 import shutil
 from dataclasses import replace
 from pathlib import Path
@@ -124,14 +125,20 @@ async def evaluate(self, req: EvalRequest, *, admin: bool = False) -> EvalSummar
             and sha == self.base_commit
             and not self._free_baseline_used
         )
-        exp = await self.engine.evaluate(
-            replace(req, commit=sha), admin=admin, free=free_baseline
-        )
-        # Consume the freebie only after the eval succeeded: an eval that
-        # raised (invalid split, infra failure) has given the agent nothing,
-        # so it must not burn the one free reference measurement.
+        # Claim the freebie BEFORE the await (asyncio is atomic between await
+        # points, so a concurrent second baseline eval sees the claim and pays)
+        # and refund it if the eval raises: a failed eval gave the agent
+        # nothing, so it must not burn the one free reference measurement.
         if free_baseline:
             self._free_baseline_used = True
+        try:
+            exp = await self.engine.evaluate(
+                replace(req, commit=sha), admin=admin, free=free_baseline
+            )
+        except BaseException:
+            if free_baseline:
+                self._free_baseline_used = False
+            raise
         # Route with the agent's real tier even when the eval was unmetered.
         result_path = self._route_results(exp, admin=admin)
         budget_remaining = None
@@ -282,31 +289,45 @@ def _route_results(self, experiment: Experiment, *, admin: bool) -> str | None:
         # agent's earlier evidence for the same commit; repeat measurements are
         # exactly the ones worth comparing. result_path in the response names
         # the dir for THIS eval.
+        results_root = self.agent_volume / "results"
+        if self._eval_seq == 0 and results_root.exists():
+            # Volume reuse (sidecar restart): resume the ordinal past every
+            # surviving dir, or the first N evals of the new session would
+            # silently wipe __e1..__eN — the exact erasure this scheme exists
+            # to prevent.
+            self._eval_seq = max(
+                (
+                    int(m.group(1))
+                    for d in results_root.iterdir()
+                    if (m := re.search(r"__e(\d+)$", d.name))
+                ),
+                default=0,
+            )
         self._eval_seq += 1
-        dest = (
-            self.agent_volume
-            / "results"
-            / f"{split}__{commit[:12]}__e{self._eval_seq}"
-        )
-        if dest.exists():  # ordinal collision only on volume reuse; never merge
+        dest = results_root / f"{split}__{commit[:12]}__e{self._eval_seq}"
+        if dest.exists():  # unreachable after the resume scan; never merge
             shutil.rmtree(dest)
         dest.mkdir(parents=True, exist_ok=True)
 
         # Aggregate summary is label-safe for both visible and partial tiers.
-        # n_scored / n_errored / score_se qualify the mean: a mean over 3
+        # n_scored / n_errored / mean_score_se qualify the mean: a mean over 3
         # scored samples of 18, or one dominated by errored zero-fills, is a
         # different measurement than a clean full-split mean, and the agent
         # (and any auditor) should see that without per-sample access.
+        # mean_score_se is named to bind it to mean_score: both are computed
+        # over the zero-filled n_samples population (score() fills errored
+        # samples with 0.0), NOT over the n_scored subset — an SE of the
+        # 3-of-18-scored mean would be a different (and larger) number.
         sample_results = experiment.result.sample_results
         filled = [
             r.score if r.score is not None else 0.0
             for r in sample_results.values()
         ]
-        score_se = None
+        mean_score_se = None
         if len(filled) > 1:
             m = sum(filled) / len(filled)
             var = sum((x - m) ** 2 for x in filled) / (len(filled) - 1)
-            score_se = (var / len(filled)) ** 0.5
+            mean_score_se = (var / len(filled)) ** 0.5
         (dest / "summary.json").write_text(
             json.dumps(
                 {
@@ -320,7 +341,7 @@ def _route_results(self, experiment: Experiment, *, admin: bool) -> str | None:
                         1 for r in sample_results.values() if r.is_error()
                     ),
                     "mean_score": experiment.result.score(),
-                    "score_se": score_se,
+                    "mean_score_se": mean_score_se,
                     "status": experiment.result.status.value,
                 },
                 indent=2,
diff --git a/vero/src/vero/harbor/verifier.py b/vero/src/vero/harbor/verifier.py
index 5b41d9f..63f588c 100644
--- a/vero/src/vero/harbor/verifier.py
+++ b/vero/src/vero/harbor/verifier.py
@@ -15,6 +15,7 @@
 import asyncio
 import json
 import logging
+import re
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Literal
@@ -38,6 +39,15 @@ class VerificationTarget:
     split: str
     reward_key: str
     sample_ids: list[int] | None = None  # None = full split
+    # Executor-model override for this target (Mode B): score the selected
+    # commit under a DIFFERENT model than the one it was optimized on. This is
+    # the transfer probe: home-model evals cannot see model-specific couplings
+    # the optimizer bakes in (measured live: three champions independently
+    # hardcoded temperature=0 and scored 0/72 on an executor that rejects it,
+    # while looking healthy on every home-model eval). None = the task's
+    # configured model. The baseline is scored under the same override, so the
+    # comparison stays like-for-like.
+    model: str | None = None
 
 
 class Verifier:
@@ -137,23 +147,29 @@ async def _finalize(self) -> dict:
         rewards: dict[str, float] = {}
         target_errors: dict[str, str] = {}
         for target in self.targets:
-            score = await self._admin_eval_score(
+            score, cause = await self._admin_eval_score(
                 task=target.task,
                 dataset_id=target.dataset_id,
                 split=target.split,
                 commit=sha,
                 sample_ids=target.sample_ids,
+                model=target.model,
                 what=f"target '{target.reward_key}'",
             )
             if score is None:
                 # Persistent failure: floor the target so reward.json still
-                # ships, and record the failure in the wrapper (echoed to the
-                # trial's durable stdout) so a floored-by-outage reward can
-                # never masquerade as a measured 0.0.
+                # ships, and record the failure WITH ITS CAUSE in the wrapper
+                # (echoed to the trial's durable stdout). A floored-by-outage
+                # reward must never masquerade as a measured 0.0, and the cause
+                # separates the two floored cases that demand opposite actions:
+                # a champion that deterministically crashes on this target's
+                # executor (a real, reportable portability failure) vs an infra
+                # outage (invalidate and re-run).
                 rewards[target.reward_key] = float(default_minimum_score)
                 target_errors[target.reward_key] = (
                     f"eval failed after {self._baseline_score_attempts} attempt(s); "
                     f"reward floored, not measured"
+                    + (f"; cause: {cause}" if cause else "")
                 )
             else:
                 rewards[target.reward_key] = score
@@ -171,17 +187,21 @@ async def _admin_eval_score(
         split: str,
         commit: str,
         sample_ids: list[int] | None = None,
+        model: str | None = None,
         what: str,
-    ) -> float | None:
+    ) -> tuple[float | None, str | None]:
         """One reward-critical admin eval with bounded retry.
 
-        Returns the eval's score with errored samples counted 0.0 (min-fill:
-        an errored sample is a failed measurement of the candidate, and
-        excluding it would reward candidates whose failures error out rather
-        than score). An eval in which NO sample scored is indistinguishable
-        from an infrastructure outage and must never quietly become 0.0, so it
-        is retried like an exception; ``None`` after the last attempt means
-        "could not measure", and the caller decides the fail-safe.
+        Returns ``(score, failure_cause)``. The score counts errored samples
+        as 0.0 (min-fill: an errored sample is a failed measurement of the
+        candidate, and excluding it would reward candidates whose failures
+        error out rather than score). An eval in which NO sample scored is
+        indistinguishable from an infrastructure outage and must never quietly
+        become 0.0, so it is retried like an exception; ``(None, cause)``
+        after the last attempt means "could not measure", and the caller
+        decides the fail-safe. ``cause`` summarizes the dominant per-sample
+        errors so the durable record can distinguish a deterministically
+        crashing candidate from an outage.
         """
         last_error: Exception | str | None = None
         for attempt in range(1, self._baseline_score_attempts + 1):
@@ -192,9 +212,13 @@ async def _admin_eval_score(
                     split=split,
                     commit=commit,
                     sample_ids=sample_ids,
+                    model=model,
                 )
                 if exp.result.score(fill_score=None) is None:
-                    last_error = "eval scored no samples (all errored or empty)"
+                    last_error = (
+                        "eval scored no samples (all errored or empty); "
+                        + self._dominant_sample_errors(exp)
+                    )
                     logger.warning(
                         "%s attempt %d/%d: %s",
                         what, attempt, self._baseline_score_attempts, last_error,
@@ -207,7 +231,7 @@ async def _admin_eval_score(
                     # other unmeasurable outcome, never bypass the loop.
                     last_error = "eval returned no aggregate score"
                     continue
-                return float(score)
+                return float(score), None
             except Exception as exc:  # noqa: BLE001 - retried, then surfaced as None
                 last_error = exc
                 logger.warning(
@@ -218,7 +242,34 @@ async def _admin_eval_score(
             "%s failed after %d attempt(s): %s",
             what, self._baseline_score_attempts, last_error,
         )
-        return None
+        return None, str(last_error) if last_error is not None else None
+
+    @staticmethod
+    def _dominant_sample_errors(exp) -> str:
+        """Frequency summary of the per-sample error strings of an experiment
+        (e.g. "12x: No verifier rewards for task ... (attempts died:
+        UnsupportedParamsError x6)"). One identical cause across every sample
+        is the signature of a deterministic candidate crash; a mixed bag points
+        at infra. Top few only: the value is the shape, not the full list.
+        Diagnostics must never fail finalize, so any surprise shape degrades
+        to a fixed string instead of raising."""
+        try:
+            counts: dict[str, int] = {}
+            for r in exp.result.sample_results.values():
+                if r.error:
+                    # Group on the CAUSE, not the sample: runner errors embed
+                    # the task name ("No verifier rewards for task 'x/y'..."),
+                    # so keying on the raw string would leave every sample in
+                    # its own 1x bucket and a deterministic crash across a
+                    # multi-task slice would read as a mixed bag.
+                    key = re.sub(r"for task '[^']*'", "for task '…'", r.error)
+                    counts[key] = counts.get(key, 0) + 1
+            if not counts:
+                return "no per-sample errors recorded"
+            top = sorted(counts.items(), key=lambda i: -i[1])[:3]
+            return "; ".join(f"{n}x: {err}" for err, n in top)
+        except Exception:  # noqa: BLE001 - diagnostics only
+            return "no per-sample errors recorded"
 
     async def _maybe_score_baseline(self, rewards: dict[str, float]) -> dict:
         """Admin-score the unmodified baseline on every target and report it.
@@ -254,12 +305,15 @@ async def _maybe_score_baseline(self, rewards: dict[str, float]) -> dict:
             try:
                 baselines: dict[str, float] = {}
                 for target in self.targets:
+                    # Same executor override as the candidate's target eval, or
+                    # the baseline comparison is not like-for-like.
                     exp = await self.engine.evaluate_admin(
                         task=target.task,
                         dataset_id=target.dataset_id,
                         split=target.split,
                         commit=self.base_commit,
                         sample_ids=target.sample_ids,
+                        model=target.model,
                     )
                     if exp.result.score(fill_score=None) is None:
                         # All-error/empty is an outage, not a 0.0 baseline: a
@@ -426,7 +480,7 @@ async def _best_from_db(self) -> str:
         for idx, (_, row) in enumerate(shortlist.iterrows()):
             commit = row["candidate_commit"]
             dataset_id = row.get("dataset_subset_dataset_id")
-            score = await self._admin_eval_score(
+            score, _cause = await self._admin_eval_score(
                 task=self.selection_task,
                 dataset_id=dataset_id,
                 split=self.selection_split,
@@ -462,7 +516,7 @@ async def _best_from_db(self) -> str:
             base_dataset_id = self.selection_dataset_id
             if base_dataset_id is None:
                 base_dataset_id = shortlist.iloc[0].get("dataset_subset_dataset_id")
-            base_score_opt = await self._admin_eval_score(
+            base_score_opt, _cause = await self._admin_eval_score(
                 task=self.selection_task,
                 dataset_id=base_dataset_id,
                 split=self.selection_split,
diff --git a/vero/tests/test_engine.py b/vero/tests/test_engine.py
index 401e6b9..3ac0022 100644
--- a/vero/tests/test_engine.py
+++ b/vero/tests/test_engine.py
@@ -182,6 +182,25 @@ async def test_free_runs_without_debiting_budget(self, monkeypatch):
         assert svc.status()[("dev", "ds1")].remaining_run_budget == 3
         assert svc.status()[("dev", "ds1")].remaining_sample_budget == 100
 
+    @pytest.mark.asyncio
+    async def test_admin_model_override_rides_task_params(self, monkeypatch):
+        # Transfer targets: evaluate_admin(model=...) must reach the eval
+        # strategy via task_params without mutating the shared run_constraints.
+        svc = _make_service(monkeypatch=monkeypatch)
+        await svc.evaluate_admin(
+            task="t", dataset_id="ds1", split="dev", commit="c1", model="openai/gpt-4o"
+        )
+        params = svc.evaluator.evaluate.await_args.kwargs["evaluation_parameters"]
+        assert params.task_params["harbor_model_override"] == "openai/gpt-4o"
+        assert svc.run_constraints.task_params.get("harbor_model_override") is None
+
+    @pytest.mark.asyncio
+    async def test_admin_no_override_keeps_shared_constraints(self, monkeypatch):
+        svc = _make_service(monkeypatch=monkeypatch)
+        await svc.evaluate_admin(task="t", dataset_id="ds1", split="dev", commit="c1")
+        params = svc.evaluator.evaluate.await_args.kwargs["evaluation_parameters"]
+        assert params is svc.run_constraints
+
     @pytest.mark.asyncio
     async def test_free_does_not_bypass_no_access_gate(self, monkeypatch):
         from vero.core.dataset import SplitAccess
diff --git a/vero/tests/test_harbor_build.py b/vero/tests/test_harbor_build.py
index 2efdeb2..a9aa8d7 100644
--- a/vero/tests/test_harbor_build.py
+++ b/vero/tests/test_harbor_build.py
@@ -363,6 +363,67 @@ def test_instruction_omits_free_baseline_claim_when_unsupported(built):
     assert "budget-free" not in text
 
 
+def test_target_model_reaches_serve_json(tmp_path, monkeypatch):
+    # Transfer probe: a target's executor-model override must survive the
+    # compile into serve.json and validate back through ServeConfig.
+    monkeypatch.setenv("VERO_SKIP_SECRET_CHECK", "1")
+    config = BuildConfigA(
+        name="vero/gsm8k-opt",
+        agent_repo=str(_agent_repo(tmp_path)),
+        task="gsm8k",
+        task_module="gsm8k_agent.vero_tasks",
+        dataset=str(_dataset(tmp_path)),
+        splits=[
+            {"split": "validation", "access": "non_viewable"},
+            {"split": "test", "access": "no_access"},
+        ],
+        budgets=[{"split": "validation", "total_run_budget": 5}],
+        selection_split="validation",
+        targets=[
+            {"split": "test", "reward_key": "reward"},
+            {"split": "test", "reward_key": "reward_4o", "model": "openai/gpt-4o"},
+        ],
+    )
+    out = compile_task(config, tmp_path / "task", vero_root=_stub_vero(tmp_path))
+    cfg = load_serve_config(out / "environment" / "sidecar" / "serve.json")
+    by_key = {t.reward_key: t for t in cfg.targets}
+    assert by_key["reward_4o"].model == "openai/gpt-4o"
+    assert by_key["reward"].model is None
+
+
+def test_instruction_renders_exhaust_budget_by_default(built):
+    text = (built / "instruction.md").read_text()
+    assert "Unspent budget is wasted" in text
+
+
+def test_instruction_omits_exhaust_budget_when_disabled(tmp_path, monkeypatch):
+    # The persistence exhortation is an instruction LEVER: off, stopping early
+    # becomes the agent's own choice, which is the ablation arm for measuring
+    # what the exhortation itself contributes.
+    monkeypatch.setenv("VERO_SKIP_SECRET_CHECK", "1")
+    config = BuildConfigA(
+        name="vero/gsm8k-opt",
+        description="optimize gsm8k",
+        agent_repo=str(_agent_repo(tmp_path)),
+        task="gsm8k",
+        task_module="gsm8k_agent.vero_tasks",
+        dataset=str(_dataset(tmp_path)),
+        splits=[
+            {"split": "validation", "access": "non_viewable"},
+            {"split": "test", "access": "no_access"},
+        ],
+        budgets=[{"split": "validation", "total_run_budget": 5}],
+        reward_mode="auto_best",
+        selection_split="validation",
+        targets=[{"split": "test", "reward_key": "reward"}],
+        instruct_exhaust_budget=False,
+    )
+    out = compile_task(config, tmp_path / "task", vero_root=_stub_vero(tmp_path))
+    text = (out / "instruction.md").read_text()
+    assert "Unspent budget is wasted" not in text
+    assert "Scores are noisy." in text  # the noise fact is not part of the lever
+
+
 def _multifidelity_config_with_splits(tmp_path, splits) -> BuildConfigB:
     # instruct_multifidelity is a Mode-B-only lever, so the multi-fidelity gate is
     # exercised through a Mode-B config (local inner_task so it compiles offline).
diff --git a/vero/tests/test_harbor_protocol.py b/vero/tests/test_harbor_protocol.py
index b7bd6f2..c24f42c 100644
--- a/vero/tests/test_harbor_protocol.py
+++ b/vero/tests/test_harbor_protocol.py
@@ -110,3 +110,17 @@ def test_advertises_subset_floor_on_non_viewable_only(self):
         by_split = {s["split"]: s for s in status.splits}
         assert by_split["validation"]["min_subset_samples"] == 5
         assert by_split["train"]["min_subset_samples"] == 1
+
+    def test_default_floor_matches_sidecar_enforcement_default(self):
+        # A caller that forgets to pass the floor must not advertise a laxer
+        # one than EvaluationSidecar enforces by default (5): agents would
+        # send sub-floor requests that get rejected.
+        budget = {
+            ("validation", "ds1"): SplitBudget(split="validation", dataset_id="ds1", total_run_budget=3),
+        }
+        status = build_status(
+            submit_enabled=False,
+            budget=budget,
+            split_accesses=[SplitAccess.non_viewable("validation")],
+        )
+        assert status.splits[0]["min_subset_samples"] == 5
diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py
index 0f53e65..b34449f 100644
--- a/vero/tests/test_harbor_runner.py
+++ b/vero/tests/test_harbor_runner.py
@@ -54,6 +54,7 @@ def _write_trial(
     trajectory: str | None = None,
     finished_at: str | None = None,
     exception_type: str | None = None,
+    exception_message: str = "",
 ):
     # Real harbor layout: <jobs>/<timestamp>/<trial>/result.json, plus a job-level
     # <jobs>/<timestamp>/result.json summary (no task_name) that collation must skip.
@@ -73,7 +74,7 @@ def _write_trial(
     if exception_type is not None:
         data["exception_info"] = {
             "exception_type": exception_type,
-            "exception_message": "",
+            "exception_message": exception_message,
             "exception_traceback": "",
         }
     (d / "result.json").write_text(json.dumps(data))
@@ -117,6 +118,19 @@ def test_no_harbor_requirement_keeps_candidate_env(self):
         cmd = _runner()._build_command("/wt", _params(), ["t0"], Path("/jobs"))
         assert "--with" not in cmd
 
+    def test_model_override_beats_configured_model(self):
+        # Transfer targets: a per-eval executor override (via task_params)
+        # wins over the task's configured model.
+        params = _params()
+        params.task_params = {"harbor_model_override": "openai/gpt-4o"}
+        cmd = _runner()._build_command("/wt", params, ["t0"], Path("/jobs"))
+        m = cmd[cmd.index("-m") + 1]
+        assert m == "openai/gpt-4o"
+
+    def test_no_override_uses_configured_model(self):
+        cmd = _runner()._build_command("/wt", _params(), ["t0"], Path("/jobs"))
+        assert cmd[cmd.index("-m") + 1] == "anthropic/x"
+
 
 class TestExtractReward:
     def test_priority_pass_then_reward_then_sole_key(self):
@@ -636,6 +650,47 @@ def test_missing_transcripts_omitted_silently(self, tmp_path):
         assert r.score == 0.0
         assert r.feedback is None
 
+    def test_empty_pane_falls_through_to_trajectory(self, tmp_path):
+        # An empty transcript file carries nothing and must not surface as ""
+        # feedback; the empty pane falls through to the trajectory.
+        runner = _fb_runner()
+        jobs = tmp_path / "jobs"
+        _write_trial(
+            jobs, "trial0", "t0", {"reward": 0.0}, pane="", trajectory='{"steps": [1]}'
+        )
+        assert self._result(runner, jobs).feedback == '{"steps": [1]}'
+
+    def test_all_empty_transcripts_yield_no_feedback(self, tmp_path):
+        runner = _fb_runner()
+        jobs = tmp_path / "jobs"
+        _write_trial(jobs, "trial0", "t0", {"reward": 0.0}, pane="", trajectory="")
+        assert self._result(runner, jobs).feedback is None
+
+    def test_no_rewards_error_sample_carries_crash_transcript(self, tmp_path):
+        # A candidate edit that crashes the agent before scoring lands in the
+        # no-verifier-rewards error branch; the transcript is the only way the
+        # optimizer can see the crash it caused.
+        runner = _fb_runner()
+        jobs = tmp_path / "jobs"
+        _write_trial(jobs, "trial0", "t0", None, pane="crash tail")
+        r = self._result(runner, jobs)
+        assert r.error is not None
+        assert r.feedback == "crash tail"
+
+    def test_no_rewards_error_names_dead_exception_types(self, tmp_path):
+        # The error string must carry WHY the attempts died: it is the one
+        # field that flows to the DB, the per-sample files, and the verifier's
+        # target_errors, and it separates a deterministic candidate crash
+        # (measured live: 72/72 UnsupportedParamsError on an off-model
+        # executor) from an infra outage.
+        runner = _fb_runner()
+        jobs = tmp_path / "jobs"
+        _write_trial(jobs, "trial0", "t0", None, exception_type="UnsupportedParamsError")
+        _write_trial(jobs, "trial1", "t0", None, exception_type="UnsupportedParamsError")
+        r = self._result(runner, jobs)
+        assert r.error is not None
+        assert "UnsupportedParamsError x2" in r.error
+
     def test_first_failed_attempt_transcript_used(self, tmp_path):
         # Two failed attempts: the FIRST one's transcript (by finished_at) is
         # attached, deterministically, regardless of rglob order.
@@ -895,3 +950,339 @@ def test_mean_zero_fills_reward_key_mismatch(self, tmp_path):
         assert r.score == 0.5
         assert r.metrics["n_dead"] == 1.0
         assert r.metrics["n_scored"] == 1.0
+
+
+class TestInfraResilience:
+    """Dead-attempt classification + bounded within-eval infra retry.
+
+    Classification is diagnostic and retry-gating only: every dead attempt
+    still scores 0.0 regardless of class (excusing infra-labeled deaths from
+    the score would let a candidate raise fake ConnectionErrors on hard tasks).
+    The retry is OFF BY DEFAULT and exists for trusted-candidate evaluations:
+    against an adversarial optimizer it is a re-roll lever, since the
+    qualifying predicate derives from exceptions raised in candidate code. It
+    re-measures only samples whose EVERY attempt died of a transient infra
+    cause, in a fresh sibling jobs dir, and stamps an audit marker on
+    recovered samples so the discarded round stays visible.
+    """
+
+    def _flow_runner(self, monkeypatch, tmp_path, rounds_by_call, **cfg):
+        """A runner whose _run_harbor writes fixture trials per call:
+        rounds_by_call[i] is a list of (trial, task, rewards, exc_type, exc_msg)
+        written into whatever jobs dir call i receives."""
+        monkeypatch.setenv("VERO_HOME_DIR", str(tmp_path / "vh"))
+        runner = HarborRunner(HarborConfig(
+            task_source="org/ds@1", agent_import_path="pkg.mod:Agent",
+            **cfg,
+        ))
+        calls = []
+
+        async def _fake_run(project_path, params, task_names, jobs_dir):
+            i = len(calls)
+            calls.append((list(task_names), Path(jobs_dir)))
+            for trial, task, rewards, exc_type, exc_msg in rounds_by_call[i]:
+                _write_trial(Path(jobs_dir), trial, task, rewards,
+                             exception_type=exc_type, exception_message=exc_msg)
+
+        monkeypatch.setattr(runner, "_run_harbor", _fake_run)
+        monkeypatch.setattr(runner, "_task_names_for", lambda p: [(0, "t0"), (1, "t1")])
+        sleeps = []
+        import asyncio as _asyncio
+        real_sleep = _asyncio.sleep
+
+        async def _fast_sleep(d, *a, **k):
+            sleeps.append(d)
+            await real_sleep(0)
+
+        monkeypatch.setattr("asyncio.sleep", _fast_sleep)
+        return runner, calls, sleeps
+
+    def test_config_rejects_zero_delay_when_retries_enabled(self):
+        # A zero delay silently nullifies the backoff and an instant retry
+        # re-enters the same outage; misconfiguration must fail loudly.
+        with pytest.raises(ValueError, match="infra_retry_delay_s"):
+            HarborConfig(task_source="org/ds@1", agent_import_path="p:m",
+                         infra_retry_rounds=1, infra_retry_delay_s=0.0)
+        # with retries off the delay is never used, so 0 is not an error
+        HarborConfig(task_source="org/ds@1", agent_import_path="p:m",
+                     infra_retry_rounds=0, infra_retry_delay_s=0.0)
+
+    def test_config_rejects_negative_rounds(self):
+        with pytest.raises(ValueError, match="infra_retry_rounds"):
+            HarborConfig(task_source="org/ds@1", agent_import_path="p:m",
+                         infra_retry_rounds=-1)
+
+    def test_infra_deaths_labeled_and_counted_but_still_zero_filled(self, tmp_path):
+        # One scored attempt, one infra death, one candidate crash: the mean
+        # zero-fills BOTH deaths (classification must never move the score),
+        # and the labels + n_dead_infra let an analyst see which zeros
+        # measured the plumbing.
+        runner = HarborRunner(HarborConfig(
+            task_source="org/ds@1", agent_import_path="pkg.mod:Agent",
+            n_attempts=3, aggregate_attempts="mean",
+        ))
+        jobs = tmp_path / "jobs"
+        _write_trial(jobs, "a", "t0", {"reward": 1.0})
+        _write_trial(jobs, "b", "t0", None, exception_type="ConnectionError")
+        _write_trial(jobs, "c", "t0", None, exception_type="UnsupportedParamsError")
+        groups = runner._trial_groups(jobs)
+        r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"])
+        assert r.score == pytest.approx(1.0 / 3)
+        assert r.output["dead_exception_types"] == {
+            "ConnectionError[infra]": 1,
+            "UnsupportedParamsError": 1,
+        }
+        assert r.metrics["n_dead"] == 2.0
+        assert r.metrics["n_dead_infra"] == 1.0
+
+    def test_forged_infra_suffix_is_neutralized(self, tmp_path):
+        # The suffix contract is load-bearing and the exception type name is
+        # candidate-authored: a class literally named "XError[infra]" must not
+        # classify as infra. Brackets are neutralized before labeling.
+        runner = HarborRunner(HarborConfig(
+            task_source="org/ds@1", agent_import_path="pkg.mod:Agent",
+            n_attempts=2, aggregate_attempts="mean",
+        ))
+        jobs = tmp_path / "jobs"
+        _write_trial(jobs, "a", "t0", {"reward": 1.0})
+        _write_trial(jobs, "b", "t0", None, exception_type="MadeUpError[infra]")
+        groups = runner._trial_groups(jobs)
+        r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"])
+        assert r.output["dead_exception_types"] == {"MadeUpError(infra)": 1}
+        assert r.metrics["n_dead_infra"] == 0.0
+
+    def test_key_budget_exhaustion_labeled_and_alarmed(self, tmp_path, caplog):
+        # litellm reports a spent key budget as a BadRequestError; only the
+        # message identifies it. It must be labeled infra (those zeros likely
+        # measure an outage) and alarmed at ERROR (every later call fails
+        # identically). The alarm text hedges: the signature is read from
+        # candidate-process exceptions and needs corroboration.
+        import logging as _logging
+
+        runner = HarborRunner(HarborConfig(
+            task_source="org/ds@1", agent_import_path="pkg.mod:Agent",
+            n_attempts=2, aggregate_attempts="mean",
+        ))
+        jobs = tmp_path / "jobs"
+        msg = "litellm.BadRequestError: Budget has been exceeded! Current cost: 102.47, Max budget: 99.0"
+        _write_trial(jobs, "a", "t0", None, exception_type="BadRequestError", exception_message=msg)
+        _write_trial(jobs, "b", "t0", None, exception_type="BadRequestError", exception_message=msg)
+        groups = runner._trial_groups(jobs)
+        with caplog.at_level(_logging.ERROR, logger="vero.harbor.runner"):
+            r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"])
+        assert r.error is not None
+        assert "BadRequestError[infra:llm-key-budget] x2" in r.error
+        assert r.output["dead_exception_types"] == {"BadRequestError[infra:llm-key-budget]": 2}
+        assert any("spend budget as exhausted" in rec.message for rec in caplog.records)
+
+    @pytest.mark.asyncio
+    async def test_retry_is_off_by_default(self, tmp_path, monkeypatch):
+        # The re-roll lever must be opt-in: with a default config, an
+        # infra-outaged sample books its cause-rich error and nothing re-runs.
+        runner, calls, sleeps = self._flow_runner(
+            monkeypatch, tmp_path,
+            rounds_by_call=[
+                [("a", "t0", {"reward": 1.0}, None, ""),
+                 ("b", "t1", None, "ConnectionError", "")],
+            ],
+        )
+        assert runner.config.infra_retry_rounds == 0
+        params = _params()
+        await runner.produce_sample_results(
+            workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res"
+        )
+        assert len(calls) == 1
+        assert sleeps == []
+
+    @pytest.mark.asyncio
+    async def test_retry_reruns_only_the_outaged_sample(self, tmp_path, monkeypatch):
+        # t0 scores; t1 loses every attempt to ConnectionError. The retry round
+        # re-runs t1 ALONE in a fresh SIBLING jobs dir after a backoff, the
+        # fresh measurement replaces the recorded outage, and the booked output
+        # carries the audit marker naming the discarded dead attempts.
+        runner, calls, sleeps = self._flow_runner(
+            monkeypatch, tmp_path,
+            rounds_by_call=[
+                [("a", "t0", {"reward": 1.0}, None, ""),
+                 ("b", "t1", None, "ConnectionError", ""),
+                 ("c", "t1", None, "ConnectionError", "")],
+                [("r1", "t1", {"reward": 0.5}, None, "")],
+            ],
+            infra_retry_rounds=1,
+        )
+        params = _params()
+        await runner.produce_sample_results(
+            workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res"
+        )
+        results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id)
+        assert results[0].score == 1.0
+        assert results[1].error is None and results[1].score == 0.5
+        assert results[1].output["infra_retry"] == {
+            "recovered_round": 1,
+            "discarded_rounds": [{"ConnectionError[infra]": 2}],
+        }
+        assert "infra_retry" not in (results[0].output or {})
+        assert len(calls) == 2
+        assert calls[1][0] == ["t1"]
+        # sibling of jobs/, never nested inside it (resume collations rglob
+        # the whole jobs dir and would pool this round's dead attempts)
+        assert calls[1][1].name == "jobs-infra-retry-1"
+        assert calls[1][1].parent == calls[0][1].parent
+        assert sleeps == [30.0]
+
+    @pytest.mark.asyncio
+    async def test_candidate_crash_and_spent_key_never_retry(self, tmp_path, monkeypatch):
+        # A crash is a result; a spent key cannot recover by waiting. Neither
+        # triggers a retry round even with retries enabled.
+        budget_msg = "Budget has been exceeded! Current cost: 102.47, Max budget: 99.0"
+        runner, calls, sleeps = self._flow_runner(
+            monkeypatch, tmp_path,
+            rounds_by_call=[
+                [("a", "t0", None, "UnsupportedParamsError", ""),
+                 ("b", "t1", None, "BadRequestError", budget_msg)],
+            ],
+            infra_retry_rounds=1,
+        )
+        params = _params()
+        await runner.produce_sample_results(
+            workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res"
+        )
+        results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id)
+        assert results[0].error is not None and results[1].error is not None
+        assert len(calls) == 1
+        assert sleeps == []
+
+    @pytest.mark.asyncio
+    async def test_mixed_cause_fully_dead_sample_never_retries(self, tmp_path, monkeypatch):
+        # EVERY dead cause must be transient-infra (all, not any): a
+        # deterministically-crashing candidate must not qualify for a re-roll
+        # by mixing one fake ConnectionError into its crashes.
+        runner, calls, sleeps = self._flow_runner(
+            monkeypatch, tmp_path,
+            rounds_by_call=[
+                [("a", "t0", {"reward": 1.0}, None, ""),
+                 ("b", "t1", None, "ConnectionError", ""),
+                 ("c", "t1", None, "UnsupportedParamsError", "")],
+            ],
+            n_attempts=2, aggregate_attempts="mean", infra_retry_rounds=1,
+        )
+        params = _params()
+        await runner.produce_sample_results(
+            workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res"
+        )
+        results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id)
+        assert results[1].error is not None
+        assert len(calls) == 1
+        assert sleeps == []
+
+    @pytest.mark.asyncio
+    async def test_persistent_outage_keeps_the_cause_rich_error(self, tmp_path, monkeypatch):
+        # The retry round produces nothing (outage persists): the durable
+        # record must keep the original infra-labeled error, not degrade to
+        # "no Harbor trial result" or raise out of the eval.
+        runner, calls, sleeps = self._flow_runner(
+            monkeypatch, tmp_path,
+            rounds_by_call=[
+                [("a", "t0", {"reward": 1.0}, None, ""),
+                 ("b", "t1", None, "ConnectionError", "")],
+                [],
+            ],
+            infra_retry_rounds=1,
+        )
+        params = _params()
+        await runner.produce_sample_results(
+            workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res"
+        )
+        results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id)
+        assert results[1].error is not None
+        assert "ConnectionError[infra]" in results[1].error
+        assert results[1].output["dead_exception_types"] == {"ConnectionError[infra]": 1}
+        assert len(calls) == 2
+
+    @pytest.mark.asyncio
+    async def test_partially_scored_sample_is_a_measurement_not_an_outage(self, tmp_path, monkeypatch):
+        # One attempt scored, one died of infra: the sample is a (noisy)
+        # measurement. Its infra death stays zero-filled in the mean and it is
+        # NOT retried; anything else would let infra flakiness re-roll scores.
+        runner, calls, sleeps = self._flow_runner(
+            monkeypatch, tmp_path,
+            rounds_by_call=[
+                [("a", "t0", {"reward": 1.0}, None, ""),
+                 ("b", "t0", None, "ConnectionError", ""),
+                 ("c", "t1", {"reward": 1.0}, None, "")],
+            ],
+            n_attempts=2, aggregate_attempts="mean", infra_retry_rounds=1,
+        )
+        params = _params()
+        await runner.produce_sample_results(
+            workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res"
+        )
+        results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id)
+        assert results[0].score == 0.5
+        assert results[0].metrics["n_dead_infra"] == 1.0
+        assert len(calls) == 1
+
+    @pytest.mark.asyncio
+    async def test_retry_round_mean_is_not_diluted_by_dead_rounds(self, tmp_path, monkeypatch):
+        # The fresh round is collated from its own sibling dir alone: the two
+        # dead attempts of round 0 must not zero-dilute the fresh mean
+        # (1.0, not 0.5), and the audit marker records what was discarded.
+        runner, calls, sleeps = self._flow_runner(
+            monkeypatch, tmp_path,
+            rounds_by_call=[
+                [("a", "t0", {"reward": 1.0}, None, ""),
+                 ("b", "t1", None, "ConnectionError", ""),
+                 ("c", "t1", None, "ConnectionError", "")],
+                [("r1", "t1", {"reward": 1.0}, None, ""),
+                 ("r2", "t1", {"reward": 1.0}, None, "")],
+            ],
+            n_attempts=2, aggregate_attempts="mean", infra_retry_rounds=1,
+        )
+        params = _params()
+        await runner.produce_sample_results(
+            workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res"
+        )
+        results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id)
+        assert results[1].score == 1.0
+        assert results[1].metrics["n_scored"] == 2.0
+        assert results[1].metrics["n_dead"] == 0.0
+        assert results[1].output["infra_retry"]["discarded_rounds"] == [
+            {"ConnectionError[infra]": 2}
+        ]
+
+    @pytest.mark.asyncio
+    async def test_multi_round_backoff_and_partial_recovery(self, tmp_path, monkeypatch):
+        # Two rounds: round 1 recovers t0 and leaves t1 outaged; round 2
+        # retries ONLY the survivor, after a LONGER (linear) backoff, in its
+        # own fresh sibling dir.
+        runner, calls, sleeps = self._flow_runner(
+            monkeypatch, tmp_path,
+            rounds_by_call=[
+                [("a", "t0", None, "ConnectionError", ""),
+                 ("b", "t1", None, "TimeoutError", "")],
+                [("r1a", "t0", {"reward": 1.0}, None, ""),
+                 ("r1b", "t1", None, "TimeoutError", "")],
+                [("r2b", "t1", {"reward": 0.5}, None, "")],
+            ],
+            infra_retry_rounds=2,
+        )
+        params = _params()
+        await runner.produce_sample_results(
+            workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res"
+        )
+        results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id)
+        assert results[0].score == 1.0
+        assert results[0].output["infra_retry"] == {
+            "recovered_round": 1,
+            "discarded_rounds": [{"ConnectionError[infra]": 1}],
+        }
+        # t1 burned TWO rounds before recovering; the audit marker must list
+        # both, not just the round immediately before recovery.
+        assert results[1].score == 0.5
+        assert results[1].output["infra_retry"] == {
+            "recovered_round": 2,
+            "discarded_rounds": [{"TimeoutError[infra]": 1}, {"TimeoutError[infra]": 1}],
+        }
+        assert [c[0] for c in calls] == [["t0", "t1"], ["t0", "t1"], ["t1"]]
+        assert [c[1].name for c in calls[1:]] == ["jobs-infra-retry-1", "jobs-infra-retry-2"]
+        assert sleeps == [30.0, 60.0]
diff --git a/vero/tests/test_harbor_serve.py b/vero/tests/test_harbor_serve.py
index 200f9c2..4cb315e 100644
--- a/vero/tests/test_harbor_serve.py
+++ b/vero/tests/test_harbor_serve.py
@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import json
 import subprocess
 import textwrap
 from pathlib import Path
@@ -16,7 +17,12 @@
 
 from vero.core.dataset.store import resolve_and_save_dataset
 from vero.evaluation.engine import EvalRequest
-from vero.harbor.serve import ServeConfigA, ServeConfigB, build_components
+from vero.harbor.serve import (
+    ServeConfigA,
+    ServeConfigB,
+    _load_or_build_ledger,
+    build_components,
+)
 
 
 def _git(path: Path, *args: str) -> str:
@@ -221,6 +227,40 @@ async def test_ledger_reloads_spent_budget_across_restart(fixture):
     assert reloaded == after, "sidecar restart must not refill spent budget"
 
 
+class TestLedgerFailClosed:
+    """A persisted ledger that exists but cannot be read fails CLOSED: spend
+    that cannot be reconstructed is treated as fully spent. The old fallback
+    (configured budgets) refunded the agent everything already spent, so any
+    crash that corrupted the flush minted budget."""
+
+    _CFGS = [{
+        "split": "validation", "dataset_id": "ds",
+        "total_run_budget": 5, "total_sample_budget": 50,
+    }]
+
+    def test_missing_file_boots_configured(self, tmp_path):
+        led = _load_or_build_ledger(self._CFGS, tmp_path / "ledger.json")
+        b = led.get("ds", "validation")
+        assert b.remaining_run_budget == 5
+        assert b.remaining_sample_budget == 50
+
+    def test_unparseable_file_fails_closed_and_keeps_evidence(self, tmp_path):
+        p = tmp_path / "ledger.json"
+        p.write_text("{definitely not json")
+        led = _load_or_build_ledger(self._CFGS, p)
+        b = led.get("ds", "validation")
+        assert b.remaining_run_budget == 0
+        assert b.remaining_sample_budget == 0
+        # the unreadable original survives for the operator to inspect
+        assert p.with_suffix(".corrupt").read_text() == "{definitely not json"
+
+    def test_malformed_entries_fail_closed(self, tmp_path):
+        p = tmp_path / "ledger.json"
+        p.write_text(json.dumps([{"no_split_key": 1}]))
+        led = _load_or_build_ledger(self._CFGS, p)
+        assert led.get("ds", "validation").remaining_run_budget == 0
+
+
 @pytest.mark.asyncio
 async def test_feedback_levers_reach_harbor_runner(fixture):
     # Lever 1 pass-through: ServeConfigB -> build_components -> HarborRunner kwargs
diff --git a/vero/tests/test_harbor_server.py b/vero/tests/test_harbor_server.py
index bafd4ee..937ecca 100644
--- a/vero/tests/test_harbor_server.py
+++ b/vero/tests/test_harbor_server.py
@@ -245,7 +245,8 @@ async def test_summary_carries_qualifiers(self, tmp_path):
         assert data["n_scored"] == 3
         assert data["n_errored"] == 0
         assert data["mean_score"] == pytest.approx(1 / 3)
-        assert data["score_se"] == pytest.approx(1 / 3)  # sd .5774 / sqrt(3)
+        # SE of mean_score, i.e. over the zero-filled n_samples population
+        assert data["mean_score_se"] == pytest.approx(1 / 3)  # sd .5774 / sqrt(3)
         # enum VALUE, not "ExperimentResultStatus.SUCCESS"
         assert data["status"] == "success"
 
@@ -263,6 +264,18 @@ async def test_reevals_get_versioned_dirs(self, tmp_path):
         assert (Path(s1.result_path) / "summary.json").exists()
         assert (Path(s2.result_path) / "summary.json").exists()
 
+    @pytest.mark.asyncio
+    async def test_volume_reuse_resumes_ordinal_past_survivors(self, tmp_path):
+        # A restarted sidecar (fresh _eval_seq) on a reused volume must not
+        # wipe the prior session's __e1..__eN evidence.
+        survivor = tmp_path / "agent_vol" / "results" / "validation__old000000000__e7"
+        survivor.mkdir(parents=True)
+        (survivor / "summary.json").write_text("{}")
+        sidecar = _sidecar(tmp_path, split="validation")
+        s = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation"))
+        assert s.result_path.endswith("__e8")
+        assert (survivor / "summary.json").exists()
+
 
 class TestKAnonymityFloor:
     """Subset evals on non_viewable splits are floored: the aggregate response
@@ -396,6 +409,22 @@ async def test_failed_free_eval_does_not_consume_freebie(self, tmp_path):
         await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation"))
         assert sidecar.engine.evaluate.await_args.kwargs["free"] is True
 
+    @pytest.mark.asyncio
+    async def test_freebie_claimed_before_eval_await(self, tmp_path):
+        # The claim must be visible DURING the eval await, or a concurrent
+        # second baseline eval would also resolve free_baseline=True and both
+        # would ride free (asyncio interleaves at await points).
+        sidecar = _sidecar(tmp_path, split="validation", base_commit="abcdef123456")
+        seen: list[bool] = []
+
+        async def _spy(req, **kwargs):
+            seen.append(sidecar._free_baseline_used)
+            return _experiment("validation")
+
+        sidecar.engine.evaluate = AsyncMock(side_effect=_spy)
+        await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation"))
+        assert seen == [True]
+
     def test_status_surfaces_free_baseline(self, tmp_path):
         sidecar = _sidecar(tmp_path, split="train", base_commit="abcdef123456")
         s = sidecar.status()
diff --git a/vero/tests/test_harbor_verifier.py b/vero/tests/test_harbor_verifier.py
index 72e4ec4..55714bd 100644
--- a/vero/tests/test_harbor_verifier.py
+++ b/vero/tests/test_harbor_verifier.py
@@ -89,7 +89,7 @@ async def test_auto_best_reranks_by_admin_score(self, tmp_path):
         )
         admin_scores = {"hi": 0.1, "lo": 0.95}
 
-        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None):
             return MagicMock(
                 result=MagicMock(score=MagicMock(return_value=admin_scores.get(commit, 0.99)))
             )
@@ -125,7 +125,7 @@ async def test_auto_best_excludes_baseline_from_ranking(self, tmp_path):
             }
         )
 
-        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None):
             return MagicMock(result=MagicMock(score=MagicMock(return_value=0.7)))
 
         engine.evaluate_admin = AsyncMock(side_effect=_admin)
@@ -172,7 +172,7 @@ async def test_lucky_subset_eval_does_not_outrank_full_split(self, tmp_path):
             }
         )
 
-        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None):
             # admin re-score agrees the full-split ranking is right
             score = {"solid": 0.7, "lucky": 0.3}.get(commit, 0.5)
             return MagicMock(result=MagicMock(score=MagicMock(return_value=score)))
@@ -212,7 +212,7 @@ async def test_all_subset_evals_still_rankable(self, tmp_path):
             }
         )
 
-        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None):
             return MagicMock(result=MagicMock(score=MagicMock(return_value=0.5)))
 
         engine.evaluate_admin = AsyncMock(side_effect=_admin)
@@ -260,7 +260,7 @@ async def test_reverts_to_base_when_no_candidate_beats_baseline(self, tmp_path):
         # agent admin-scores 0.2 on the selection split; base admin-scores 0.3;
         # the reverted base scores 0.35 on the target split (distinct values so the
         # assertions can tell the target eval apart from the floor comparison).
-        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None):
             if commit == "base":
                 score = 0.35 if split == "validation" else 0.3
             else:
@@ -296,7 +296,7 @@ async def test_exact_tie_reverts_to_base(self, tmp_path):
         engine = MagicMock()
         engine.db.get_experiments_df.return_value = self._df()
 
-        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None):
             return MagicMock(result=MagicMock(score=MagicMock(return_value=0.3)))  # all equal
 
         engine.evaluate_admin = AsyncMock(side_effect=_admin)
@@ -327,7 +327,7 @@ async def test_floor_noop_without_base_commit(self, tmp_path):
             }
         )
 
-        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None):
             return MagicMock(result=MagicMock(score=MagicMock(return_value=0.5)))
 
         engine.evaluate_admin = AsyncMock(side_effect=_admin)
@@ -349,7 +349,7 @@ async def test_keeps_candidate_that_beats_baseline(self, tmp_path):
         engine = MagicMock()
         engine.db.get_experiments_df.return_value = self._df()
 
-        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None):
             score = 0.3 if commit == "base" else 0.6  # agent genuinely improves
             return MagicMock(result=MagicMock(score=MagicMock(return_value=score)))
 
@@ -374,7 +374,7 @@ async def test_floor_off_ships_least_bad_candidate(self, tmp_path):
         engine = MagicMock()
         engine.db.get_experiments_df.return_value = self._df()
 
-        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None):
             return MagicMock(result=MagicMock(score=MagicMock(return_value=0.2)))
 
         engine.evaluate_admin = AsyncMock(side_effect=_admin)
@@ -476,7 +476,7 @@ async def test_candidates_present_keeps_normal_selection(self, tmp_path):
             }
         )
 
-        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None):
             return MagicMock(result=MagicMock(score=MagicMock(return_value=0.5)))
 
         engine.evaluate_admin = AsyncMock(side_effect=_admin)
@@ -733,7 +733,7 @@ async def test_floor_fail_safe_reverts_on_unmeasurable_baseline(self, tmp_path):
             }
         )
 
-        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None):
             if commit == "base" and split == "validation":
                 raise RuntimeError("nested run crashed")
             return MagicMock(result=MagicMock(score=MagicMock(return_value=0.9)))
@@ -755,6 +755,90 @@ async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
         assert engine.evaluate_admin.await_args.kwargs["commit"] == "base"
         assert engine.evaluate_admin.await_args.kwargs["split"] == "test"
 
+    @pytest.mark.asyncio
+    async def test_transfer_target_model_reaches_champion_and_baseline_evals(self, tmp_path):
+        # A target's executor-model override must apply to BOTH the champion
+        # eval and the baseline eval, or the comparison is not like-for-like.
+        self._submit(tmp_path)
+        engine = MagicMock()
+        engine.evaluate_admin = AsyncMock(
+            return_value=MagicMock(result=MagicMock(score=MagicMock(return_value=0.5)))
+        )
+        v = Verifier(
+            engine=engine,
+            admin_volume=tmp_path,
+            reward_mode="submit",
+            score_baseline=True,
+            base_commit="base",
+            targets=[VerificationTarget(
+                task="t", dataset_id="ds", split="test",
+                reward_key="reward_4o", model="openai/gpt-4o",
+            )],
+        )
+        await v.finalize()
+        models = [c.kwargs["model"] for c in engine.evaluate_admin.await_args_list]
+        commits = [c.kwargs["commit"] for c in engine.evaluate_admin.await_args_list]
+        assert models == ["openai/gpt-4o", "openai/gpt-4o"]
+        assert set(commits) == {"cand", "base"}
+
+    @pytest.mark.asyncio
+    async def test_floored_target_records_dominant_crash_cause(self, tmp_path):
+        # An all-errored target eval floors the reward AND names the dominant
+        # per-sample cause in target_errors: a champion that deterministically
+        # crashes on this target's executor must be distinguishable from an
+        # infra outage in the one durable record.
+        self._submit(tmp_path)
+        engine = MagicMock()
+        crash = "No verifier rewards for task 't'. (attempts died: UnsupportedParamsError x6)"
+        exp = MagicMock()
+        exp.result.score = MagicMock(return_value=None)
+        exp.result.sample_results = {
+            i: MagicMock(error=crash) for i in range(3)
+        }
+        engine.evaluate_admin = AsyncMock(return_value=exp)
+        v = Verifier(
+            engine=engine,
+            admin_volume=tmp_path,
+            reward_mode="submit",
+            baseline_score_attempts=2,
+            targets=[VerificationTarget(task="t", dataset_id="ds", split="test", reward_key="reward")],
+        )
+        result = await v.finalize()
+        assert result["rewards"] == {"reward": 0.0}
+        assert "UnsupportedParamsError x6" in result["target_errors"]["reward"]
+        assert "3x:" in result["target_errors"]["reward"]
+
+    @pytest.mark.asyncio
+    async def test_crash_cause_clusters_across_task_names(self, tmp_path):
+        # Runner error strings embed the task name; a slice spanning several
+        # tasks that all die of the SAME exception must still cluster as one
+        # dominant cause (grouping normalizes the task name away), or the
+        # deterministic-crash signature degrades into 1x singletons.
+        self._submit(tmp_path)
+        engine = MagicMock()
+        exp = MagicMock()
+        exp.result.score = MagicMock(return_value=None)
+        exp.result.sample_results = {
+            i: MagicMock(
+                error=(
+                    f"No verifier rewards for task 'org/task-{i}'. "
+                    f"(attempts died: UnsupportedParamsError x6)"
+                )
+            )
+            for i in range(3)
+        }
+        engine.evaluate_admin = AsyncMock(return_value=exp)
+        v = Verifier(
+            engine=engine,
+            admin_volume=tmp_path,
+            reward_mode="submit",
+            baseline_score_attempts=2,
+            targets=[VerificationTarget(task="t", dataset_id="ds", split="test", reward_key="reward")],
+        )
+        result = await v.finalize()
+        assert "3x:" in result["target_errors"]["reward"]
+        assert "UnsupportedParamsError x6" in result["target_errors"]["reward"]
+
     @pytest.mark.asyncio
     async def test_target_eval_retries_then_floors_with_error_marker(self, tmp_path):
         # A persistently failing target eval floors the reward (reward.json