feat(harbor): feedback levers: failure transcripts, multi-fidelity screening, per-attempt detail

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)

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 <admin_volume>/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.

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.

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 '<org>/<name>' 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",