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