fix: [no-ticket] harbor - honest measurement signals (summary qualifiers, versioned re-evals, dead-attempt causes)

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 {}

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,
             )

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.

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