diff --git a/vero/src/vero/harbor/build/compiler.py b/vero/src/vero/harbor/build/compiler.py
index 7d1605a..98f2073 100644
--- a/vero/src/vero/harbor/build/compiler.py
+++ b/vero/src/vero/harbor/build/compiler.py
@@ -367,6 +367,10 @@ def compile_task(
         description=config.description,
         mode=config.mode,
         timeout=config.timeout,
+        # The verifier phase runs the whole finalize battery (shortlist
+        # re-scores + floor + targets + baseline attempts), not one eval;
+        # unset falls back to `timeout` for backward compatibility.
+        verifier_timeout=config.verifier_timeout or config.timeout,
         secrets=config.secrets,
         read_only_paths=config.read_only_paths,
         base_image_main=config.base_image_main,
diff --git a/vero/src/vero/harbor/build/config.py b/vero/src/vero/harbor/build/config.py
index 07520f2..4cb4a54 100644
--- a/vero/src/vero/harbor/build/config.py
+++ b/vero/src/vero/harbor/build/config.py
@@ -82,6 +82,15 @@ class _BuildConfigBase(BaseModel):
     timeout: int = 1800
     max_concurrency: int = 8
 
+    # Wall-clock budget for the VERIFIER phase (Harbor's [verifier] timeout_sec).
+    # Finalize is not one eval: it runs up to rescore_top_k shortlist re-scores
+    # + 1 floor eval + len(targets) target evals + len(targets) x
+    # baseline_score_attempts baseline evals, each a full nested run in Mode B.
+    # Sizing this at one eval's duration kills finalize mid-flight and the trial
+    # ships NO reward.json. Defaults to `timeout` when unset; size it as
+    # (rescore_top_k + 1 + 3 x len(targets)) x a single eval's duration + slack.
+    verifier_timeout: int | None = None
+
 
 class BuildConfigA(_BuildConfigBase):
     """Mode A: vero runs inference + scoring against a saved dataset."""
diff --git a/vero/src/vero/harbor/build/templates/task.toml.j2 b/vero/src/vero/harbor/build/templates/task.toml.j2
index c037e22..143d234 100644
--- a/vero/src/vero/harbor/build/templates/task.toml.j2
+++ b/vero/src/vero/harbor/build/templates/task.toml.j2
@@ -13,8 +13,11 @@ user = "agent"
 # Shared mode: Harbor runs tests/test.sh in `main` with the whole env (incl. the
 # eval-sidecar) still up. The verifier runs as root, reads the admin token, and
 # calls the sidecar's `finalize` endpoint to score the selected commit.
+# timeout_sec must cover the WHOLE finalize battery (shortlist re-scores +
+# floor + targets + baseline attempts), each a full nested eval, or Harbor
+# kills finalize mid-flight and the trial ships no reward.json.
 environment_mode = "shared"
-timeout_sec = {{ timeout }}
+timeout_sec = {{ verifier_timeout }}
 
 [environment]
 # Compose-based environment: environment/docker-compose.yaml adds the eval-sidecar
diff --git a/vero/src/vero/harbor/verifier.py b/vero/src/vero/harbor/verifier.py
index 35efdfa..5b41d9f 100644
--- a/vero/src/vero/harbor/verifier.py
+++ b/vero/src/vero/harbor/verifier.py
@@ -12,6 +12,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
 from dataclasses import dataclass
@@ -75,13 +76,34 @@ def __init__(
         # candidate even when every candidate regressed, shipping a regression
         # (observed live: a weak inner model, every candidate below baseline).
         self.auto_best_baseline_floor = auto_best_baseline_floor
-        # Baseline scoring is retried this many times total before its outcome is
-        # reported as an error; the nested eval can fail transiently (a nested
-        # harbor run crashing right after a large eval), and a single blip must
-        # not silently drop the regression check.
+        # Every reward-critical finalize eval (targets, shortlist re-scores,
+        # floor, baseline) is retried this many times total: the nested eval can
+        # fail transiently (a nested harbor run crashing right after a large
+        # eval), and a single blip must not abort finalize; a trial that ships
+        # no reward.json loses its result entirely.
         self._baseline_score_attempts = max(1, baseline_score_attempts)
+        # Finalize is idempotent: Harbor may retry the verifier (or an operator
+        # may re-POST /finalize), and a replayed finalize must return the FIRST
+        # completed result verbatim. Re-running it would re-rank against a DB
+        # that now also contains the first finalize's own admin evals, so a
+        # retry could select a DIFFERENT champion than the one already reported.
+        self._finalize_lock = asyncio.Lock()
+        self._finalize_result: dict | None = None
 
     async def finalize(self) -> dict:
+        """Idempotent entry point: the first completed finalize is cached and
+        replayed verbatim on any retry (see __init__ for why re-running would
+        be unsound). The lock serializes concurrent calls so exactly one
+        finalize ever computes."""
+        async with self._finalize_lock:
+            if self._finalize_result is not None:
+                logger.info("finalize: replaying cached result (idempotent)")
+                return self._finalize_result
+            result = await self._finalize()
+            self._finalize_result = result
+            return result
+
+    async def _finalize(self) -> dict:
         """Select the commit, score it on every target, and score the baseline.
 
         Returns a wrapper ``{"rewards": {reward_key: score}, "baseline": {...}}``.
@@ -113,20 +135,90 @@ async def finalize(self) -> dict:
             return {"rewards": rewards, "baseline": {"skipped": "no candidate commit"}}
         logger.info(f"Verifier selected commit {sha} (mode={self.reward_mode})")
         rewards: dict[str, float] = {}
+        target_errors: dict[str, str] = {}
         for target in self.targets:
-            exp = await self.engine.evaluate_admin(
+            score = await self._admin_eval_score(
                 task=target.task,
                 dataset_id=target.dataset_id,
                 split=target.split,
                 commit=sha,
                 sample_ids=target.sample_ids,
+                what=f"target '{target.reward_key}'",
             )
-            score = exp.result.score()
-            rewards[target.reward_key] = (
-                float(score) if score is not None else default_minimum_score
-            )
+            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.
+                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"
+                )
+            else:
+                rewards[target.reward_key] = score
         baseline = await self._maybe_score_baseline(rewards)
-        return {"rewards": rewards, "baseline": baseline}
+        result = {"rewards": rewards, "baseline": baseline}
+        if target_errors:
+            result["target_errors"] = target_errors
+        return result
+
+    async def _admin_eval_score(
+        self,
+        *,
+        task: str | None,
+        dataset_id: str,
+        split: str,
+        commit: str,
+        sample_ids: list[int] | None = None,
+        what: str,
+    ) -> float | 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.
+        """
+        last_error: Exception | str | None = None
+        for attempt in range(1, self._baseline_score_attempts + 1):
+            try:
+                exp = await self.engine.evaluate_admin(
+                    task=task,
+                    dataset_id=dataset_id,
+                    split=split,
+                    commit=commit,
+                    sample_ids=sample_ids,
+                )
+                if exp.result.score(fill_score=None) is None:
+                    last_error = "eval scored no samples (all errored or empty)"
+                    logger.warning(
+                        "%s attempt %d/%d: %s",
+                        what, attempt, self._baseline_score_attempts, last_error,
+                    )
+                    continue
+                score = exp.result.score()
+                if score is None:
+                    # Should be unreachable (the strict check above already
+                    # passed), but a None here must consume a retry like any
+                    # other unmeasurable outcome, never bypass the loop.
+                    last_error = "eval returned no aggregate score"
+                    continue
+                return float(score)
+            except Exception as exc:  # noqa: BLE001 - retried, then surfaced as None
+                last_error = exc
+                logger.warning(
+                    "%s attempt %d/%d failed: %s",
+                    what, attempt, self._baseline_score_attempts, exc,
+                )
+        logger.error(
+            "%s failed after %d attempt(s): %s",
+            what, self._baseline_score_attempts, last_error,
+        )
+        return None
 
     async def _maybe_score_baseline(self, rewards: dict[str, float]) -> dict:
         """Admin-score the unmodified baseline on every target and report it.
@@ -169,6 +261,14 @@ async def _maybe_score_baseline(self, rewards: dict[str, float]) -> dict:
                         commit=self.base_commit,
                         sample_ids=target.sample_ids,
                     )
+                    if exp.result.score(fill_score=None) is None:
+                        # All-error/empty is an outage, not a 0.0 baseline: a
+                        # zero here would fake a huge candidate improvement.
+                        # Raise into the retry loop instead.
+                        raise RuntimeError(
+                            f"baseline eval on '{target.reward_key}' scored no "
+                            f"samples (all errored or empty)"
+                        )
                     score = exp.result.score()
                     baselines[target.reward_key] = (
                         float(score) if score is not None else default_minimum_score
@@ -276,31 +376,73 @@ async def _best_from_db(self) -> str:
             if len(full_split_df) > 0:
                 split_df = full_split_df
         # Shortlist by recorded score (cheap, agent-influenced -> not trusted as
-        # final), one row per candidate (highest recorded score wins the slot).
-        ranked = split_df.sort_values(
+        # final). Recorded evals are POOLED before shortlisting, in two steps:
+        # every eval of the same commit averages into that commit's score, then
+        # commits with the same git TREE (identical content) collapse into one
+        # candidate group scored by the group mean. Max-over-rows selection made
+        # every re-measurement an independent lottery draw, and one live
+        # optimizer farmed exactly that ("distinct empty commits = clean
+        # independent lottery tickets") while another refused to re-measure its
+        # champion to protect a lucky draw. Pooling makes re-measurement
+        # variance-REDUCING (as statistics wants) instead of max-inflating, and
+        # tree-dedup stops identical content from stuffing the top-K shortlist
+        # or collecting several admin re-score draws.
+        agg: dict[str, tuple[str, str]] = {
+            "mean_score": ("mean_score", "mean"),
+            "candidate_created_at": ("candidate_created_at", "max"),
+        }
+        if "dataset_subset_dataset_id" in split_df.columns:
+            agg["dataset_subset_dataset_id"] = ("dataset_subset_dataset_id", "first")
+        per_commit = (
+            split_df.groupby("candidate_commit").agg(**agg).reset_index()
+        )
+        trees: dict[str, str] = {}
+        for commit in per_commit["candidate_commit"]:
+            # Unresolvable tree (non-git workspace, unknown sha) falls back to
+            # the commit itself: no pooling across commits, never a crash.
+            trees[commit] = (await self._tree_of(commit)) or commit
+        per_commit["_tree"] = per_commit["candidate_commit"].map(trees)
+        # Newest commit represents its tree group (any member is equivalent
+        # content-wise; newest keeps logs intuitive).
+        per_commit = per_commit.sort_values(
+            by=["candidate_created_at"], ascending=[False]
+        )
+        group_agg: dict[str, tuple[str, str]] = {
+            "mean_score": ("mean_score", "mean"),
+            "candidate_commit": ("candidate_commit", "first"),
+            "candidate_created_at": ("candidate_created_at", "first"),
+        }
+        if "dataset_subset_dataset_id" in per_commit.columns:
+            group_agg["dataset_subset_dataset_id"] = (
+                "dataset_subset_dataset_id", "first",
+            )
+        pooled = per_commit.groupby("_tree", sort=False).agg(**group_agg)
+        ranked = pooled.sort_values(
             by=["mean_score", "candidate_created_at"], ascending=[False, False]
         )
-        ranked = ranked.drop_duplicates(subset=["candidate_commit"], keep="first")
         shortlist = ranked.head(max(1, self.rescore_top_k))
 
         rescored: list[tuple[float, int, str]] = []
         for idx, (_, row) in enumerate(shortlist.iterrows()):
             commit = row["candidate_commit"]
             dataset_id = row.get("dataset_subset_dataset_id")
-            exp = await self.engine.evaluate_admin(
+            score = await self._admin_eval_score(
                 task=self.selection_task,
                 dataset_id=dataset_id,
                 split=self.selection_split,
                 commit=commit,
+                what=f"auto_best re-score of {commit}",
             )
-            score = exp.result.score()
-            admin_score = float(score) if score is not None else default_minimum_score
+            # A candidate whose re-score persistently fails is unscorable, not
+            # zero-scored: it keeps the floor value and cannot win on a fluke,
+            # but its failure does not abort the other candidates' re-scores.
+            admin_score = score if score is not None else float(default_minimum_score)
             # Tie-break by shortlist position (already ordered by recorded score
             # then recency), so ties resolve deterministically without depending on
             # the type of candidate_created_at (a datetime in the real DB).
             rescored.append((admin_score, idx, commit))
             logger.info(
-                "auto_best re-score: commit=%s admin_score=%s (recorded=%s)",
+                "auto_best re-score: commit=%s admin_score=%s (pooled recorded=%s)",
                 commit,
                 admin_score,
                 row["mean_score"],
@@ -320,14 +462,25 @@ 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_exp = await self.engine.evaluate_admin(
+            base_score_opt = await self._admin_eval_score(
                 task=self.selection_task,
                 dataset_id=base_dataset_id,
                 split=self.selection_split,
                 commit=self.base_commit,
+                what="auto_best floor (baseline)",
             )
-            base_s = base_exp.result.score()
-            base_score = float(base_s) if base_s is not None else default_minimum_score
+            if base_score_opt is None:
+                # The floor exists to stop shipped regressions; shipping a
+                # candidate with the floor check unmeasured re-opens exactly
+                # that hole. Fail safe: revert to the seed.
+                logger.error(
+                    "auto_best floor: baseline could not be measured; failing "
+                    "safe to base_commit %s instead of shipping unverified "
+                    "candidate %s.",
+                    self.base_commit, best_commit,
+                )
+                return self.base_commit
+            base_score = base_score_opt
             if best_score <= base_score:
                 logger.info(
                     "auto_best floor: best candidate %s (admin_score=%s) does not beat "
@@ -340,3 +493,17 @@ async def _best_from_db(self) -> str:
                 best_commit, best_score, base_score,
             )
         return best_commit
+
+    async def _tree_of(self, commit: str) -> str | None:
+        """Git tree hash for a commit via the engine's workspace, or None when
+        it cannot be resolved (non-git workspace, unknown sha). None makes the
+        caller treat the commit as its own pooling group: degraded, never wrong."""
+        workspace = getattr(self.engine.evaluator, "workspace", None)
+        tree_hash = getattr(workspace, "tree_hash", None)
+        if tree_hash is None:
+            return None
+        try:
+            return await tree_hash(commit)
+        except Exception:  # noqa: BLE001 - pooling is an optimization, not a gate
+            logger.warning("could not resolve tree hash for commit %s", commit)
+            return None
diff --git a/vero/src/vero/workspace/git.py b/vero/src/vero/workspace/git.py
index 93df894..c349332 100644
--- a/vero/src/vero/workspace/git.py
+++ b/vero/src/vero/workspace/git.py
@@ -316,6 +316,12 @@ async def resolve_ref(self, ref: str) -> str:
         """Resolve a git ref (branch, tag, short hash) to a full commit hash."""
         return await self._git("rev-parse", ref)
 
+    async def tree_hash(self, ref: str) -> str:
+        """Tree hash of a commit: identical content -> identical tree, whatever
+        the commit metadata says. Lets selection pool re-committed identical
+        trees instead of treating each sha as a fresh candidate."""
+        return await self._git("rev-parse", f"{ref}^{{tree}}")
+
     async def current_branch(self) -> str | None:
         """Return current branch name, or None if detached HEAD."""
         try:
diff --git a/vero/tests/test_harbor_verifier.py b/vero/tests/test_harbor_verifier.py
index 264cdc3..72e4ec4 100644
--- a/vero/tests/test_harbor_verifier.py
+++ b/vero/tests/test_harbor_verifier.py
@@ -610,3 +610,208 @@ async def test_missing_base_commit_warns(self, tmp_path, caplog):
         assert rewards == {"accuracy": 0.9}
         assert not (tmp_path / "baseline.json").exists()
         assert any("base_commit is not set" in m for m in caplog.messages)
+
+
+class TestSelectionIntegrity:
+    """PR: finalize idempotency, pooled shortlisting, retrying reward-critical
+    evals, and the fail-safe floor. Each behavior traces to a live incident:
+    optimizers farmed max-over-rows selection with re-measure commits, and a
+    disk-full trial shipped no reward.json because finalize was single-shot."""
+
+    def _submit(self, tmp_path, commit="cand"):
+        (tmp_path / "submission.json").write_text(json.dumps({"commit": commit}))
+
+    @pytest.mark.asyncio
+    async def test_finalize_is_idempotent(self, tmp_path):
+        # A retried finalize must replay the first result verbatim, not
+        # recompute (a re-run would re-rank against a DB polluted by the first
+        # finalize's own admin evals and could crown a different champion).
+        self._submit(tmp_path)
+        engine = _engine([0.8, 0.1])  # a recompute would consume the 0.1
+        v = Verifier(
+            engine=engine,
+            admin_volume=tmp_path,
+            reward_mode="submit",
+            targets=[VerificationTarget(task="t", dataset_id="ds", split="test", reward_key="reward")],
+        )
+        first = await v.finalize()
+        second = await v.finalize()
+        assert first == second
+        assert engine.evaluate_admin.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_same_commit_re_measures_pool_to_mean(self, tmp_path):
+        # Re-measuring a commit must reduce variance, not mint lottery draws:
+        # A's [0.9, 0.1] pools to 0.5 and loses the only shortlist slot to
+        # B's 0.6. Max-over-rows would have shortlisted A on the lucky 0.9.
+        engine = MagicMock()
+        engine.db.get_experiments_df.return_value = pd.DataFrame(
+            {
+                "dataset_subset_split": ["validation"] * 3,
+                "dataset_subset_dataset_id": ["ds1"] * 3,
+                "candidate_commit": ["A", "A", "B"],
+                "mean_score": [0.9, 0.1, 0.6],
+                "candidate_created_at": [1, 2, 3],
+            }
+        )
+        engine.evaluate_admin = AsyncMock(
+            side_effect=lambda **kw: MagicMock(
+                result=MagicMock(score=MagicMock(return_value=0.7))
+            )
+        )
+        engine.evaluator.workspace.tree_hash = AsyncMock(side_effect=lambda ref: ref + "-tree")
+        v = Verifier(
+            engine=engine,
+            admin_volume=tmp_path,
+            reward_mode="auto_best",
+            selection_split="validation",
+            selection_task="math",
+            rescore_top_k=1,
+            auto_best_baseline_floor=False,
+            targets=[VerificationTarget(task="math", dataset_id="ds1", split="test", reward_key="reward")],
+        )
+        await v.finalize()
+        rescored = [c.kwargs["commit"] for c in engine.evaluate_admin.await_args_list]
+        assert "A" not in rescored
+        assert "B" in rescored
+
+    @pytest.mark.asyncio
+    async def test_identical_trees_pool_and_do_not_stuff_shortlist(self, tmp_path):
+        # A1/A2 are the same content (same git tree) recommitted: they must
+        # collapse into ONE candidate group (pooled 0.85), so B still gets a
+        # top-2 shortlist slot and the group is re-scored exactly once.
+        engine = MagicMock()
+        engine.db.get_experiments_df.return_value = pd.DataFrame(
+            {
+                "dataset_subset_split": ["validation"] * 3,
+                "dataset_subset_dataset_id": ["ds1"] * 3,
+                "candidate_commit": ["A1", "A2", "B"],
+                "mean_score": [0.9, 0.8, 0.55],
+                "candidate_created_at": [1, 2, 3],
+            }
+        )
+        engine.evaluate_admin = AsyncMock(
+            side_effect=lambda **kw: MagicMock(
+                result=MagicMock(score=MagicMock(return_value=0.7))
+            )
+        )
+        trees = {"A1": "T", "A2": "T", "B": "TB"}
+        engine.evaluator.workspace.tree_hash = AsyncMock(side_effect=lambda ref: trees[ref])
+        v = Verifier(
+            engine=engine,
+            admin_volume=tmp_path,
+            reward_mode="auto_best",
+            selection_split="validation",
+            selection_task="math",
+            rescore_top_k=2,
+            auto_best_baseline_floor=False,
+            targets=[VerificationTarget(task="math", dataset_id="ds1", split="test", reward_key="reward")],
+        )
+        await v.finalize()
+        # count only selection-split re-scores (the winner also gets a final
+        # eval on the TARGET split, which is not a shortlist draw)
+        rescored = [
+            c.kwargs["commit"]
+            for c in engine.evaluate_admin.await_args_list
+            if c.kwargs["split"] == "validation"
+        ]
+        assert "B" in rescored
+        assert len([c for c in rescored if c in ("A1", "A2")]) == 1
+
+    @pytest.mark.asyncio
+    async def test_floor_fail_safe_reverts_on_unmeasurable_baseline(self, tmp_path):
+        # If the floor's baseline eval cannot be measured, shipping the
+        # candidate would re-open the shipped-regression hole: revert to seed.
+        engine = MagicMock()
+        engine.db.get_experiments_df.return_value = pd.DataFrame(
+            {
+                "dataset_subset_split": ["validation"],
+                "dataset_subset_dataset_id": ["ds1"],
+                "candidate_commit": ["cand"],
+                "mean_score": [0.9],
+                "candidate_created_at": [1],
+            }
+        )
+
+        async def _admin(*, task, dataset_id, split, commit, sample_ids=None):
+            if commit == "base" and split == "validation":
+                raise RuntimeError("nested run crashed")
+            return MagicMock(result=MagicMock(score=MagicMock(return_value=0.9)))
+
+        engine.evaluate_admin = AsyncMock(side_effect=_admin)
+        engine.evaluator.workspace.tree_hash = AsyncMock(side_effect=lambda ref: ref + "-tree")
+        v = Verifier(
+            engine=engine,
+            admin_volume=tmp_path,
+            reward_mode="auto_best",
+            selection_split="validation",
+            selection_task="math",
+            base_commit="base",
+            baseline_score_attempts=2,
+            targets=[VerificationTarget(task="math", dataset_id="ds1", split="test", reward_key="reward")],
+        )
+        await v.finalize()
+        # the final (target) eval ran on the SEED, not the unverified candidate
+        assert engine.evaluate_admin.await_args.kwargs["commit"] == "base"
+        assert engine.evaluate_admin.await_args.kwargs["split"] == "test"
+
+    @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
+        # must still ship) and records the outage in the wrapper so a floored
+        # reward can never masquerade as a measured 0.0.
+        self._submit(tmp_path)
+        engine = MagicMock()
+        engine.evaluate_admin = AsyncMock(side_effect=RuntimeError("boom"))
+        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 "reward" in result["target_errors"]
+        assert engine.evaluate_admin.await_count == 2  # retried before flooring
+
+    @pytest.mark.asyncio
+    async def test_target_eval_transient_failure_recovers(self, tmp_path):
+        self._submit(tmp_path)
+        engine = MagicMock()
+        healthy = MagicMock(result=MagicMock(score=MagicMock(return_value=0.8)))
+        engine.evaluate_admin = AsyncMock(side_effect=[RuntimeError("blip"), healthy])
+        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.8}
+        assert "target_errors" not in result
+
+    @pytest.mark.asyncio
+    async def test_all_error_eval_is_retried_not_zeroed(self, tmp_path):
+        # An eval in which no sample scored (score(fill_score=None) is None)
+        # is an outage: retry it, never let it quietly become a measured 0.0.
+        self._submit(tmp_path)
+        engine = MagicMock()
+        all_error = MagicMock(
+            result=MagicMock(
+                score=MagicMock(side_effect=lambda fill_score=0.0: None if fill_score is None else 0.0)
+            )
+        )
+        healthy = MagicMock(result=MagicMock(score=MagicMock(return_value=0.7)))
+        engine.evaluate_admin = AsyncMock(side_effect=[all_error, healthy])
+        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.7}
+        assert engine.evaluate_admin.await_count == 2