diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py index 0ec4024..7b6eede 100644 --- a/vero/src/vero/harbor/runner.py +++ b/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 diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py index ab02f79..3b90e66 100644 --- a/vero/tests/test_harbor_runner.py +++ b/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