fix: [no-ticket] harbor - ops integrity (exhaust-budget instruction lever, fail-closed ledger restore)

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

@@ -243,6 +243,7 @@ def _serve_config(
         "submit_enabled": config.submit_enabled,
         "score_baseline": config.score_baseline,
         "k_anonymity_floor": config.k_anonymity_floor,
+        "instruct_exhaust_budget": config.instruct_exhaust_budget,
         "agent_volume": AGENT_VOLUME,
         "admin_volume": ADMIN_VOLUME,
         "admin_token_path": TOKEN_PATH,
@@ -405,6 +406,7 @@ def compile_task(
         and {"sample_ids", "num_samples"}
         <= {f.name for f in dataclasses.fields(EvalRequest)}
         and any(s.access == "viewable" for s in config.splits),
+        exhaust_budget=config.instruct_exhaust_budget,
     )
     _render(jenv, "task.toml.j2", out / "task.toml", **ctx)
     _render(jenv, "instruction.md.j2", out / "instruction.md", **ctx)

vero/src/vero/harbor/build/config.py

@@ -72,6 +72,13 @@ class _BuildConfigBase(BaseModel):
     # mean_score, so singleton subsets would hand back per-sample labels.
     k_anonymity_floor: int = 5
 
+    # Instruction lever: render the "unspent budget is wasted" persistence
+    # bullet that tells the optimizer to keep spending (re-measure the champion,
+    # try one more variant) instead of stopping early. On by default (current
+    # behavior); off makes stopping-early a choice the agent arrives at itself,
+    # which is the ablation arm for measuring what the exhortation contributes.
+    instruct_exhaust_budget: bool = True
+
     # write-access: paths in the target repo the optimizer may NOT edit
     # (the scorer, by default). Applied as unix perms in main before the agent runs.
     read_only_paths: list[str] = Field(default_factory=list)

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.

vero/src/vero/harbor/protocol.py

@@ -104,7 +104,10 @@ def build_status(
     split_accesses: list[SplitAccess],
     base_commit: str | None = None,
     free_baseline_available: bool = False,
-    k_anonymity_floor: int = 1,
+    # Default matches EvaluationSidecar's enforcement default: a caller that
+    # forgets to pass the floor must not advertise a laxer one than the
+    # sidecar enforces (agents would send sub-floor requests that 400).
+    k_anonymity_floor: int = 5,
 ) -> StatusSummary:
     """Build the agent-facing status from the budget ledger + split tiers.
 

vero/src/vero/harbor/runner.py

@@ -411,6 +411,12 @@ def _out(output: dict) -> dict:
         if not rewards:
             return SampleResult(
                 error=f"No verifier rewards for task '{task_name}'.",
+                # The agent died before scoring. A candidate edit that CRASHES
+                # the agent lands here, and "no verifier rewards" alone gives
+                # the optimizer no way to see its own crash; the transcript
+                # does. Passed as score 0.0: an unscored attempt counts as a
+                # failure everywhere else too.
+                feedback=self._failure_feedback(0.0, attempts),
                 output=_out(
                     {"task_name": task_name, "trial_name": trial.get("trial_name")}
                 ),
@@ -553,6 +559,12 @@ def _read_transcript_tail(self, trial_dir: Path) -> str | None:
                 data = path.read_bytes()
             except OSError:
                 continue
+            # An empty transcript carries nothing: keep looking (an empty pane
+            # must fall through to the trajectory), and if every candidate is
+            # empty return None so the caller tries the next failed attempt
+            # instead of emitting "" as feedback.
+            if not data:
+                continue
             # errors="replace": a multibyte char straddling the cap boundary is
             # rendered as U+FFFD rather than crashing the collation.
             return data[-self.feedback_max_bytes :].decode("utf-8", errors="replace")

vero/src/vero/harbor/serve.py

@@ -10,6 +10,7 @@
 
 import json
 import logging
+import shutil
 from pathlib import Path
 from typing import Annotated, Literal
 
@@ -82,6 +83,10 @@ class _ServeConfigBase(BaseModel):
     # EvaluationSidecar.k_anonymity_floor for the leak this closes.
     k_anonymity_floor: int = 5
 
+    # Consumed at COMPILE time (the instruction's exhaust-budget bullet);
+    # recorded here so serve.json mirrors build.yaml, like instruct_multifidelity.
+    instruct_exhaust_budget: bool = True
+
     # volumes / token
     agent_volume: str
     admin_volume: str
@@ -155,8 +160,12 @@ def _load_or_build_ledger(
     a sidecar restart would reset all spent budget to full, letting the agent
     regain its full evaluation budget by triggering a restart. On startup we
     reconstruct each SplitBudget and restore its persisted ``remaining_*`` values.
-    Falls back to the configured budgets if the file is missing or unreadable
-    (fail-safe to the configured budget, never to unlimited).
+
+    A MISSING file is a fresh boot: configured budgets. A file that exists but
+    cannot be parsed fails CLOSED: metered budgets restore with zero remaining.
+    The old fallback (configured budgets) refunded the agent everything already
+    spent, so any crash that corrupted the flush minted budget; spend that
+    cannot be read must be treated as fully spent, never as never-happened.
     """
     if persist_path.exists():
         try:
@@ -181,11 +190,28 @@ def _load_or_build_ledger(
             )
             return BudgetLedger(budgets, persist_path=persist_path)
         except (json.JSONDecodeError, KeyError, OSError) as e:
-            logger.warning(
-                "Could not reload persisted ledger %s (%s); using configured budgets.",
+            logger.error(
+                "Persisted ledger %s exists but is unreadable (%s); failing "
+                "CLOSED: metered agent budgets restore as exhausted. Admin and "
+                "finalize are unaffected. The unreadable file is preserved at "
+                "%s; delete ledger.json deliberately to boot fresh.",
                 persist_path,
                 e,
+                persist_path.with_suffix(".corrupt"),
             )
+            try:  # keep the evidence: the next flush overwrites persist_path
+                shutil.copyfile(persist_path, persist_path.with_suffix(".corrupt"))
+            except OSError:
+                pass
+            budgets = []
+            for cfg in budget_cfgs:
+                b = SplitBudget(**cfg)
+                if b.total_sample_budget is not None:
+                    b.remaining_sample_budget = 0
+                if b.total_run_budget is not None:
+                    b.remaining_run_budget = 0
+                budgets.append(b)
+            return BudgetLedger(budgets, persist_path=persist_path)
     return BudgetLedger(
         [SplitBudget(**b) for b in budget_cfgs], persist_path=persist_path
     )

vero/src/vero/harbor/server.py

@@ -11,6 +11,7 @@
 
 import json
 import logging
+import re
 import shutil
 from dataclasses import replace
 from pathlib import Path
@@ -124,14 +125,20 @@ async def evaluate(self, req: EvalRequest, *, admin: bool = False) -> EvalSummar
             and sha == self.base_commit
             and not self._free_baseline_used
         )
-        exp = await self.engine.evaluate(
-            replace(req, commit=sha), admin=admin, free=free_baseline
-        )
-        # Consume the freebie only after the eval succeeded: an eval that
-        # raised (invalid split, infra failure) has given the agent nothing,
-        # so it must not burn the one free reference measurement.
+        # Claim the freebie BEFORE the await (asyncio is atomic between await
+        # points, so a concurrent second baseline eval sees the claim and pays)
+        # and refund it if the eval raises: a failed eval gave the agent
+        # nothing, so it must not burn the one free reference measurement.
         if free_baseline:
             self._free_baseline_used = True
+        try:
+            exp = await self.engine.evaluate(
+                replace(req, commit=sha), admin=admin, free=free_baseline
+            )
+        except BaseException:
+            if free_baseline:
+                self._free_baseline_used = False
+            raise
         # Route with the agent's real tier even when the eval was unmetered.
         result_path = self._route_results(exp, admin=admin)
         budget_remaining = None
@@ -282,31 +289,45 @@ def _route_results(self, experiment: Experiment, *, admin: bool) -> str | None:
         # agent's earlier evidence for the same commit; repeat measurements are
         # exactly the ones worth comparing. result_path in the response names
         # the dir for THIS eval.
+        results_root = self.agent_volume / "results"
+        if self._eval_seq == 0 and results_root.exists():
+            # Volume reuse (sidecar restart): resume the ordinal past every
+            # surviving dir, or the first N evals of the new session would
+            # silently wipe __e1..__eN — the exact erasure this scheme exists
+            # to prevent.
+            self._eval_seq = max(
+                (
+                    int(m.group(1))
+                    for d in results_root.iterdir()
+                    if (m := re.search(r"__e(\d+)$", d.name))
+                ),
+                default=0,
+            )
         self._eval_seq += 1
-        dest = (
-            self.agent_volume
-            / "results"
-            / f"{split}__{commit[:12]}__e{self._eval_seq}"
-        )
-        if dest.exists():  # ordinal collision only on volume reuse; never merge
+        dest = results_root / f"{split}__{commit[:12]}__e{self._eval_seq}"
+        if dest.exists():  # unreachable after the resume scan; never merge
             shutil.rmtree(dest)
         dest.mkdir(parents=True, exist_ok=True)
 
         # Aggregate summary is label-safe for both visible and partial tiers.
-        # n_scored / n_errored / score_se qualify the mean: a mean over 3
+        # n_scored / n_errored / mean_score_se qualify the mean: a mean over 3
         # scored samples of 18, or one dominated by errored zero-fills, is a
         # different measurement than a clean full-split mean, and the agent
         # (and any auditor) should see that without per-sample access.
+        # mean_score_se is named to bind it to mean_score: both are computed
+        # over the zero-filled n_samples population (score() fills errored
+        # samples with 0.0), NOT over the n_scored subset — an SE of the
+        # 3-of-18-scored mean would be a different (and larger) number.
         sample_results = experiment.result.sample_results
         filled = [
             r.score if r.score is not None else 0.0
             for r in sample_results.values()
         ]
-        score_se = None
+        mean_score_se = None
         if len(filled) > 1:
             m = sum(filled) / len(filled)
             var = sum((x - m) ** 2 for x in filled) / (len(filled) - 1)
-            score_se = (var / len(filled)) ** 0.5
+            mean_score_se = (var / len(filled)) ** 0.5
         (dest / "summary.json").write_text(
             json.dumps(
                 {
@@ -320,7 +341,7 @@ def _route_results(self, experiment: Experiment, *, admin: bool) -> str | None:
                         1 for r in sample_results.values() if r.is_error()
                     ),
                     "mean_score": experiment.result.score(),
-                    "score_se": score_se,
+                    "mean_score_se": mean_score_se,
                     "status": experiment.result.status.value,
                 },
                 indent=2,

vero/tests/test_harbor_build.py

@@ -363,6 +363,39 @@ def test_instruction_omits_free_baseline_claim_when_unsupported(built):
     assert "budget-free" not in text
 
 
+def test_instruction_renders_exhaust_budget_by_default(built):
+    text = (built / "instruction.md").read_text()
+    assert "Unspent budget is wasted" in text
+
+
+def test_instruction_omits_exhaust_budget_when_disabled(tmp_path, monkeypatch):
+    # The persistence exhortation is an instruction LEVER: off, stopping early
+    # becomes the agent's own choice, which is the ablation arm for measuring
+    # what the exhortation itself contributes.
+    monkeypatch.setenv("VERO_SKIP_SECRET_CHECK", "1")
+    config = BuildConfigA(
+        name="vero/gsm8k-opt",
+        description="optimize gsm8k",
+        agent_repo=str(_agent_repo(tmp_path)),
+        task="gsm8k",
+        task_module="gsm8k_agent.vero_tasks",
+        dataset=str(_dataset(tmp_path)),
+        splits=[
+            {"split": "validation", "access": "non_viewable"},
+            {"split": "test", "access": "no_access"},
+        ],
+        budgets=[{"split": "validation", "total_run_budget": 5}],
+        reward_mode="auto_best",
+        selection_split="validation",
+        targets=[{"split": "test", "reward_key": "reward"}],
+        instruct_exhaust_budget=False,
+    )
+    out = compile_task(config, tmp_path / "task", vero_root=_stub_vero(tmp_path))
+    text = (out / "instruction.md").read_text()
+    assert "Unspent budget is wasted" not in text
+    assert "Scores are noisy." in text  # the noise fact is not part of the lever
+
+
 def _multifidelity_config_with_splits(tmp_path, splits) -> BuildConfigB:
     # instruct_multifidelity is a Mode-B-only lever, so the multi-fidelity gate is
     # exercised through a Mode-B config (local inner_task so it compiles offline).

vero/tests/test_harbor_protocol.py

@@ -110,3 +110,17 @@ def test_advertises_subset_floor_on_non_viewable_only(self):
         by_split = {s["split"]: s for s in status.splits}
         assert by_split["validation"]["min_subset_samples"] == 5
         assert by_split["train"]["min_subset_samples"] == 1
+
+    def test_default_floor_matches_sidecar_enforcement_default(self):
+        # A caller that forgets to pass the floor must not advertise a laxer
+        # one than EvaluationSidecar enforces by default (5): agents would
+        # send sub-floor requests that get rejected.
+        budget = {
+            ("validation", "ds1"): SplitBudget(split="validation", dataset_id="ds1", total_run_budget=3),
+        }
+        status = build_status(
+            submit_enabled=False,
+            budget=budget,
+            split_accesses=[SplitAccess.non_viewable("validation")],
+        )
+        assert status.splits[0]["min_subset_samples"] == 5

vero/tests/test_harbor_runner.py

@@ -636,6 +636,33 @@ def test_missing_transcripts_omitted_silently(self, tmp_path):
         assert r.score == 0.0
         assert r.feedback is None
 
+    def test_empty_pane_falls_through_to_trajectory(self, tmp_path):
+        # An empty transcript file carries nothing and must not surface as ""
+        # feedback; the empty pane falls through to the trajectory.
+        runner = _fb_runner()
+        jobs = tmp_path / "jobs"
+        _write_trial(
+            jobs, "trial0", "t0", {"reward": 0.0}, pane="", trajectory='{"steps": [1]}'
+        )
+        assert self._result(runner, jobs).feedback == '{"steps": [1]}'
+
+    def test_all_empty_transcripts_yield_no_feedback(self, tmp_path):
+        runner = _fb_runner()
+        jobs = tmp_path / "jobs"
+        _write_trial(jobs, "trial0", "t0", {"reward": 0.0}, pane="", trajectory="")
+        assert self._result(runner, jobs).feedback is None
+
+    def test_no_rewards_error_sample_carries_crash_transcript(self, tmp_path):
+        # A candidate edit that crashes the agent before scoring lands in the
+        # no-verifier-rewards error branch; the transcript is the only way the
+        # optimizer can see the crash it caused.
+        runner = _fb_runner()
+        jobs = tmp_path / "jobs"
+        _write_trial(jobs, "trial0", "t0", None, pane="crash tail")
+        r = self._result(runner, jobs)
+        assert r.error is not None
+        assert r.feedback == "crash tail"
+
     def test_first_failed_attempt_transcript_used(self, tmp_path):
         # Two failed attempts: the FIRST one's transcript (by finished_at) is
         # attached, deterministically, regardless of rglob order.

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