fix: [no-ticket] harbor - floored rewards name their cause (candidate crash vs infra outage)

vero/src/vero/harbor/runner.py

@@ -409,8 +409,28 @@ 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 = ""
+            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"
+                    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})"
             return SampleResult(
-                error=f"No verifier rewards for task '{task_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

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
@@ -137,7 +138,7 @@ 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,
@@ -147,13 +148,18 @@ async def _finalize(self) -> dict:
             )
             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
@@ -172,16 +178,19 @@ async def _admin_eval_score(
         commit: str,
         sample_ids: list[int] | 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):
@@ -194,7 +203,10 @@ async def _admin_eval_score(
                     sample_ids=sample_ids,
                 )
                 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 +219,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 +230,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.
@@ -426,7 +465,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 +501,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,

vero/tests/test_harbor_runner.py

@@ -663,6 +663,20 @@ def test_no_rewards_error_sample_carries_crash_transcript(self, tmp_path):
         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.

vero/tests/test_harbor_verifier.py

@@ -755,6 +755,64 @@ 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_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