refactor(harbor): split Mode A / Mode B into discriminated-union configs

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,22 +193,39 @@ 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.
 
         Unlike :meth:`evaluate` (which is bound to ``default_task`` and metered),
         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,
             split=split,
             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]:

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

vero/src/vero/harbor/build/__init__.py

@@ -1,6 +1,17 @@
 """The `vero harbor build` compiler: BuildConfig -> a runnable Harbor task dir."""
 
 from vero.harbor.build.compiler import compile_task
-from vero.harbor.build.config import BuildConfig
+from vero.harbor.build.config import (
+    BuildConfig,
+    BuildConfigA,
+    BuildConfigB,
+    load_build_config,
+)
 
-__all__ = ["BuildConfig", "compile_task"]
+__all__ = [
+    "BuildConfig",
+    "BuildConfigA",
+    "BuildConfigB",
+    "compile_task",
+    "load_build_config",
+]

vero/src/vero/harbor/build/compiler.py

@@ -18,7 +18,7 @@
 from jinja2 import Environment, FileSystemLoader
 
 from vero.evaluation.engine import EvalRequest
-from vero.harbor.build.config import BuildConfig
+from vero.harbor.build.config import BuildConfigA, BuildConfigB
 from vero.harbor.protocol import StatusSummary
 
 logger = logging.getLogger(__name__)
@@ -210,24 +210,14 @@ def _validate_partition_names(
         )
 
 
-def _serve_config(config: BuildConfig, dataset_id: str | None, base_commit: str) -> dict:
-    harbor = None
-    if config.harbor is not None:
-        # Local inner task -> baked sidecar-only path; registry ref -> pass through.
-        harbor = {**config.harbor}
-        if config.inner_task:
-            harbor["task_source"] = INNER_TASK
-    targets = [
-        {
-            "task": config.task,
-            "dataset_id": dataset_id,
-            "split": t.split,
-            "reward_key": t.reward_key,
-            "sample_ids": t.sample_ids,
-        }
-        for t in config.targets
-    ]
-    return {
+def _serve_config(
+    config: BuildConfigA | BuildConfigB, dataset_id: str | None, base_commit: str
+) -> dict:
+    # Fields common to both ServeConfig variants. Mode-specific keys are added
+    # below so the wrong-mode key never reaches the (extra="forbid") ServeConfig.
+    task = config.task if isinstance(config, BuildConfigA) else None
+    common = {
+        "mode": config.mode,
         "repo_path": AGENT_BASELINE,
         "agent_repo_path": WORK_AGENT,
         "session_id": SESSION_ID,
@@ -237,33 +227,61 @@ def _serve_config(config: BuildConfig, dataset_id: str | None, base_commit: str)
             {"split": b.split, "dataset_id": dataset_id, **b.model_dump(exclude={"split"}, exclude_none=True)}
             for b in config.budgets
         ],
-        "task": config.task,
-        "task_project": config.task_project,
-        "task_module": config.task_module,
-        "harbor": harbor,
         "reward_mode": config.reward_mode,
         "selection_split": config.selection_split,
-        "targets": targets,
+        "targets": [
+            {
+                "task": task,
+                "dataset_id": dataset_id,
+                "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,
-        "feedback_transcripts": config.feedback_transcripts,
-        "feedback_max_bytes": config.feedback_max_bytes,
-        "instruct_multifidelity": config.instruct_multifidelity,
-        "expose_attempt_detail": config.expose_attempt_detail,
+        "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,
         "timeout": config.timeout,
-        "sample_timeout": config.sample_timeout,
         "max_concurrency": config.max_concurrency,
         "host": "0.0.0.0",
         "port": 8000,
     }
+    if isinstance(config, BuildConfigA):
+        return {
+            **common,
+            "task": config.task,
+            "task_project": config.task_project,
+            "task_module": config.task_module,
+            "sample_timeout": config.sample_timeout,
+        }
+    # Mode B: local inner task -> baked sidecar-only path; registry ref passes through.
+    harbor = None
+    if config.harbor is not None:
+        harbor = {**config.harbor}
+        if config.inner_task:
+            harbor["task_source"] = INNER_TASK
+    return {
+        **common,
+        "harbor": harbor,
+        "feedback_transcripts": config.feedback_transcripts,
+        "feedback_max_bytes": config.feedback_max_bytes,
+        "instruct_multifidelity": config.instruct_multifidelity,
+        "expose_attempt_detail": config.expose_attempt_detail,
+    }
 
 
 def compile_task(
-    config: BuildConfig, out_dir: Path | str, *, vero_root: Path | None = None
+    config: BuildConfigA | BuildConfigB,
+    out_dir: Path | str,
+    *,
+    vero_root: Path | None = None,
 ) -> Path:
     """Compile ``config`` into a Harbor task directory at ``out_dir``."""
     import json
@@ -272,23 +290,10 @@ def compile_task(
 
     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),
-            )
+    # The Mode-A "you set a Mode-B-only lever" warning (and its ServeConfig twin)
+    # is superseded by the Mode-A / Mode-B type split: a Mode-A config that sets
+    # feedback_transcripts / expose_attempt_detail is now a load-time
+    # ValidationError, so the condition is structurally impossible here.
 
     out = Path(out_dir)
     if out.exists():
@@ -315,7 +320,7 @@ def compile_task(
     # the dataset into vh before the dir is torn down, so cleanup is safe.
     with tempfile.TemporaryDirectory() as tmp_str:
         tmp = Path(tmp_str)
-        if config.mode == "A":
+        if isinstance(config, BuildConfigA):
             if not config.dataset:
                 raise ValueError("Mode A requires a dataset.")
             dataset_id = _register(config.dataset, vh, tmp)
@@ -365,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,
@@ -373,7 +382,9 @@ def compile_task(
         selection_split=config.selection_split,
         submit_enabled=config.submit_enabled,
         eval_num_samples=None,
-        bake_inner_task=bool(config.inner_task),
+        # inner_task / instruct_multifidelity are Mode-B-only fields; on a
+        # Mode-A config they do not exist, so read them only when present.
+        bake_inner_task=bool(getattr(config, "inner_task", None)),
         # The free-baseline bullet may only render when the sidecar shipping in
         # this same tree actually grants the free eval; the feature lives on a
         # different PR chain than the compiler, and an instruction that promises
@@ -392,10 +403,11 @@ def compile_task(
         # 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
+        multifidelity=getattr(config, "instruct_multifidelity", False)
         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)