fix: [no-ticket] harbor - scoring integrity (dead attempts, best-rank monotonicity, strict reward key)

vero/src/vero/harbor/runner.py

@@ -274,19 +274,26 @@ def _trial_groups(self, jobs_dir: Path) -> dict[str, list[dict]]:
             )
         return groups
 
-    @staticmethod
-    def _trial_rank(data: dict, result_json: Path) -> tuple:
+    def _trial_rank(self, data: dict, result_json: Path) -> tuple:
         """Sort key for picking the best of several trials of one task. Higher wins:
-        prefer a clean trial with rewards, then any trial with rewards, then the most
-        recent attempt (finished_at, falling back to file mtime)."""
-        has_rewards = bool((data.get("verifier_result") or {}).get("rewards"))
+        a clean scored trial first, then any scored trial, then the HIGHER REWARD,
+        then recency (finished_at, falling back to file mtime).
+
+        The reward must be part of the key: with concurrent attempts, finish order
+        is nondeterministic, so ranking on recency alone made 'best' mean "the last
+        clean attempt to finish", and a later clean 0.0 could silently replace an
+        earlier clean 1.0. 'best' has to be monotone in the attempt scores
+        (max-of-score, pass@k-like) or a passing trial can be clobbered."""
+        rewards = (data.get("verifier_result") or {}).get("rewards") or {}
+        reward = self._extract_reward(rewards) if rewards else None
+        has_rewards = reward is not None
         clean = has_rewards and not data.get("exception_info")
         finished_at = data.get("finished_at") or ""
         try:
             mtime = result_json.stat().st_mtime
         except OSError:
             mtime = 0.0
-        return (clean, has_rewards, finished_at, mtime)
+        return (clean, has_rewards, reward if has_rewards else -1.0, finished_at, mtime)
 
     def _sample_result(
         self,
@@ -315,51 +322,62 @@ def _out(output: dict) -> dict:
             if attempt_detail is not None:
                 output["attempts"] = attempt_detail
             return output
-        # Mean aggregation across attempts: average the reward over every SCORED
-        # attempt, dirty or clean. Harbor can record an exception (agent timeout,
-        # non-zero agent exit) and still run the verifier, so such an attempt
-        # carries a real measured 0.0; dropping it would estimate
-        # P(pass | attempt finished cleanly), which is non-monotone (one pass plus
-        # two timeouts would score 1.0) and systematically forgives candidates
-        # that make the agent slower. Only attempts with no rewards at all
-        # (failed before the verifier scored) are excluded. Falls through to the
-        # single best trial when nothing scored. `attempts` may also be present
-        # under 'best' aggregation (collation loads them for the feedback
-        # levers), so the mean path is gated on the config, not their presence.
+        # Mean aggregation across attempts: average the reward over every attempt
+        # that RAN. Harbor can record an exception (agent timeout, non-zero agent
+        # exit) and still run the verifier, so such an attempt carries a real
+        # measured 0.0. An attempt that died BEFORE the verifier scored it
+        # (crash, rate limit) is also a real, failed attempt and counts as 0.0:
+        # dropping it would estimate P(pass | attempt survived to scoring), which
+        # a candidate can game by dying early on hard tasks. Measured live: a
+        # no-retry candidate outscored its retry-hardened successors purely
+        # through dropped rate-limited attempts, and won selection on the
+        # artifact. n_dead in the metrics records how many zeros came from
+        # unscored attempts so infra noise stays visible. A sample where NO
+        # attempt scored falls through to the single-trial path (which errors):
+        # an all-dead sample is an outage to investigate, never a silent 0.0.
+        # `attempts` may also be present under 'best' aggregation (collation
+        # loads them for the feedback levers), so the mean path is gated on the
+        # config, not their presence.
         if attempts and self.config.aggregate_attempts == "mean":
-            scored_trials = [
-                t for t in attempts if (t.get("verifier_result") or {}).get("rewards")
-            ]
-            if scored_trials:
-                scored = [
-                    self._extract_reward((t.get("verifier_result") or {}).get("rewards"))
-                    for t in scored_trials
-                ]
-                n_clean = sum(
-                    1 for t in scored_trials if not t.get("exception_info")
-                )
-                if len(scored) < self.config.n_attempts:
-                    # Fewer measurements than configured (attempts died before
-                    # scoring, or the nested run was cut off): the mean is
-                    # noisier than the config promises. Never let k shrink
-                    # silently; n_scored in the metrics records the actual k.
+            measured: list[float] = []
+            n_scored = 0
+            n_dead = 0
+            n_clean = 0
+            for t in attempts:
+                rewards = (t.get("verifier_result") or {}).get("rewards") or {}
+                reward = self._extract_reward(rewards) if rewards else None
+                if reward is not None:
+                    measured.append(reward)
+                    n_scored += 1
+                    if not t.get("exception_info"):
+                        n_clean += 1
+                else:
+                    measured.append(0.0)
+                    n_dead += 1
+            if n_scored:
+                if len(measured) < self.config.n_attempts or n_dead:
+                    # Fewer or dirtier measurements than the config promises:
+                    # the mean is noisier (or partly zero-filled). Never let
+                    # that happen silently; the metrics carry the actual counts.
                     logger.warning(
-                        f"Task '{task_name}': mean over {len(scored)} scored "
-                        f"attempt(s) of {self.config.n_attempts} configured."
+                        f"Task '{task_name}': mean over {len(measured)} "
+                        f"attempt(s) of {self.config.n_attempts} configured "
+                        f"({n_scored} scored, {n_dead} dead counted 0.0)."
                     )
-                mean = sum(scored) / len(scored)
+                mean = sum(measured) / len(measured)
                 return SampleResult(
                     score=mean,
                     feedback=self._failure_feedback(mean, attempts),
                     metrics={
                         "reward_mean": mean,
                         "n_attempts": float(len(attempts)),
-                        "n_scored": float(len(scored)),
+                        "n_scored": float(n_scored),
+                        "n_dead": float(n_dead),
                         "n_clean": float(n_clean),
                     },
                     output=_out({
                         "task_name": task_name,
-                        "attempt_scores": scored,
+                        "attempt_scores": measured,
                         "aggregate": "mean",
                     }),
                     **common,
@@ -374,6 +392,21 @@ def _out(output: dict) -> dict:
                 **common,
             )
         score = self._extract_reward(rewards)
+        if score is None:
+            # The verifier scored, but not on the configured metric (or on
+            # several unrecognized ones). Scoring a substitute metric, or an
+            # average, would silently change what the number means: error loud.
+            return SampleResult(
+                error=(
+                    f"Rewards for task '{task_name}' carry no usable metric "
+                    f"(reward_key={self.config.reward_key!r}, "
+                    f"keys={sorted(rewards)})."
+                ),
+                output=_out(
+                    {"task_name": task_name, "trial_name": trial.get("trial_name")}
+                ),
+                **common,
+            )
         return SampleResult(
             score=score,
             feedback=self._failure_feedback(score, attempts),
@@ -386,12 +419,28 @@ def _out(output: dict) -> dict:
             **common,
         )
 
-    def _extract_reward(self, rewards: dict) -> float:
-        for key in (self.config.reward_key, "pass", "reward"):
-            if key and key in rewards:
+    def _extract_reward(self, rewards: dict) -> float | None:
+        """Reward for one trial's rewards dict, or None when no unambiguous
+        metric is present.
+
+        A configured reward_key is a contract: a rewards dict missing it is an
+        unscorable measurement (None), never a silent fallback to another key.
+        Falling back would let attempts within one mean be scored on different
+        metrics, and averaging arbitrary keys would let a candidate inflate its
+        score by emitting easy auxiliary metrics (lint, partial credit) beside
+        the real one. Without a configured key, 'pass' then 'reward' are
+        accepted, then a sole remaining key (unambiguous); several unrecognized
+        keys are refused (None), not averaged.
+        """
+        if self.config.reward_key:
+            value = rewards.get(self.config.reward_key)
+            return None if value is None else float(value)
+        for key in ("pass", "reward"):
+            if key in rewards:
                 return float(rewards[key])
-        values = [float(v) for v in rewards.values()]
-        return sum(values) / len(values) if values else 0.0
+        if len(rewards) == 1:
+            return float(next(iter(rewards.values())))
+        return None
 
     def _attempt_detail(self, attempts: list[dict] | None) -> list[dict] | None:
         """Lever 3: one {reward, exception} entry per attempt, in attempt order
@@ -437,8 +486,12 @@ def _failure_feedback(
             return None
         for attempt in attempts:
             rewards = (attempt.get("verifier_result") or {}).get("rewards")
-            if not rewards or self._extract_reward(rewards) != 0.0:
+            reward = self._extract_reward(rewards) if rewards else None
+            if reward is not None and reward != 0.0:
                 continue
+            # reward is None = the attempt died before the verifier scored it.
+            # It counts as 0.0 in mean aggregation, so its transcript (which
+            # shows the crash) is fair feedback material like any failure.
             trial_dir = attempt.get("_trial_dir")
             if not trial_dir:
                 continue

vero/tests/test_harbor_runner.py

@@ -101,15 +101,25 @@ def test_local_source(self, tmp_path):
 
 
 class TestExtractReward:
-    def test_priority_pass_then_reward_then_mean(self):
+    def test_priority_pass_then_reward_then_sole_key(self):
         r = _runner()
         assert r._extract_reward({"pass": 1.0, "reward": 0.0}) == 1.0
         assert r._extract_reward({"reward": 0.7}) == 0.7
-        assert r._extract_reward({"a": 0.2, "b": 0.4}) == pytest.approx(0.3)
+        assert r._extract_reward({"accuracy": 0.9}) == 0.9  # sole key: unambiguous
+
+    def test_several_unknown_keys_refused_not_averaged(self):
+        # Averaging arbitrary keys would let a candidate inflate its score by
+        # emitting easy auxiliary metrics beside the real one.
+        assert _runner()._extract_reward({"a": 0.2, "b": 0.4}) is None
 
     def test_reward_key_override(self):
         assert _runner(reward_key="acc")._extract_reward({"acc": 0.9, "pass": 0.0}) == 0.9
 
+    def test_configured_key_is_strict_no_fallback(self):
+        # A configured reward_key missing from the dict is unscorable (None),
+        # never a silent substitution of 'pass'/'reward'.
+        assert _runner(reward_key="acc")._extract_reward({"pass": 1.0}) is None
+
 
 class TestCollate:
     @pytest.mark.asyncio
@@ -302,12 +312,13 @@ def test_resume_with_nothing_ran_skips_guard(self, tmp_path, monkeypatch):
 
 
 class TestMeanAttemptAggregation:
-    """aggregate_attempts='mean': average the reward across every SCORED
-    attempt, dirty or clean (de-noising; estimates per-attempt pass
-    probability). Harbor scores timed-out attempts 0.0 while also recording
-    the exception; those must count, or the mean forgives slow candidates.
-    Default 'best' keeps the existing latest-clean behavior, which inflates
-    toward pass@k.
+    """aggregate_attempts='mean': average the reward across every attempt
+    that RAN (de-noising; estimates per-attempt pass probability). Harbor
+    scores timed-out attempts 0.0 while also recording the exception; those
+    must count, or the mean forgives slow candidates. Attempts that died
+    BEFORE scoring count 0.0 too (n_dead in metrics), or dying early becomes
+    a scoring exploit. Default 'best' picks the single highest-scoring clean
+    trial (pass@k-like).
     """
 
     def _write(self, run, trial, task, rewards=None, exc=False):
@@ -334,9 +345,11 @@ def test_mean_averages_clean_attempts(self, tmp_path):
         assert r.score == 0.5
         assert r.metrics["n_scored"] == 2.0
 
-    def test_mean_excludes_attempts_without_rewards(self, tmp_path):
-        # An attempt that died before the verifier scored it carries no
-        # measurement; it is excluded (but still counted in n_attempts).
+    def test_mean_zero_fills_attempts_without_rewards(self, tmp_path):
+        # An attempt that died before the verifier scored it is a real, failed
+        # attempt and counts 0.0. Excluding it would estimate
+        # P(pass | attempt survived), which rewards dying early on hard tasks:
+        # a live optimizer won selection on exactly that artifact.
         runner = HarborRunner(HarborConfig(
             task_source="org/ds", agent_import_path="p:m",
             n_attempts=2, aggregate_attempts="mean",
@@ -346,10 +359,38 @@ def test_mean_excludes_attempts_without_rewards(self, tmp_path):
         self._write(run, "t0bad", "t0", exc=True)
         groups = runner._trial_groups(jobs)
         r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"])
-        assert r.score == 1.0
+        assert r.score == 0.5
         assert r.metrics["n_scored"] == 1.0
+        assert r.metrics["n_dead"] == 1.0
         assert r.metrics["n_attempts"] == 2.0
 
+    def test_mean_all_attempts_dead_errors_not_zero(self, tmp_path):
+        # Every attempt died before scoring: that is an outage to investigate,
+        # not a silent 0.0 measurement; the sample must surface as an error.
+        runner = HarborRunner(HarborConfig(
+            task_source="org/ds", agent_import_path="p:m",
+            n_attempts=2, aggregate_attempts="mean",
+        ))
+        jobs = tmp_path / "jobs"; run = jobs / "2026-01-01__00-00-00"
+        self._write(run, "t0a", "t0", exc=True)
+        self._write(run, "t0b", "t0", exc=True)
+        groups = runner._trial_groups(jobs)
+        r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"])
+        assert r.error is not None
+        assert r.score is None
+
+    def test_best_rank_is_monotone_in_reward(self, tmp_path):
+        # 'best' must never let a later clean 0.0 clobber an earlier clean 1.0:
+        # the reward is part of the rank, recency only breaks ties.
+        runner = _runner()
+        jobs = tmp_path / "jobs"; run = jobs / "2026-01-01__00-00-00"
+        # _write derives finished_at from len(trial)%10: "t0a" -> 00:03,
+        # "t0badlate" -> 00:09, so the 0.0 trial genuinely finishes LATER.
+        self._write(run, "t0a", "t0", rewards={"reward": 1.0})
+        self._write(run, "t0badlate", "t0", rewards={"reward": 0.0})
+        trials = runner._load_trials(jobs)
+        assert (trials["t0"]["verifier_result"]["rewards"]["reward"]) == 1.0
+
     def test_mean_counts_scored_exception_attempts(self, tmp_path):
         # The live-GAIA shape: harbor records AgentTimeoutError but still runs
         # the verifier, so the attempt has BOTH exception_info and a scored 0.0.
@@ -482,7 +523,10 @@ def test_partial_k_mean_warns(self, tmp_path, caplog):
         with caplog.at_level("WARNING", logger="vero.harbor.runner"):
             r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"])
         assert r.metrics["n_scored"] == 2.0
-        assert any("2 scored attempt(s) of 3 configured" in m for m in caplog.messages)
+        assert any(
+            "2 attempt(s) of 3 configured (2 scored, 0 dead" in m
+            for m in caplog.messages
+        )
 
 
 def _fb_runner(**kwargs):
@@ -790,3 +834,24 @@ def test_flag_off_leaves_output_without_attempts(self, tmp_path):
         ))
         mean = self._result(mean_runner, jobs)
         assert "attempts" not in mean.output
+
+
+class TestMeanRewardKeyMismatch:
+    def test_mean_zero_fills_reward_key_mismatch(self, tmp_path):
+        # An attempt whose rewards LACK the configured key is unscorable on the
+        # configured metric and counts 0.0 in the mean (n_dead), exactly like
+        # dying pre-verifier: falling back to another key would score attempts
+        # within one mean on different metrics.
+        runner = HarborRunner(HarborConfig(
+            task_source="org/ds", agent_import_path="p:m",
+            n_attempts=2, aggregate_attempts="mean", reward_key="acc",
+        ))
+        jobs = tmp_path / "jobs"; run = jobs / "2026-01-01__00-00-00"
+        w = TestMeanAttemptAggregation()
+        w._write(run, "t0a", "t0", rewards={"acc": 1.0})
+        w._write(run, "t0b", "t0", rewards={"other": 1.0})
+        groups = runner._trial_groups(jobs)
+        r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"])
+        assert r.score == 0.5
+        assert r.metrics["n_dead"] == 1.0
+        assert r.metrics["n_scored"] == 1.0