From d0d7cc5ed009fdcd7bc350e4d5cf527cb8928625 Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Thu, 9 Jul 2026 02:36:05 +0300 Subject: [PATCH 1/8] fix(harbor): ops integrity (exhaust-budget instruction lever, ledger fails closed on corrupt restore) Two operational fixes: 1. instruct_exhaust_budget (default True): the instruction's "unspent budget is wasted" persistence bullet becomes a build-config lever, like instruct_multifidelity. On preserves current behavior; off makes stopping-early the agent's own choice, which is the ablation arm for measuring what the exhortation itself contributes to optimizer persistence. The "scores are noisy" fact stays unconditional. 2. _load_or_build_ledger fails CLOSED on a persisted ledger that exists but cannot be parsed: metered budgets restore with zero remaining, and the unreadable file is preserved as ledger.corrupt for the operator. The old fallback restored the CONFIGURED budgets, which refunded the agent everything already spent, so any crash that corrupted the flush minted budget. A missing file is still a fresh boot; admin and finalize are unaffected either way. Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/harbor/build/compiler.py | 2 + vero/src/vero/harbor/build/config.py | 7 ++++ .../harbor/build/templates/instruction.md.j2 | 4 +- vero/src/vero/harbor/serve.py | 34 +++++++++++++-- vero/tests/test_harbor_build.py | 33 +++++++++++++++ vero/tests/test_harbor_serve.py | 42 ++++++++++++++++++- 6 files changed, 116 insertions(+), 6 deletions(-) diff --git a/vero/src/vero/harbor/build/compiler.py b/vero/src/vero/harbor/build/compiler.py index 2021399..61123b8 100644 --- a/vero/src/vero/harbor/build/compiler.py +++ b/vero/src/vero/harbor/build/compiler.py @@ -243,6 +243,7 @@ def _serve_config( "submit_enabled": config.submit_enabled, "score_baseline": config.score_baseline, "k_anonymity_floor": config.k_anonymity_floor, + "instruct_exhaust_budget": config.instruct_exhaust_budget, "agent_volume": AGENT_VOLUME, "admin_volume": ADMIN_VOLUME, "admin_token_path": TOKEN_PATH, @@ -405,6 +406,7 @@ def compile_task( and {"sample_ids", "num_samples"} <= {f.name for f in dataclasses.fields(EvalRequest)} and any(s.access == "viewable" for s in config.splits), + exhaust_budget=config.instruct_exhaust_budget, ) _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 8cf6c9a..5b242f3 100644 --- a/vero/src/vero/harbor/build/config.py +++ b/vero/src/vero/harbor/build/config.py @@ -72,6 +72,13 @@ class _BuildConfigBase(BaseModel): # mean_score, so singleton subsets would hand back per-sample labels. k_anonymity_floor: int = 5 + # Instruction lever: render the "unspent budget is wasted" persistence + # bullet that tells the optimizer to keep spending (re-measure the champion, + # try one more variant) instead of stopping early. On by default (current + # behavior); off makes stopping-early a choice the agent arrives at itself, + # which is the ablation arm for measuring what the exhortation contributes. + instruct_exhaust_budget: bool = True + # 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. read_only_paths: list[str] = Field(default_factory=list) diff --git a/vero/src/vero/harbor/build/templates/instruction.md.j2 b/vero/src/vero/harbor/build/templates/instruction.md.j2 index df25032..6447398 100644 --- a/vero/src/vero/harbor/build/templates/instruction.md.j2 +++ b/vero/src/vero/harbor/build/templates/instruction.md.j2 @@ -51,9 +51,11 @@ in rough proportion to its size, so it is the cheap way to triage ideas: from a regression. Repeat baseline evals are metered. `vero harbor status` shows the baseline sha and whether the free eval is still available. {% endif %} -- Scores are noisy. Unspent budget is wasted: if you finish with evals left, spend +- Scores are noisy. +{% if exhaust_budget %}- Unspent budget is wasted: if you finish with evals left, spend them re-measuring your best candidate to confirm its score, or trying one more variant, rather than stopping early. +{% endif %} - The test split is hidden: you cannot evaluate it, and its labels never reach this container. Trying to read it will fail. - The scorer is locked. Only the eval sidecar scores. diff --git a/vero/src/vero/harbor/serve.py b/vero/src/vero/harbor/serve.py index 00eb327..52c73e0 100644 --- a/vero/src/vero/harbor/serve.py +++ b/vero/src/vero/harbor/serve.py @@ -10,6 +10,7 @@ import json import logging +import shutil from pathlib import Path from typing import Annotated, Literal @@ -82,6 +83,10 @@ class _ServeConfigBase(BaseModel): # EvaluationSidecar.k_anonymity_floor for the leak this closes. k_anonymity_floor: int = 5 + # Consumed at COMPILE time (the instruction's exhaust-budget bullet); + # recorded here so serve.json mirrors build.yaml, like instruct_multifidelity. + instruct_exhaust_budget: bool = True + # volumes / token agent_volume: str admin_volume: str @@ -155,8 +160,12 @@ def _load_or_build_ledger( a sidecar restart would reset all spent budget to full, letting the agent regain its full evaluation budget by triggering a restart. On startup we reconstruct each SplitBudget and restore its persisted ``remaining_*`` values. - Falls back to the configured budgets if the file is missing or unreadable - (fail-safe to the configured budget, never to unlimited). + + A MISSING file is a fresh boot: configured budgets. A file that exists but + cannot be parsed fails CLOSED: metered budgets restore with zero remaining. + The old fallback (configured budgets) refunded the agent everything already + spent, so any crash that corrupted the flush minted budget; spend that + cannot be read must be treated as fully spent, never as never-happened. """ if persist_path.exists(): try: @@ -181,11 +190,28 @@ def _load_or_build_ledger( ) return BudgetLedger(budgets, persist_path=persist_path) except (json.JSONDecodeError, KeyError, OSError) as e: - logger.warning( - "Could not reload persisted ledger %s (%s); using configured budgets.", + logger.error( + "Persisted ledger %s exists but is unreadable (%s); failing " + "CLOSED: metered agent budgets restore as exhausted. Admin and " + "finalize are unaffected. The unreadable file is preserved at " + "%s; delete ledger.json deliberately to boot fresh.", persist_path, e, + persist_path.with_suffix(".corrupt"), ) + try: # keep the evidence: the next flush overwrites persist_path + shutil.copyfile(persist_path, persist_path.with_suffix(".corrupt")) + except OSError: + pass + budgets = [] + for cfg in budget_cfgs: + b = SplitBudget(**cfg) + if b.total_sample_budget is not None: + b.remaining_sample_budget = 0 + if b.total_run_budget is not None: + b.remaining_run_budget = 0 + budgets.append(b) + return BudgetLedger(budgets, persist_path=persist_path) return BudgetLedger( [SplitBudget(**b) for b in budget_cfgs], persist_path=persist_path ) diff --git a/vero/tests/test_harbor_build.py b/vero/tests/test_harbor_build.py index 2efdeb2..46983bf 100644 --- a/vero/tests/test_harbor_build.py +++ b/vero/tests/test_harbor_build.py @@ -363,6 +363,39 @@ def test_instruction_omits_free_baseline_claim_when_unsupported(built): assert "budget-free" not in text +def test_instruction_renders_exhaust_budget_by_default(built): + text = (built / "instruction.md").read_text() + assert "Unspent budget is wasted" in text + + +def test_instruction_omits_exhaust_budget_when_disabled(tmp_path, monkeypatch): + # The persistence exhortation is an instruction LEVER: off, stopping early + # becomes the agent's own choice, which is the ablation arm for measuring + # what the exhortation itself contributes. + monkeypatch.setenv("VERO_SKIP_SECRET_CHECK", "1") + config = BuildConfigA( + name="vero/gsm8k-opt", + description="optimize gsm8k", + agent_repo=str(_agent_repo(tmp_path)), + task="gsm8k", + task_module="gsm8k_agent.vero_tasks", + dataset=str(_dataset(tmp_path)), + splits=[ + {"split": "validation", "access": "non_viewable"}, + {"split": "test", "access": "no_access"}, + ], + budgets=[{"split": "validation", "total_run_budget": 5}], + reward_mode="auto_best", + selection_split="validation", + targets=[{"split": "test", "reward_key": "reward"}], + instruct_exhaust_budget=False, + ) + out = compile_task(config, tmp_path / "task", vero_root=_stub_vero(tmp_path)) + text = (out / "instruction.md").read_text() + assert "Unspent budget is wasted" not in text + assert "Scores are noisy." in text # the noise fact is not part of the lever + + def _multifidelity_config_with_splits(tmp_path, splits) -> BuildConfigB: # instruct_multifidelity is a Mode-B-only lever, so the multi-fidelity gate is # exercised through a Mode-B config (local inner_task so it compiles offline). diff --git a/vero/tests/test_harbor_serve.py b/vero/tests/test_harbor_serve.py index 200f9c2..4cb315e 100644 --- a/vero/tests/test_harbor_serve.py +++ b/vero/tests/test_harbor_serve.py @@ -8,6 +8,7 @@ from __future__ import annotations +import json import subprocess import textwrap from pathlib import Path @@ -16,7 +17,12 @@ from vero.core.dataset.store import resolve_and_save_dataset from vero.evaluation.engine import EvalRequest -from vero.harbor.serve import ServeConfigA, ServeConfigB, build_components +from vero.harbor.serve import ( + ServeConfigA, + ServeConfigB, + _load_or_build_ledger, + build_components, +) def _git(path: Path, *args: str) -> str: @@ -221,6 +227,40 @@ async def test_ledger_reloads_spent_budget_across_restart(fixture): assert reloaded == after, "sidecar restart must not refill spent budget" +class TestLedgerFailClosed: + """A persisted ledger that exists but cannot be read fails CLOSED: spend + that cannot be reconstructed is treated as fully spent. The old fallback + (configured budgets) refunded the agent everything already spent, so any + crash that corrupted the flush minted budget.""" + + _CFGS = [{ + "split": "validation", "dataset_id": "ds", + "total_run_budget": 5, "total_sample_budget": 50, + }] + + def test_missing_file_boots_configured(self, tmp_path): + led = _load_or_build_ledger(self._CFGS, tmp_path / "ledger.json") + b = led.get("ds", "validation") + assert b.remaining_run_budget == 5 + assert b.remaining_sample_budget == 50 + + def test_unparseable_file_fails_closed_and_keeps_evidence(self, tmp_path): + p = tmp_path / "ledger.json" + p.write_text("{definitely not json") + led = _load_or_build_ledger(self._CFGS, p) + b = led.get("ds", "validation") + assert b.remaining_run_budget == 0 + assert b.remaining_sample_budget == 0 + # the unreadable original survives for the operator to inspect + assert p.with_suffix(".corrupt").read_text() == "{definitely not json" + + def test_malformed_entries_fail_closed(self, tmp_path): + p = tmp_path / "ledger.json" + p.write_text(json.dumps([{"no_split_key": 1}])) + led = _load_or_build_ledger(self._CFGS, p) + assert led.get("ds", "validation").remaining_run_budget == 0 + + @pytest.mark.asyncio async def test_feedback_levers_reach_harbor_runner(fixture): # Lever 1 pass-through: ServeConfigB -> build_components -> HarborRunner kwargs From bd1d1fec9ca3c9c72e59a31ce821c702920da82c Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Thu, 9 Jul 2026 02:39:22 +0300 Subject: [PATCH 2/8] fix(harbor): feedback robustness (review follow-ups from #30) Two Greptile P2s on #30, fixed at the stack tip: - An empty transcript file no longer surfaces as "" feedback: empty candidates are skipped (an empty pane falls through to the trajectory), and if everything is empty the search moves to the next failed attempt rather than short-circuiting on "". - The no-verifier-rewards error branch (agent died before scoring) now attaches the failure transcript like any failed sample: a candidate edit that crashes the agent lands exactly here, and the transcript is the only way the optimizer can see the crash it caused. Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/harbor/runner.py | 12 ++++++++++++ vero/tests/test_harbor_runner.py | 27 +++++++++++++++++++++++++++ 2 files changed, 39 insertions(+) diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py index f262261..d5265c7 100644 --- a/vero/src/vero/harbor/runner.py +++ b/vero/src/vero/harbor/runner.py @@ -411,6 +411,12 @@ def _out(output: dict) -> dict: if not rewards: return SampleResult( error=f"No verifier rewards for task '{task_name}'.", + # The agent died before scoring. A candidate edit that CRASHES + # the agent lands here, and "no verifier rewards" alone gives + # the optimizer no way to see its own crash; the transcript + # does. Passed as score 0.0: an unscored attempt counts as a + # failure everywhere else too. + feedback=self._failure_feedback(0.0, attempts), output=_out( {"task_name": task_name, "trial_name": trial.get("trial_name")} ), @@ -553,6 +559,12 @@ def _read_transcript_tail(self, trial_dir: Path) -> str | None: data = path.read_bytes() except OSError: continue + # An empty transcript carries nothing: keep looking (an empty pane + # must fall through to the trajectory), and if every candidate is + # empty return None so the caller tries the next failed attempt + # instead of emitting "" as feedback. + if not data: + 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") diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py index 0f53e65..9642403 100644 --- a/vero/tests/test_harbor_runner.py +++ b/vero/tests/test_harbor_runner.py @@ -636,6 +636,33 @@ def test_missing_transcripts_omitted_silently(self, tmp_path): assert r.score == 0.0 assert r.feedback is None + def test_empty_pane_falls_through_to_trajectory(self, tmp_path): + # An empty transcript file carries nothing and must not surface as "" + # feedback; the empty pane falls through to the trajectory. + runner = _fb_runner() + jobs = tmp_path / "jobs" + _write_trial( + jobs, "trial0", "t0", {"reward": 0.0}, pane="", trajectory='{"steps": [1]}' + ) + assert self._result(runner, jobs).feedback == '{"steps": [1]}' + + def test_all_empty_transcripts_yield_no_feedback(self, tmp_path): + runner = _fb_runner() + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", {"reward": 0.0}, pane="", trajectory="") + assert self._result(runner, jobs).feedback is None + + def test_no_rewards_error_sample_carries_crash_transcript(self, tmp_path): + # A candidate edit that crashes the agent before scoring lands in the + # no-verifier-rewards error branch; the transcript is the only way the + # optimizer can see the crash it caused. + runner = _fb_runner() + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", None, pane="crash tail") + r = self._result(runner, jobs) + assert r.error is not None + assert r.feedback == "crash tail" + 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. From e1b1d6b4522c0c7d75e955139710080cf3896648 Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Thu, 9 Jul 2026 02:50:51 +0300 Subject: [PATCH 3/8] fix(harbor): review follow-ups from #34/#35 (freebie claim atomicity, floor default, ordinal resume, SE naming) - Free-baseline flag is claimed BEFORE the eval await and refunded on failure: setting it only after success reopened a window where two concurrent baseline evals both resolved free (asyncio interleaves at await points). Claim-then-refund keeps both properties: concurrent callers see the claim, and a failed eval does not burn the freebie. - build_status defaults k_anonymity_floor to 5, matching the sidecar's enforcement default: a caller that forgets to pass the floor must not advertise a laxer one than gets enforced. - _route_results resumes the eval ordinal past surviving __eN dirs on a reused volume: a restarted sidecar started back at e1 and silently wiped the prior session's evidence, the exact erasure the versioned dirs exist to prevent. - score_se renamed to mean_score_se and documented: it is the SE of the zero-filled mean_score over n_samples, not of the n_scored subset. Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/harbor/protocol.py | 5 ++- vero/src/vero/harbor/server.py | 53 +++++++++++++++++++++--------- vero/tests/test_harbor_protocol.py | 14 ++++++++ vero/tests/test_harbor_server.py | 31 ++++++++++++++++- 4 files changed, 85 insertions(+), 18 deletions(-) diff --git a/vero/src/vero/harbor/protocol.py b/vero/src/vero/harbor/protocol.py index c43f101..156c8e3 100644 --- a/vero/src/vero/harbor/protocol.py +++ b/vero/src/vero/harbor/protocol.py @@ -104,7 +104,10 @@ def build_status( split_accesses: list[SplitAccess], base_commit: str | None = None, free_baseline_available: bool = False, - k_anonymity_floor: int = 1, + # Default matches EvaluationSidecar's enforcement default: a caller that + # forgets to pass the floor must not advertise a laxer one than the + # sidecar enforces (agents would send sub-floor requests that 400). + k_anonymity_floor: int = 5, ) -> StatusSummary: """Build the agent-facing status from the budget ledger + split tiers. diff --git a/vero/src/vero/harbor/server.py b/vero/src/vero/harbor/server.py index d29b7a4..4dff350 100644 --- a/vero/src/vero/harbor/server.py +++ b/vero/src/vero/harbor/server.py @@ -11,6 +11,7 @@ import json import logging +import re import shutil from dataclasses import replace from pathlib import Path @@ -124,14 +125,20 @@ async def evaluate(self, req: EvalRequest, *, admin: bool = False) -> EvalSummar and sha == self.base_commit and not self._free_baseline_used ) - exp = await self.engine.evaluate( - replace(req, commit=sha), admin=admin, free=free_baseline - ) - # Consume the freebie only after the eval succeeded: an eval that - # raised (invalid split, infra failure) has given the agent nothing, - # so it must not burn the one free reference measurement. + # Claim the freebie BEFORE the await (asyncio is atomic between await + # points, so a concurrent second baseline eval sees the claim and pays) + # and refund it if the eval raises: a failed eval gave the agent + # nothing, so it must not burn the one free reference measurement. if free_baseline: self._free_baseline_used = True + try: + exp = await self.engine.evaluate( + replace(req, commit=sha), admin=admin, free=free_baseline + ) + except BaseException: + if free_baseline: + self._free_baseline_used = False + raise # Route with the agent's real tier even when the eval was unmetered. result_path = self._route_results(exp, admin=admin) budget_remaining = None @@ -282,31 +289,45 @@ def _route_results(self, experiment: Experiment, *, admin: bool) -> str | None: # 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. + results_root = self.agent_volume / "results" + if self._eval_seq == 0 and results_root.exists(): + # Volume reuse (sidecar restart): resume the ordinal past every + # surviving dir, or the first N evals of the new session would + # silently wipe __e1..__eN — the exact erasure this scheme exists + # to prevent. + self._eval_seq = max( + ( + int(m.group(1)) + for d in results_root.iterdir() + if (m := re.search(r"__e(\d+)$", d.name)) + ), + default=0, + ) 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 + dest = results_root / f"{split}__{commit[:12]}__e{self._eval_seq}" + if dest.exists(): # unreachable after the resume scan; 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 + # n_scored / n_errored / mean_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. + # mean_score_se is named to bind it to mean_score: both are computed + # over the zero-filled n_samples population (score() fills errored + # samples with 0.0), NOT over the n_scored subset — an SE of the + # 3-of-18-scored mean would be a different (and larger) number. 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 + mean_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 + mean_score_se = (var / len(filled)) ** 0.5 (dest / "summary.json").write_text( json.dumps( { @@ -320,7 +341,7 @@ def _route_results(self, experiment: Experiment, *, admin: bool) -> str | None: 1 for r in sample_results.values() if r.is_error() ), "mean_score": experiment.result.score(), - "score_se": score_se, + "mean_score_se": mean_score_se, "status": experiment.result.status.value, }, indent=2, diff --git a/vero/tests/test_harbor_protocol.py b/vero/tests/test_harbor_protocol.py index b7bd6f2..c24f42c 100644 --- a/vero/tests/test_harbor_protocol.py +++ b/vero/tests/test_harbor_protocol.py @@ -110,3 +110,17 @@ def test_advertises_subset_floor_on_non_viewable_only(self): by_split = {s["split"]: s for s in status.splits} assert by_split["validation"]["min_subset_samples"] == 5 assert by_split["train"]["min_subset_samples"] == 1 + + def test_default_floor_matches_sidecar_enforcement_default(self): + # A caller that forgets to pass the floor must not advertise a laxer + # one than EvaluationSidecar enforces by default (5): agents would + # send sub-floor requests that get rejected. + budget = { + ("validation", "ds1"): SplitBudget(split="validation", dataset_id="ds1", total_run_budget=3), + } + status = build_status( + submit_enabled=False, + budget=budget, + split_accesses=[SplitAccess.non_viewable("validation")], + ) + assert status.splits[0]["min_subset_samples"] == 5 diff --git a/vero/tests/test_harbor_server.py b/vero/tests/test_harbor_server.py index bafd4ee..937ecca 100644 --- a/vero/tests/test_harbor_server.py +++ b/vero/tests/test_harbor_server.py @@ -245,7 +245,8 @@ async def test_summary_carries_qualifiers(self, tmp_path): 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) + # SE of mean_score, i.e. over the zero-filled n_samples population + assert data["mean_score_se"] == pytest.approx(1 / 3) # sd .5774 / sqrt(3) # enum VALUE, not "ExperimentResultStatus.SUCCESS" assert data["status"] == "success" @@ -263,6 +264,18 @@ async def test_reevals_get_versioned_dirs(self, tmp_path): assert (Path(s1.result_path) / "summary.json").exists() assert (Path(s2.result_path) / "summary.json").exists() + @pytest.mark.asyncio + async def test_volume_reuse_resumes_ordinal_past_survivors(self, tmp_path): + # A restarted sidecar (fresh _eval_seq) on a reused volume must not + # wipe the prior session's __e1..__eN evidence. + survivor = tmp_path / "agent_vol" / "results" / "validation__old000000000__e7" + survivor.mkdir(parents=True) + (survivor / "summary.json").write_text("{}") + sidecar = _sidecar(tmp_path, split="validation") + s = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + assert s.result_path.endswith("__e8") + assert (survivor / "summary.json").exists() + class TestKAnonymityFloor: """Subset evals on non_viewable splits are floored: the aggregate response @@ -396,6 +409,22 @@ async def test_failed_free_eval_does_not_consume_freebie(self, tmp_path): await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) assert sidecar.engine.evaluate.await_args.kwargs["free"] is True + @pytest.mark.asyncio + async def test_freebie_claimed_before_eval_await(self, tmp_path): + # The claim must be visible DURING the eval await, or a concurrent + # second baseline eval would also resolve free_baseline=True and both + # would ride free (asyncio interleaves at await points). + sidecar = _sidecar(tmp_path, split="validation", base_commit="abcdef123456") + seen: list[bool] = [] + + async def _spy(req, **kwargs): + seen.append(sidecar._free_baseline_used) + return _experiment("validation") + + sidecar.engine.evaluate = AsyncMock(side_effect=_spy) + await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + assert seen == [True] + def test_status_surfaces_free_baseline(self, tmp_path): sidecar = _sidecar(tmp_path, split="train", base_commit="abcdef123456") s = sidecar.status() From 4b59b0befa0756ebf3921aff7966371ff7674b13 Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Thu, 9 Jul 2026 10:41:44 +0300 Subject: [PATCH 4/8] fix(harbor): floored rewards name their cause (deterministic candidate crash vs infra outage) Measured live in the transfer matrix: three champions scored 0/72 on an off-model executor because their optimizers hardcoded temperature=0, which that provider rejects. The harness handled it safely (rewards floored, finalize shipped) but the durable record could not say WHY, and "why" decides opposite actions: a deterministic candidate crash is a real, reportable portability failure; an infra outage means invalidate and re-run. Two changes: - Collation's no-verifier-rewards error string now names the dead attempts' exception types ("attempts died: UnsupportedParamsError x6"). The error string is the one field that flows to the DB, the per-sample files, and the verifier. - _admin_eval_score returns (score, failure_cause); a floored target's target_errors entry carries the dominant per-sample causes (frequency summary, top 3). Diagnostics are fail-safe: any surprise result shape degrades to a fixed string, never fails finalize. Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/harbor/runner.py | 22 +++++++++- vero/src/vero/harbor/verifier.py | 66 ++++++++++++++++++++++-------- vero/tests/test_harbor_runner.py | 14 +++++++ vero/tests/test_harbor_verifier.py | 27 ++++++++++++ 4 files changed, 111 insertions(+), 18 deletions(-) diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py index d5265c7..d76d0e4 100644 --- a/vero/src/vero/harbor/runner.py +++ b/vero/src/vero/harbor/runner.py @@ -409,8 +409,28 @@ def _out(output: dict) -> dict: ) rewards = (trial.get("verifier_result") or {}).get("rewards") or {} if not rewards: + # Name the cause in the error string itself: it is the one field + # that flows everywhere (DB, per-sample files, the verifier's + # target_errors), and "no verifier rewards" alone cannot separate + # a champion that crashes deterministically on this executor from + # an infra outage — a distinction that decides whether the cell is + # a measurement or a re-run. + cause = "" + if attempts: + dead = {} + for t in attempts: + if (t.get("verifier_result") or {}).get("rewards"): + continue + exc = (t.get("exception_info") or {}).get("exception_type") + key = exc or "no_rewards_recorded" + dead[key] = dead.get(key, 0) + 1 + if dead: + causes = ", ".join( + f"{k} x{v}" for k, v in sorted(dead.items(), key=lambda i: -i[1]) + ) + cause = f" (attempts died: {causes})" return SampleResult( - error=f"No verifier rewards for task '{task_name}'.", + error=f"No verifier rewards for task '{task_name}'.{cause}", # The agent died before scoring. A candidate edit that CRASHES # the agent lands here, and "no verifier rewards" alone gives # the optimizer no way to see its own crash; the transcript diff --git a/vero/src/vero/harbor/verifier.py b/vero/src/vero/harbor/verifier.py index 5b41d9f..813a1bf 100644 --- a/vero/src/vero/harbor/verifier.py +++ b/vero/src/vero/harbor/verifier.py @@ -137,7 +137,7 @@ async def _finalize(self) -> dict: rewards: dict[str, float] = {} target_errors: dict[str, str] = {} for target in self.targets: - score = await self._admin_eval_score( + score, cause = await self._admin_eval_score( task=target.task, dataset_id=target.dataset_id, split=target.split, @@ -147,13 +147,18 @@ async def _finalize(self) -> dict: ) if score is None: # Persistent failure: floor the target so reward.json still - # ships, and record the failure in the wrapper (echoed to the - # trial's durable stdout) so a floored-by-outage reward can - # never masquerade as a measured 0.0. + # ships, and record the failure WITH ITS CAUSE in the wrapper + # (echoed to the trial's durable stdout). A floored-by-outage + # reward must never masquerade as a measured 0.0, and the cause + # separates the two floored cases that demand opposite actions: + # a champion that deterministically crashes on this target's + # executor (a real, reportable portability failure) vs an infra + # outage (invalidate and re-run). rewards[target.reward_key] = float(default_minimum_score) target_errors[target.reward_key] = ( f"eval failed after {self._baseline_score_attempts} attempt(s); " f"reward floored, not measured" + + (f"; cause: {cause}" if cause else "") ) else: rewards[target.reward_key] = score @@ -172,16 +177,19 @@ async def _admin_eval_score( commit: str, sample_ids: list[int] | None = None, what: str, - ) -> float | None: + ) -> tuple[float | None, str | None]: """One reward-critical admin eval with bounded retry. - Returns the eval's score with errored samples counted 0.0 (min-fill: - an errored sample is a failed measurement of the candidate, and - excluding it would reward candidates whose failures error out rather - than score). An eval in which NO sample scored is indistinguishable - from an infrastructure outage and must never quietly become 0.0, so it - is retried like an exception; ``None`` after the last attempt means - "could not measure", and the caller decides the fail-safe. + Returns ``(score, failure_cause)``. The score counts errored samples + as 0.0 (min-fill: an errored sample is a failed measurement of the + candidate, and excluding it would reward candidates whose failures + error out rather than score). An eval in which NO sample scored is + indistinguishable from an infrastructure outage and must never quietly + become 0.0, so it is retried like an exception; ``(None, cause)`` + after the last attempt means "could not measure", and the caller + decides the fail-safe. ``cause`` summarizes the dominant per-sample + errors so the durable record can distinguish a deterministically + crashing candidate from an outage. """ last_error: Exception | str | None = None for attempt in range(1, self._baseline_score_attempts + 1): @@ -194,7 +202,10 @@ async def _admin_eval_score( sample_ids=sample_ids, ) if exp.result.score(fill_score=None) is None: - last_error = "eval scored no samples (all errored or empty)" + last_error = ( + "eval scored no samples (all errored or empty); " + + self._dominant_sample_errors(exp) + ) logger.warning( "%s attempt %d/%d: %s", what, attempt, self._baseline_score_attempts, last_error, @@ -207,7 +218,7 @@ async def _admin_eval_score( # other unmeasurable outcome, never bypass the loop. last_error = "eval returned no aggregate score" continue - return float(score) + return float(score), None except Exception as exc: # noqa: BLE001 - retried, then surfaced as None last_error = exc logger.warning( @@ -218,7 +229,28 @@ async def _admin_eval_score( "%s failed after %d attempt(s): %s", what, self._baseline_score_attempts, last_error, ) - return None + return None, str(last_error) if last_error is not None else None + + @staticmethod + def _dominant_sample_errors(exp) -> str: + """Frequency summary of the per-sample error strings of an experiment + (e.g. "12x: No verifier rewards for task ... (attempts died: + UnsupportedParamsError x6)"). One identical cause across every sample + is the signature of a deterministic candidate crash; a mixed bag points + at infra. Top few only: the value is the shape, not the full list. + Diagnostics must never fail finalize, so any surprise shape degrades + to a fixed string instead of raising.""" + try: + counts: dict[str, int] = {} + for r in exp.result.sample_results.values(): + if r.error: + counts[r.error] = counts.get(r.error, 0) + 1 + if not counts: + return "no per-sample errors recorded" + top = sorted(counts.items(), key=lambda i: -i[1])[:3] + return "; ".join(f"{n}x: {err}" for err, n in top) + except Exception: # noqa: BLE001 - diagnostics only + return "no per-sample errors recorded" async def _maybe_score_baseline(self, rewards: dict[str, float]) -> dict: """Admin-score the unmodified baseline on every target and report it. @@ -426,7 +458,7 @@ async def _best_from_db(self) -> str: for idx, (_, row) in enumerate(shortlist.iterrows()): commit = row["candidate_commit"] dataset_id = row.get("dataset_subset_dataset_id") - score = await self._admin_eval_score( + score, _cause = await self._admin_eval_score( task=self.selection_task, dataset_id=dataset_id, split=self.selection_split, @@ -462,7 +494,7 @@ async def _best_from_db(self) -> str: base_dataset_id = self.selection_dataset_id if base_dataset_id is None: base_dataset_id = shortlist.iloc[0].get("dataset_subset_dataset_id") - base_score_opt = await self._admin_eval_score( + base_score_opt, _cause = await self._admin_eval_score( task=self.selection_task, dataset_id=base_dataset_id, split=self.selection_split, diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py index 9642403..76ee062 100644 --- a/vero/tests/test_harbor_runner.py +++ b/vero/tests/test_harbor_runner.py @@ -663,6 +663,20 @@ def test_no_rewards_error_sample_carries_crash_transcript(self, tmp_path): assert r.error is not None assert r.feedback == "crash tail" + def test_no_rewards_error_names_dead_exception_types(self, tmp_path): + # The error string must carry WHY the attempts died: it is the one + # field that flows to the DB, the per-sample files, and the verifier's + # target_errors, and it separates a deterministic candidate crash + # (measured live: 72/72 UnsupportedParamsError on an off-model + # executor) from an infra outage. + runner = _fb_runner() + jobs = tmp_path / "jobs" + _write_trial(jobs, "trial0", "t0", None, exception_type="UnsupportedParamsError") + _write_trial(jobs, "trial1", "t0", None, exception_type="UnsupportedParamsError") + r = self._result(runner, jobs) + assert r.error is not None + assert "UnsupportedParamsError x2" in r.error + 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. diff --git a/vero/tests/test_harbor_verifier.py b/vero/tests/test_harbor_verifier.py index 72e4ec4..3d000a4 100644 --- a/vero/tests/test_harbor_verifier.py +++ b/vero/tests/test_harbor_verifier.py @@ -755,6 +755,33 @@ async def _admin(*, task, dataset_id, split, commit, sample_ids=None): assert engine.evaluate_admin.await_args.kwargs["commit"] == "base" assert engine.evaluate_admin.await_args.kwargs["split"] == "test" + @pytest.mark.asyncio + async def test_floored_target_records_dominant_crash_cause(self, tmp_path): + # An all-errored target eval floors the reward AND names the dominant + # per-sample cause in target_errors: a champion that deterministically + # crashes on this target's executor must be distinguishable from an + # infra outage in the one durable record. + self._submit(tmp_path) + engine = MagicMock() + crash = "No verifier rewards for task 't'. (attempts died: UnsupportedParamsError x6)" + exp = MagicMock() + exp.result.score = MagicMock(return_value=None) + exp.result.sample_results = { + i: MagicMock(error=crash) for i in range(3) + } + engine.evaluate_admin = AsyncMock(return_value=exp) + v = Verifier( + engine=engine, + admin_volume=tmp_path, + reward_mode="submit", + baseline_score_attempts=2, + targets=[VerificationTarget(task="t", dataset_id="ds", split="test", reward_key="reward")], + ) + result = await v.finalize() + assert result["rewards"] == {"reward": 0.0} + assert "UnsupportedParamsError x6" in result["target_errors"]["reward"] + assert "3x:" in result["target_errors"]["reward"] + @pytest.mark.asyncio async def test_target_eval_retries_then_floors_with_error_marker(self, tmp_path): # A persistently failing target eval floors the reward (reward.json From 1472e04c7a436402449cf683182a605b3333f418 Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Thu, 9 Jul 2026 15:25:51 +0300 Subject: [PATCH 5/8] fix(harbor): dominant-cause grouping normalizes task names so cross-task crashes cluster Greptile follow-up on #37: _dominant_sample_errors keyed on the raw error string, which embeds the task name, so identical exceptions across a multi-task slice landed in 1x singletons instead of one dominant cause. Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/harbor/verifier.py | 9 ++++++++- vero/tests/test_harbor_verifier.py | 31 ++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/vero/src/vero/harbor/verifier.py b/vero/src/vero/harbor/verifier.py index 813a1bf..7285668 100644 --- a/vero/src/vero/harbor/verifier.py +++ b/vero/src/vero/harbor/verifier.py @@ -15,6 +15,7 @@ import asyncio import json import logging +import re from dataclasses import dataclass from pathlib import Path from typing import Literal @@ -244,7 +245,13 @@ def _dominant_sample_errors(exp) -> str: counts: dict[str, int] = {} for r in exp.result.sample_results.values(): if r.error: - counts[r.error] = counts.get(r.error, 0) + 1 + # Group on the CAUSE, not the sample: runner errors embed + # the task name ("No verifier rewards for task 'x/y'..."), + # so keying on the raw string would leave every sample in + # its own 1x bucket and a deterministic crash across a + # multi-task slice would read as a mixed bag. + key = re.sub(r"for task '[^']*'", "for task '…'", r.error) + counts[key] = counts.get(key, 0) + 1 if not counts: return "no per-sample errors recorded" top = sorted(counts.items(), key=lambda i: -i[1])[:3] diff --git a/vero/tests/test_harbor_verifier.py b/vero/tests/test_harbor_verifier.py index 3d000a4..dade7c3 100644 --- a/vero/tests/test_harbor_verifier.py +++ b/vero/tests/test_harbor_verifier.py @@ -782,6 +782,37 @@ async def test_floored_target_records_dominant_crash_cause(self, tmp_path): assert "UnsupportedParamsError x6" in result["target_errors"]["reward"] assert "3x:" in result["target_errors"]["reward"] + @pytest.mark.asyncio + async def test_crash_cause_clusters_across_task_names(self, tmp_path): + # Runner error strings embed the task name; a slice spanning several + # tasks that all die of the SAME exception must still cluster as one + # dominant cause (grouping normalizes the task name away), or the + # deterministic-crash signature degrades into 1x singletons. + self._submit(tmp_path) + engine = MagicMock() + exp = MagicMock() + exp.result.score = MagicMock(return_value=None) + exp.result.sample_results = { + i: MagicMock( + error=( + f"No verifier rewards for task 'org/task-{i}'. " + f"(attempts died: UnsupportedParamsError x6)" + ) + ) + for i in range(3) + } + engine.evaluate_admin = AsyncMock(return_value=exp) + v = Verifier( + engine=engine, + admin_volume=tmp_path, + reward_mode="submit", + baseline_score_attempts=2, + targets=[VerificationTarget(task="t", dataset_id="ds", split="test", reward_key="reward")], + ) + result = await v.finalize() + assert "3x:" in result["target_errors"]["reward"] + assert "UnsupportedParamsError x6" in result["target_errors"]["reward"] + @pytest.mark.asyncio async def test_target_eval_retries_then_floors_with_error_marker(self, tmp_path): # A persistently failing target eval floors the reward (reward.json From de76acf7eb4772dc539b091b437c0641624008de Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Thu, 9 Jul 2026 10:49:25 +0300 Subject: [PATCH 6/8] feat(harbor): transfer targets: per-target executor-model override at finalize Home-model evals cannot see model-specific couplings the optimizer bakes in. Measured live in the wave-1 transfer matrix: three of five champions independently hardcoded temperature=0 (a variance trick on their home model, gpt-4.1-mini) and scored 0/72 on claude-opus-4-8, which rejects it, while looking healthy on every eval the optimization loop ever ran. The portability failure was invisible until a separate, manual, after-the-fact probe. VerificationTarget gains `model`: a target with an executor override scores the selected commit under a model it was NOT optimized on, in the same finalize battery as its home-model reward. The baseline is scored under the same override so the comparison stays like-for-like. Plumbing: build.yaml TargetSpec -> compiler -> serve.json _TargetCfg -> VerificationTarget -> engine.evaluate_admin(model=...) -> task_params ["harbor_model_override"] -> HarborRunner -m flag. The override rides task_params, so Mode A ignores it and the runner needs no new state; the shared run_constraints are copied, never mutated. Test fakes of evaluate_admin widened to accept the new kwarg (a strict signature turned the new call into a retried TypeError, flooring rewards, which is itself a nice demonstration of the floor's fail-safe). Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/evaluation/engine.py | 19 +++++++++- vero/src/vero/harbor/build/compiler.py | 1 + vero/src/vero/harbor/build/config.py | 6 ++++ vero/src/vero/harbor/runner.py | 8 +++-- vero/src/vero/harbor/serve.py | 2 ++ vero/src/vero/harbor/verifier.py | 15 ++++++++ vero/tests/test_engine.py | 19 ++++++++++ vero/tests/test_harbor_build.py | 28 +++++++++++++++ vero/tests/test_harbor_runner.py | 13 +++++++ vero/tests/test_harbor_verifier.py | 48 ++++++++++++++++++++------ 10 files changed, 145 insertions(+), 14 deletions(-) diff --git a/vero/src/vero/evaluation/engine.py b/vero/src/vero/evaluation/engine.py index 201f51c..b2f2196 100644 --- a/vero/src/vero/evaluation/engine.py +++ b/vero/src/vero/evaluation/engine.py @@ -193,6 +193,7 @@ async def evaluate_admin( split: str, commit: str, sample_ids: list[int] | None = None, + model: str | None = None, ) -> Experiment: """Admin/verifier evaluation: explicit ``task``, no budget, no allowlist. @@ -200,7 +201,23 @@ async def evaluate_admin( this scores an arbitrary ``(task, dataset_id, split)`` — including held-out tasks/splits the agent never had access to. Used by the verifier to score the selected commit on its configured targets. + + ``model`` overrides the executor model for this one eval (rides + ``task_params`` so the eval strategy can honor it; the Mode-B + HarborRunner does, the Mode-A vero-task path ignores it). Used for + transfer targets: scoring the champion under a model it was NOT + optimized on. """ + params = self.run_constraints + if model is not None: + params = params.model_copy( + update={ + "task_params": { + **(params.task_params or {}), + "harbor_model_override": model, + } + } + ) return await self.evaluator.evaluate( commit=commit, dataset_id=dataset_id, @@ -208,7 +225,7 @@ async def evaluate_admin( task=task, sample_ids=sample_ids, db=self.db, - evaluation_parameters=self.run_constraints, + evaluation_parameters=params, ) def status(self) -> dict[tuple[str, str], SplitBudget]: diff --git a/vero/src/vero/harbor/build/compiler.py b/vero/src/vero/harbor/build/compiler.py index 61123b8..dc42431 100644 --- a/vero/src/vero/harbor/build/compiler.py +++ b/vero/src/vero/harbor/build/compiler.py @@ -236,6 +236,7 @@ def _serve_config( "split": t.split, "reward_key": t.reward_key, "sample_ids": t.sample_ids, + "model": t.model, } for t in config.targets ], diff --git a/vero/src/vero/harbor/build/config.py b/vero/src/vero/harbor/build/config.py index 5b242f3..6893847 100644 --- a/vero/src/vero/harbor/build/config.py +++ b/vero/src/vero/harbor/build/config.py @@ -36,6 +36,12 @@ class TargetSpec(BaseModel): split: str reward_key: str = "reward" sample_ids: list[int] | None = None + # Executor-model override for this target (transfer probe; Mode B only): + # score the selected commit AND the baseline under a model it was not + # optimized on, so model-specific couplings (measured live: hardcoded + # temperature=0 crashing 72/72 on an executor that rejects it) surface at + # finalize instead of one substrate away. + model: str | None = None class _BuildConfigBase(BaseModel): diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py index d76d0e4..4c6042b 100644 --- a/vero/src/vero/harbor/runner.py +++ b/vero/src/vero/harbor/runner.py @@ -129,8 +129,12 @@ def _build_command( "--n-attempts", str(c.n_attempts), "--max-retries", str(c.max_retries), ] - if c.model: - cmd += ["-m", c.model] + # Per-eval executor override (transfer targets): the verifier scores + # the champion under a model it was not optimized on. Rides + # task_params so it needs no runner state. + model = (params.task_params or {}).get("harbor_model_override") or c.model + if model: + cmd += ["-m", str(model)] for task_name in task_names: cmd += ["-i", task_name] cmd += ["--jobs-dir", str(jobs_dir), *c.extra_args] diff --git a/vero/src/vero/harbor/serve.py b/vero/src/vero/harbor/serve.py index 52c73e0..bd42329 100644 --- a/vero/src/vero/harbor/serve.py +++ b/vero/src/vero/harbor/serve.py @@ -43,6 +43,8 @@ class _TargetCfg(BaseModel): split: str reward_key: str = "reward" sample_ids: list[int] | None = None + # Executor-model override for this target (transfer probe; Mode B only). + model: str | None = None class _ServeConfigBase(BaseModel): diff --git a/vero/src/vero/harbor/verifier.py b/vero/src/vero/harbor/verifier.py index 7285668..63f588c 100644 --- a/vero/src/vero/harbor/verifier.py +++ b/vero/src/vero/harbor/verifier.py @@ -39,6 +39,15 @@ class VerificationTarget: split: str reward_key: str sample_ids: list[int] | None = None # None = full split + # Executor-model override for this target (Mode B): score the selected + # commit under a DIFFERENT model than the one it was optimized on. This is + # the transfer probe: home-model evals cannot see model-specific couplings + # the optimizer bakes in (measured live: three champions independently + # hardcoded temperature=0 and scored 0/72 on an executor that rejects it, + # while looking healthy on every home-model eval). None = the task's + # configured model. The baseline is scored under the same override, so the + # comparison stays like-for-like. + model: str | None = None class Verifier: @@ -144,6 +153,7 @@ async def _finalize(self) -> dict: split=target.split, commit=sha, sample_ids=target.sample_ids, + model=target.model, what=f"target '{target.reward_key}'", ) if score is None: @@ -177,6 +187,7 @@ async def _admin_eval_score( split: str, commit: str, sample_ids: list[int] | None = None, + model: str | None = None, what: str, ) -> tuple[float | None, str | None]: """One reward-critical admin eval with bounded retry. @@ -201,6 +212,7 @@ async def _admin_eval_score( split=split, commit=commit, sample_ids=sample_ids, + model=model, ) if exp.result.score(fill_score=None) is None: last_error = ( @@ -293,12 +305,15 @@ async def _maybe_score_baseline(self, rewards: dict[str, float]) -> dict: try: baselines: dict[str, float] = {} for target in self.targets: + # Same executor override as the candidate's target eval, or + # the baseline comparison is not like-for-like. exp = await self.engine.evaluate_admin( task=target.task, dataset_id=target.dataset_id, split=target.split, commit=self.base_commit, sample_ids=target.sample_ids, + model=target.model, ) if exp.result.score(fill_score=None) is None: # All-error/empty is an outage, not a 0.0 baseline: a diff --git a/vero/tests/test_engine.py b/vero/tests/test_engine.py index 401e6b9..3ac0022 100644 --- a/vero/tests/test_engine.py +++ b/vero/tests/test_engine.py @@ -182,6 +182,25 @@ async def test_free_runs_without_debiting_budget(self, monkeypatch): assert svc.status()[("dev", "ds1")].remaining_run_budget == 3 assert svc.status()[("dev", "ds1")].remaining_sample_budget == 100 + @pytest.mark.asyncio + async def test_admin_model_override_rides_task_params(self, monkeypatch): + # Transfer targets: evaluate_admin(model=...) must reach the eval + # strategy via task_params without mutating the shared run_constraints. + svc = _make_service(monkeypatch=monkeypatch) + await svc.evaluate_admin( + task="t", dataset_id="ds1", split="dev", commit="c1", model="openai/gpt-4o" + ) + params = svc.evaluator.evaluate.await_args.kwargs["evaluation_parameters"] + assert params.task_params["harbor_model_override"] == "openai/gpt-4o" + assert svc.run_constraints.task_params.get("harbor_model_override") is None + + @pytest.mark.asyncio + async def test_admin_no_override_keeps_shared_constraints(self, monkeypatch): + svc = _make_service(monkeypatch=monkeypatch) + await svc.evaluate_admin(task="t", dataset_id="ds1", split="dev", commit="c1") + params = svc.evaluator.evaluate.await_args.kwargs["evaluation_parameters"] + assert params is svc.run_constraints + @pytest.mark.asyncio async def test_free_does_not_bypass_no_access_gate(self, monkeypatch): from vero.core.dataset import SplitAccess diff --git a/vero/tests/test_harbor_build.py b/vero/tests/test_harbor_build.py index 46983bf..a9aa8d7 100644 --- a/vero/tests/test_harbor_build.py +++ b/vero/tests/test_harbor_build.py @@ -363,6 +363,34 @@ def test_instruction_omits_free_baseline_claim_when_unsupported(built): assert "budget-free" not in text +def test_target_model_reaches_serve_json(tmp_path, monkeypatch): + # Transfer probe: a target's executor-model override must survive the + # compile into serve.json and validate back through ServeConfig. + monkeypatch.setenv("VERO_SKIP_SECRET_CHECK", "1") + config = BuildConfigA( + name="vero/gsm8k-opt", + agent_repo=str(_agent_repo(tmp_path)), + task="gsm8k", + task_module="gsm8k_agent.vero_tasks", + dataset=str(_dataset(tmp_path)), + splits=[ + {"split": "validation", "access": "non_viewable"}, + {"split": "test", "access": "no_access"}, + ], + budgets=[{"split": "validation", "total_run_budget": 5}], + selection_split="validation", + targets=[ + {"split": "test", "reward_key": "reward"}, + {"split": "test", "reward_key": "reward_4o", "model": "openai/gpt-4o"}, + ], + ) + out = compile_task(config, tmp_path / "task", vero_root=_stub_vero(tmp_path)) + cfg = load_serve_config(out / "environment" / "sidecar" / "serve.json") + by_key = {t.reward_key: t for t in cfg.targets} + assert by_key["reward_4o"].model == "openai/gpt-4o" + assert by_key["reward"].model is None + + def test_instruction_renders_exhaust_budget_by_default(built): text = (built / "instruction.md").read_text() assert "Unspent budget is wasted" in text diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py index 76ee062..222e6d0 100644 --- a/vero/tests/test_harbor_runner.py +++ b/vero/tests/test_harbor_runner.py @@ -117,6 +117,19 @@ def test_no_harbor_requirement_keeps_candidate_env(self): cmd = _runner()._build_command("/wt", _params(), ["t0"], Path("/jobs")) assert "--with" not in cmd + def test_model_override_beats_configured_model(self): + # Transfer targets: a per-eval executor override (via task_params) + # wins over the task's configured model. + params = _params() + params.task_params = {"harbor_model_override": "openai/gpt-4o"} + cmd = _runner()._build_command("/wt", params, ["t0"], Path("/jobs")) + m = cmd[cmd.index("-m") + 1] + assert m == "openai/gpt-4o" + + def test_no_override_uses_configured_model(self): + cmd = _runner()._build_command("/wt", _params(), ["t0"], Path("/jobs")) + assert cmd[cmd.index("-m") + 1] == "anthropic/x" + class TestExtractReward: def test_priority_pass_then_reward_then_sole_key(self): diff --git a/vero/tests/test_harbor_verifier.py b/vero/tests/test_harbor_verifier.py index dade7c3..55714bd 100644 --- a/vero/tests/test_harbor_verifier.py +++ b/vero/tests/test_harbor_verifier.py @@ -89,7 +89,7 @@ async def test_auto_best_reranks_by_admin_score(self, tmp_path): ) admin_scores = {"hi": 0.1, "lo": 0.95} - async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None): return MagicMock( result=MagicMock(score=MagicMock(return_value=admin_scores.get(commit, 0.99))) ) @@ -125,7 +125,7 @@ async def test_auto_best_excludes_baseline_from_ranking(self, tmp_path): } ) - async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None): return MagicMock(result=MagicMock(score=MagicMock(return_value=0.7))) engine.evaluate_admin = AsyncMock(side_effect=_admin) @@ -172,7 +172,7 @@ async def test_lucky_subset_eval_does_not_outrank_full_split(self, tmp_path): } ) - async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=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))) @@ -212,7 +212,7 @@ async def test_all_subset_evals_still_rankable(self, tmp_path): } ) - async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None): return MagicMock(result=MagicMock(score=MagicMock(return_value=0.5))) engine.evaluate_admin = AsyncMock(side_effect=_admin) @@ -260,7 +260,7 @@ async def test_reverts_to_base_when_no_candidate_beats_baseline(self, tmp_path): # agent admin-scores 0.2 on the selection split; base admin-scores 0.3; # the reverted base scores 0.35 on the target split (distinct values so the # assertions can tell the target eval apart from the floor comparison). - async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None): if commit == "base": score = 0.35 if split == "validation" else 0.3 else: @@ -296,7 +296,7 @@ async def test_exact_tie_reverts_to_base(self, tmp_path): engine = MagicMock() engine.db.get_experiments_df.return_value = self._df() - async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None): return MagicMock(result=MagicMock(score=MagicMock(return_value=0.3))) # all equal engine.evaluate_admin = AsyncMock(side_effect=_admin) @@ -327,7 +327,7 @@ async def test_floor_noop_without_base_commit(self, tmp_path): } ) - async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None): return MagicMock(result=MagicMock(score=MagicMock(return_value=0.5))) engine.evaluate_admin = AsyncMock(side_effect=_admin) @@ -349,7 +349,7 @@ async def test_keeps_candidate_that_beats_baseline(self, tmp_path): engine = MagicMock() engine.db.get_experiments_df.return_value = self._df() - async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None): score = 0.3 if commit == "base" else 0.6 # agent genuinely improves return MagicMock(result=MagicMock(score=MagicMock(return_value=score))) @@ -374,7 +374,7 @@ async def test_floor_off_ships_least_bad_candidate(self, tmp_path): engine = MagicMock() engine.db.get_experiments_df.return_value = self._df() - async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None): return MagicMock(result=MagicMock(score=MagicMock(return_value=0.2))) engine.evaluate_admin = AsyncMock(side_effect=_admin) @@ -476,7 +476,7 @@ async def test_candidates_present_keeps_normal_selection(self, tmp_path): } ) - async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None): return MagicMock(result=MagicMock(score=MagicMock(return_value=0.5))) engine.evaluate_admin = AsyncMock(side_effect=_admin) @@ -733,7 +733,7 @@ async def test_floor_fail_safe_reverts_on_unmeasurable_baseline(self, tmp_path): } ) - async def _admin(*, task, dataset_id, split, commit, sample_ids=None): + async def _admin(*, task, dataset_id, split, commit, sample_ids=None, model=None): if commit == "base" and split == "validation": raise RuntimeError("nested run crashed") return MagicMock(result=MagicMock(score=MagicMock(return_value=0.9))) @@ -755,6 +755,32 @@ async def _admin(*, task, dataset_id, split, commit, sample_ids=None): assert engine.evaluate_admin.await_args.kwargs["commit"] == "base" assert engine.evaluate_admin.await_args.kwargs["split"] == "test" + @pytest.mark.asyncio + async def test_transfer_target_model_reaches_champion_and_baseline_evals(self, tmp_path): + # A target's executor-model override must apply to BOTH the champion + # eval and the baseline eval, or the comparison is not like-for-like. + self._submit(tmp_path) + engine = MagicMock() + engine.evaluate_admin = AsyncMock( + return_value=MagicMock(result=MagicMock(score=MagicMock(return_value=0.5))) + ) + v = Verifier( + engine=engine, + admin_volume=tmp_path, + reward_mode="submit", + score_baseline=True, + base_commit="base", + targets=[VerificationTarget( + task="t", dataset_id="ds", split="test", + reward_key="reward_4o", model="openai/gpt-4o", + )], + ) + await v.finalize() + models = [c.kwargs["model"] for c in engine.evaluate_admin.await_args_list] + commits = [c.kwargs["commit"] for c in engine.evaluate_admin.await_args_list] + assert models == ["openai/gpt-4o", "openai/gpt-4o"] + assert set(commits) == {"cand", "base"} + @pytest.mark.asyncio async def test_floored_target_records_dominant_crash_cause(self, tmp_path): # An all-errored target eval floors the reward AND names the dominant From d0a426b2569b3e2124cb3195a2520b6bd1fd1295 Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Thu, 9 Jul 2026 16:37:30 +0300 Subject: [PATCH 7/8] feat(harbor): infra resilience: dead-attempt classification, opt-in outage retry, key-budget alarm Three infra failure modes measured live in the E5 matrix runs, fixed at the measurement layer: - Dead attempts are classified infra vs candidate (conservative exception-type allowlist + the litellm key-budget message signature). Labels flow into dead_exception_types, error strings, and a new n_dead_infra metric. Classification never moves a score: every dead attempt still zero-fills, or faking infra would excuse failures. Exception type names are candidate-authored, so brackets are neutralized before labeling (a class named 'XError[infra]' cannot walk in pre-suffixed). - An OPT-IN, bounded, backoff-spaced within-eval retry re-measures samples whose every attempt died of a transient infra cause (the 65-second DNS blip that killed 44/72 attempts of one eval). Off by default: against an adversarial optimizer the qualifying predicate is a re-roll lever, since a stochastic candidate that raises allowlisted exceptions on failing attempts converts all-bad rounds into fresh draws. Retry rounds run in fresh sibling jobs dirs (nesting would pool dead attempts into later resumed means) and recovered samples carry an infra_retry audit marker naming the discarded attempts. - An ERROR-level alarm names key-budget exhaustion (a spent key fails every later call identically; two matrix cells of budget-exceeded zeros were nearly booked as a portability finding), with hedged wording since the signature reads candidate-process exceptions. Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/harbor/config.py | 26 +++ vero/src/vero/harbor/runner.py | 244 ++++++++++++++++++++++-- vero/tests/test_harbor_runner.py | 314 ++++++++++++++++++++++++++++++- 3 files changed, 572 insertions(+), 12 deletions(-) diff --git a/vero/src/vero/harbor/config.py b/vero/src/vero/harbor/config.py index dea857f..3642227 100644 --- a/vero/src/vero/harbor/config.py +++ b/vero/src/vero/harbor/config.py @@ -40,6 +40,32 @@ class HarborConfig: # without running anything). None keeps the current behavior: the # candidate env supplies harbor, and is trusted to. harbor_requirement: str | None = None + # Bounded within-eval retry for infra-destroyed samples. A sample whose + # EVERY attempt died of a transient infrastructure cause (connection, + # timeout, rate limit, 5xx) was never measured at all: re-run it after a + # backoff instead of booking the outage as a permanent error. Measured + # live: a 65-second host DNS blip killed 44 of 72 attempts of one eval + # with ConnectionError, and nothing in the record distinguished the blip + # from a bad candidate. + # + # OFF BY DEFAULT, and it must stay off when the candidate is an + # adversarial optimizer. The qualifying predicate is built from exception + # types raised inside candidate code, and agents are stochastic: a + # candidate that raises an allowlisted exception whenever an attempt is + # going badly loses nothing on partially-good samples (its fakes + # zero-fill like honest failures) but converts every all-bad sample from + # a booked 0.0 into a fresh re-roll. That is one-sided selection over + # attempt sets, exactly what the zero-fill invariant exists to prevent. + # Enable only for trusted-candidate evaluations (frozen agents, + # operator-run matrices), where re-measuring an outage is pure signal + # recovery. Candidate crashes and exhausted key budgets never retry + # regardless (a crash is a result; a spent key cannot recover by + # waiting), and recovered samples carry an ``infra_retry`` audit marker. + infra_retry_rounds: int = 0 + # Backoff before retry round N is N times this many seconds. Transient + # infra needs time, not immediacy: instant retries burned 6 of 8 run + # attempts inside one live DNS blip. + infra_retry_delay_s: float = 30.0 extra_args: list[str] = field(default_factory=list) # passthrough harbor run flags def __post_init__(self) -> None: diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py index 4c6042b..51a9e16 100644 --- a/vero/src/vero/harbor/runner.py +++ b/vero/src/vero/harbor/runner.py @@ -10,6 +10,7 @@ from __future__ import annotations +import asyncio import json import logging from pathlib import Path @@ -31,6 +32,68 @@ logger = logging.getLogger(__name__) +# Dead-attempt classification. An attempt that died of one of these exception +# types never got a fair shot at the task: the model endpoint, the network, or +# the key quota failed before the candidate could be measured. Classification +# is diagnostic and retry-gating ONLY: every dead attempt still scores 0.0 +# regardless of class, because excusing infra-labeled deaths from the score +# would hand the candidate a lever (raise ConnectionError on hard tasks and +# have those attempts dropped). Conservative allowlist: anything unlisted +# counts against the candidate. +_INFRA_EXCEPTION_TYPES = frozenset({ + "APIConnectionError", + "APITimeoutError", + "ConnectTimeout", + "ConnectionError", + "InternalServerError", + "RateLimitError", + "ReadTimeout", + "ServiceUnavailableError", + "Timeout", + "TimeoutError", +}) +_INFRA_SUFFIX = "[infra]" +# litellm surfaces a spent key budget as a BadRequestError; only the message +# distinguishes it from a candidate-caused bad request. Infra, but PERSISTENT: +# waiting cannot refill a key, so it alarms instead of retrying. (Measured +# live: a key crossed its spend cap mid-matrix and two cells of BadRequestError +# zeros were nearly booked as a portability finding.) +_KEY_BUDGET_MARKER = "budget has been exceeded" +_KEY_BUDGET_SUFFIX = "[infra:llm-key-budget]" + + +def _dead_attempt_label(exception_info: dict | None) -> str: + """Cause label for one dead attempt; infra causes carry a class suffix so + every downstream record (metrics, error strings, per-sample output) shows + at a glance whether the deaths measure the candidate or the plumbing. An + attempt can die with no recorded exception at all (the verifier simply + produced no rewards); keep it countable.""" + info = exception_info or {} + exc = info.get("exception_type") + if not exc: + return "no_rewards_recorded" + # The class suffixes below are load-bearing (retry gating, the + # n_dead_infra metric, the key-budget alarm) and round-trip through + # persisted output, while the exception type name is candidate-authored: + # a class literally named "XError[infra]" must not walk in pre-suffixed. + # Neutralize brackets before classifying; genuine types contain none. + exc = exc.replace("[", "(").replace("]", ")") + if exc == "BadRequestError" and _KEY_BUDGET_MARKER in ( + info.get("exception_message") or "" + ).lower(): + return f"{exc}{_KEY_BUDGET_SUFFIX}" + if exc in _INFRA_EXCEPTION_TYPES: + return f"{exc}{_INFRA_SUFFIX}" + return exc + + +def _is_infra(label: str) -> bool: + return label.endswith((_INFRA_SUFFIX, _KEY_BUDGET_SUFFIX)) + + +def _is_transient_infra(label: str) -> bool: + return label.endswith(_INFRA_SUFFIX) + class HarborRunner: """Mode-B EvalStrategy: nested `harbor run` + collate -> SampleResults.""" @@ -76,6 +139,147 @@ async def produce_sample_results( str(workspace.project_path), params, [t for _, t in pending], jobs_dir ) self._collate(jobs_dir, pairs, params, ran=[t for _, t in pending]) + await self._retry_transient_infra(workspace, params, pairs, jobs_dir) + + async def _retry_transient_infra( + self, + workspace: Workspace, + params: EvaluationParameters, + pairs: list[tuple[int, str]], + jobs_dir: Path, + ) -> None: + """Bounded re-run of samples that were outaged, not measured. + + Scope is deliberately narrow: only samples whose EVERY attempt died of + a transient infra cause qualify. A partially scored sample is a noisy + measurement whose infra-dead attempts stay zero-filled; a candidate + crash is a result; an exhausted key budget cannot recover by waiting. + Measured live: a 65-second host DNS blip killed 44 of 72 attempts of + one eval with ConnectionError, and with no pause between attempts the + whole burst fit inside the blip, booking a near-zero that nothing in + the record distinguished from a bad candidate. + + OFF BY DEFAULT (see HarborConfig.infra_retry_rounds), because against + an adversarial candidate this is a re-roll lever: the qualifying + predicate is built from exception types raised inside candidate code, + so a stochastic candidate that raises an allowlisted exception + whenever an attempt is going badly converts its all-bad samples from + booked zeros into fresh re-rolls. Enable only for trusted-candidate + evaluations. Two record-integrity rules hold either way: each round + runs in a fresh SIBLING of the jobs dir and collates from there alone + (never inside it: resume-path collations rglob the whole jobs dir, and + nested round dirs would pool this round's dead attempts into a later + resumed mean), and a recovered sample's booked output carries an + ``infra_retry`` audit marker naming the discarded dead attempts, so + the re-measurement is never invisible in the durable record. + """ + for round_no in range(1, self.config.infra_retry_rounds + 1): + retry: list[tuple[int, str]] = [] + prior_dead: dict[int, dict[str, int]] = {} + for sid, t in pairs: + dead = self._transient_infra_dead(params, sid) + if dead: + retry.append((sid, t)) + prior_dead[sid] = dead + if not retry: + return + delay = self.config.infra_retry_delay_s * round_no + logger.warning( + f"{len(retry)} sample(s) lost every attempt to transient infra " + f"causes; retry round {round_no}/{self.config.infra_retry_rounds} " + f"in {delay:.0f}s." + ) + await asyncio.sleep(delay) + # Sibling of the jobs dir, never a subdir (see docstring), with a + # globally fresh ordinal: a dir left over from an earlier eval of + # the same result_dir must never leak its stale trials into this + # round's collation. + ordinal = len(list(jobs_dir.parent.glob("jobs-infra-retry-*"))) + 1 + round_dir = jobs_dir.parent / f"jobs-infra-retry-{ordinal}" + await self._run_harbor( + str(workspace.project_path), params, [t for _, t in retry], round_dir + ) + # Collate only what the round actually produced: overwriting a + # cause-rich infra error with "no Harbor trial result" (because the + # retry round itself died early) would degrade the record. + produced = set(self._load_trials(round_dir)) + got = [(sid, t) for sid, t in retry if t in produced] + if not got: + logger.warning( + f"infra retry round {round_no} produced no trials; keeping " + f"the recorded errors." + ) + continue + self._collate(round_dir, got, params, ran=[t for _, t in got]) + self._mark_recovered(params, got, prior_dead, round_no) + + def _mark_recovered( + self, + params: EvaluationParameters, + got: list[tuple[int, str]], + prior_dead: dict[int, dict[str, int]], + round_no: int, + ) -> None: + """Stamp the audit marker onto samples the retry round recovered. + + A successful retry rebooks the sample from the fresh round alone, so + without this the durable record would show a clean measurement with no + trace that a full round of dead attempts was discarded, and a + discarded round is exactly the kind of fact an auditor of the scores + must be able to see.""" + for sid, _ in got: + result = self._existing(params, sid) + if result is None or result.is_error(): + continue # still failed; the error itself is the audit trail + output = result.output if isinstance(result.output, dict) else {} + output["infra_retry"] = { + "round": round_no, + "discarded_dead_attempts": prior_dead.get(sid, {}), + } + result.output = output + save_sample_result( + get_vero_home_dir() / "sessions", + params.session_id, + params.result_id, + sample_id=sid, + result=result, + ) + + def _transient_infra_dead( + self, params: EvaluationParameters, sample_id: int + ) -> dict[str, int] | None: + """The persisted dead-cause dict when the sample was never measured + (it errored AND every dead attempt carries a transient-infra label), + else None. Samples with no structured cause record are not retryable: + the conservative default is "measured", the same fail-closed direction + as zero-filling.""" + existing = self._existing(params, sample_id) + if existing is None or not existing.is_error(): + return None + output = existing.output if isinstance(existing.output, dict) else {} + dead = output.get("dead_exception_types") or {} + if dead and all(_is_transient_infra(k) for k in dead): + return dead + return None + + @staticmethod + def _alarm_key_budget(task_name: str, dead: dict[str, int]) -> None: + """The one infra cause that deserves its own alarm: an exhausted key + budget fails every subsequent LLM call identically, so from the first + occurrence onward the run is measuring the outage, not the agent, and + neither retrying nor waiting fixes it. ERROR level: this is the line + an operator greps for after a run full of inexplicable zeros.""" + n = sum(v for k, v in dead.items() if k.endswith(_KEY_BUDGET_SUFFIX)) + if n: + logger.error( + f"Task '{task_name}': {n} attempt(s) report the LLM key's " + f"spend budget as exhausted. If real, every call on this key " + f"fails the same way until it is refilled or swapped, and " + f"scores recorded meanwhile measure the outage, not the " + f"candidate. The signature is read from candidate-process " + f"exceptions, so corroborate against the key's own spend " + f"records before invalidating results." + ) # ------------------------------------------------------------------ # Task selection (host-side; just task names) @@ -209,6 +413,9 @@ def _collate( self.config.aggregate_attempts == "mean" or self.feedback_transcripts or self.expose_attempt_detail + # The infra retry decides from per-attempt death causes, which are + # only recorded when attempts are loaded at collation. + or self.config.infra_retry_rounds > 0 ) groups = self._trial_groups(jobs_dir) if need_attempts else {} for sample_id, task_name in pairs: @@ -374,10 +581,7 @@ 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" + key = _dead_attempt_label(t.get("exception_info")) dead_types[key] = dead_types.get(key, 0) + 1 if n_scored: if len(measured) < self.config.n_attempts or n_dead: @@ -398,6 +602,7 @@ def _out(output: dict) -> dict: if dead_types: # dict, not metrics: metrics are float-valued by contract. mean_output["dead_exception_types"] = dead_types + self._alarm_key_budget(task_name, dead_types) return SampleResult( score=mean, feedback=self._failure_feedback(mean, attempts), @@ -406,6 +611,15 @@ def _out(output: dict) -> dict: "n_attempts": float(len(attempts)), "n_scored": float(n_scored), "n_dead": float(n_dead), + # Zeros with an infra-labeled cause: still counted in + # the mean (see the classification note at module top) + # but split out for the analyst deciding whether a + # cell's zeros measured the plumbing. Labels derive + # from candidate-process exceptions: corroborating + # evidence for invalidating a cell, not proof. + "n_dead_infra": float( + sum(v for k, v in dead_types.items() if _is_infra(k)) + ), "n_clean": float(n_clean), }, output=_out(mean_output), @@ -417,22 +631,32 @@ def _out(output: dict) -> dict: # that flows everywhere (DB, per-sample files, the verifier's # target_errors), and "no verifier rewards" alone cannot separate # a champion that crashes deterministically on this executor from - # an infra outage — a distinction that decides whether the cell is + # an infra outage: a distinction that decides whether the cell is # a measurement or a re-run. cause = "" + dead: dict[str, int] = {} if attempts: - dead = {} for t in attempts: if (t.get("verifier_result") or {}).get("rewards"): continue - exc = (t.get("exception_info") or {}).get("exception_type") - key = exc or "no_rewards_recorded" + key = _dead_attempt_label(t.get("exception_info")) dead[key] = dead.get(key, 0) + 1 if dead: causes = ", ".join( f"{k} x{v}" for k, v in sorted(dead.items(), key=lambda i: -i[1]) ) cause = f" (attempts died: {causes})" + self._alarm_key_budget(task_name, dead) + no_rewards_output = { + "task_name": task_name, + "trial_name": trial.get("trial_name"), + } + if dead: + # Structured twin of the error string above: the within-eval + # infra retry reads this to decide whether the sample was + # measured or merely outaged (string parsing would be the + # fragile alternative). + no_rewards_output["dead_exception_types"] = dead return SampleResult( error=f"No verifier rewards for task '{task_name}'.{cause}", # The agent died before scoring. A candidate edit that CRASHES @@ -441,9 +665,7 @@ def _out(output: dict) -> dict: # does. Passed as score 0.0: an unscored attempt counts as a # failure everywhere else too. feedback=self._failure_feedback(0.0, attempts), - output=_out( - {"task_name": task_name, "trial_name": trial.get("trial_name")} - ), + output=_out(no_rewards_output), **common, ) score = self._extract_reward(rewards) diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py index 222e6d0..ba7c655 100644 --- a/vero/tests/test_harbor_runner.py +++ b/vero/tests/test_harbor_runner.py @@ -54,6 +54,7 @@ def _write_trial( trajectory: str | None = None, finished_at: str | None = None, exception_type: str | None = None, + exception_message: str = "", ): # Real harbor layout: ///result.json, plus a job-level # //result.json summary (no task_name) that collation must skip. @@ -73,7 +74,7 @@ def _write_trial( if exception_type is not None: data["exception_info"] = { "exception_type": exception_type, - "exception_message": "", + "exception_message": exception_message, "exception_traceback": "", } (d / "result.json").write_text(json.dumps(data)) @@ -949,3 +950,314 @@ def test_mean_zero_fills_reward_key_mismatch(self, tmp_path): assert r.score == 0.5 assert r.metrics["n_dead"] == 1.0 assert r.metrics["n_scored"] == 1.0 + + +class TestInfraResilience: + """Dead-attempt classification + bounded within-eval infra retry. + + Classification is diagnostic and retry-gating only: every dead attempt + still scores 0.0 regardless of class (excusing infra-labeled deaths from + the score would let a candidate raise fake ConnectionErrors on hard tasks). + The retry is OFF BY DEFAULT and exists for trusted-candidate evaluations: + against an adversarial optimizer it is a re-roll lever, since the + qualifying predicate derives from exceptions raised in candidate code. It + re-measures only samples whose EVERY attempt died of a transient infra + cause, in a fresh sibling jobs dir, and stamps an audit marker on + recovered samples so the discarded round stays visible. + """ + + def _flow_runner(self, monkeypatch, tmp_path, rounds_by_call, **cfg): + """A runner whose _run_harbor writes fixture trials per call: + rounds_by_call[i] is a list of (trial, task, rewards, exc_type, exc_msg) + written into whatever jobs dir call i receives.""" + monkeypatch.setenv("VERO_HOME_DIR", str(tmp_path / "vh")) + runner = HarborRunner(HarborConfig( + task_source="org/ds@1", agent_import_path="pkg.mod:Agent", + **cfg, + )) + calls = [] + + async def _fake_run(project_path, params, task_names, jobs_dir): + i = len(calls) + calls.append((list(task_names), Path(jobs_dir))) + for trial, task, rewards, exc_type, exc_msg in rounds_by_call[i]: + _write_trial(Path(jobs_dir), trial, task, rewards, + exception_type=exc_type, exception_message=exc_msg) + + monkeypatch.setattr(runner, "_run_harbor", _fake_run) + monkeypatch.setattr(runner, "_task_names_for", lambda p: [(0, "t0"), (1, "t1")]) + sleeps = [] + import asyncio as _asyncio + real_sleep = _asyncio.sleep + + async def _fast_sleep(d, *a, **k): + sleeps.append(d) + await real_sleep(0) + + monkeypatch.setattr("asyncio.sleep", _fast_sleep) + return runner, calls, sleeps + + def test_infra_deaths_labeled_and_counted_but_still_zero_filled(self, tmp_path): + # One scored attempt, one infra death, one candidate crash: the mean + # zero-fills BOTH deaths (classification must never move the score), + # and the labels + n_dead_infra let an analyst see which zeros + # measured the plumbing. + runner = HarborRunner(HarborConfig( + task_source="org/ds@1", agent_import_path="pkg.mod:Agent", + n_attempts=3, aggregate_attempts="mean", + )) + jobs = tmp_path / "jobs" + _write_trial(jobs, "a", "t0", {"reward": 1.0}) + _write_trial(jobs, "b", "t0", None, exception_type="ConnectionError") + _write_trial(jobs, "c", "t0", None, exception_type="UnsupportedParamsError") + groups = runner._trial_groups(jobs) + r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"]) + assert r.score == pytest.approx(1.0 / 3) + assert r.output["dead_exception_types"] == { + "ConnectionError[infra]": 1, + "UnsupportedParamsError": 1, + } + assert r.metrics["n_dead"] == 2.0 + assert r.metrics["n_dead_infra"] == 1.0 + + def test_forged_infra_suffix_is_neutralized(self, tmp_path): + # The suffix contract is load-bearing and the exception type name is + # candidate-authored: a class literally named "XError[infra]" must not + # classify as infra. Brackets are neutralized before labeling. + runner = HarborRunner(HarborConfig( + task_source="org/ds@1", agent_import_path="pkg.mod:Agent", + n_attempts=2, aggregate_attempts="mean", + )) + jobs = tmp_path / "jobs" + _write_trial(jobs, "a", "t0", {"reward": 1.0}) + _write_trial(jobs, "b", "t0", None, exception_type="MadeUpError[infra]") + groups = runner._trial_groups(jobs) + r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"]) + assert r.output["dead_exception_types"] == {"MadeUpError(infra)": 1} + assert r.metrics["n_dead_infra"] == 0.0 + + def test_key_budget_exhaustion_labeled_and_alarmed(self, tmp_path, caplog): + # litellm reports a spent key budget as a BadRequestError; only the + # message identifies it. It must be labeled infra (those zeros likely + # measure an outage) and alarmed at ERROR (every later call fails + # identically). The alarm text hedges: the signature is read from + # candidate-process exceptions and needs corroboration. + import logging as _logging + + runner = HarborRunner(HarborConfig( + task_source="org/ds@1", agent_import_path="pkg.mod:Agent", + n_attempts=2, aggregate_attempts="mean", + )) + jobs = tmp_path / "jobs" + msg = "litellm.BadRequestError: Budget has been exceeded! Current cost: 102.47, Max budget: 99.0" + _write_trial(jobs, "a", "t0", None, exception_type="BadRequestError", exception_message=msg) + _write_trial(jobs, "b", "t0", None, exception_type="BadRequestError", exception_message=msg) + groups = runner._trial_groups(jobs) + with caplog.at_level(_logging.ERROR, logger="vero.harbor.runner"): + r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"]) + assert r.error is not None + assert "BadRequestError[infra:llm-key-budget] x2" in r.error + assert r.output["dead_exception_types"] == {"BadRequestError[infra:llm-key-budget]": 2} + assert any("spend budget as exhausted" in rec.message for rec in caplog.records) + + @pytest.mark.asyncio + async def test_retry_is_off_by_default(self, tmp_path, monkeypatch): + # The re-roll lever must be opt-in: with a default config, an + # infra-outaged sample books its cause-rich error and nothing re-runs. + runner, calls, sleeps = self._flow_runner( + monkeypatch, tmp_path, + rounds_by_call=[ + [("a", "t0", {"reward": 1.0}, None, ""), + ("b", "t1", None, "ConnectionError", "")], + ], + ) + assert runner.config.infra_retry_rounds == 0 + params = _params() + await runner.produce_sample_results( + workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res" + ) + assert len(calls) == 1 + assert sleeps == [] + + @pytest.mark.asyncio + async def test_retry_reruns_only_the_outaged_sample(self, tmp_path, monkeypatch): + # t0 scores; t1 loses every attempt to ConnectionError. The retry round + # re-runs t1 ALONE in a fresh SIBLING jobs dir after a backoff, the + # fresh measurement replaces the recorded outage, and the booked output + # carries the audit marker naming the discarded dead attempts. + runner, calls, sleeps = self._flow_runner( + monkeypatch, tmp_path, + rounds_by_call=[ + [("a", "t0", {"reward": 1.0}, None, ""), + ("b", "t1", None, "ConnectionError", ""), + ("c", "t1", None, "ConnectionError", "")], + [("r1", "t1", {"reward": 0.5}, None, "")], + ], + infra_retry_rounds=1, + ) + params = _params() + await runner.produce_sample_results( + workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res" + ) + results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id) + assert results[0].score == 1.0 + assert results[1].error is None and results[1].score == 0.5 + assert results[1].output["infra_retry"] == { + "round": 1, + "discarded_dead_attempts": {"ConnectionError[infra]": 2}, + } + assert "infra_retry" not in (results[0].output or {}) + assert len(calls) == 2 + assert calls[1][0] == ["t1"] + # sibling of jobs/, never nested inside it (resume collations rglob + # the whole jobs dir and would pool this round's dead attempts) + assert calls[1][1].name == "jobs-infra-retry-1" + assert calls[1][1].parent == calls[0][1].parent + assert sleeps == [30.0] + + @pytest.mark.asyncio + async def test_candidate_crash_and_spent_key_never_retry(self, tmp_path, monkeypatch): + # A crash is a result; a spent key cannot recover by waiting. Neither + # triggers a retry round even with retries enabled. + budget_msg = "Budget has been exceeded! Current cost: 102.47, Max budget: 99.0" + runner, calls, sleeps = self._flow_runner( + monkeypatch, tmp_path, + rounds_by_call=[ + [("a", "t0", None, "UnsupportedParamsError", ""), + ("b", "t1", None, "BadRequestError", budget_msg)], + ], + infra_retry_rounds=1, + ) + params = _params() + await runner.produce_sample_results( + workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res" + ) + results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id) + assert results[0].error is not None and results[1].error is not None + assert len(calls) == 1 + assert sleeps == [] + + @pytest.mark.asyncio + async def test_mixed_cause_fully_dead_sample_never_retries(self, tmp_path, monkeypatch): + # EVERY dead cause must be transient-infra (all, not any): a + # deterministically-crashing candidate must not qualify for a re-roll + # by mixing one fake ConnectionError into its crashes. + runner, calls, sleeps = self._flow_runner( + monkeypatch, tmp_path, + rounds_by_call=[ + [("a", "t0", {"reward": 1.0}, None, ""), + ("b", "t1", None, "ConnectionError", ""), + ("c", "t1", None, "UnsupportedParamsError", "")], + ], + n_attempts=2, aggregate_attempts="mean", infra_retry_rounds=1, + ) + params = _params() + await runner.produce_sample_results( + workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res" + ) + results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id) + assert results[1].error is not None + assert len(calls) == 1 + assert sleeps == [] + + @pytest.mark.asyncio + async def test_persistent_outage_keeps_the_cause_rich_error(self, tmp_path, monkeypatch): + # The retry round produces nothing (outage persists): the durable + # record must keep the original infra-labeled error, not degrade to + # "no Harbor trial result" or raise out of the eval. + runner, calls, sleeps = self._flow_runner( + monkeypatch, tmp_path, + rounds_by_call=[ + [("a", "t0", {"reward": 1.0}, None, ""), + ("b", "t1", None, "ConnectionError", "")], + [], + ], + infra_retry_rounds=1, + ) + params = _params() + await runner.produce_sample_results( + workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res" + ) + results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id) + assert results[1].error is not None + assert "ConnectionError[infra]" in results[1].error + assert results[1].output["dead_exception_types"] == {"ConnectionError[infra]": 1} + assert len(calls) == 2 + + @pytest.mark.asyncio + async def test_partially_scored_sample_is_a_measurement_not_an_outage(self, tmp_path, monkeypatch): + # One attempt scored, one died of infra: the sample is a (noisy) + # measurement. Its infra death stays zero-filled in the mean and it is + # NOT retried; anything else would let infra flakiness re-roll scores. + runner, calls, sleeps = self._flow_runner( + monkeypatch, tmp_path, + rounds_by_call=[ + [("a", "t0", {"reward": 1.0}, None, ""), + ("b", "t0", None, "ConnectionError", ""), + ("c", "t1", {"reward": 1.0}, None, "")], + ], + n_attempts=2, aggregate_attempts="mean", infra_retry_rounds=1, + ) + params = _params() + await runner.produce_sample_results( + workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res" + ) + results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id) + assert results[0].score == 0.5 + assert results[0].metrics["n_dead_infra"] == 1.0 + assert len(calls) == 1 + + @pytest.mark.asyncio + async def test_retry_round_mean_is_not_diluted_by_dead_rounds(self, tmp_path, monkeypatch): + # The fresh round is collated from its own sibling dir alone: the two + # dead attempts of round 0 must not zero-dilute the fresh mean + # (1.0, not 0.5), and the audit marker records what was discarded. + runner, calls, sleeps = self._flow_runner( + monkeypatch, tmp_path, + rounds_by_call=[ + [("a", "t0", {"reward": 1.0}, None, ""), + ("b", "t1", None, "ConnectionError", ""), + ("c", "t1", None, "ConnectionError", "")], + [("r1", "t1", {"reward": 1.0}, None, ""), + ("r2", "t1", {"reward": 1.0}, None, "")], + ], + n_attempts=2, aggregate_attempts="mean", infra_retry_rounds=1, + ) + params = _params() + await runner.produce_sample_results( + workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res" + ) + results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id) + assert results[1].score == 1.0 + assert results[1].metrics["n_scored"] == 2.0 + assert results[1].metrics["n_dead"] == 0.0 + assert results[1].output["infra_retry"]["discarded_dead_attempts"] == { + "ConnectionError[infra]": 2 + } + + @pytest.mark.asyncio + async def test_multi_round_backoff_and_partial_recovery(self, tmp_path, monkeypatch): + # Two rounds: round 1 recovers t0 and leaves t1 outaged; round 2 + # retries ONLY the survivor, after a LONGER (linear) backoff, in its + # own fresh sibling dir. + runner, calls, sleeps = self._flow_runner( + monkeypatch, tmp_path, + rounds_by_call=[ + [("a", "t0", None, "ConnectionError", ""), + ("b", "t1", None, "TimeoutError", "")], + [("r1a", "t0", {"reward": 1.0}, None, ""), + ("r1b", "t1", None, "TimeoutError", "")], + [("r2b", "t1", {"reward": 0.5}, None, "")], + ], + infra_retry_rounds=2, + ) + params = _params() + await runner.produce_sample_results( + workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res" + ) + results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id) + assert results[0].score == 1.0 and results[0].output["infra_retry"]["round"] == 1 + assert results[1].score == 0.5 and results[1].output["infra_retry"]["round"] == 2 + assert [c[0] for c in calls] == [["t0", "t1"], ["t0", "t1"], ["t1"]] + assert [c[1].name for c in calls[1:]] == ["jobs-infra-retry-1", "jobs-infra-retry-2"] + assert sleeps == [30.0, 60.0] From 46f8c07a32936cd852d8f5a9b98ff37ac6300231 Mon Sep 17 00:00:00 2001 From: Shehab Yasser Date: Thu, 9 Jul 2026 17:18:23 +0300 Subject: [PATCH 8/8] fix(harbor): review follow-ups: validate retry knobs, full multi-round audit history - HarborConfig rejects infra_retry_delay_s <= 0 when retries are enabled (a zero delay silently nullified the backoff) and negative infra_retry_rounds. - The infra_retry audit marker now lists EVERY discarded round in order (discarded_rounds), not just the one immediately before recovery; recovered_round names when the sample finally measured. Co-Authored-By: Claude Opus 4.8 (1M context) --- vero/src/vero/harbor/config.py | 12 ++++++++++ vero/src/vero/harbor/runner.py | 20 +++++++++------- vero/tests/test_harbor_runner.py | 39 ++++++++++++++++++++++++++------ 3 files changed, 56 insertions(+), 15 deletions(-) diff --git a/vero/src/vero/harbor/config.py b/vero/src/vero/harbor/config.py index 3642227..f6b0b4b 100644 --- a/vero/src/vero/harbor/config.py +++ b/vero/src/vero/harbor/config.py @@ -76,6 +76,18 @@ def __post_init__(self) -> None: f"aggregate_attempts must be 'best' or 'mean', got " f"{self.aggregate_attempts!r}" ) + if self.infra_retry_rounds < 0: + raise ValueError( + f"infra_retry_rounds must be >= 0, got {self.infra_retry_rounds}" + ) + # A zero (or negative) delay silently nullifies the backoff, and an + # instant retry re-enters the same outage: 6 of 8 run attempts once + # burned inside a single live DNS blip for exactly this reason. + if self.infra_retry_rounds > 0 and self.infra_retry_delay_s <= 0: + raise ValueError( + f"infra_retry_delay_s must be > 0 when infra retries are " + f"enabled, got {self.infra_retry_delay_s}" + ) @property def is_registry(self) -> bool: diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py index 51a9e16..c12d927 100644 --- a/vero/src/vero/harbor/runner.py +++ b/vero/src/vero/harbor/runner.py @@ -173,14 +173,17 @@ async def _retry_transient_infra( ``infra_retry`` audit marker naming the discarded dead attempts, so the re-measurement is never invisible in the durable record. """ + # Every discarded round per sample, in round order: a sample that + # rides several retry rounds before recovering must surface ALL the + # rounds it burned, not just the last one before recovery. + discard_history: dict[int, list[dict[str, int]]] = {} for round_no in range(1, self.config.infra_retry_rounds + 1): retry: list[tuple[int, str]] = [] - prior_dead: dict[int, dict[str, int]] = {} for sid, t in pairs: dead = self._transient_infra_dead(params, sid) if dead: retry.append((sid, t)) - prior_dead[sid] = dead + discard_history.setdefault(sid, []).append(dead) if not retry: return delay = self.config.infra_retry_delay_s * round_no @@ -211,30 +214,31 @@ async def _retry_transient_infra( ) continue self._collate(round_dir, got, params, ran=[t for _, t in got]) - self._mark_recovered(params, got, prior_dead, round_no) + self._mark_recovered(params, got, discard_history, round_no) def _mark_recovered( self, params: EvaluationParameters, got: list[tuple[int, str]], - prior_dead: dict[int, dict[str, int]], + discard_history: dict[int, list[dict[str, int]]], round_no: int, ) -> None: """Stamp the audit marker onto samples the retry round recovered. A successful retry rebooks the sample from the fresh round alone, so without this the durable record would show a clean measurement with no - trace that a full round of dead attempts was discarded, and a + trace that full rounds of dead attempts were discarded, and a discarded round is exactly the kind of fact an auditor of the scores - must be able to see.""" + must be able to see. ``discarded_rounds`` lists every burned round in + order, not just the one immediately before recovery.""" for sid, _ in got: result = self._existing(params, sid) if result is None or result.is_error(): continue # still failed; the error itself is the audit trail output = result.output if isinstance(result.output, dict) else {} output["infra_retry"] = { - "round": round_no, - "discarded_dead_attempts": prior_dead.get(sid, {}), + "recovered_round": round_no, + "discarded_rounds": discard_history.get(sid, []), } result.output = output save_sample_result( diff --git a/vero/tests/test_harbor_runner.py b/vero/tests/test_harbor_runner.py index ba7c655..b34449f 100644 --- a/vero/tests/test_harbor_runner.py +++ b/vero/tests/test_harbor_runner.py @@ -997,6 +997,21 @@ async def _fast_sleep(d, *a, **k): monkeypatch.setattr("asyncio.sleep", _fast_sleep) return runner, calls, sleeps + def test_config_rejects_zero_delay_when_retries_enabled(self): + # A zero delay silently nullifies the backoff and an instant retry + # re-enters the same outage; misconfiguration must fail loudly. + with pytest.raises(ValueError, match="infra_retry_delay_s"): + HarborConfig(task_source="org/ds@1", agent_import_path="p:m", + infra_retry_rounds=1, infra_retry_delay_s=0.0) + # with retries off the delay is never used, so 0 is not an error + HarborConfig(task_source="org/ds@1", agent_import_path="p:m", + infra_retry_rounds=0, infra_retry_delay_s=0.0) + + def test_config_rejects_negative_rounds(self): + with pytest.raises(ValueError, match="infra_retry_rounds"): + HarborConfig(task_source="org/ds@1", agent_import_path="p:m", + infra_retry_rounds=-1) + def test_infra_deaths_labeled_and_counted_but_still_zero_filled(self, tmp_path): # One scored attempt, one infra death, one candidate crash: the mean # zero-fills BOTH deaths (classification must never move the score), @@ -1103,8 +1118,8 @@ async def test_retry_reruns_only_the_outaged_sample(self, tmp_path, monkeypatch) assert results[0].score == 1.0 assert results[1].error is None and results[1].score == 0.5 assert results[1].output["infra_retry"] == { - "round": 1, - "discarded_dead_attempts": {"ConnectionError[infra]": 2}, + "recovered_round": 1, + "discarded_rounds": [{"ConnectionError[infra]": 2}], } assert "infra_retry" not in (results[0].output or {}) assert len(calls) == 2 @@ -1231,9 +1246,9 @@ async def test_retry_round_mean_is_not_diluted_by_dead_rounds(self, tmp_path, mo assert results[1].score == 1.0 assert results[1].metrics["n_scored"] == 2.0 assert results[1].metrics["n_dead"] == 0.0 - assert results[1].output["infra_retry"]["discarded_dead_attempts"] == { - "ConnectionError[infra]": 2 - } + assert results[1].output["infra_retry"]["discarded_rounds"] == [ + {"ConnectionError[infra]": 2} + ] @pytest.mark.asyncio async def test_multi_round_backoff_and_partial_recovery(self, tmp_path, monkeypatch): @@ -1256,8 +1271,18 @@ async def test_multi_round_backoff_and_partial_recovery(self, tmp_path, monkeypa workspace=MagicMock(project_path="/wt"), params=params, result_dir=tmp_path / "res" ) results = load_all_sample_results(get_vero_home_dir() / "sessions", "s", params.result_id) - assert results[0].score == 1.0 and results[0].output["infra_retry"]["round"] == 1 - assert results[1].score == 0.5 and results[1].output["infra_retry"]["round"] == 2 + assert results[0].score == 1.0 + assert results[0].output["infra_retry"] == { + "recovered_round": 1, + "discarded_rounds": [{"ConnectionError[infra]": 1}], + } + # t1 burned TWO rounds before recovering; the audit marker must list + # both, not just the round immediately before recovery. + assert results[1].score == 0.5 + assert results[1].output["infra_retry"] == { + "recovered_round": 2, + "discarded_rounds": [{"TimeoutError[infra]": 1}, {"TimeoutError[infra]": 1}], + } assert [c[0] for c in calls] == [["t0", "t1"], ["t0", "t1"], ["t1"]] assert [c[1].name for c in calls[1:]] == ["jobs-infra-retry-1", "jobs-infra-retry-2"] assert sleeps == [30.0, 60.0]