diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py index 5bd3754..f262261 100644 --- a/vero/src/vero/harbor/runner.py +++ b/vero/src/vero/harbor/runner.py @@ -353,6 +353,12 @@ def _out(output: dict) -> dict: n_scored = 0 n_dead = 0 n_clean = 0 + # Dead attempts are not interchangeable: a rate-limited attempt is + # infra noise that retunes with capacity, while a crash points at + # the candidate (or a task bug), and dead attempts cluster hard by + # cause in practice. n_dead alone hides that, so record the + # exception type behind every zero-filled attempt. + dead_types: dict[str, int] = {} for t in attempts: rewards = (t.get("verifier_result") or {}).get("rewards") or {} reward = self._extract_reward(rewards) if rewards else None @@ -364,6 +370,11 @@ 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" + dead_types[key] = dead_types.get(key, 0) + 1 if n_scored: if len(measured) < self.config.n_attempts or n_dead: # Fewer or dirtier measurements than the config promises: @@ -375,6 +386,14 @@ def _out(output: dict) -> dict: f"({n_scored} scored, {n_dead} dead counted 0.0)." ) mean = sum(measured) / len(measured) + mean_output = { + "task_name": task_name, + "attempt_scores": measured, + "aggregate": "mean", + } + if dead_types: + # dict, not metrics: metrics are float-valued by contract. + mean_output["dead_exception_types"] = dead_types return SampleResult( score=mean, feedback=self._failure_feedback(mean, attempts), @@ -385,11 +404,7 @@ def _out(output: dict) -> dict: "n_dead": float(n_dead), "n_clean": float(n_clean), }, - output=_out({ - "task_name": task_name, - "attempt_scores": measured, - "aggregate": "mean", - }), + output=_out(mean_output), **common, ) rewards = (trial.get("verifier_result") or {}).get("rewards") or {} diff --git a/vero/src/vero/harbor/server.py b/vero/src/vero/harbor/server.py index f74f0b3..d29b7a4 100644 --- a/vero/src/vero/harbor/server.py +++ b/vero/src/vero/harbor/server.py @@ -82,6 +82,7 @@ def __init__( # but at >= k times the sample budget per label. <= 1 disables the floor. self.k_anonymity_floor = k_anonymity_floor self._free_baseline_used = False + self._eval_seq = 0 # per-eval ordinal for result-dir versioning # ------------------------------------------------------------------ # Handlers (the HTTP layer resolves `admin` from auth and calls these) @@ -275,24 +276,52 @@ def _route_results(self, experiment: Experiment, *, admin: bool) -> str | None: return None commit = experiment.run.candidate.commit - dest = self.agent_volume / "results" / f"{split}__{commit[:12]}" - # Recreate the dir so it reflects exactly this metered run. The dir is keyed - # only on (split, commit[:12]); a prior eval of the same commit on a larger - # sample set would otherwise leave stale per-sample files behind that this - # run did not produce, and result_path would surface them as if they were. - if dest.exists(): + # Every metered eval gets its own versioned dir. Keying on + # (split, commit) alone forced a wipe-and-rewrite, so a re-measurement + # (a multifidelity confirm, a noise re-eval of the champion) erased the + # agent's earlier evidence for the same commit; repeat measurements are + # exactly the ones worth comparing. result_path in the response names + # the dir for THIS eval. + self._eval_seq += 1 + dest = ( + self.agent_volume + / "results" + / f"{split}__{commit[:12]}__e{self._eval_seq}" + ) + if dest.exists(): # ordinal collision only on volume reuse; never merge shutil.rmtree(dest) dest.mkdir(parents=True, exist_ok=True) # Aggregate summary is label-safe for both visible and partial tiers. + # n_scored / n_errored / score_se qualify the mean: a mean over 3 + # scored samples of 18, or one dominated by errored zero-fills, is a + # different measurement than a clean full-split mean, and the agent + # (and any auditor) should see that without per-sample access. + sample_results = experiment.result.sample_results + filled = [ + r.score if r.score is not None else 0.0 + for r in sample_results.values() + ] + score_se = None + if len(filled) > 1: + m = sum(filled) / len(filled) + var = sum((x - m) ** 2 for x in filled) / (len(filled) - 1) + score_se = (var / len(filled)) ** 0.5 (dest / "summary.json").write_text( json.dumps( { "split": split, "commit": commit, - "n_samples": len(experiment.result.sample_results), + "n_samples": len(sample_results), + "n_scored": sum( + 1 for r in sample_results.values() if r.score is not None + ), + "n_errored": sum( + 1 for r in sample_results.values() if r.is_error() + ), "mean_score": experiment.result.score(), - "status": str(experiment.result.status), + "score_se": score_se, + "status": experiment.result.status.value, }, indent=2, ) diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py index 4fe93db..0f53e65 100644 --- a/vero/tests/test_harbor_runner.py +++ b/vero/tests/test_harbor_runner.py @@ -382,6 +382,28 @@ def test_mean_zero_fills_attempts_without_rewards(self, tmp_path): assert r.metrics["n_dead"] == 1.0 assert r.metrics["n_attempts"] == 2.0 + def test_mean_records_dead_exception_types(self, tmp_path): + # n_dead alone hides WHY attempts died, and cause matters: rate-limit + # deaths are infra noise, crashes point at the candidate, and deaths + # cluster hard by cause (E1: 110/129 UnicodeDecodeErrors sat on two + # tasks). Every zero-filled attempt gets its exception type counted. + runner = HarborRunner(HarborConfig( + task_source="org/ds", agent_import_path="p:m", + n_attempts=3, aggregate_attempts="mean", + )) + jobs = tmp_path / "jobs"; run = jobs / "2026-01-01__00-00-00" + self._write(run, "t0a", "t0", rewards={"reward": 1.0}) + self._write(run, "t0bad", "t0", exc=True) # exception_type "X" + self._write(run, "t0gone", "t0") # no rewards AND no exception recorded + groups = runner._trial_groups(jobs) + r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"]) + assert r.output["dead_exception_types"] == {"X": 1, "no_rewards_recorded": 1} + # all-clean samples carry no key at all + self._write(run, "t1a", "t1", rewards={"reward": 1.0}) + groups = runner._trial_groups(jobs) + r1 = runner._sample_result(groups["t1"][0], 1, "t1", _params(), attempts=groups["t1"]) + assert "dead_exception_types" not in r1.output + 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. diff --git a/vero/tests/test_harbor_server.py b/vero/tests/test_harbor_server.py index 741365d..bafd4ee 100644 --- a/vero/tests/test_harbor_server.py +++ b/vero/tests/test_harbor_server.py @@ -1,6 +1,7 @@ """Tests for vero.harbor.server.EvaluationSidecar — handlers, tier-routing, submit.""" import json +from pathlib import Path from unittest.mock import AsyncMock, MagicMock import pytest @@ -83,7 +84,7 @@ async def test_visible_split_writes_full_per_sample(self, tmp_path): ) summary = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="train")) - dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456" + dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456__e1" assert (dest / "summary.json").exists() assert {(dest / f"{i}.json").exists() for i in range(3)} == {True} assert summary.result_path == str(dest) @@ -94,7 +95,7 @@ async def test_partial_split_writes_summary_only_no_labels(self, tmp_path): sidecar = _sidecar(tmp_path, split="validation") # non_viewable -> partial summary = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) - dest = tmp_path / "agent_vol" / "results" / "validation__abcdef123456" + dest = tmp_path / "agent_vol" / "results" / "validation__abcdef123456__e1" assert (dest / "summary.json").exists() # NO per-sample files -> the label-bearing feedback never reaches the agent assert not list(dest.glob("[0-9]*.json")) @@ -128,7 +129,7 @@ async def test_feedback_reaches_agent_on_viewable_only(self, tmp_path): tmp_path, split="train", accesses=[SplitAccess.viewable("train")] ) await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="train")) - dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456" + dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456__e1" assert "secret-0" in (dest / "0.json").read_text() @pytest.mark.asyncio @@ -172,7 +173,7 @@ async def test_attempts_reach_agent_on_viewable_only(self, tmp_path): return_value=self._experiment_with_attempts("train") ) await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="train")) - dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456" + dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456__e1" blob = json.loads((dest / "0.json").read_text()) assert blob["output"]["attempts"] == [ {"reward": 0.0, "exception": "SecretTimeoutError"} @@ -229,6 +230,40 @@ def test_status_reports_submit_and_splits(self, tmp_path): assert status.splits[0]["remaining_run_budget"] == 5 +class TestHonestSummary: + """summary.json must qualify its mean: how many samples actually scored, + how many errored, and the standard error — a mean over 3-of-18 scored + samples is a different measurement than a clean full-split mean.""" + + @pytest.mark.asyncio + async def test_summary_carries_qualifiers(self, tmp_path): + sidecar = _sidecar(tmp_path, split="validation") + s = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + data = json.loads((Path(s.result_path) / "summary.json").read_text()) + # scores are [0.0, 1.0, 0.0] + assert data["n_samples"] == 3 + assert data["n_scored"] == 3 + assert data["n_errored"] == 0 + assert data["mean_score"] == pytest.approx(1 / 3) + assert data["score_se"] == pytest.approx(1 / 3) # sd .5774 / sqrt(3) + # enum VALUE, not "ExperimentResultStatus.SUCCESS" + assert data["status"] == "success" + + @pytest.mark.asyncio + async def test_reevals_get_versioned_dirs(self, tmp_path): + # Repeat measurements of one commit are exactly the evidence worth + # comparing (multifidelity confirms, champion re-evals); the second + # eval must not erase the first. + sidecar = _sidecar(tmp_path, split="validation") + s1 = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + s2 = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + assert s1.result_path != s2.result_path + assert s1.result_path.endswith("__e1") + assert s2.result_path.endswith("__e2") + assert (Path(s1.result_path) / "summary.json").exists() + assert (Path(s2.result_path) / "summary.json").exists() + + class TestKAnonymityFloor: """Subset evals on non_viewable splits are floored: the aggregate response carries mean_score, so a singleton subset returns that sample's score @@ -327,7 +362,7 @@ async def test_first_baseline_eval_is_unmetered_but_not_admin(self, tmp_path): assert sidecar.engine.evaluate.await_args.kwargs["free"] is True assert sidecar.engine.evaluate.await_args.kwargs["admin"] is False # and results were routed with the agent tier (summary written) - dest = tmp_path / "agent_vol" / "results" / "validation__abcdef123456" + dest = tmp_path / "agent_vol" / "results" / "validation__abcdef123456__e1" assert (dest / "summary.json").exists() @pytest.mark.asyncio