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/runner.py b/vero/src/vero/harbor/runner.py
index 4c6042b..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)
@@ -209,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:
@@ -374,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:
@@ -398,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),
@@ -406,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),
@@ -417,22 +635,32 @@ def _out(output: dict) -> dict:
             # 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
+            # an infra outage: a distinction that decides whether the cell is
             # a measurement or a re-run.
             cause = ""
+            dead: dict[str, int] = {}
             if attempts:
-                dead = {}
                 for t in attempts:
                     if (t.get("verifier_result") or {}).get("rewards"):
                         continue
-                    exc = (t.get("exception_info") or {}).get("exception_type")
-                    key = exc or "no_rewards_recorded"
+                    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}'.{cause}",
                 # The agent died before scoring. A candidate edit that CRASHES
@@ -441,9 +669,7 @@ def _out(output: dict) -> dict:
                 # 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(
-                    {"task_name": task_name, "trial_name": trial.get("trial_name")}
-                ),
+                output=_out(no_rewards_output),
                 **common,
             )
         score = self._extract_reward(rewards)
diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py
index 222e6d0..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))
@@ -949,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]