From fc04cd64a38940a90d80e525c6fa44f528f8d97b Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Sun, 5 Jul 2026 13:23:24 +0300 Subject: [PATCH 1/4] feat(harbor): transcript feedback for failed viewable samples (feedback_transcripts) When enabled, collation attaches the last feedback_max_bytes bytes of a failed sample's trial transcript (agent/terminus_2.pane, falling back to agent/trajectory.json) to SampleResult.feedback: the first failed attempt only, passed samples carry nothing, missing transcripts are omitted silently. Exposure to the agent stays gated by the sidecar's tier routing (per-sample files are written only for viewable splits), so nothing can leak to non_viewable or no_access tiers. Plumbed build.yaml -> serve.json -> ServeConfig -> HarborRunner, mirroring score_baseline. Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/harbor/build/compiler.py | 2 + vero/src/vero/harbor/build/config.py | 6 ++ vero/src/vero/harbor/runner.py | 82 ++++++++++++++-- vero/src/vero/harbor/serve.py | 11 ++- vero/tests/test_harbor_build.py | 28 ++++++ vero/tests/test_harbor_runner.py | 131 ++++++++++++++++++++++++- vero/tests/test_harbor_serve.py | 18 ++++ vero/tests/test_harbor_server.py | 33 +++++++ 8 files changed, 296 insertions(+), 15 deletions(-) diff --git a/vero/src/vero/harbor/build/compiler.py b/vero/src/vero/harbor/build/compiler.py index 06c6c04..fbcdd9d 100644 --- a/vero/src/vero/harbor/build/compiler.py +++ b/vero/src/vero/harbor/build/compiler.py @@ -246,6 +246,8 @@ def _serve_config(config: BuildConfig, dataset_id: str | None, base_commit: str) "base_commit": base_commit, "submit_enabled": config.submit_enabled, "score_baseline": config.score_baseline, + "feedback_transcripts": config.feedback_transcripts, + "feedback_max_bytes": config.feedback_max_bytes, "agent_volume": AGENT_VOLUME, "admin_volume": ADMIN_VOLUME, "admin_token_path": TOKEN_PATH, diff --git a/vero/src/vero/harbor/build/config.py b/vero/src/vero/harbor/build/config.py index fcd6a84..974dede 100644 --- a/vero/src/vero/harbor/build/config.py +++ b/vero/src/vero/harbor/build/config.py @@ -70,6 +70,12 @@ class BuildConfig(BaseModel): # write it to /baseline.json, so a candidate that generalizes # WORSE than the untouched repo is visible as a regression. score_baseline: bool = False + # Lever 1 (Mode B): each FAILED sample (reward 0) of an eval carries the + # tail of its trial transcript in the per-sample `feedback` field. Rides + # the per-sample result files the sidecar writes ONLY for viewable splits, + # so it can never surface for non_viewable / no_access tiers. + feedback_transcripts: bool = False + feedback_max_bytes: int = 3000 # write-access: paths in the target repo the optimizer may NOT edit # (the scorer, by default). Applied as unix perms in main before the agent runs. diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py index c7916d6..ccd0557 100644 --- a/vero/src/vero/harbor/runner.py +++ b/vero/src/vero/harbor/runner.py @@ -35,8 +35,20 @@ class HarborRunner: """Mode-B EvalStrategy: nested `harbor run` + collate -> SampleResults.""" - def __init__(self, config: HarborConfig): + def __init__( + self, + config: HarborConfig, + *, + feedback_transcripts: bool = False, + feedback_max_bytes: int = 3000, + ): self.config = config + # Lever 1: attach the transcript tail of a FAILED sample's trial to its + # SampleResult.feedback. Whether the agent ever sees it is decided by + # the sidecar's tier routing (per-sample files are viewable-only), not + # here; this only controls whether the field is filled at collation. + self.feedback_transcripts = feedback_transcripts + self.feedback_max_bytes = feedback_max_bytes async def produce_sample_results( self, @@ -174,11 +186,10 @@ def _collate( f"canonical '/' form; refusing to score all " f"samples 0." ) - groups = ( - self._trial_groups(jobs_dir) - if self.config.aggregate_attempts == "mean" - else {} + need_attempts = ( + self.config.aggregate_attempts == "mean" or self.feedback_transcripts ) + groups = self._trial_groups(jobs_dir) if need_attempts else {} for sample_id, task_name in pairs: if self._is_done(params, sample_id): continue # already collated successfully (resume); errors are redone @@ -235,7 +246,16 @@ def _trial_groups(self, jobs_dir: Path) -> dict[str, list[dict]]: task_name = data.get("task_name") if not task_name: continue + # Transcripts (agent/terminus_2.pane etc.) live next to result.json; + # keep the dir so feedback can find them after the path is dropped. + data["_trial_dir"] = str(result_json.parent) groups.setdefault(task_name, []).append(data) + # rglob order is undefined; sort each group so "first attempt" is a + # stable notion (feedback uses the first failed attempt's transcript). + for attempts in groups.values(): + attempts.sort( + key=lambda d: (d.get("finished_at") or "", d.get("trial_name") or "") + ) return groups @staticmethod @@ -281,8 +301,10 @@ def _sample_result( # 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. - if attempts: + # 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. + if attempts and self.config.aggregate_attempts == "mean": scored_trials = [ t for t in attempts if (t.get("verifier_result") or {}).get("rewards") ] @@ -303,10 +325,12 @@ def _sample_result( f"Task '{task_name}': mean over {len(scored)} scored " f"attempt(s) of {self.config.n_attempts} configured." ) + mean = sum(scored) / len(scored) return SampleResult( - score=sum(scored) / len(scored), + score=mean, + feedback=self._failure_feedback(mean, attempts), metrics={ - "reward_mean": sum(scored) / len(scored), + "reward_mean": mean, "n_attempts": float(len(attempts)), "n_scored": float(len(scored)), "n_clean": float(n_clean), @@ -325,8 +349,10 @@ def _sample_result( output={"task_name": task_name, "trial_name": trial.get("trial_name")}, **common, ) + score = self._extract_reward(rewards) return SampleResult( - score=self._extract_reward(rewards), + score=score, + feedback=self._failure_feedback(score, attempts), metrics={k: float(v) for k, v in rewards.items()}, output={ "task_name": task_name, @@ -343,6 +369,42 @@ def _extract_reward(self, rewards: dict) -> float: values = [float(v) for v in rewards.values()] return sum(values) / len(values) if values else 0.0 + def _failure_feedback( + self, score: float, attempts: list[dict] | None + ) -> str | None: + """Lever 1: transcript tail for a failed sample (score 0.0). + + Uses the FIRST failed attempt only (attempts are sorted at load): the + first failure is the cheapest reproducible one, and one tail per sample + bounds the payload. Passed samples, and everything with the lever off, + return None (the field serializes as null either way, so responses are + byte-identical to before when disabled). + """ + if not self.feedback_transcripts or score != 0.0 or not attempts: + return None + for attempt in attempts: + rewards = (attempt.get("verifier_result") or {}).get("rewards") + if not rewards or self._extract_reward(rewards) != 0.0: + continue + trial_dir = attempt.get("_trial_dir") + if not trial_dir: + return None + return self._read_transcript_tail(Path(trial_dir)) + return None + + def _read_transcript_tail(self, trial_dir: Path) -> str | None: + """Last ``feedback_max_bytes`` of the trial's transcript: the terminal + pane when present, else the trajectory; None (field omitted) when the + trial recorded neither.""" + for rel in ("agent/terminus_2.pane", "agent/trajectory.json"): + path = trial_dir / rel + try: + data = path.read_bytes() + except OSError: + continue + return data[-self.feedback_max_bytes :].decode("utf-8", errors="replace") + return None + def _existing(self, params: EvaluationParameters, sample_id: int) -> SampleResult | None: return load_sample_result( get_vero_home_dir() / "sessions", diff --git a/vero/src/vero/harbor/serve.py b/vero/src/vero/harbor/serve.py index 962805a..7989df3 100644 --- a/vero/src/vero/harbor/serve.py +++ b/vero/src/vero/harbor/serve.py @@ -77,6 +77,11 @@ class ServeConfig(BaseModel): # auto_best never ships a candidate that fails to beat the untouched baseline # on the selection split; it reverts to base_commit instead (needs base_commit). auto_best_baseline_floor: bool = True + # Lever 1 (Mode B): failed samples carry their trial-transcript tail in the + # per-sample `feedback` field. Exposure stays gated by the sidecar's tier + # routing (per-sample files are written only for viewable splits). + feedback_transcripts: bool = False + feedback_max_bytes: int = 3000 # volumes / token agent_volume: str @@ -189,7 +194,11 @@ async def build_components(config: ServeConfig) -> tuple[EvaluationSidecar, Veri from vero.harbor.runner import HarborRunner from vero.harbor.config import HarborConfig - eval_strategy = HarborRunner(HarborConfig(**config.harbor)) + eval_strategy = HarborRunner( + HarborConfig(**config.harbor), + feedback_transcripts=config.feedback_transcripts, + feedback_max_bytes=config.feedback_max_bytes, + ) evaluator = Evaluator( workspace, diff --git a/vero/tests/test_harbor_build.py b/vero/tests/test_harbor_build.py index 089921c..d366fdc 100644 --- a/vero/tests/test_harbor_build.py +++ b/vero/tests/test_harbor_build.py @@ -155,6 +155,34 @@ def test_score_baseline_true_through_compile_task(tmp_path, monkeypatch): assert raw["score_baseline"] is True +def test_feedback_transcripts_reach_serve_json(built): + # Lever 1 plumbing: the flags must be in the compiler <-> serve contract + # (default off) and validate through ServeConfig. + raw = json.loads((built / "environment" / "sidecar" / "serve.json").read_text()) + assert raw["feedback_transcripts"] is False + assert raw["feedback_max_bytes"] == 3000 + cfg = ServeConfig.from_file(built / "environment" / "sidecar" / "serve.json") + assert cfg.feedback_transcripts is False + assert cfg.feedback_max_bytes == 3000 + + +def test_feedback_transcripts_configured_through_yaml(): + # Through the actual YAML path, mirroring the score_baseline exemplar. + from vero.harbor.build.compiler import _serve_config + + config = BuildConfig.model_validate(yaml.safe_load( + "name: o/n\n" + "agent_repo: .\n" + "splits:\n" + " - {split: validation, access: viewable}\n" + "feedback_transcripts: true\n" + "feedback_max_bytes: 512\n" + )) + raw = _serve_config(config, "ds", "sha") + assert raw["feedback_transcripts"] is True + assert raw["feedback_max_bytes"] == 512 + + def test_rendered_files_parse(built): tomllib.loads((built / "task.toml").read_text()) # valid TOML compose = yaml.safe_load((built / "environment/docker-compose.yaml").read_text()) diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py index c18ee6d..1613aec 100644 --- a/vero/tests/test_harbor_runner.py +++ b/vero/tests/test_harbor_runner.py @@ -44,16 +44,34 @@ def _params(): ) -def _write_trial(jobs_dir: Path, trial: str, task_name: str, rewards: dict): +def _write_trial( + jobs_dir: Path, + trial: str, + task_name: str, + rewards: dict, + *, + pane: str | None = None, + trajectory: str | None = None, + finished_at: str | None = None, +): # Real harbor layout: ///result.json, plus a job-level # //result.json summary (no task_name) that collation must skip. + # Transcripts (when present) live at /agent/terminus_2.pane and + # /agent/trajectory.json, next to result.json. run = jobs_dir / "2026-01-01__00-00-00" d = run / trial d.mkdir(parents=True, exist_ok=True) (run / "result.json").write_text(json.dumps({"job": "summary"})) # job-level, no task_name - (d / "result.json").write_text( - json.dumps({"task_name": task_name, "trial_name": trial, "verifier_result": {"rewards": rewards}}) - ) + data = {"task_name": task_name, "trial_name": trial, "verifier_result": {"rewards": rewards}} + if finished_at is not None: + data["finished_at"] = finished_at + (d / "result.json").write_text(json.dumps(data)) + if pane is not None: + (d / "agent").mkdir(exist_ok=True) + (d / "agent" / "terminus_2.pane").write_text(pane) + if trajectory is not None: + (d / "agent").mkdir(exist_ok=True) + (d / "agent" / "trajectory.json").write_text(trajectory) class TestBuildCommand: @@ -454,3 +472,108 @@ def test_partial_k_mean_warns(self, tmp_path, caplog): 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) + + +def _fb_runner(**kwargs): + return HarborRunner( + HarborConfig(task_source="org/ds@1", agent_import_path="pkg.mod:Agent"), + feedback_transcripts=True, + **kwargs, + ) + + +class TestTranscriptFeedback: + """Lever 1 (feedback_transcripts): a FAILED sample (reward 0) carries the + tail of its trial transcript in SampleResult.feedback. Population rules are + tested here; the hidden-split gate (per-sample files are viewable-only) is + the sidecar's and is covered in test_harbor_server.""" + + def _result(self, runner, jobs, task="t0"): + trials = runner._load_trials(jobs) + groups = runner._trial_groups(jobs) + return runner._sample_result( + trials.get(task), 0, task, _params(), attempts=groups.get(task) + ) + + @pytest.mark.asyncio + async def test_failed_carries_pane_tail_passed_does_not(self, tmp_path, monkeypatch): + monkeypatch.setenv("VERO_HOME_DIR", str(tmp_path / "vh")) + runner = _fb_runner() + params = _params() + result_dir = tmp_path / "result" + _write_trial(result_dir / "jobs", "trial0", "t0", {"reward": 0.0}, pane="failing tail") + _write_trial(result_dir / "jobs", "trial1", "t1", {"reward": 1.0}, pane="passing tail") + monkeypatch.setattr(runner, "_task_names_for", lambda p: [(0, "t0"), (1, "t1")]) + runner._run_harbor = AsyncMock() + ws = MagicMock(project_path="/wt") + await runner.produce_sample_results(workspace=ws, params=params, result_dir=result_dir) + results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id) + assert results[0].score == 0.0 + assert results[0].feedback == "failing tail" + assert results[1].score == 1.0 + assert results[1].feedback is None # passed samples carry no feedback + + def test_flag_off_leaves_feedback_unset(self, tmp_path): + runner = _runner() # default: feedback_transcripts=False + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}, pane="failing tail") + assert self._result(runner, jobs).feedback is None + + def test_byte_cap_keeps_last_bytes_only(self, tmp_path): + runner = _fb_runner(feedback_max_bytes=16) + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}, pane="A" * 100 + "TAIL-OF-THE-PANE") + r = self._result(runner, jobs) + assert r.feedback == "TAIL-OF-THE-PANE" + assert len(r.feedback.encode()) <= 16 + + def test_falls_back_to_trajectory_when_pane_missing(self, tmp_path): + runner = _fb_runner() + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}, trajectory='{"steps": []}') + assert self._result(runner, jobs).feedback == '{"steps": []}' + + def test_missing_transcripts_omitted_silently(self, tmp_path): + runner = _fb_runner() + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}) # no pane, no trajectory + r = self._result(runner, jobs) + assert r.score == 0.0 + assert r.feedback is None + + 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. + runner = HarborRunner( + HarborConfig( + task_source="org/ds@1", agent_import_path="pkg.mod:Agent", + n_attempts=2, aggregate_attempts="mean", + ), + feedback_transcripts=True, + ) + jobs = tmp_path / "jobs" + _write_trial(jobs, "zz-early", "t0", {"reward": 0.0}, pane="first attempt", + finished_at="2026-01-01T00:01:00") + _write_trial(jobs, "aa-late", "t0", {"reward": 0.0}, pane="second attempt", + finished_at="2026-01-01T00:09:00") + r = self._result(runner, jobs) + assert r.score == 0.0 + assert r.feedback == "first attempt" + + def test_partially_passing_mean_sample_gets_no_feedback(self, tmp_path): + # Failed means reward 0; a mean of [1.0, 0.0] is not a failed sample. + runner = HarborRunner( + HarborConfig( + task_source="org/ds@1", agent_import_path="pkg.mod:Agent", + n_attempts=2, aggregate_attempts="mean", + ), + feedback_transcripts=True, + ) + jobs = tmp_path / "jobs" + _write_trial(jobs, "a", "t0", {"reward": 1.0}, pane="p1", + finished_at="2026-01-01T00:01:00") + _write_trial(jobs, "b", "t0", {"reward": 0.0}, pane="p2", + finished_at="2026-01-01T00:02:00") + r = self._result(runner, jobs) + assert r.score == 0.5 + assert r.feedback is None diff --git a/vero/tests/test_harbor_serve.py b/vero/tests/test_harbor_serve.py index 655c389..49c8d57 100644 --- a/vero/tests/test_harbor_serve.py +++ b/vero/tests/test_harbor_serve.py @@ -221,6 +221,24 @@ async def test_ledger_reloads_spent_budget_across_restart(fixture): assert reloaded == after, "sidecar restart must not refill spent budget" +@pytest.mark.asyncio +async def test_feedback_levers_reach_harbor_runner(fixture): + # Lever 1 pass-through: ServeConfig -> build_components -> HarborRunner kwargs + # (mirrors how score_baseline reaches the Verifier). + agent_dir, head, task_dir, dataset_id, tmp = fixture + config = _serve_config(agent_dir, head, task_dir, dataset_id, tmp).model_copy( + update={ + "harbor": {"task_source": "org/x", "agent_import_path": "p:C"}, + "feedback_transcripts": True, + "feedback_max_bytes": 512, + } + ) + sidecar, _, _ = await build_components(config) + runner = sidecar.engine.evaluator.eval_strategy + assert runner.feedback_transcripts is True + assert runner.feedback_max_bytes == 512 + + def test_mode_b_sample_timeout_warns(caplog): # Setting sample_timeout in Mode B is a no-op (nested harbor tasks use their # own timeouts); the author must be told rather than silently ignored. diff --git a/vero/tests/test_harbor_server.py b/vero/tests/test_harbor_server.py index dadb1c7..df987d8 100644 --- a/vero/tests/test_harbor_server.py +++ b/vero/tests/test_harbor_server.py @@ -105,6 +105,39 @@ async def test_admin_eval_writes_nothing_to_agent_volume(self, tmp_path): assert summary.budget_remaining is None +class TestFeedbackTierGate: + """Lever 1 hidden-split safety: per-sample feedback (transcript tails) rides + the per-sample result files, and _route_results writes those ONLY for + viewable splits. Nothing feedback-bearing may ever land on the agent volume + for a non_viewable or no_access split, regardless of any collation flag.""" + + @pytest.mark.asyncio + async def test_feedback_reaches_agent_on_viewable_only(self, tmp_path): + sidecar = _sidecar( + 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" + assert "secret-0" in (dest / "0.json").read_text() + + @pytest.mark.asyncio + async def test_feedback_never_written_for_non_viewable(self, tmp_path): + sidecar = _sidecar(tmp_path, split="validation") # non_viewable + await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + for f in (tmp_path / "agent_vol").rglob("*"): + if f.is_file(): + blob = f.read_text() + assert "secret-" not in blob + assert "feedback" not in blob + + @pytest.mark.asyncio + async def test_feedback_never_written_for_no_access(self, tmp_path): + sidecar = _sidecar(tmp_path, split="test") # no_access + await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="test")) + agent_vol = tmp_path / "agent_vol" + assert not agent_vol.exists() or not list(agent_vol.rglob("*.json")) + + class TestSubmit: @pytest.mark.asyncio async def test_submit_records_nomination(self, tmp_path): From 261ef48f0519c8633a4983c0d62c0c82a4f76d36 Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Sun, 5 Jul 2026 13:25:44 +0300 Subject: [PATCH 2/4] feat(harbor): instruction teaches multi-fidelity screening (instruct_multifidelity) When enabled, the compiled instruction gains a section teaching the optimizer to triage rough ideas on subset evals (--num-samples / --sample-ids) and spend full-split evals only on survivors, stating the true economics: every eval debits one run-budget unit regardless of size, while the sample budget is debited only for the samples actually run, and subset aggregates are noisier. The section is gated on introspecting EvalRequest for the subset-eval fields (the same merge-order-truthfulness pattern as the free-baseline bullet), so it can never render against a sidecar that lacks subset evals. Plumbed build.yaml -> serve.json -> ServeConfig for contract parity; consumption is compile-time. Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/harbor/build/compiler.py | 9 ++ vero/src/vero/harbor/build/config.py | 5 ++ .../harbor/build/templates/instruction.md.j2 | 17 ++++ vero/src/vero/harbor/serve.py | 5 ++ vero/tests/test_harbor_build.py | 90 +++++++++++++++++++ 5 files changed, 126 insertions(+) diff --git a/vero/src/vero/harbor/build/compiler.py b/vero/src/vero/harbor/build/compiler.py index fbcdd9d..3a269aa 100644 --- a/vero/src/vero/harbor/build/compiler.py +++ b/vero/src/vero/harbor/build/compiler.py @@ -17,6 +17,7 @@ from jinja2 import Environment, FileSystemLoader +from vero.evaluation.engine import EvalRequest from vero.harbor.build.config import BuildConfig from vero.harbor.protocol import StatusSummary @@ -248,6 +249,7 @@ def _serve_config(config: BuildConfig, dataset_id: str | None, base_commit: str) "score_baseline": config.score_baseline, "feedback_transcripts": config.feedback_transcripts, "feedback_max_bytes": config.feedback_max_bytes, + "instruct_multifidelity": config.instruct_multifidelity, "agent_volume": AGENT_VOLUME, "admin_volume": ADMIN_VOLUME, "admin_token_path": TOKEN_PATH, @@ -360,6 +362,13 @@ def compile_task( # instruction truthful under any merge order. free_baseline="free_baseline_available" in {f.name for f in dataclasses.fields(StatusSummary)}, + # Same merge-order-truthfulness introspection for the multi-fidelity + # section: it may only render when the sidecar shipping in this tree + # actually accepts subset evals (sample_ids / num_samples on the eval + # request), or the instruction would teach a knob that 400s. + multifidelity=config.instruct_multifidelity + and {"sample_ids", "num_samples"} + <= {f.name for f in dataclasses.fields(EvalRequest)}, ) _render(jenv, "task.toml.j2", out / "task.toml", **ctx) _render(jenv, "instruction.md.j2", out / "instruction.md", **ctx) diff --git a/vero/src/vero/harbor/build/config.py b/vero/src/vero/harbor/build/config.py index 974dede..43b770a 100644 --- a/vero/src/vero/harbor/build/config.py +++ b/vero/src/vero/harbor/build/config.py @@ -76,6 +76,11 @@ class BuildConfig(BaseModel): # so it can never surface for non_viewable / no_access tiers. feedback_transcripts: bool = False feedback_max_bytes: int = 3000 + # Lever 2: the compiled instruction teaches multi-fidelity screening (triage + # rough ideas on subset evals via num_samples / sample_ids, confirm survivors + # on the full split). Renders only when the sidecar in the same tree actually + # accepts subset evals; see the compiler's ctx gate. + instruct_multifidelity: bool = False # write-access: paths in the target repo the optimizer may NOT edit # (the scorer, by default). Applied as unix perms in main before the agent runs. diff --git a/vero/src/vero/harbor/build/templates/instruction.md.j2 b/vero/src/vero/harbor/build/templates/instruction.md.j2 index 441de7e..df25032 100644 --- a/vero/src/vero/harbor/build/templates/instruction.md.j2 +++ b/vero/src/vero/harbor/build/templates/instruction.md.j2 @@ -22,6 +22,23 @@ scored on the hidden test split at the end. Only commits *other than the seeded baseline* are selectable: baseline evals create no candidate, so make sure at least one eval is of a commit that contains your changes.{% endif %} +{% if multifidelity %} +## Screen cheaply, confirm expensively + +`vero harbor eval` also accepts `--num-samples N` (the first N samples of the +split) or `--sample-ids 0,3,7` (specific samples). A subset eval finishes faster +in rough proportion to its size, so it is the cheap way to triage ideas: + +- Cost: every eval, subset or full, spends 1 unit of the split's run budget. + The split's sample budget (when it has one) is debited only for the samples + actually run, so subsets stretch a sample-metered budget much further. +- Noise: a subset aggregate is noisier than a full-split score. Treat subset + results as a coarse filter, not a ranking. +- Strategy: screen rough ideas on a small fixed subset (the same sample ids + every time, so scores stay comparable), then spend full-split evals only on + the survivors. The final selection compares full-split scores. + +{% endif %} ## Rules - Budget is finite and metered per split — spend it wisely. diff --git a/vero/src/vero/harbor/serve.py b/vero/src/vero/harbor/serve.py index 7989df3..e63e04b 100644 --- a/vero/src/vero/harbor/serve.py +++ b/vero/src/vero/harbor/serve.py @@ -82,6 +82,11 @@ class ServeConfig(BaseModel): # routing (per-sample files are written only for viewable splits). feedback_transcripts: bool = False feedback_max_bytes: int = 3000 + # Lever 2: consumed at COMPILE time (the instruction's multi-fidelity + # section); recorded here so serve.json mirrors build.yaml. The sidecar's + # subset-eval support itself is unconditional (EvalRequest.num_samples / + # sample_ids), so there is nothing to toggle at serve time. + instruct_multifidelity: bool = False # volumes / token agent_volume: str diff --git a/vero/tests/test_harbor_build.py b/vero/tests/test_harbor_build.py index d366fdc..bebe0fa 100644 --- a/vero/tests/test_harbor_build.py +++ b/vero/tests/test_harbor_build.py @@ -13,6 +13,7 @@ import pytest import yaml +from vero.evaluation.engine import EvalRequest from vero.harbor.build import BuildConfig, compile_task from vero.harbor.protocol import StatusSummary from vero.harbor.serve import ServeConfig @@ -24,6 +25,13 @@ f.name for f in dataclasses.fields(StatusSummary) } +# Whether the sidecar in THIS tree accepts subset evals (num_samples / +# sample_ids on the eval request); the multi-fidelity instruction section is +# gated on it, same merge-order pattern as the free-baseline bullet. +_HAS_SUBSET_EVALS = {"sample_ids", "num_samples"} <= { + f.name for f in dataclasses.fields(EvalRequest) +} + def _stub_vero(root: Path) -> Path: """A minimal stand-in for the vero source tree (compiler just copies it).""" @@ -285,6 +293,88 @@ def test_instruction_omits_free_baseline_claim_when_unsupported(built): assert "budget-free" not in text +def _multifidelity_config(tmp_path) -> BuildConfig: + return BuildConfig( + name="vero/gsm8k-opt", + agent_repo=str(_agent_repo(tmp_path)), + mode="A", + task="gsm8k", + dataset=str(_dataset(tmp_path)), + splits=[{"split": "validation", "access": "non_viewable"}], + instruct_multifidelity=True, + ) + + +@pytest.mark.skipif( + not _HAS_SUBSET_EVALS, reason="sidecar in this tree has no subset evals" +) +def test_instruction_teaches_multifidelity_when_enabled(tmp_path, monkeypatch): + monkeypatch.setenv("VERO_SKIP_SECRET_CHECK", "1") + out = compile_task( + _multifidelity_config(tmp_path), tmp_path / "task", vero_root=_stub_vero(tmp_path) + ) + text = (out / "instruction.md").read_text() + assert "--num-samples" in text + assert "--sample-ids" in text + # ...and it must state the true budget economics: a subset eval still costs + # a full run-budget unit, while the sample budget scales with subset size. + assert "1 unit of the split's run budget" in text + assert "debited only for the samples" in text + assert "noisier" in text + + +def test_instruction_omits_multifidelity_by_default(built): + text = (built / "instruction.md").read_text() + assert "--num-samples" not in text + assert "Screen cheaply" not in text + + +def test_multifidelity_gate_suppresses_section_without_subset_evals(tmp_path, monkeypatch): + # Merge-order guard, exercised the way the free-baseline gate is designed: + # against a sidecar whose EvalRequest lacks subset-eval fields, the section + # must not render even when the build flag asks for it, or the instruction + # would teach a knob the eval endpoint rejects. + import dataclasses as dc + + from vero.harbor.build import compiler + + @dc.dataclass + class _LegacyEvalRequest: + dataset_id: str + split: str + + monkeypatch.setenv("VERO_SKIP_SECRET_CHECK", "1") + monkeypatch.setattr(compiler, "EvalRequest", _LegacyEvalRequest) + out = compile_task( + _multifidelity_config(tmp_path), tmp_path / "task", vero_root=_stub_vero(tmp_path) + ) + text = (out / "instruction.md").read_text() + assert "--num-samples" not in text + assert "Screen cheaply" not in text + + +def test_instruct_multifidelity_reaches_serve_json(built): + raw = json.loads((built / "environment" / "sidecar" / "serve.json").read_text()) + assert raw["instruct_multifidelity"] is False # default off + assert ServeConfig.from_file( + built / "environment" / "sidecar" / "serve.json" + ).instruct_multifidelity is False + + +def test_instruct_multifidelity_configured_through_yaml(): + from vero.harbor.build.compiler import _serve_config + + config = BuildConfig.model_validate(yaml.safe_load( + "name: o/n\n" + "agent_repo: .\n" + "splits:\n" + " - {split: validation, access: non_viewable}\n" + "instruct_multifidelity: true\n" + )) + raw = _serve_config(config, "ds", "sha") + assert raw["instruct_multifidelity"] is True + + def test_instruction_tells_agent_to_spend_whole_budget(built): # Two live runs ended with nearly half the eval budget unspent; the # instruction must state that unspent evals are wasted and re-measurement From 8e1d1e91a6cc31f7a42f05919c38b22beaf28d97 Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Sun, 5 Jul 2026 13:32:49 +0300 Subject: [PATCH 3/4] feat(harbor): per-attempt reward and exception detail (expose_attempt_detail) When enabled, each collated sample's output carries an attempts list, one {reward, exception} entry per attempt in stable attempt order: reward is None when the attempt died before the verifier scored it, exception is the recorded exception class name (None for clean attempts). Collation now loads the attempt groups under 'best' aggregation too when a lever needs them, without touching best-mode scoring. Exposure rides the same viewable-only per-sample files as transcript feedback, so nothing reaches non_viewable or no_access tiers. Plumbed build.yaml -> serve.json -> ServeConfig -> HarborRunner, mirroring score_baseline. Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/harbor/build/compiler.py | 1 + vero/src/vero/harbor/build/config.py | 4 ++ vero/src/vero/harbor/runner.py | 48 +++++++++++++-- vero/src/vero/harbor/serve.py | 4 ++ vero/tests/test_harbor_build.py | 21 +++++++ vero/tests/test_harbor_runner.py | 82 +++++++++++++++++++++++++- vero/tests/test_harbor_serve.py | 2 + vero/tests/test_harbor_server.py | 43 ++++++++++++++ 8 files changed, 197 insertions(+), 8 deletions(-) diff --git a/vero/src/vero/harbor/build/compiler.py b/vero/src/vero/harbor/build/compiler.py index 3a269aa..4227dae 100644 --- a/vero/src/vero/harbor/build/compiler.py +++ b/vero/src/vero/harbor/build/compiler.py @@ -250,6 +250,7 @@ def _serve_config(config: BuildConfig, dataset_id: str | None, base_commit: str) "feedback_transcripts": config.feedback_transcripts, "feedback_max_bytes": config.feedback_max_bytes, "instruct_multifidelity": config.instruct_multifidelity, + "expose_attempt_detail": config.expose_attempt_detail, "agent_volume": AGENT_VOLUME, "admin_volume": ADMIN_VOLUME, "admin_token_path": TOKEN_PATH, diff --git a/vero/src/vero/harbor/build/config.py b/vero/src/vero/harbor/build/config.py index 43b770a..9e23af6 100644 --- a/vero/src/vero/harbor/build/config.py +++ b/vero/src/vero/harbor/build/config.py @@ -81,6 +81,10 @@ class BuildConfig(BaseModel): # on the full split). Renders only when the sidecar in the same tree actually # accepts subset evals; see the compiler's ctx gate. instruct_multifidelity: bool = False + # Lever 3 (Mode B): each sample's output carries an `attempts` list, one + # {reward, exception} entry per attempt. Same viewable-only exposure as + # feedback_transcripts. + expose_attempt_detail: bool = False # write-access: paths in the target repo the optimizer may NOT edit # (the scorer, by default). Applied as unix perms in main before the agent runs. diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py index ccd0557..a0a57c2 100644 --- a/vero/src/vero/harbor/runner.py +++ b/vero/src/vero/harbor/runner.py @@ -41,6 +41,7 @@ def __init__( *, feedback_transcripts: bool = False, feedback_max_bytes: int = 3000, + expose_attempt_detail: bool = False, ): self.config = config # Lever 1: attach the transcript tail of a FAILED sample's trial to its @@ -49,6 +50,10 @@ def __init__( # here; this only controls whether the field is filled at collation. self.feedback_transcripts = feedback_transcripts self.feedback_max_bytes = feedback_max_bytes + # Lever 3: attach a per-attempt {reward, exception} list to each + # sample's output. Same tier gate as feedback: filled at collation, + # exposed only via the viewable-split per-sample files. + self.expose_attempt_detail = expose_attempt_detail async def produce_sample_results( self, @@ -187,7 +192,9 @@ def _collate( f"samples 0." ) need_attempts = ( - self.config.aggregate_attempts == "mean" or self.feedback_transcripts + self.config.aggregate_attempts == "mean" + or self.feedback_transcripts + or self.expose_attempt_detail ) groups = self._trial_groups(jobs_dir) if need_attempts else {} for sample_id, task_name in pairs: @@ -293,6 +300,12 @@ def _sample_result( return SampleResult( error=f"No Harbor trial result for task '{task_name}'.", **common ) + attempt_detail = self._attempt_detail(attempts) + + 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 @@ -335,18 +348,20 @@ def _sample_result( "n_scored": float(len(scored)), "n_clean": float(n_clean), }, - output={ + output=_out({ "task_name": task_name, "attempt_scores": scored, "aggregate": "mean", - }, + }), **common, ) rewards = (trial.get("verifier_result") or {}).get("rewards") or {} if not rewards: return SampleResult( error=f"No verifier rewards for task '{task_name}'.", - output={"task_name": task_name, "trial_name": trial.get("trial_name")}, + output=_out( + {"task_name": task_name, "trial_name": trial.get("trial_name")} + ), **common, ) score = self._extract_reward(rewards) @@ -354,11 +369,11 @@ def _sample_result( score=score, feedback=self._failure_feedback(score, attempts), metrics={k: float(v) for k, v in rewards.items()}, - output={ + output=_out({ "task_name": task_name, "trial_name": trial.get("trial_name"), "rewards": rewards, - }, + }), **common, ) @@ -369,6 +384,27 @@ def _extract_reward(self, rewards: dict) -> float: values = [float(v) for v in rewards.values()] return sum(values) / len(values) if values else 0.0 + def _attempt_detail(self, attempts: list[dict] | None) -> list[dict] | None: + """Lever 3: one {reward, exception} entry per attempt, in attempt order + (sorted at load). reward is None when the attempt died before the + verifier scored it; exception is the recorded exception class name + (None for clean attempts). Off (or no attempts loaded) returns None, + which leaves the output dict without an 'attempts' key at all.""" + if not self.expose_attempt_detail or not attempts: + return None + detail = [] + for attempt in attempts: + rewards = (attempt.get("verifier_result") or {}).get("rewards") + detail.append( + { + "reward": self._extract_reward(rewards) if rewards else None, + "exception": (attempt.get("exception_info") or {}).get( + "exception_type" + ), + } + ) + return detail + def _failure_feedback( self, score: float, attempts: list[dict] | None ) -> str | None: diff --git a/vero/src/vero/harbor/serve.py b/vero/src/vero/harbor/serve.py index e63e04b..5cbd626 100644 --- a/vero/src/vero/harbor/serve.py +++ b/vero/src/vero/harbor/serve.py @@ -82,6 +82,9 @@ class ServeConfig(BaseModel): # routing (per-sample files are written only for viewable splits). feedback_transcripts: bool = False feedback_max_bytes: int = 3000 + # Lever 3 (Mode B): sample output carries a per-attempt {reward, exception} + # list. Same viewable-only exposure path as feedback_transcripts. + expose_attempt_detail: bool = False # Lever 2: consumed at COMPILE time (the instruction's multi-fidelity # section); recorded here so serve.json mirrors build.yaml. The sidecar's # subset-eval support itself is unconditional (EvalRequest.num_samples / @@ -203,6 +206,7 @@ async def build_components(config: ServeConfig) -> tuple[EvaluationSidecar, Veri HarborConfig(**config.harbor), feedback_transcripts=config.feedback_transcripts, feedback_max_bytes=config.feedback_max_bytes, + expose_attempt_detail=config.expose_attempt_detail, ) evaluator = Evaluator( diff --git a/vero/tests/test_harbor_build.py b/vero/tests/test_harbor_build.py index bebe0fa..df90f03 100644 --- a/vero/tests/test_harbor_build.py +++ b/vero/tests/test_harbor_build.py @@ -191,6 +191,27 @@ def test_feedback_transcripts_configured_through_yaml(): assert raw["feedback_max_bytes"] == 512 +def test_expose_attempt_detail_reaches_serve_json(built): + raw = json.loads((built / "environment" / "sidecar" / "serve.json").read_text()) + assert raw["expose_attempt_detail"] is False # default off + cfg = ServeConfig.from_file(built / "environment" / "sidecar" / "serve.json") + assert cfg.expose_attempt_detail is False + + +def test_expose_attempt_detail_configured_through_yaml(): + from vero.harbor.build.compiler import _serve_config + + config = BuildConfig.model_validate(yaml.safe_load( + "name: o/n\n" + "agent_repo: .\n" + "splits:\n" + " - {split: validation, access: viewable}\n" + "expose_attempt_detail: true\n" + )) + raw = _serve_config(config, "ds", "sha") + assert raw["expose_attempt_detail"] is True + + def test_rendered_files_parse(built): tomllib.loads((built / "task.toml").read_text()) # valid TOML compose = yaml.safe_load((built / "environment/docker-compose.yaml").read_text()) diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py index 1613aec..60fe690 100644 --- a/vero/tests/test_harbor_runner.py +++ b/vero/tests/test_harbor_runner.py @@ -48,11 +48,12 @@ def _write_trial( jobs_dir: Path, trial: str, task_name: str, - rewards: dict, + rewards: dict | None, *, pane: str | None = None, trajectory: str | None = None, finished_at: str | None = None, + exception_type: str | None = None, ): # Real harbor layout: ///result.json, plus a job-level # //result.json summary (no task_name) that collation must skip. @@ -62,9 +63,19 @@ def _write_trial( d = run / trial d.mkdir(parents=True, exist_ok=True) (run / "result.json").write_text(json.dumps({"job": "summary"})) # job-level, no task_name - data = {"task_name": task_name, "trial_name": trial, "verifier_result": {"rewards": rewards}} + data = { + "task_name": task_name, + "trial_name": trial, + "verifier_result": {"rewards": rewards} if rewards is not None else None, + } if finished_at is not None: data["finished_at"] = finished_at + if exception_type is not None: + data["exception_info"] = { + "exception_type": exception_type, + "exception_message": "", + "exception_traceback": "", + } (d / "result.json").write_text(json.dumps(data)) if pane is not None: (d / "agent").mkdir(exist_ok=True) @@ -577,3 +588,70 @@ def test_partially_passing_mean_sample_gets_no_feedback(self, tmp_path): r = self._result(runner, jobs) assert r.score == 0.5 assert r.feedback is None + + +class TestAttemptDetail: + """Lever 3 (expose_attempt_detail): sample output carries an `attempts` + list, one {reward, exception} entry per attempt. Population rules here; + the viewable-only exposure gate is the sidecar's (test_harbor_server).""" + + def _result(self, runner, jobs, task="t0"): + trials = runner._load_trials(jobs) + groups = runner._trial_groups(jobs) + return runner._sample_result( + trials.get(task), 0, task, _params(), attempts=groups.get(task) + ) + + def test_one_entry_per_attempt_with_exception_names(self, tmp_path): + runner = HarborRunner( + HarborConfig( + task_source="org/ds@1", agent_import_path="pkg.mod:Agent", + n_attempts=3, aggregate_attempts="mean", + ), + expose_attempt_detail=True, + ) + jobs = tmp_path / "jobs" + _write_trial(jobs, "a", "t0", {"reward": 1.0}, finished_at="2026-01-01T00:01:00") + _write_trial(jobs, "b", "t0", {"reward": 0.0}, finished_at="2026-01-01T00:02:00", + exception_type="AgentTimeoutError") + _write_trial(jobs, "c", "t0", None, finished_at="2026-01-01T00:03:00", + exception_type="RuntimeError") + r = self._result(runner, jobs) + assert r.output["attempts"] == [ + {"reward": 1.0, "exception": None}, + {"reward": 0.0, "exception": "AgentTimeoutError"}, + {"reward": None, "exception": "RuntimeError"}, + ] + + @pytest.mark.asyncio + async def test_best_mode_collates_attempts_end_to_end(self, tmp_path, monkeypatch): + # 'best' aggregation does not need the attempt groups for scoring, so + # this pins that _collate still loads them when the lever asks for it. + monkeypatch.setenv("VERO_HOME_DIR", str(tmp_path / "vh")) + runner = HarborRunner( + HarborConfig(task_source="org/ds@1", agent_import_path="pkg.mod:Agent"), + expose_attempt_detail=True, + ) + params = _params() + result_dir = tmp_path / "result" + _write_trial(result_dir / "jobs", "trial0", "t0", {"reward": 1.0}) + monkeypatch.setattr(runner, "_task_names_for", lambda p: [(0, "t0")]) + runner._run_harbor = AsyncMock() + ws = MagicMock(project_path="/wt") + await runner.produce_sample_results(workspace=ws, params=params, result_dir=result_dir) + results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id) + assert results[0].score == 1.0 # best-mode scoring untouched + assert results[0].output["attempts"] == [{"reward": 1.0, "exception": None}] + + def test_flag_off_leaves_output_without_attempts(self, tmp_path): + jobs = tmp_path / "jobs" + _write_trial(jobs, "a", "t0", {"reward": 1.0}, finished_at="2026-01-01T00:01:00") + _write_trial(jobs, "b", "t0", {"reward": 0.0}, finished_at="2026-01-01T00:02:00") + best = self._result(_runner(), jobs) + assert "attempts" not in best.output + mean_runner = HarborRunner(HarborConfig( + task_source="org/ds@1", agent_import_path="pkg.mod:Agent", + n_attempts=2, aggregate_attempts="mean", + )) + mean = self._result(mean_runner, jobs) + assert "attempts" not in mean.output diff --git a/vero/tests/test_harbor_serve.py b/vero/tests/test_harbor_serve.py index 49c8d57..3125ada 100644 --- a/vero/tests/test_harbor_serve.py +++ b/vero/tests/test_harbor_serve.py @@ -231,12 +231,14 @@ async def test_feedback_levers_reach_harbor_runner(fixture): "harbor": {"task_source": "org/x", "agent_import_path": "p:C"}, "feedback_transcripts": True, "feedback_max_bytes": 512, + "expose_attempt_detail": True, } ) sidecar, _, _ = await build_components(config) runner = sidecar.engine.evaluator.eval_strategy assert runner.feedback_transcripts is True assert runner.feedback_max_bytes == 512 + assert runner.expose_attempt_detail is True def test_mode_b_sample_timeout_warns(caplog): diff --git a/vero/tests/test_harbor_server.py b/vero/tests/test_harbor_server.py index df987d8..9a64115 100644 --- a/vero/tests/test_harbor_server.py +++ b/vero/tests/test_harbor_server.py @@ -138,6 +138,49 @@ async def test_feedback_never_written_for_no_access(self, tmp_path): assert not agent_vol.exists() or not list(agent_vol.rglob("*.json")) +class TestAttemptDetailTierGate: + """Lever 3 hidden-split safety: the per-attempt {reward, exception} list + lives in sample output, which reaches the agent only through the + viewable-only per-sample files. Same enforcement layer as feedback.""" + + def _experiment_with_attempts(self, split): + exp = _experiment(split) + for sr in exp.result.sample_results.values(): + sr.output = { + "task_name": "t0", + "attempts": [{"reward": 0.0, "exception": "SecretTimeoutError"}], + } + return exp + + @pytest.mark.asyncio + async def test_attempts_reach_agent_on_viewable_only(self, tmp_path): + sidecar = _sidecar( + tmp_path, split="train", accesses=[SplitAccess.viewable("train")] + ) + sidecar.engine.evaluate = AsyncMock( + return_value=self._experiment_with_attempts("train") + ) + await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="train")) + dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456" + blob = json.loads((dest / "0.json").read_text()) + assert blob["output"]["attempts"] == [ + {"reward": 0.0, "exception": "SecretTimeoutError"} + ] + + @pytest.mark.asyncio + async def test_attempts_never_written_for_non_viewable(self, tmp_path): + sidecar = _sidecar(tmp_path, split="validation") # non_viewable + sidecar.engine.evaluate = AsyncMock( + return_value=self._experiment_with_attempts("validation") + ) + await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + for f in (tmp_path / "agent_vol").rglob("*"): + if f.is_file(): + blob = f.read_text() + assert "SecretTimeoutError" not in blob + assert "attempts" not in blob + + class TestSubmit: @pytest.mark.asyncio async def test_submit_records_nomination(self, tmp_path): From 230add816132c59857b23ac5daf08deffc4f4eb1 Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Sun, 5 Jul 2026 14:22:01 +0300 Subject: [PATCH 4/4] fix(harbor): close multi-fidelity hidden-split leak, subset-eval shortlist pollution, and feedback-cap edge cases Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/harbor/build/compiler.py | 30 +++++- vero/src/vero/harbor/build/config.py | 7 +- vero/src/vero/harbor/runner.py | 57 +++++++++-- vero/src/vero/harbor/serve.py | 22 ++++ vero/src/vero/harbor/verifier.py | 16 +++ vero/tests/test_harbor_build.py | 122 +++++++++++++++++++++- vero/tests/test_harbor_runner.py | 135 +++++++++++++++++++++++++ vero/tests/test_harbor_server.py | 12 +++ vero/tests/test_harbor_verifier.py | 85 ++++++++++++++++ 9 files changed, 472 insertions(+), 14 deletions(-) diff --git a/vero/src/vero/harbor/build/compiler.py b/vero/src/vero/harbor/build/compiler.py index 4227dae..b91f02f 100644 --- a/vero/src/vero/harbor/build/compiler.py +++ b/vero/src/vero/harbor/build/compiler.py @@ -271,6 +271,25 @@ def compile_task( from vero.core.constants import PACKAGE_DIR vero_root = vero_root or PACKAGE_DIR + + # Mode A ignores the Mode-B-only feedback levers (they ride the nested + # `harbor run` collation, which Mode A never runs). Warn loudly at build time + # so a config that sets them in Mode A learns they will do nothing, rather + # than silently getting no feedback. + if config.mode == "A": + mode_b_only = [ + n + for n in ("feedback_transcripts", "expose_attempt_detail") + if getattr(config, n) + ] + if mode_b_only: + logger.warning( + "Mode A build sets Mode-B-only lever(s) %s; these have no effect " + "in Mode A (they ride the nested `harbor run` collation) and will " + "be ignored.", + ", ".join(mode_b_only), + ) + out = Path(out_dir) if out.exists(): shutil.rmtree(out) @@ -366,10 +385,17 @@ def compile_task( # Same merge-order-truthfulness introspection for the multi-fidelity # section: it may only render when the sidecar shipping in this tree # actually accepts subset evals (sample_ids / num_samples on the eval - # request), or the instruction would teach a knob that 400s. + # request), or the instruction would teach a knob that 400s. It also + # requires at least one VIEWABLE evaluable split: on a non_viewable + # split the sidecar returns mean_score inline, so a 1-sample subset eval + # (which the multi-fidelity section teaches) recovers that single + # sample's exact score, defeating the non_viewable contract. Only a + # viewable split is safe to screen with subsets. no_access splits are + # not agent-evaluable at all, so they never count either. multifidelity=config.instruct_multifidelity and {"sample_ids", "num_samples"} - <= {f.name for f in dataclasses.fields(EvalRequest)}, + <= {f.name for f in dataclasses.fields(EvalRequest)} + and any(s.access == "viewable" for s in config.splits), ) _render(jenv, "task.toml.j2", out / "task.toml", **ctx) _render(jenv, "instruction.md.j2", out / "instruction.md", **ctx) diff --git a/vero/src/vero/harbor/build/config.py b/vero/src/vero/harbor/build/config.py index 9e23af6..bc79cd2 100644 --- a/vero/src/vero/harbor/build/config.py +++ b/vero/src/vero/harbor/build/config.py @@ -11,7 +11,7 @@ from typing import Literal import yaml -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field class SplitAccessSpec(BaseModel): @@ -36,6 +36,11 @@ class TargetSpec(BaseModel): class BuildConfig(BaseModel): """Inputs to `vero harbor build`.""" + # Reject unknown top-level keys so a mistyped lever fails loudly at load + # time instead of silently disabling the feature: pydantic's default is to + # ignore extras, which would turn `feeback_transcripts: true` into a no-op. + model_config = ConfigDict(extra="forbid") + # identity name: str = Field(description="Harbor task name, 'org/name' format.") description: str = "" diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py index a0a57c2..0ec4024 100644 --- a/vero/src/vero/harbor/runner.py +++ b/vero/src/vero/harbor/runner.py @@ -259,9 +259,18 @@ def _trial_groups(self, jobs_dir: Path) -> dict[str, list[dict]]: groups.setdefault(task_name, []).append(data) # rglob order is undefined; sort each group so "first attempt" is a # stable notion (feedback uses the first failed attempt's transcript). + # An attempt with no finished_at must sort LAST, not first: an empty + # string would sort ahead of every real ISO timestamp and mislabel a + # timestamp-less attempt as the "first". The leading bool puts present + # timestamps first (False < True), then orders by the timestamp, and + # finally tie-breaks on the stable trial_name. for attempts in groups.values(): attempts.sort( - key=lambda d: (d.get("finished_at") or "", d.get("trial_name") or "") + key=lambda d: ( + d.get("finished_at") is None, + d.get("finished_at") or "", + d.get("trial_name") or "", + ) ) return groups @@ -410,34 +419,64 @@ def _failure_feedback( ) -> str | None: """Lever 1: transcript tail for a failed sample (score 0.0). - Uses the FIRST failed attempt only (attempts are sorted at load): the - first failure is the cheapest reproducible one, and one tail per sample - bounds the payload. Passed samples, and everything with the lever off, - return None (the field serializes as null either way, so responses are - byte-identical to before when disabled). + Walks the failed attempts in load order (attempts are sorted at load) + and returns the FIRST one with a readable transcript tail: the earliest + failure is the cheapest reproducible one, and one tail per sample bounds + the payload. A failed attempt with no recorded trial dir, or whose trial + recorded no transcript, does not end the search: the next failed attempt + is tried before giving up. Passed samples, and everything with the lever + off, return None (the field serializes as null either way, so responses + are byte-identical to before when disabled). """ if not self.feedback_transcripts or score != 0.0 or not attempts: return None + # feedback_max_bytes <= 0 means "no feedback", never "unbounded": a bare + # data[-0:] slice would return the WHOLE transcript, so the cap must be + # positive to emit anything at all. + if self.feedback_max_bytes <= 0: + return None for attempt in attempts: rewards = (attempt.get("verifier_result") or {}).get("rewards") if not rewards or self._extract_reward(rewards) != 0.0: continue trial_dir = attempt.get("_trial_dir") if not trial_dir: - return None - return self._read_transcript_tail(Path(trial_dir)) + continue + tail = self._read_transcript_tail(Path(trial_dir)) + if tail is not None: + return tail return None def _read_transcript_tail(self, trial_dir: Path) -> str | None: """Last ``feedback_max_bytes`` of the trial's transcript: the terminal pane when present, else the trajectory; None (field omitted) when the - trial recorded neither.""" + trial recorded neither. + + A non-positive cap emits nothing (matches _failure_feedback's guard). + The transcript path is confined to the trial dir: a symlinked transcript + file, or a resolved path that escapes the trial dir, is skipped silently + so a hostile trial layout cannot exfiltrate files outside its own dir. + """ + if self.feedback_max_bytes <= 0: + return None + trial_root = trial_dir.resolve() for rel in ("agent/terminus_2.pane", "agent/trajectory.json"): path = trial_dir / rel + # Reject symlinks outright, and any path that resolves outside the + # trial dir, before touching the bytes. + if path.is_symlink(): + continue + try: + resolved = path.resolve() + resolved.relative_to(trial_root) + except (OSError, ValueError): + continue try: data = path.read_bytes() except OSError: continue + # errors="replace": a multibyte char straddling the cap boundary is + # rendered as U+FFFD rather than crashing the collation. return data[-self.feedback_max_bytes :].decode("utf-8", errors="replace") return None diff --git a/vero/src/vero/harbor/serve.py b/vero/src/vero/harbor/serve.py index 5cbd626..2c11ed5 100644 --- a/vero/src/vero/harbor/serve.py +++ b/vero/src/vero/harbor/serve.py @@ -172,6 +172,27 @@ def _warn_mode_b_sample_timeout(config: ServeConfig) -> None: ) +def _warn_mode_a_ignores_feedback_levers(config: ServeConfig) -> None: + """The transcript-feedback / attempt-detail levers ride the Mode-B nested + `harbor run` collation (HarborRunner). Mode A (config.harbor is None) never + builds a HarborRunner, so these do nothing there; say so rather than let an + author think feedback is on. + """ + if config.harbor is not None: + return + mode_b_only = [ + n + for n in ("feedback_transcripts", "expose_attempt_detail") + if getattr(config, n) + ] + if mode_b_only: + logger.warning( + "Mode A serve config sets Mode-B-only lever(s) %s; these have no " + "effect in Mode A (no nested `harbor run`) and will be ignored.", + ", ".join(mode_b_only), + ) + + async def build_components(config: ServeConfig) -> tuple[EvaluationSidecar, Verifier, str]: """Assemble the sidecar + verifier (sharing one engine) and the admin token.""" vero_home = get_vero_home_dir() @@ -191,6 +212,7 @@ async def build_components(config: ServeConfig) -> tuple[EvaluationSidecar, Veri ) _warn_mode_b_sample_timeout(config) + _warn_mode_a_ignores_feedback_levers(config) workspace = await GitWorkspace.create(config.repo_path) diff --git a/vero/src/vero/harbor/verifier.py b/vero/src/vero/harbor/verifier.py index d0d996c..35efdfa 100644 --- a/vero/src/vero/harbor/verifier.py +++ b/vero/src/vero/harbor/verifier.py @@ -259,6 +259,22 @@ async def _best_from_db(self) -> str: f"auto_best mode but no candidate experiments on split " f"'{self.selection_split}'." ) + # Rank on FULL-split evals only, WHEN ANY EXIST. A subset eval + # (num_samples / sample_ids, taught by the multi-fidelity lever) records + # a mean over a handful of samples, so a lucky small subset can inflate a + # candidate's recorded score and push it into the top-K shortlist over a + # genuinely better full-split candidate. A full-split eval is recorded + # with dataset_subset_sample_ids = None (DatasetSubset.is_full_set); any + # non-null value is a subset. If at least one candidate has a full-split + # eval, subset evals are dropped for ranking so they cannot displace it. + # If EVERY eval is a subset, there is no full-split candidate to protect, + # so the subset evals are the only ranking signal and are kept (the + # winner is still decided by an admin re-score on the full split, so this + # only controls which commits enter the shortlist). + if "dataset_subset_sample_ids" in split_df.columns: + full_split_df = split_df[split_df["dataset_subset_sample_ids"].isna()] + if len(full_split_df) > 0: + split_df = full_split_df # Shortlist by recorded score (cheap, agent-influenced -> not trusted as # final), one row per candidate (highest recorded score wins the slot). ranked = split_df.sort_values( diff --git a/vero/tests/test_harbor_build.py b/vero/tests/test_harbor_build.py index df90f03..45908ef 100644 --- a/vero/tests/test_harbor_build.py +++ b/vero/tests/test_harbor_build.py @@ -314,18 +314,31 @@ def test_instruction_omits_free_baseline_claim_when_unsupported(built): assert "budget-free" not in text -def _multifidelity_config(tmp_path) -> BuildConfig: +def _multifidelity_config_with_splits(tmp_path, splits) -> BuildConfig: return BuildConfig( name="vero/gsm8k-opt", agent_repo=str(_agent_repo(tmp_path)), mode="A", task="gsm8k", dataset=str(_dataset(tmp_path)), - splits=[{"split": "validation", "access": "non_viewable"}], + splits=splits, instruct_multifidelity=True, ) +def _multifidelity_config(tmp_path) -> BuildConfig: + # Includes a viewable split so the section renders: multi-fidelity is gated + # on a viewable evaluable split existing (subset evals on a non_viewable + # split would leak per-sample scores; see the compiler ctx gate). + return _multifidelity_config_with_splits( + tmp_path, + [ + {"split": "train", "access": "viewable"}, + {"split": "validation", "access": "non_viewable"}, + ], + ) + + @pytest.mark.skipif( not _HAS_SUBSET_EVALS, reason="sidecar in this tree has no subset evals" ) @@ -350,6 +363,54 @@ def test_instruction_omits_multifidelity_by_default(built): assert "Screen cheaply" not in text +@pytest.mark.skipif( + not _HAS_SUBSET_EVALS, reason="sidecar in this tree has no subset evals" +) +def test_multifidelity_suppressed_when_only_non_viewable_evaluable(tmp_path, monkeypatch): + # Hidden-split leak guard: on a non_viewable split the sidecar returns + # mean_score inline, so a 1-sample subset eval (which the section teaches) + # recovers that sample's exact score. With no viewable evaluable split the + # section must NOT render even when instruct_multifidelity is set. + monkeypatch.setenv("VERO_SKIP_SECRET_CHECK", "1") + out = compile_task( + _multifidelity_config_with_splits( + tmp_path, + [ + {"split": "validation", "access": "non_viewable"}, + {"split": "test", "access": "no_access"}, + ], + ), + tmp_path / "task", + vero_root=_stub_vero(tmp_path), + ) + text = (out / "instruction.md").read_text() + assert "--num-samples" not in text + assert "Screen cheaply" not in text + + +@pytest.mark.skipif( + not _HAS_SUBSET_EVALS, reason="sidecar in this tree has no subset evals" +) +def test_multifidelity_rendered_when_a_viewable_split_exists(tmp_path, monkeypatch): + # A viewable split is safe to screen with subsets (its per-sample results are + # already visible), so the section renders. + monkeypatch.setenv("VERO_SKIP_SECRET_CHECK", "1") + out = compile_task( + _multifidelity_config_with_splits( + tmp_path, + [ + {"split": "train", "access": "viewable"}, + {"split": "validation", "access": "non_viewable"}, + ], + ), + tmp_path / "task", + vero_root=_stub_vero(tmp_path), + ) + text = (out / "instruction.md").read_text() + assert "--num-samples" in text + assert "Screen cheaply" in text + + def test_multifidelity_gate_suppresses_section_without_subset_evals(tmp_path, monkeypatch): # Merge-order guard, exercised the way the free-baseline gate is designed: # against a sidecar whose EvalRequest lacks subset-eval fields, the section @@ -493,3 +554,60 @@ def test_skip_env_var_bypasses_check(self, monkeypatch): lambda ts: {"org/task-a"}, ) compiler._validate_partition_names({"train": ["bare-name"]}, "org/bench") # no raise + + +class TestUnknownFieldRejection: + """A mistyped lever key must fail loudly at load, not silently disable the + feature: pydantic's default ignores extras, so `feeback_transcripts: true` + would compile a task with feedback OFF and no warning.""" + + def test_typo_lever_key_rejected(self): + from pydantic import ValidationError + + with pytest.raises(ValidationError): + BuildConfig.model_validate(yaml.safe_load( + "name: o/n\n" + "agent_repo: .\n" + "splits:\n" + " - {split: validation, access: non_viewable}\n" + "feeback_transcripts: true\n" # typo: feeback (missing 'd') + )) + + def test_known_keys_still_accepted(self): + cfg = BuildConfig.model_validate(yaml.safe_load( + "name: o/n\n" + "agent_repo: .\n" + "splits:\n" + " - {split: validation, access: non_viewable}\n" + "feedback_transcripts: true\n" + )) + assert cfg.feedback_transcripts is True + + +class TestModeAIgnoresFeedbackLevers: + """Mode A ignores the Mode-B-only feedback levers (they ride the nested + `harbor run` collation). compile_task must warn so an author does not think + feedback is on when it is not.""" + + def test_mode_a_with_feedback_lever_warns(self, tmp_path, monkeypatch, caplog): + monkeypatch.setenv("VERO_SKIP_SECRET_CHECK", "1") + config = BuildConfig( + name="vero/gsm8k-opt", + agent_repo=str(_agent_repo(tmp_path)), + mode="A", + task="gsm8k", + dataset=str(_dataset(tmp_path)), + splits=[{"split": "validation", "access": "non_viewable"}], + feedback_transcripts=True, + expose_attempt_detail=True, + ) + with caplog.at_level("WARNING", logger="vero.harbor.build.compiler"): + compile_task(config, tmp_path / "task", vero_root=_stub_vero(tmp_path)) + joined = " ".join(caplog.messages) + assert "Mode-B-only" in joined + assert "feedback_transcripts" in joined + assert "expose_attempt_detail" in joined + + def test_mode_a_without_feedback_lever_does_not_warn(self, built, caplog): + # `built` fixture is a Mode A config with the levers off: no warning. + assert not any("Mode-B-only" in m for m in caplog.messages) diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py index 60fe690..ab02f79 100644 --- a/vero/tests/test_harbor_runner.py +++ b/vero/tests/test_harbor_runner.py @@ -590,6 +590,141 @@ def test_partially_passing_mean_sample_gets_no_feedback(self, tmp_path): assert r.feedback is None +class TestTranscriptFeedbackEdgeCases: + """Byte-cap boundary + feedback_max_bytes<=0 + path-confinement + next-attempt + fallback. The tail must be exactly capped (never over), never unbounded when + the cap is 0, must not crash on a multibyte char straddling the boundary, must + refuse a symlinked / escaping transcript, and must try the next failed attempt + when the first has no transcript.""" + + def _result(self, runner, jobs, task="t0"): + trials = runner._load_trials(jobs) + groups = runner._trial_groups(jobs) + return runner._sample_result( + trials.get(task), 0, task, _params(), attempts=groups.get(task) + ) + + def test_exact_length_at_cap_returns_full(self, tmp_path): + # A transcript exactly cap bytes long is returned whole (not truncated). + pane = "B" * 16 + runner = _fb_runner(feedback_max_bytes=16) + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}, pane=pane) + r = self._result(runner, jobs) + assert r.feedback == pane + assert len(r.feedback.encode()) == 16 + + def test_one_byte_over_cap_truncates_to_cap(self, tmp_path): + # 17 bytes with a 16-byte cap keeps only the last 16 bytes. + runner = _fb_runner(feedback_max_bytes=16) + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}, pane="X" + "Y" * 16) + r = self._result(runner, jobs) + assert r.feedback == "Y" * 16 + assert len(r.feedback.encode()) == 16 + + def test_multibyte_char_straddling_cap_does_not_crash(self, tmp_path): + # A 3-byte U+2603 (snowman) straddles the cap boundary. The slice cuts + # mid-character; errors="replace" must render it without crashing. + runner = _fb_runner(feedback_max_bytes=4) + jobs = tmp_path / "jobs" + # 6 bytes: 'AAA' + a 3-byte char -> last 4 bytes cut the char mid-sequence + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}, pane="AAA☃") + r = self._result(runner, jobs) # must not raise + assert r.feedback is not None + assert len(r.feedback.encode()) <= 8 # replacement chars may re-expand slightly + + def test_zero_cap_emits_no_feedback(self, tmp_path): + # feedback_max_bytes=0 means "no feedback", NOT the whole transcript. + runner = _fb_runner(feedback_max_bytes=0) + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}, pane="should not leak") + r = self._result(runner, jobs) + assert r.feedback is None + + def test_negative_cap_emits_no_feedback(self, tmp_path): + runner = _fb_runner(feedback_max_bytes=-5) + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}, pane="should not leak") + r = self._result(runner, jobs) + assert r.feedback is None + + def test_symlinked_transcript_is_refused(self, tmp_path): + # A symlinked transcript file must be skipped silently (field omitted). + runner = _fb_runner() + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}) # no real transcript + # place a secret outside the trial dir and symlink the pane path to it + secret = tmp_path / "secret.txt" + secret.write_text("SECRET-OUTSIDE") + trial_dir = jobs / "2026-01-01__00-00-00" / "trial0" + (trial_dir / "agent").mkdir(exist_ok=True) + (trial_dir / "agent" / "terminus_2.pane").symlink_to(secret) + r = self._result(runner, jobs) + assert r.feedback is None # symlink refused, nothing leaked + + def test_escaping_transcript_is_refused(self, tmp_path): + # A trajectory that is a symlink to a file outside the trial dir is also + # refused (path resolves outside trial_root). + runner = _fb_runner() + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}) + outside = tmp_path / "outside.json" + outside.write_text('{"leak": true}') + trial_dir = jobs / "2026-01-01__00-00-00" / "trial0" + (trial_dir / "agent").mkdir(exist_ok=True) + (trial_dir / "agent" / "trajectory.json").symlink_to(outside) + r = self._result(runner, jobs) + assert r.feedback is None + + def test_next_failed_attempt_used_when_first_has_no_transcript(self, tmp_path): + # First failed attempt records no transcript; the second failed attempt's + # transcript is used instead of giving up. + runner = HarborRunner( + HarborConfig( + task_source="org/ds@1", agent_import_path="pkg.mod:Agent", + n_attempts=2, aggregate_attempts="mean", + ), + feedback_transcripts=True, + ) + jobs = tmp_path / "jobs" + _write_trial(jobs, "zz-early", "t0", {"reward": 0.0}, # no pane/trajectory + finished_at="2026-01-01T00:01:00") + _write_trial(jobs, "aa-late", "t0", {"reward": 0.0}, pane="second tail", + finished_at="2026-01-01T00:09:00") + r = self._result(runner, jobs) + assert r.score == 0.0 + assert r.feedback == "second tail" + + +class TestAttemptSortOrder: + """Attempts missing finished_at must sort LAST, not first: an empty-string + timestamp would sort ahead of every real ISO timestamp and mislabel a + timestamp-less attempt as the "first" (which feedback keys off).""" + + def test_missing_finished_at_sorts_last(self, tmp_path): + runner = _runner() + jobs = tmp_path / "jobs" + # one attempt with a real timestamp, one with none + _write_trial(jobs, "with-ts", "t0", {"reward": 0.0}, + finished_at="2026-01-01T00:05:00") + _write_trial(jobs, "no-ts", "t0", {"reward": 0.0}) # finished_at absent + groups = runner._trial_groups(jobs) + names = [a.get("trial_name") for a in groups["t0"]] + assert names == ["with-ts", "no-ts"] # timestamped first, missing last + + def test_feedback_uses_timestamped_attempt_over_timeless_one(self, tmp_path): + runner = _fb_runner() + jobs = tmp_path / "jobs" + _write_trial(jobs, "no-ts", "t0", {"reward": 0.0}, pane="timeless tail") + _write_trial(jobs, "with-ts", "t0", {"reward": 0.0}, pane="timestamped tail", + finished_at="2026-01-01T00:05:00") + trials = runner._load_trials(jobs) + groups = runner._trial_groups(jobs) + r = runner._sample_result(trials.get("t0"), 0, "t0", _params(), attempts=groups.get("t0")) + assert r.feedback == "timestamped tail" + + class TestAttemptDetail: """Lever 3 (expose_attempt_detail): sample output carries an `attempts` list, one {reward, exception} entry per attempt. Population rules here; diff --git a/vero/tests/test_harbor_server.py b/vero/tests/test_harbor_server.py index 9a64115..57f233f 100644 --- a/vero/tests/test_harbor_server.py +++ b/vero/tests/test_harbor_server.py @@ -180,6 +180,18 @@ async def test_attempts_never_written_for_non_viewable(self, tmp_path): assert "SecretTimeoutError" not in blob assert "attempts" not in blob + @pytest.mark.asyncio + async def test_attempts_never_written_for_no_access(self, tmp_path): + # no_access is admin/verifier only: no per-sample files at all, so the + # attempt detail can never surface (mirrors the feedback no_access case). + sidecar = _sidecar(tmp_path, split="test") # no_access + sidecar.engine.evaluate = AsyncMock( + return_value=self._experiment_with_attempts("test") + ) + await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="test")) + agent_vol = tmp_path / "agent_vol" + assert not agent_vol.exists() or not list(agent_vol.rglob("*.json")) + class TestSubmit: @pytest.mark.asyncio diff --git a/vero/tests/test_harbor_verifier.py b/vero/tests/test_harbor_verifier.py index 735763a..264cdc3 100644 --- a/vero/tests/test_harbor_verifier.py +++ b/vero/tests/test_harbor_verifier.py @@ -146,6 +146,91 @@ async def _admin(*, task, dataset_id, split, commit, sample_ids=None): assert engine.evaluate_admin.await_args.kwargs["commit"] == "agent" +class TestSubsetEvalShortlistFilter: + """auto_best ranks the shortlist on FULL-split evals only. A subset eval + (num_samples / sample_ids) records a mean over a few samples, so a lucky + 1-sample eval can inflate a candidate's recorded score and steal a shortlist + slot from a genuinely better full-split candidate. Full-split evals have + dataset_subset_sample_ids = None; subset evals carry a list and are ignored + for ranking (the admin re-score still runs on the full split).""" + + @pytest.mark.asyncio + async def test_lucky_subset_eval_does_not_outrank_full_split(self, tmp_path): + # 'lucky' has a high 1-sample eval (recorded 0.95) but a low full-split + # eval (0.30). 'solid' has a higher full-split eval (0.70). With + # rescore_top_k=1 only one commit is shortlisted; ranking on full-split + # evals must shortlist 'solid', not 'lucky'. + engine = MagicMock() + engine.db.get_experiments_df.return_value = pd.DataFrame( + { + "dataset_subset_split": ["validation", "validation", "validation"], + "dataset_subset_dataset_id": ["ds1", "ds1", "ds1"], + "dataset_subset_sample_ids": [[0], None, None], # lucky subset, then full splits + "candidate_commit": ["lucky", "lucky", "solid"], + "mean_score": [0.95, 0.30, 0.70], + "candidate_created_at": [3, 1, 2], + } + ) + + async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + # admin re-score agrees the full-split ranking is right + score = {"solid": 0.7, "lucky": 0.3}.get(commit, 0.5) + return MagicMock(result=MagicMock(score=MagicMock(return_value=score))) + + engine.evaluate_admin = AsyncMock(side_effect=_admin) + v = Verifier( + engine=engine, + admin_volume=tmp_path, + reward_mode="auto_best", + selection_split="validation", + selection_task="math", + rescore_top_k=1, + auto_best_baseline_floor=False, + targets=[VerificationTarget(task="math", dataset_id="ds1", split="test", reward_key="reward")], + ) + await v.finalize() + # 'solid' is selected (target-scored); 'lucky' never entered the shortlist + rescored = [c.kwargs["commit"] for c in engine.evaluate_admin.await_args_list] + assert "lucky" not in rescored + assert engine.evaluate_admin.await_args.kwargs["commit"] == "solid" + + @pytest.mark.asyncio + async def test_all_subset_evals_still_rankable(self, tmp_path): + # When EVERY eval is a subset, there is no full-split candidate to + # protect, so the subset evals are the only ranking signal and are kept + # (a legitimate all-subset workflow must still select a candidate). The + # admin re-score on the full split remains the trust anchor. + engine = MagicMock() + engine.db.get_experiments_df.return_value = pd.DataFrame( + { + "dataset_subset_split": ["validation", "validation"], + "dataset_subset_dataset_id": ["ds1", "ds1"], + "dataset_subset_sample_ids": [[0], [0, 1]], + "candidate_commit": ["a", "b"], + "mean_score": [0.9, 0.8], + "candidate_created_at": [1, 2], + } + ) + + async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + return MagicMock(result=MagicMock(score=MagicMock(return_value=0.5))) + + engine.evaluate_admin = AsyncMock(side_effect=_admin) + v = Verifier( + engine=engine, + admin_volume=tmp_path, + reward_mode="auto_best", + selection_split="validation", + selection_task="math", + auto_best_baseline_floor=False, + targets=[VerificationTarget(task="math", dataset_id="ds1", split="test", reward_key="reward")], + ) + rewards = (await v.finalize())["rewards"] + # a candidate was selected and target-scored (not floored) + assert rewards == {"reward": 0.5} + engine.evaluate_admin.assert_awaited() + + class TestAutoBestBaselineFloor: """auto_best never ships a candidate that fails to beat the baseline.