diff --git a/vero/src/vero/evaluation/engine.py b/vero/src/vero/evaluation/engine.py index 69b9483..b2f2196 100644 --- a/vero/src/vero/evaluation/engine.py +++ b/vero/src/vero/evaluation/engine.py @@ -146,8 +146,18 @@ def resolve_samples(self, req: EvalRequest) -> tuple[list[int] | None, int]: # Evaluation # ------------------------------------------------------------------ - async def evaluate(self, req: EvalRequest, *, admin: bool = False) -> Experiment: - """Meter (unless admin) and run one evaluation; return the full Experiment. + async def evaluate( + self, req: EvalRequest, *, admin: bool = False, free: bool = False + ) -> Experiment: + """Meter (unless admin or free) and run one evaluation; return the full + Experiment. + + ``admin`` and ``free`` are distinct authorities and must stay separate: + ``admin`` grants ACCESS (bypasses the tier gate and the ledger), while + ``free`` only waives the BUDGET debit for an otherwise ordinary agent + eval (the sidecar's one free baseline eval). A free eval still runs as + the agent: conflating the two let the free-baseline path evaluate + ``no_access`` splits and return their aggregate score to the agent. ``no_access`` gating is EXPLICIT and fail-closed: when ``split_accesses`` is configured, the split's tier is resolved (an unlisted split defaults @@ -163,7 +173,7 @@ async def evaluate(self, req: EvalRequest, *, admin: bool = False) -> Experiment f"and cannot be evaluated by the agent." ) sample_ids, n = self.resolve_samples(req) - if not admin: + if not admin and not free: await self.budget.reserve(req.dataset_id, req.split, n) return await self.evaluator.evaluate( commit=req.commit, @@ -183,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. @@ -190,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, @@ -198,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/app.py b/vero/src/vero/harbor/app.py index 16a53ec..f8e2432 100644 --- a/vero/src/vero/harbor/app.py +++ b/vero/src/vero/harbor/app.py @@ -17,7 +17,7 @@ from vero.evaluation.engine import EvalRequest from vero.exceptions import ExperimentBudgetExceeded, InvalidSplitError from vero.harbor.auth import check_admin -from vero.harbor.server import SubmitDisabledError +from vero.harbor.server import KAnonymityError, SubmitDisabledError from vero.harbor.verifier import NoCandidateError if TYPE_CHECKING: @@ -58,6 +58,10 @@ def create_app( SubmitDisabledError, lambda r, e: JSONResponse(status_code=409, content={"error": str(e)}), ) + app.add_exception_handler( + KAnonymityError, + lambda r, e: JSONResponse(status_code=400, content={"error": str(e)}), + ) app.add_exception_handler( NoCandidateError, lambda r, e: JSONResponse(status_code=409, content={"error": str(e)}), diff --git a/vero/src/vero/harbor/build/compiler.py b/vero/src/vero/harbor/build/compiler.py index 7d1605a..dc42431 100644 --- a/vero/src/vero/harbor/build/compiler.py +++ b/vero/src/vero/harbor/build/compiler.py @@ -236,12 +236,15 @@ def _serve_config( "split": t.split, "reward_key": t.reward_key, "sample_ids": t.sample_ids, + "model": t.model, } for t in config.targets ], "base_commit": base_commit, "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, @@ -367,6 +370,10 @@ def compile_task( description=config.description, mode=config.mode, timeout=config.timeout, + # The verifier phase runs the whole finalize battery (shortlist + # re-scores + floor + targets + baseline attempts), not one eval; + # unset falls back to `timeout` for backward compatibility. + verifier_timeout=config.verifier_timeout or config.timeout, secrets=config.secrets, read_only_paths=config.read_only_paths, base_image_main=config.base_image_main, @@ -400,6 +407,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 07520f2..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): @@ -67,6 +73,18 @@ class _BuildConfigBase(BaseModel): # WORSE than the untouched repo is visible as a regression. score_baseline: bool = False + # Minimum sample count for agent-chosen subset evals of non_viewable splits + # (full-split evals always pass; <=1 disables). Aggregate responses carry + # 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) @@ -82,6 +100,15 @@ class _BuildConfigBase(BaseModel): timeout: int = 1800 max_concurrency: int = 8 + # Wall-clock budget for the VERIFIER phase (Harbor's [verifier] timeout_sec). + # Finalize is not one eval: it runs up to rescore_top_k shortlist re-scores + # + 1 floor eval + len(targets) target evals + len(targets) x + # baseline_score_attempts baseline evals, each a full nested run in Mode B. + # Sizing this at one eval's duration kills finalize mid-flight and the trial + # ships NO reward.json. Defaults to `timeout` when unset; size it as + # (rescore_top_k + 1 + 3 x len(targets)) x a single eval's duration + slack. + verifier_timeout: int | None = None + class BuildConfigA(_BuildConfigBase): """Mode A: vero runs inference + scoring against a saved dataset.""" 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/build/templates/task.toml.j2 b/vero/src/vero/harbor/build/templates/task.toml.j2 index c037e22..143d234 100644 --- a/vero/src/vero/harbor/build/templates/task.toml.j2 +++ b/vero/src/vero/harbor/build/templates/task.toml.j2 @@ -13,8 +13,11 @@ user = "agent" # Shared mode: Harbor runs tests/test.sh in `main` with the whole env (incl. the # eval-sidecar) still up. The verifier runs as root, reads the admin token, and # calls the sidecar's `finalize` endpoint to score the selected commit. +# timeout_sec must cover the WHOLE finalize battery (shortlist re-scores + +# floor + targets + baseline attempts), each a full nested eval, or Harbor +# kills finalize mid-flight and the trial ships no reward.json. environment_mode = "shared" -timeout_sec = {{ timeout }} +timeout_sec = {{ verifier_timeout }} [environment] # Compose-based environment: environment/docker-compose.yaml adds the eval-sidecar diff --git a/vero/src/vero/harbor/config.py b/vero/src/vero/harbor/config.py index 387c132..f6b0b4b 100644 --- a/vero/src/vero/harbor/config.py +++ b/vero/src/vero/harbor/config.py @@ -30,6 +30,42 @@ class HarborConfig: # estimates pass probability instead of pass@k (which "best" # inflates toward). aggregate_attempts: str = "best" + # Trusted source for the nested `harbor` CLI, as a uv requirement spec + # (e.g. "harbor==0.1.17" or a pinned git URL). When set, the runner layers + # it over the candidate env with `uv run --with`, whose ephemeral overlay + # takes precedence for both the console script and sys.path — so the + # orchestrator that scores the candidate resolves from THIS spec, not from + # whatever the candidate's own pyproject/uv.lock pin (which the agent + # controls, and could point at a fork that fabricates trial results + # 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: @@ -40,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/protocol.py b/vero/src/vero/harbor/protocol.py index 20c0500..156c8e3 100644 --- a/vero/src/vero/harbor/protocol.py +++ b/vero/src/vero/harbor/protocol.py @@ -104,6 +104,10 @@ def build_status( split_accesses: list[SplitAccess], base_commit: str | None = None, free_baseline_available: bool = False, + # 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. @@ -119,6 +123,15 @@ def build_status( "dataset_id": dataset_id, "tier": str(tier), "agent_evaluable": tier != SplitAccessLevel.no_access, + # Subset evals below this sample count are rejected on + # non_viewable splits (full-split evals always pass). Advertised + # so an agent plans subset probes instead of burning budget on + # requests the sidecar will refuse. + "min_subset_samples": ( + k_anonymity_floor + if tier == SplitAccessLevel.non_viewable + else 1 + ), "remaining_sample_budget": b.remaining_sample_budget, "remaining_run_budget": b.remaining_run_budget, } diff --git a/vero/src/vero/harbor/runner.py b/vero/src/vero/harbor/runner.py index 7b6eede..c12d927 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,151 @@ 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. + """ + # 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]] = [] + for sid, t in pairs: + dead = self._transient_infra_dead(params, sid) + if dead: + retry.append((sid, t)) + discard_history.setdefault(sid, []).append(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, discard_history, round_no) + + def _mark_recovered( + self, + params: EvaluationParameters, + got: list[tuple[int, str]], + 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 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. ``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"] = { + "recovered_round": round_no, + "discarded_rounds": discard_history.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) @@ -109,8 +317,18 @@ def _build_command( jobs_dir: Path, ) -> list[str]: c = self.config - cmd = [ - "uv", "run", "--project", project_path, + cmd = ["uv", "run", "--project", project_path] + # Resolve the nested harbor from the trusted spec rather than the + # candidate's lockfile (see HarborConfig.harbor_requirement). This + # raises the bar from "edit one pyproject line" to "tamper with the + # orchestrator in-process at runtime"; it is not a full boundary — + # the agent's code still imports into the nested harbor process via + # --agent-import-path, so a hostile candidate can in principle still + # forge results from inside. Full isolation needs an out-of-process + # verifier and is tracked separately. + if c.harbor_requirement: + cmd += ["--with", c.harbor_requirement] + cmd += [ "harbor", "run", *c.source_args(), "--agent-import-path", c.agent_import_path, @@ -119,8 +337,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] @@ -195,6 +417,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: @@ -343,6 +568,12 @@ def _out(output: dict) -> dict: n_scored = 0 n_dead = 0 n_clean = 0 + # Dead attempts are not interchangeable: a rate-limited attempt is + # infra noise that retunes with capacity, while a crash points at + # the candidate (or a task bug), and dead attempts cluster hard by + # cause in practice. n_dead alone hides that, so record the + # exception type behind every zero-filled attempt. + dead_types: dict[str, int] = {} for t in attempts: rewards = (t.get("verifier_result") or {}).get("rewards") or {} reward = self._extract_reward(rewards) if rewards else None @@ -354,6 +585,8 @@ def _out(output: dict) -> dict: else: measured.append(0.0) n_dead += 1 + 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: # Fewer or dirtier measurements than the config promises: @@ -365,6 +598,15 @@ def _out(output: dict) -> dict: f"({n_scored} scored, {n_dead} dead counted 0.0)." ) mean = sum(measured) / len(measured) + mean_output = { + "task_name": task_name, + "attempt_scores": measured, + "aggregate": "mean", + } + if dead_types: + # dict, not metrics: metrics are float-valued by contract. + mean_output["dead_exception_types"] = dead_types + self._alarm_key_budget(task_name, dead_types) return SampleResult( score=mean, feedback=self._failure_feedback(mean, attempts), @@ -373,22 +615,61 @@ 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({ - "task_name": task_name, - "attempt_scores": measured, - "aggregate": "mean", - }), + output=_out(mean_output), **common, ) 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 = "" + dead: dict[str, int] = {} + if attempts: + for t in attempts: + if (t.get("verifier_result") or {}).get("rewards"): + continue + 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}'.", - output=_out( - {"task_name": task_name, "trial_name": trial.get("trial_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 + # 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(no_rewards_output), **common, ) score = self._extract_reward(rewards) @@ -528,6 +809,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/src/vero/harbor/serve.py b/vero/src/vero/harbor/serve.py index 603b650..bd42329 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 @@ -42,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): @@ -77,6 +80,15 @@ class _ServeConfigBase(BaseModel): # on the selection split; it reverts to base_commit instead (needs base_commit). auto_best_baseline_floor: bool = True + # Minimum sample count for agent-chosen subset evals of non_viewable + # splits (full-split evals always pass; <=1 disables). See + # 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 @@ -150,8 +162,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: @@ -176,11 +192,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 ) @@ -240,6 +273,10 @@ async def build_components( eval_strategy=eval_strategy, ) + split_accesses = [ + SplitAccess(split=s.split, access=SplitAccessLevel(s.access)) + for s in config.split_accesses + ] db = ExperimentDatabase(id=config.session_id) # shared by engine (writes) + verifier (reads) engine = EvaluationEngine( evaluator=evaluator, @@ -253,12 +290,12 @@ async def build_components( ), session_id=config.session_id, vero_home=vero_home, + # The engine-side no_access gate is only armed when split_accesses is + # set. Without it the ledger was the sole gate (no_access splits are + # unbudgeted, so reserve() raised) — and every unmetered path (admin, + # the free baseline eval) walked straight past it. + split_accesses=split_accesses, ) - - split_accesses = [ - SplitAccess(split=s.split, access=SplitAccessLevel(s.access)) - for s in config.split_accesses - ] sidecar = EvaluationSidecar( engine=engine, split_accesses=split_accesses, @@ -267,6 +304,7 @@ async def build_components( admin_volume=Path(config.admin_volume), submit_enabled=config.submit_enabled, base_commit=config.base_commit, + k_anonymity_floor=config.k_anonymity_floor, ) verifier = Verifier( engine=engine, diff --git a/vero/src/vero/harbor/server.py b/vero/src/vero/harbor/server.py index 1f99f88..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 @@ -38,6 +39,11 @@ class SubmitDisabledError(RuntimeError): """Raised when submit() is called but the task does not use submit selection.""" +class KAnonymityError(RuntimeError): + """Raised when an agent eval on a non_viewable split selects fewer samples + than the k-anonymity floor allows.""" + + class EvaluationSidecar: """Agent-facing handlers over the EvaluationEngine. @@ -56,6 +62,7 @@ def __init__( admin_volume: Path, submit_enabled: bool = False, base_commit: str | None = None, + k_anonymity_floor: int = 5, ): self.engine = engine self.split_accesses = split_accesses @@ -64,13 +71,42 @@ def __init__( self.admin_volume = Path(admin_volume) self.submit_enabled = submit_enabled self.base_commit = base_commit + # Minimum sample count for an agent-chosen SUBSET eval of a non_viewable + # split. The aggregate response carries mean_score, so a singleton subset + # returns the sample's label-derived score verbatim, and n singleton + # evals reconstruct the split's per-sample labels wholesale. The floor + # applies only to proper subsets (sample_ids/num_samples): a full-split + # eval reveals exactly the intended aggregate, so it always passes and a + # split smaller than the floor degrades to full-split-only rather than + # becoming unevaluable. This is a cost multiplier, not a proof: means of + # k-sized overlapping subsets still admit reconstruction by elimination, + # but at >= k times the sample budget per label. <= 1 disables the floor. + self.k_anonymity_floor = k_anonymity_floor self._free_baseline_used = False + self._eval_seq = 0 # per-eval ordinal for result-dir versioning # ------------------------------------------------------------------ # Handlers (the HTTP layer resolves `admin` from auth and calls these) # ------------------------------------------------------------------ async def evaluate(self, req: EvalRequest, *, admin: bool = False) -> EvalSummary: + # k-anonymity floor, checked before any work (no commit transfer, no + # budget debit, no eval) so a rejected request costs the agent nothing. + # sample_ids is None exactly when the request covers the full split + # (resolve_samples collapses a covering num_samples to None too), and a + # full-split aggregate is the intended surface, so only proper subsets + # are floored. + if not admin and self.k_anonymity_floor > 1: + tier = tier_for_split(req.split, self.split_accesses) + if tier == SplitAccessLevel.non_viewable: + sample_ids, n = self.engine.resolve_samples(req) + if sample_ids is not None and n < self.k_anonymity_floor: + raise KAnonymityError( + f"Evals on non_viewable split '{req.split}' must cover " + f"at least {self.k_anonymity_floor} samples (or the " + f"whole split); got {n}. Aggregate scores over smaller " + f"subsets would reveal per-sample results." + ) sha = await self._transfer_commit(req.commit) # The agent's FIRST eval of the seeded baseline is budget-free. The # baseline is the reference every candidate is implicitly compared to, @@ -80,17 +116,29 @@ async def evaluate(self, req: EvalRequest, *, admin: bool = False) -> EvalSummar # an optimizer that skipped the reference could not tell a no-op edit # from an improvement and quit with budget unspent). Capped at one: # later baseline evals debit normally, so free compute is bounded. + # `free` waives only the budget debit; the eval still runs as the agent + # (tier gates apply), so the free baseline cannot touch no_access + # splits — riding the admin flag here did exactly that. free_baseline = ( not admin and self.base_commit is not None and sha == self.base_commit and not self._free_baseline_used ) + # 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 - exp = await self.engine.evaluate( - replace(req, commit=sha), admin=admin or free_baseline - ) + 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 @@ -125,6 +173,7 @@ def status(self) -> StatusSummary: free_baseline_available=( self.base_commit is not None and not self._free_baseline_used ), + k_anonymity_floor=self.k_anonymity_floor, ) def list_experiments(self) -> list[dict]: @@ -234,24 +283,66 @@ def _route_results(self, experiment: Experiment, *, admin: bool) -> str | None: return None commit = experiment.run.candidate.commit - dest = self.agent_volume / "results" / f"{split}__{commit[:12]}" - # Recreate the dir so it reflects exactly this metered run. The dir is keyed - # only on (split, commit[:12]); a prior eval of the same commit on a larger - # sample set would otherwise leave stale per-sample files behind that this - # run did not produce, and result_path would surface them as if they were. - if dest.exists(): + # Every metered eval gets its own versioned dir. Keying on + # (split, commit) alone forced a wipe-and-rewrite, so a re-measurement + # (a multifidelity confirm, a noise re-eval of the champion) erased the + # agent's earlier evidence for the same commit; repeat measurements are + # exactly the ones worth comparing. result_path in the response names + # the dir for THIS eval. + 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 = 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 / 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() + ] + 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) + mean_score_se = (var / len(filled)) ** 0.5 (dest / "summary.json").write_text( json.dumps( { "split": split, "commit": commit, - "n_samples": len(experiment.result.sample_results), + "n_samples": len(sample_results), + "n_scored": sum( + 1 for r in sample_results.values() if r.score is not None + ), + "n_errored": sum( + 1 for r in sample_results.values() if r.is_error() + ), "mean_score": experiment.result.score(), - "status": str(experiment.result.status), + "mean_score_se": mean_score_se, + "status": experiment.result.status.value, }, indent=2, ) diff --git a/vero/src/vero/harbor/verifier.py b/vero/src/vero/harbor/verifier.py index 35efdfa..63f588c 100644 --- a/vero/src/vero/harbor/verifier.py +++ b/vero/src/vero/harbor/verifier.py @@ -12,8 +12,10 @@ from __future__ import annotations +import asyncio import json import logging +import re from dataclasses import dataclass from pathlib import Path from typing import Literal @@ -37,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: @@ -75,13 +86,34 @@ def __init__( # candidate even when every candidate regressed, shipping a regression # (observed live: a weak inner model, every candidate below baseline). self.auto_best_baseline_floor = auto_best_baseline_floor - # Baseline scoring is retried this many times total before its outcome is - # reported as an error; the nested eval can fail transiently (a nested - # harbor run crashing right after a large eval), and a single blip must - # not silently drop the regression check. + # Every reward-critical finalize eval (targets, shortlist re-scores, + # floor, baseline) is retried this many times total: the nested eval can + # fail transiently (a nested harbor run crashing right after a large + # eval), and a single blip must not abort finalize; a trial that ships + # no reward.json loses its result entirely. self._baseline_score_attempts = max(1, baseline_score_attempts) + # Finalize is idempotent: Harbor may retry the verifier (or an operator + # may re-POST /finalize), and a replayed finalize must return the FIRST + # completed result verbatim. Re-running it would re-rank against a DB + # that now also contains the first finalize's own admin evals, so a + # retry could select a DIFFERENT champion than the one already reported. + self._finalize_lock = asyncio.Lock() + self._finalize_result: dict | None = None async def finalize(self) -> dict: + """Idempotent entry point: the first completed finalize is cached and + replayed verbatim on any retry (see __init__ for why re-running would + be unsound). The lock serializes concurrent calls so exactly one + finalize ever computes.""" + async with self._finalize_lock: + if self._finalize_result is not None: + logger.info("finalize: replaying cached result (idempotent)") + return self._finalize_result + result = await self._finalize() + self._finalize_result = result + return result + + async def _finalize(self) -> dict: """Select the commit, score it on every target, and score the baseline. Returns a wrapper ``{"rewards": {reward_key: score}, "baseline": {...}}``. @@ -113,20 +145,131 @@ async def finalize(self) -> dict: return {"rewards": rewards, "baseline": {"skipped": "no candidate commit"}} logger.info(f"Verifier selected commit {sha} (mode={self.reward_mode})") rewards: dict[str, float] = {} + target_errors: dict[str, str] = {} for target in self.targets: - exp = await self.engine.evaluate_admin( + score, cause = await self._admin_eval_score( task=target.task, dataset_id=target.dataset_id, split=target.split, commit=sha, sample_ids=target.sample_ids, + model=target.model, + what=f"target '{target.reward_key}'", ) - score = exp.result.score() - rewards[target.reward_key] = ( - float(score) if score is not None else default_minimum_score - ) + if score is None: + # Persistent failure: floor the target so reward.json still + # 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 baseline = await self._maybe_score_baseline(rewards) - return {"rewards": rewards, "baseline": baseline} + result = {"rewards": rewards, "baseline": baseline} + if target_errors: + result["target_errors"] = target_errors + return result + + async def _admin_eval_score( + self, + *, + task: str | None, + dataset_id: str, + 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. + + 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): + try: + exp = await self.engine.evaluate_admin( + task=task, + dataset_id=dataset_id, + split=split, + commit=commit, + sample_ids=sample_ids, + model=model, + ) + if exp.result.score(fill_score=None) is None: + 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, + ) + continue + score = exp.result.score() + if score is None: + # Should be unreachable (the strict check above already + # passed), but a None here must consume a retry like any + # other unmeasurable outcome, never bypass the loop. + last_error = "eval returned no aggregate score" + continue + return float(score), None + except Exception as exc: # noqa: BLE001 - retried, then surfaced as None + last_error = exc + logger.warning( + "%s attempt %d/%d failed: %s", + what, attempt, self._baseline_score_attempts, exc, + ) + logger.error( + "%s failed after %d attempt(s): %s", + what, self._baseline_score_attempts, last_error, + ) + 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: + # 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] + 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. @@ -162,13 +305,24 @@ 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 + # zero here would fake a huge candidate improvement. + # Raise into the retry loop instead. + raise RuntimeError( + f"baseline eval on '{target.reward_key}' scored no " + f"samples (all errored or empty)" + ) score = exp.result.score() baselines[target.reward_key] = ( float(score) if score is not None else default_minimum_score @@ -276,31 +430,73 @@ async def _best_from_db(self) -> str: if len(full_split_df) > 0: split_df = full_split_df # Shortlist by recorded score (cheap, agent-influenced -> not trusted as - # final), one row per candidate (highest recorded score wins the slot). - ranked = split_df.sort_values( + # final). Recorded evals are POOLED before shortlisting, in two steps: + # every eval of the same commit averages into that commit's score, then + # commits with the same git TREE (identical content) collapse into one + # candidate group scored by the group mean. Max-over-rows selection made + # every re-measurement an independent lottery draw, and one live + # optimizer farmed exactly that ("distinct empty commits = clean + # independent lottery tickets") while another refused to re-measure its + # champion to protect a lucky draw. Pooling makes re-measurement + # variance-REDUCING (as statistics wants) instead of max-inflating, and + # tree-dedup stops identical content from stuffing the top-K shortlist + # or collecting several admin re-score draws. + agg: dict[str, tuple[str, str]] = { + "mean_score": ("mean_score", "mean"), + "candidate_created_at": ("candidate_created_at", "max"), + } + if "dataset_subset_dataset_id" in split_df.columns: + agg["dataset_subset_dataset_id"] = ("dataset_subset_dataset_id", "first") + per_commit = ( + split_df.groupby("candidate_commit").agg(**agg).reset_index() + ) + trees: dict[str, str] = {} + for commit in per_commit["candidate_commit"]: + # Unresolvable tree (non-git workspace, unknown sha) falls back to + # the commit itself: no pooling across commits, never a crash. + trees[commit] = (await self._tree_of(commit)) or commit + per_commit["_tree"] = per_commit["candidate_commit"].map(trees) + # Newest commit represents its tree group (any member is equivalent + # content-wise; newest keeps logs intuitive). + per_commit = per_commit.sort_values( + by=["candidate_created_at"], ascending=[False] + ) + group_agg: dict[str, tuple[str, str]] = { + "mean_score": ("mean_score", "mean"), + "candidate_commit": ("candidate_commit", "first"), + "candidate_created_at": ("candidate_created_at", "first"), + } + if "dataset_subset_dataset_id" in per_commit.columns: + group_agg["dataset_subset_dataset_id"] = ( + "dataset_subset_dataset_id", "first", + ) + pooled = per_commit.groupby("_tree", sort=False).agg(**group_agg) + ranked = pooled.sort_values( by=["mean_score", "candidate_created_at"], ascending=[False, False] ) - ranked = ranked.drop_duplicates(subset=["candidate_commit"], keep="first") shortlist = ranked.head(max(1, self.rescore_top_k)) rescored: list[tuple[float, int, str]] = [] for idx, (_, row) in enumerate(shortlist.iterrows()): commit = row["candidate_commit"] dataset_id = row.get("dataset_subset_dataset_id") - exp = await self.engine.evaluate_admin( + score, _cause = await self._admin_eval_score( task=self.selection_task, dataset_id=dataset_id, split=self.selection_split, commit=commit, + what=f"auto_best re-score of {commit}", ) - score = exp.result.score() - admin_score = float(score) if score is not None else default_minimum_score + # A candidate whose re-score persistently fails is unscorable, not + # zero-scored: it keeps the floor value and cannot win on a fluke, + # but its failure does not abort the other candidates' re-scores. + admin_score = score if score is not None else float(default_minimum_score) # Tie-break by shortlist position (already ordered by recorded score # then recency), so ties resolve deterministically without depending on # the type of candidate_created_at (a datetime in the real DB). rescored.append((admin_score, idx, commit)) logger.info( - "auto_best re-score: commit=%s admin_score=%s (recorded=%s)", + "auto_best re-score: commit=%s admin_score=%s (pooled recorded=%s)", commit, admin_score, row["mean_score"], @@ -320,14 +516,25 @@ 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_exp = await self.engine.evaluate_admin( + base_score_opt, _cause = await self._admin_eval_score( task=self.selection_task, dataset_id=base_dataset_id, split=self.selection_split, commit=self.base_commit, + what="auto_best floor (baseline)", ) - base_s = base_exp.result.score() - base_score = float(base_s) if base_s is not None else default_minimum_score + if base_score_opt is None: + # The floor exists to stop shipped regressions; shipping a + # candidate with the floor check unmeasured re-opens exactly + # that hole. Fail safe: revert to the seed. + logger.error( + "auto_best floor: baseline could not be measured; failing " + "safe to base_commit %s instead of shipping unverified " + "candidate %s.", + self.base_commit, best_commit, + ) + return self.base_commit + base_score = base_score_opt if best_score <= base_score: logger.info( "auto_best floor: best candidate %s (admin_score=%s) does not beat " @@ -340,3 +547,17 @@ async def _best_from_db(self) -> str: best_commit, best_score, base_score, ) return best_commit + + async def _tree_of(self, commit: str) -> str | None: + """Git tree hash for a commit via the engine's workspace, or None when + it cannot be resolved (non-git workspace, unknown sha). None makes the + caller treat the commit as its own pooling group: degraded, never wrong.""" + workspace = getattr(self.engine.evaluator, "workspace", None) + tree_hash = getattr(workspace, "tree_hash", None) + if tree_hash is None: + return None + try: + return await tree_hash(commit) + except Exception: # noqa: BLE001 - pooling is an optimization, not a gate + logger.warning("could not resolve tree hash for commit %s", commit) + return None diff --git a/vero/src/vero/workspace/git.py b/vero/src/vero/workspace/git.py index 93df894..c349332 100644 --- a/vero/src/vero/workspace/git.py +++ b/vero/src/vero/workspace/git.py @@ -316,6 +316,12 @@ async def resolve_ref(self, ref: str) -> str: """Resolve a git ref (branch, tag, short hash) to a full commit hash.""" return await self._git("rev-parse", ref) + async def tree_hash(self, ref: str) -> str: + """Tree hash of a commit: identical content -> identical tree, whatever + the commit metadata says. Lets selection pool re-committed identical + trees instead of treating each sha as a fresh candidate.""" + return await self._git("rev-parse", f"{ref}^{{tree}}") + async def current_branch(self) -> str | None: """Return current branch name, or None if detached HEAD.""" try: diff --git a/vero/tests/test_engine.py b/vero/tests/test_engine.py index 0349adf..3ac0022 100644 --- a/vero/tests/test_engine.py +++ b/vero/tests/test_engine.py @@ -162,3 +162,54 @@ async def test_non_viewable_split_still_evaluable(self, monkeypatch): ) svc.evaluator.evaluate.assert_awaited_once() assert svc.status()[("dev", "ds1")].remaining_run_budget == 2 + + +class TestFreeEval: + """`free` waives only the budget debit; every access gate still applies. + + free and admin are distinct authorities: the sidecar's free baseline eval + once rode the admin flag and could thereby evaluate no_access splits and + hand the agent their aggregate score.""" + + @pytest.mark.asyncio + async def test_free_runs_without_debiting_budget(self, monkeypatch): + svc = _make_service(monkeypatch=monkeypatch) + await svc.evaluate( + EvalRequest(dataset_id="ds1", split="dev", commit="c1", num_samples=10), + free=True, + ) + svc.evaluator.evaluate.assert_awaited_once() + 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 + + svc = _make_service(monkeypatch=monkeypatch) + svc.split_accesses = [SplitAccess.no_access("test")] + with pytest.raises(InvalidSplitError): + await svc.evaluate( + EvalRequest(dataset_id="ds1", split="test", commit="c1", num_samples=10), + free=True, + ) + svc.evaluator.evaluate.assert_not_awaited() diff --git a/vero/tests/test_harbor_build.py b/vero/tests/test_harbor_build.py index 2efdeb2..a9aa8d7 100644 --- a/vero/tests/test_harbor_build.py +++ b/vero/tests/test_harbor_build.py @@ -363,6 +363,67 @@ 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 + + +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_protocol.py b/vero/tests/test_harbor_protocol.py index e636f46..c24f42c 100644 --- a/vero/tests/test_harbor_protocol.py +++ b/vero/tests/test_harbor_protocol.py @@ -91,3 +91,36 @@ def test_lists_budgeted_splits_with_tier(self): assert by_split["validation"]["tier"] == str(SplitAccessLevel.non_viewable) assert by_split["validation"]["agent_evaluable"] is True assert by_split["validation"]["remaining_run_budget"] == 3 + + def test_advertises_subset_floor_on_non_viewable_only(self): + budget = { + ("train", "ds1"): SplitBudget(split="train", dataset_id="ds1", total_run_budget=10), + ("validation", "ds1"): SplitBudget(split="validation", dataset_id="ds1", total_run_budget=3), + } + accesses = [ + SplitAccess.viewable("train"), + SplitAccess.non_viewable("validation"), + ] + status = build_status( + submit_enabled=False, + budget=budget, + split_accesses=accesses, + k_anonymity_floor=5, + ) + 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_runner.py b/vero/tests/test_harbor_runner.py index 3b90e66..b34449f 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)) @@ -99,6 +100,37 @@ def test_local_source(self, tmp_path): assert "-p" in cmd and str(tmp_path) in cmd assert "-d" not in cmd + def test_harbor_requirement_layers_trusted_cli(self): + # The overlay must come before the `harbor` executable name so uv + # resolves the CLI from the trusted spec, not the candidate's lockfile. + runner = HarborRunner( + HarborConfig( + task_source="org/ds@1", + agent_import_path="pkg.mod:Agent", + harbor_requirement="harbor==0.1.17", + ) + ) + cmd = runner._build_command("/wt", _params(), ["t0"], Path("/jobs")) + assert cmd[:6] == ["uv", "run", "--project", "/wt", "--with", "harbor==0.1.17"] + assert cmd[6] == "harbor" + + 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): @@ -364,6 +396,28 @@ def test_mean_zero_fills_attempts_without_rewards(self, tmp_path): assert r.metrics["n_dead"] == 1.0 assert r.metrics["n_attempts"] == 2.0 + def test_mean_records_dead_exception_types(self, tmp_path): + # n_dead alone hides WHY attempts died, and cause matters: rate-limit + # deaths are infra noise, crashes point at the candidate, and deaths + # cluster hard by cause (E1: 110/129 UnicodeDecodeErrors sat on two + # tasks). Every zero-filled attempt gets its exception type counted. + runner = HarborRunner(HarborConfig( + task_source="org/ds", agent_import_path="p:m", + n_attempts=3, aggregate_attempts="mean", + )) + jobs = tmp_path / "jobs"; run = jobs / "2026-01-01__00-00-00" + self._write(run, "t0a", "t0", rewards={"reward": 1.0}) + self._write(run, "t0bad", "t0", exc=True) # exception_type "X" + self._write(run, "t0gone", "t0") # no rewards AND no exception recorded + groups = runner._trial_groups(jobs) + r = runner._sample_result(groups["t0"][0], 0, "t0", _params(), attempts=groups["t0"]) + assert r.output["dead_exception_types"] == {"X": 1, "no_rewards_recorded": 1} + # all-clean samples carry no key at all + self._write(run, "t1a", "t1", rewards={"reward": 1.0}) + groups = runner._trial_groups(jobs) + r1 = runner._sample_result(groups["t1"][0], 1, "t1", _params(), attempts=groups["t1"]) + assert "dead_exception_types" not in r1.output + def test_mean_all_attempts_dead_errors_not_zero(self, tmp_path): # Every attempt died before scoring: that is an outage to investigate, # not a silent 0.0 measurement; the sample must surface as an error. @@ -596,6 +650,47 @@ 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_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. @@ -855,3 +950,339 @@ 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_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), + # 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"] == { + "recovered_round": 1, + "discarded_rounds": [{"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_rounds"] == [ + {"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 + 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] 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 diff --git a/vero/tests/test_harbor_server.py b/vero/tests/test_harbor_server.py index 57f233f..937ecca 100644 --- a/vero/tests/test_harbor_server.py +++ b/vero/tests/test_harbor_server.py @@ -1,6 +1,7 @@ """Tests for vero.harbor.server.EvaluationSidecar — handlers, tier-routing, submit.""" import json +from pathlib import Path from unittest.mock import AsyncMock, MagicMock import pytest @@ -41,9 +42,19 @@ def _experiment(split: str, commit: str = "abcdef123456") -> Experiment: ) -def _sidecar(tmp_path, *, split, submit_enabled=False, accesses=None, base_commit=None): +def _sidecar( + tmp_path, + *, + split, + submit_enabled=False, + accesses=None, + base_commit=None, + k_anonymity_floor=5, +): engine = MagicMock() engine.evaluate = AsyncMock(return_value=_experiment(split)) + # Full-split request by default (sample_ids None); floor tests override. + engine.resolve_samples = MagicMock(return_value=(None, 3)) engine.budget = BudgetLedger( [SplitBudget(split=split, dataset_id="ds1", total_run_budget=5, total_sample_budget=100)] ) @@ -56,6 +67,7 @@ def _sidecar(tmp_path, *, split, submit_enabled=False, accesses=None, base_commi admin_volume=tmp_path / "admin_vol", submit_enabled=submit_enabled, base_commit=base_commit, + k_anonymity_floor=k_anonymity_floor, ) # Stub the git transfer (integration-tested separately); pin the sha. sidecar._transfer_commit = AsyncMock(return_value="abcdef123456") @@ -72,7 +84,7 @@ async def test_visible_split_writes_full_per_sample(self, tmp_path): ) summary = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="train")) - dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456" + dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456__e1" assert (dest / "summary.json").exists() assert {(dest / f"{i}.json").exists() for i in range(3)} == {True} assert summary.result_path == str(dest) @@ -83,7 +95,7 @@ async def test_partial_split_writes_summary_only_no_labels(self, tmp_path): sidecar = _sidecar(tmp_path, split="validation") # non_viewable -> partial summary = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) - dest = tmp_path / "agent_vol" / "results" / "validation__abcdef123456" + dest = tmp_path / "agent_vol" / "results" / "validation__abcdef123456__e1" assert (dest / "summary.json").exists() # NO per-sample files -> the label-bearing feedback never reaches the agent assert not list(dest.glob("[0-9]*.json")) @@ -117,7 +129,7 @@ async def test_feedback_reaches_agent_on_viewable_only(self, tmp_path): tmp_path, split="train", accesses=[SplitAccess.viewable("train")] ) await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="train")) - dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456" + dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456__e1" assert "secret-0" in (dest / "0.json").read_text() @pytest.mark.asyncio @@ -161,7 +173,7 @@ async def test_attempts_reach_agent_on_viewable_only(self, tmp_path): return_value=self._experiment_with_attempts("train") ) await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="train")) - dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456" + dest = tmp_path / "agent_vol" / "results" / "train__abcdef123456__e1" blob = json.loads((dest / "0.json").read_text()) assert blob["output"]["attempts"] == [ {"reward": 0.0, "exception": "SecretTimeoutError"} @@ -218,6 +230,135 @@ def test_status_reports_submit_and_splits(self, tmp_path): assert status.splits[0]["remaining_run_budget"] == 5 +class TestHonestSummary: + """summary.json must qualify its mean: how many samples actually scored, + how many errored, and the standard error — a mean over 3-of-18 scored + samples is a different measurement than a clean full-split mean.""" + + @pytest.mark.asyncio + async def test_summary_carries_qualifiers(self, tmp_path): + sidecar = _sidecar(tmp_path, split="validation") + s = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + data = json.loads((Path(s.result_path) / "summary.json").read_text()) + # scores are [0.0, 1.0, 0.0] + assert data["n_samples"] == 3 + assert data["n_scored"] == 3 + assert data["n_errored"] == 0 + assert data["mean_score"] == pytest.approx(1 / 3) + # 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" + + @pytest.mark.asyncio + async def test_reevals_get_versioned_dirs(self, tmp_path): + # Repeat measurements of one commit are exactly the evidence worth + # comparing (multifidelity confirms, champion re-evals); the second + # eval must not erase the first. + sidecar = _sidecar(tmp_path, split="validation") + s1 = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + s2 = await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + assert s1.result_path != s2.result_path + assert s1.result_path.endswith("__e1") + assert s2.result_path.endswith("__e2") + assert (Path(s1.result_path) / "summary.json").exists() + assert (Path(s2.result_path) / "summary.json").exists() + + @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 + carries mean_score, so a singleton subset returns that sample's score + verbatim, and n singleton evals reconstruct the split's labels wholesale.""" + + def _floored(self, tmp_path, **kw): + return _sidecar(tmp_path, split="validation", **kw) + + @pytest.mark.asyncio + async def test_subset_below_floor_rejected_before_any_work(self, tmp_path): + from vero.harbor.server import KAnonymityError + + sidecar = self._floored(tmp_path) + sidecar.engine.resolve_samples = MagicMock(return_value=([0], 1)) + with pytest.raises(KAnonymityError): + await sidecar.evaluate( + EvalRequest(dataset_id="ds1", split="validation", sample_ids=[0]) + ) + # rejected up front: no commit transfer, no eval, no budget debit + sidecar._transfer_commit.assert_not_awaited() + sidecar.engine.evaluate.assert_not_awaited() + + @pytest.mark.asyncio + async def test_subset_at_floor_allowed(self, tmp_path): + sidecar = self._floored(tmp_path) + sidecar.engine.resolve_samples = MagicMock(return_value=(list(range(5)), 5)) + await sidecar.evaluate( + EvalRequest(dataset_id="ds1", split="validation", sample_ids=list(range(5))) + ) + sidecar.engine.evaluate.assert_awaited_once() + + @pytest.mark.asyncio + async def test_full_split_always_passes(self, tmp_path): + # sample_ids None == the whole split; its aggregate is the intended + # surface, so a split smaller than the floor stays evaluable. + sidecar = self._floored(tmp_path) + sidecar.engine.resolve_samples = MagicMock(return_value=(None, 3)) + await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + sidecar.engine.evaluate.assert_awaited_once() + + @pytest.mark.asyncio + async def test_viewable_split_not_floored(self, tmp_path): + # per-sample results are already exposed on viewable splits; nothing to protect + sidecar = _sidecar( + tmp_path, split="train", accesses=[SplitAccess.viewable("train")] + ) + sidecar.engine.resolve_samples = MagicMock(return_value=([0], 1)) + await sidecar.evaluate( + EvalRequest(dataset_id="ds1", split="train", sample_ids=[0]) + ) + sidecar.engine.evaluate.assert_awaited_once() + + @pytest.mark.asyncio + async def test_admin_exempt(self, tmp_path): + sidecar = self._floored(tmp_path) + sidecar.engine.resolve_samples = MagicMock(return_value=([0], 1)) + await sidecar.evaluate( + EvalRequest(dataset_id="ds1", split="validation", sample_ids=[0]), + admin=True, + ) + sidecar.engine.evaluate.assert_awaited_once() + + @pytest.mark.asyncio + async def test_floor_of_one_disables(self, tmp_path): + sidecar = self._floored(tmp_path, k_anonymity_floor=1) + sidecar.engine.resolve_samples = MagicMock(return_value=([0], 1)) + await sidecar.evaluate( + EvalRequest(dataset_id="ds1", split="validation", sample_ids=[0]) + ) + sidecar.engine.evaluate.assert_awaited_once() + + def test_status_advertises_floor(self, tmp_path): + sidecar = self._floored(tmp_path) + assert sidecar.status().splits[0]["min_subset_samples"] == 5 + + def test_status_floor_is_one_for_viewable(self, tmp_path): + sidecar = _sidecar( + tmp_path, split="train", accesses=[SplitAccess.viewable("train")] + ) + assert sidecar.status().splits[0]["min_subset_samples"] == 1 + + class TestFreeBaselineEval: """The agent's first eval of the seeded baseline is budget-free: it is the reference every candidate is compared to and can never win selection, so @@ -226,13 +367,15 @@ class TestFreeBaselineEval: no-op edit from an improvement, and quit with budget unspent).""" @pytest.mark.asyncio - async def test_first_baseline_eval_is_unmetered(self, tmp_path): + async def test_first_baseline_eval_is_unmetered_but_not_admin(self, tmp_path): sidecar = _sidecar(tmp_path, split="validation", base_commit="abcdef123456") await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) - # engine.evaluate was called with admin=True (bypasses the ledger) - assert sidecar.engine.evaluate.await_args.kwargs["admin"] is True - # but results were routed with the agent tier (summary written) - dest = tmp_path / "agent_vol" / "results" / "validation__abcdef123456" + # free waives the ledger; admin stays False so every access gate applies + # (riding admin=True here let the free baseline evaluate no_access splits) + assert sidecar.engine.evaluate.await_args.kwargs["free"] is True + assert sidecar.engine.evaluate.await_args.kwargs["admin"] is False + # and results were routed with the agent tier (summary written) + dest = tmp_path / "agent_vol" / "results" / "validation__abcdef123456__e1" assert (dest / "summary.json").exists() @pytest.mark.asyncio @@ -240,19 +383,47 @@ async def test_second_baseline_eval_is_metered(self, tmp_path): sidecar = _sidecar(tmp_path, split="validation", base_commit="abcdef123456") await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) - assert sidecar.engine.evaluate.await_args.kwargs["admin"] is False + assert sidecar.engine.evaluate.await_args.kwargs["free"] is False @pytest.mark.asyncio async def test_non_baseline_commit_always_metered(self, tmp_path): sidecar = _sidecar(tmp_path, split="validation", base_commit="other000000") await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) - assert sidecar.engine.evaluate.await_args.kwargs["admin"] is False + assert sidecar.engine.evaluate.await_args.kwargs["free"] is False @pytest.mark.asyncio async def test_no_base_commit_never_free(self, tmp_path): sidecar = _sidecar(tmp_path, split="validation") # base_commit=None await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) - assert sidecar.engine.evaluate.await_args.kwargs["admin"] is False + assert sidecar.engine.evaluate.await_args.kwargs["free"] is False + + @pytest.mark.asyncio + async def test_failed_free_eval_does_not_consume_freebie(self, tmp_path): + sidecar = _sidecar(tmp_path, split="validation", base_commit="abcdef123456") + sidecar.engine.evaluate = AsyncMock(side_effect=RuntimeError("infra")) + with pytest.raises(RuntimeError): + await sidecar.evaluate(EvalRequest(dataset_id="ds1", split="validation")) + # the failed eval gave the agent nothing; the free reference remains + assert sidecar.status().free_baseline_available is True + sidecar.engine.evaluate = AsyncMock(return_value=_experiment("validation")) + 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") diff --git a/vero/tests/test_harbor_verifier.py b/vero/tests/test_harbor_verifier.py index 264cdc3..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) @@ -610,3 +610,292 @@ async def test_missing_base_commit_warns(self, tmp_path, caplog): assert rewards == {"accuracy": 0.9} assert not (tmp_path / "baseline.json").exists() assert any("base_commit is not set" in m for m in caplog.messages) + + +class TestSelectionIntegrity: + """PR: finalize idempotency, pooled shortlisting, retrying reward-critical + evals, and the fail-safe floor. Each behavior traces to a live incident: + optimizers farmed max-over-rows selection with re-measure commits, and a + disk-full trial shipped no reward.json because finalize was single-shot.""" + + def _submit(self, tmp_path, commit="cand"): + (tmp_path / "submission.json").write_text(json.dumps({"commit": commit})) + + @pytest.mark.asyncio + async def test_finalize_is_idempotent(self, tmp_path): + # A retried finalize must replay the first result verbatim, not + # recompute (a re-run would re-rank against a DB polluted by the first + # finalize's own admin evals and could crown a different champion). + self._submit(tmp_path) + engine = _engine([0.8, 0.1]) # a recompute would consume the 0.1 + v = Verifier( + engine=engine, + admin_volume=tmp_path, + reward_mode="submit", + targets=[VerificationTarget(task="t", dataset_id="ds", split="test", reward_key="reward")], + ) + first = await v.finalize() + second = await v.finalize() + assert first == second + assert engine.evaluate_admin.await_count == 1 + + @pytest.mark.asyncio + async def test_same_commit_re_measures_pool_to_mean(self, tmp_path): + # Re-measuring a commit must reduce variance, not mint lottery draws: + # A's [0.9, 0.1] pools to 0.5 and loses the only shortlist slot to + # B's 0.6. Max-over-rows would have shortlisted A on the lucky 0.9. + engine = MagicMock() + engine.db.get_experiments_df.return_value = pd.DataFrame( + { + "dataset_subset_split": ["validation"] * 3, + "dataset_subset_dataset_id": ["ds1"] * 3, + "candidate_commit": ["A", "A", "B"], + "mean_score": [0.9, 0.1, 0.6], + "candidate_created_at": [1, 2, 3], + } + ) + engine.evaluate_admin = AsyncMock( + side_effect=lambda **kw: MagicMock( + result=MagicMock(score=MagicMock(return_value=0.7)) + ) + ) + engine.evaluator.workspace.tree_hash = AsyncMock(side_effect=lambda ref: ref + "-tree") + v = Verifier( + engine=engine, + admin_volume=tmp_path, + reward_mode="auto_best", + selection_split="validation", + selection_task="math", + rescore_top_k=1, + auto_best_baseline_floor=False, + targets=[VerificationTarget(task="math", dataset_id="ds1", split="test", reward_key="reward")], + ) + await v.finalize() + rescored = [c.kwargs["commit"] for c in engine.evaluate_admin.await_args_list] + assert "A" not in rescored + assert "B" in rescored + + @pytest.mark.asyncio + async def test_identical_trees_pool_and_do_not_stuff_shortlist(self, tmp_path): + # A1/A2 are the same content (same git tree) recommitted: they must + # collapse into ONE candidate group (pooled 0.85), so B still gets a + # top-2 shortlist slot and the group is re-scored exactly once. + engine = MagicMock() + engine.db.get_experiments_df.return_value = pd.DataFrame( + { + "dataset_subset_split": ["validation"] * 3, + "dataset_subset_dataset_id": ["ds1"] * 3, + "candidate_commit": ["A1", "A2", "B"], + "mean_score": [0.9, 0.8, 0.55], + "candidate_created_at": [1, 2, 3], + } + ) + engine.evaluate_admin = AsyncMock( + side_effect=lambda **kw: MagicMock( + result=MagicMock(score=MagicMock(return_value=0.7)) + ) + ) + trees = {"A1": "T", "A2": "T", "B": "TB"} + engine.evaluator.workspace.tree_hash = AsyncMock(side_effect=lambda ref: trees[ref]) + v = Verifier( + engine=engine, + admin_volume=tmp_path, + reward_mode="auto_best", + selection_split="validation", + selection_task="math", + rescore_top_k=2, + auto_best_baseline_floor=False, + targets=[VerificationTarget(task="math", dataset_id="ds1", split="test", reward_key="reward")], + ) + await v.finalize() + # count only selection-split re-scores (the winner also gets a final + # eval on the TARGET split, which is not a shortlist draw) + rescored = [ + c.kwargs["commit"] + for c in engine.evaluate_admin.await_args_list + if c.kwargs["split"] == "validation" + ] + assert "B" in rescored + assert len([c for c in rescored if c in ("A1", "A2")]) == 1 + + @pytest.mark.asyncio + async def test_floor_fail_safe_reverts_on_unmeasurable_baseline(self, tmp_path): + # If the floor's baseline eval cannot be measured, shipping the + # candidate would re-open the shipped-regression hole: revert to seed. + engine = MagicMock() + engine.db.get_experiments_df.return_value = pd.DataFrame( + { + "dataset_subset_split": ["validation"], + "dataset_subset_dataset_id": ["ds1"], + "candidate_commit": ["cand"], + "mean_score": [0.9], + "candidate_created_at": [1], + } + ) + + 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))) + + engine.evaluate_admin = AsyncMock(side_effect=_admin) + engine.evaluator.workspace.tree_hash = AsyncMock(side_effect=lambda ref: ref + "-tree") + v = Verifier( + engine=engine, + admin_volume=tmp_path, + reward_mode="auto_best", + selection_split="validation", + selection_task="math", + base_commit="base", + baseline_score_attempts=2, + targets=[VerificationTarget(task="math", dataset_id="ds1", split="test", reward_key="reward")], + ) + await v.finalize() + # the final (target) eval ran on the SEED, not the unverified candidate + 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 + # 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_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 + # must still ship) and records the outage in the wrapper so a floored + # reward can never masquerade as a measured 0.0. + self._submit(tmp_path) + engine = MagicMock() + engine.evaluate_admin = AsyncMock(side_effect=RuntimeError("boom")) + 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 "reward" in result["target_errors"] + assert engine.evaluate_admin.await_count == 2 # retried before flooring + + @pytest.mark.asyncio + async def test_target_eval_transient_failure_recovers(self, tmp_path): + self._submit(tmp_path) + engine = MagicMock() + healthy = MagicMock(result=MagicMock(score=MagicMock(return_value=0.8))) + engine.evaluate_admin = AsyncMock(side_effect=[RuntimeError("blip"), healthy]) + 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.8} + assert "target_errors" not in result + + @pytest.mark.asyncio + async def test_all_error_eval_is_retried_not_zeroed(self, tmp_path): + # An eval in which no sample scored (score(fill_score=None) is None) + # is an outage: retry it, never let it quietly become a measured 0.0. + self._submit(tmp_path) + engine = MagicMock() + all_error = MagicMock( + result=MagicMock( + score=MagicMock(side_effect=lambda fill_score=0.0: None if fill_score is None else 0.0) + ) + ) + healthy = MagicMock(result=MagicMock(score=MagicMock(return_value=0.7))) + engine.evaluate_admin = AsyncMock(side_effect=[all_error, healthy]) + 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.7} + assert engine.evaluate_admin.await_count == 2