Bench: harbor + swe-bench generation mode with runtime hardening (+ a templated dataset card)

config.example.yaml

@@ -144,6 +144,11 @@ output:
   # These files are excluded from resume detection, conversion, README generation, and uploads.
   failures_dir: ./failures
 
+  # Harbor bench-mode intermediates (downloaded sources, raw trials, normalized sessions).
+  # Defaults to a `bench` dir beside traces_dir (parallel to sandbox/failures), never inside
+  # the dataset. Set a path to override.
+  bench_dir: null
+
   # Used in the generated trace README.
   pretty_name: "My Agent Traces"
 
@@ -154,6 +159,23 @@ publish:
   hf_token: null
   private: false
 
+# Benchmark mode (`teich generate --mode bench`). Each source is run by a backend selected
+# by `type`: `harbor` (the harbor package; needs teich[harbor], Python 3.12+) or `swe-bench`
+# (the swebench package; needs teich[swe]). teich runs the configured agent on each task/
+# instance, harvests the plain native trace into output/{passed,failed,borderline}/, and
+# writes scores + provenance to output/metadata/<stem>.json. Intermediates live under
+# output.bench_dir (a sibling `bench` dir by default), outside the dataset/uploads.
+# `--resume` skips already-harvested tasks; `--refresh` re-downloads remote harbor sources
+# (no effect on swe-bench, which uses Hugging Face's own dataset cache).
+# Keep bench its own project: dedicated output.traces_dir + publish.repo_id.
+bench:
+  sources: []
+    # - { type: harbor,    source: terminal-bench@2.0 }          # registry spec
+    # - { type: harbor,    source: ./local-tasks }               # local task dir / dir of task dirs
+    # - { type: swe-bench, source: SWE-bench/SWE-bench_Verified, split: test }
+    # - { type: swe-bench, source: SWE-bench/SWE-bench_Lite, instances: [django__django-12345] }
+    #   # optional per-source: repo (git/HF registry), version, instances [..], backend (harbor)
+
 # Number of prompts to run in parallel.
 max_concurrency: 1
 

pyproject.toml

@@ -20,6 +20,7 @@ dependencies = [
     "rich>=13.0",
     "datasets>=2.19.0",
     "huggingface_hub>=0.23.0",
+    "jinja2>=3.1",
     "fastapi>=0.110",
     "uvicorn>=0.29",
     "websockets>=12",
@@ -28,6 +29,10 @@ dependencies = [
 [project.optional-dependencies]
 studio = ["fastapi>=0.110", "uvicorn>=0.29", "websockets>=12"]
 dev = ["pytest>=8.0", "ruff>=0.4", "jinja2>=3.1", "httpx>=0.27"]
+# Per-backend bench extras (install only what you run); `bench` enables both.
+harbor = ["harbor>=0.15.0 ; python_full_version >= '3.12'"]
+swe = ["swebench>=4.1.0", "jinja2>=3.1"]
+bench = ["teich[harbor,swe]"]
 
 [project.scripts]
 teich = "teich.cli:main"

src/teich/bench/__init__.py

@@ -0,0 +1,5 @@
+"""Benchmark (Harbor-format) task generation for ``teich generate --mode bench``."""
+
+from .runner import run_bench
+
+__all__ = ["run_bench"]

src/teich/bench/backends/__init__.py

@@ -0,0 +1,50 @@
+"""Bench backends: a registry mapping a source ``type`` to its backend implementation."""
+
+from __future__ import annotations
+
+from .base import (
+    BENCH_SPLITS,
+    BenchBackend,
+    BenchRun,
+    BenchTask,
+    bench_root,
+    bench_stem,
+    existing_output,
+    harvest,
+    primary_score,
+    route_split,
+    source_id,
+)
+from .harbor import HarborBackend
+from .swebench import SweBenchBackend
+
+# type -> backend factory.
+_BACKENDS: dict[str, type] = {
+    HarborBackend.type: HarborBackend,
+    SweBenchBackend.type: SweBenchBackend,
+}
+
+
+def get_backend(source_type: str) -> BenchBackend:
+    """Instantiate the backend for a source ``type``; clear error for an unknown type."""
+    cls = _BACKENDS.get(source_type)
+    if cls is None:
+        supported = ", ".join(sorted(_BACKENDS))
+        raise RuntimeError(f"Unknown bench source type {source_type!r}; supported: {supported}.")
+    return cls()
+
+
+__all__ = [
+    "BENCH_SPLITS",
+    "BenchBackend",
+    "BenchRun",
+    "BenchTask",
+    "bench_root",
+    "bench_stem",
+    "existing_output",
+    "get_backend",
+    "harvest",
+    "primary_score",
+    "route_split",
+    "source_id",
+]

src/teich/bench/backends/base.py

@@ -0,0 +1,227 @@
+"""Pluggable bench-backend abstraction.
+
+A benchmark *source* (declared in ``bench.sources`` with a ``type``) is executed by a
+backend (harbor, swe-bench). Each backend turns a task into a ``BenchRun`` — the agent's
+plain native trace plus a rewards dict — and the *harvest* here (route by score into
+``passed``/``failed``/``borderline`` + a per-task ``metadata/`` sidecar) is backend-agnostic.
+
+Trace/metadata filenames are namespaced by source (``bench-<source>-<task>``) so two sources
+in one dataset can never collide; an unfinished task writes nothing and is retried on resume.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Iterable, Protocol
+
+from ...converter import convert_traces_to_training_data
+
+if TYPE_CHECKING:
+    from ...config import BenchSource, Config
+
+BENCH_SPLITS = ("passed", "failed", "borderline")
+
+
+@dataclass
+class BenchTask:
+    """One unit of work within a source (a harbor task dir, a swe-bench instance, ...)."""
+
+    id: str
+    raw: Any = None
+
+
+@dataclass
+class BenchRun:
+    """A backend's result for one task: the native trace + scores + provenance metadata."""
+
+    native_lines: list[str]
+    rewards: dict[str, float] | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+class BenchBackend(Protocol):
+    """What a benchmark backend must provide. The driver + harvest are shared."""
+
+    type: str
+
+    def require(self) -> None:
+        """Raise RuntimeError with an install hint if the backend's extra is missing."""
+
+    def tasks(self, cfg: Config, source: BenchSource, *, refresh: bool = False) -> Iterable[BenchTask]:
+        """Resolve a source into its tasks/instances (``refresh`` re-fetches remote data)."""
+
+    def run(self, cfg: Config, source: BenchSource, task: BenchTask) -> BenchRun:
+        """Run the agent on one task in its environment and return its trace + rewards."""
+
+
+def bench_root(cfg: Config) -> Path:
+    """Working dir for backends' intermediates (downloads, sessions, trials).
+
+    A ``bench`` dir beside ``traces_dir`` (parallel to sandbox/failures, outside the
+    dataset); overridable via ``output.bench_dir``. Re-checked here (not just at config
+    load) so a ``--output`` override can't leave ``bench_dir`` inside the dataset.
+    """
+    root = Path(cfg.output.bench_dir) if cfg.output.bench_dir is not None else cfg.output.traces_dir.parent / "bench"
+    # Guard both the explicit override and the computed default: a bench root at/under traces_dir
+    # (e.g. output.traces_dir named ``bench``, so the sibling default collides with it) would get
+    # raw trials/sessions uploaded and misclassified as dataset rows.
+    traces = cfg.output.traces_dir.resolve()
+    if traces == root.resolve() or traces in root.resolve().parents:
+        raise RuntimeError(
+            f"bench working dir ({root}) must be outside output.traces_dir "
+            f"({cfg.output.traces_dir}); raw trials/sessions there would be uploaded and "
+            "misclassified as dataset rows. Set output.bench_dir explicitly or rename traces_dir."
+        )
+    return root
+
+
+def slug(value: str) -> str:
+    """Filesystem-safe slug (``terminal-bench@2.0`` -> ``terminal-bench-2.0``)."""
+    return re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip("-") or "x"
+
+
+def _instances_key(instances: list[str] | None) -> str | None:
+    """A bounded, order-independent discriminator for an ``instances`` subset.
+
+    The full comma-joined list would otherwise flow through ``source_id`` into per-task
+    filenames, Docker tags, and container/env-file names; a dozen ~22-char SWE-bench IDs
+    blow past the 255-byte path-component and 128-char Docker-tag limits and crash the run
+    before any result is written. Short lists stay readable; longer ones collapse to a
+    stable hash (sorted, so listing order doesn't spawn a distinct id).
+    """
+    if not instances:
+        return None
+    joined = ",".join(sorted(instances))
+    if len(joined) <= 40:
+        return joined
+    return "i" + hashlib.sha1(joined.encode("utf-8")).hexdigest()[:8]
+
+
+def source_id(source: BenchSource) -> str:
+    """Stable identifier for a source, used to namespace its output files.
+
+    Keyed on the discriminating knobs (not just ``source``): two sources that share a spec
+    but differ in ``repo``/``version``/``split``/``instances``/``backend`` get distinct ids,
+    so they can't overwrite each other's traces or wrongly resume-skip. A suffix is appended
+    only when a field diverges from its default, so the common single-field source keeps a
+    clean, stable id. ``type`` is intentionally excluded (it is recorded in ``metadata`` and
+    the existing namespacing contract is type-independent).
+    """
+    extras = [
+        slug(part)
+        for part in (
+            source.repo,
+            source.version,
+            source.split,
+            _instances_key(source.instances),
+            source.backend if source.backend != "docker" else None,
+        )
+        if part
+    ]
+    base_id = slug(source.source)
+    return base_id if not extras else f"{base_id}-{slug('-'.join(extras))}"
+
+
+def bench_stem(source: BenchSource, task_id: str) -> str:
+    """Per-task dataset stem, namespaced by source (the co-mingle guard keys on ``bench-``)."""
+    return f"bench-{source_id(source)}-{slug(task_id)}"
+
+
+def existing_output(cfg: Config, stem: str) -> Path | None:
+    """The harvested trace for ``stem`` (in any split), or None — used for ``--resume``."""
+    for split in BENCH_SPLITS:
+        path = cfg.output.traces_dir / split / f"{stem}.jsonl"
+        if path.is_file() and path.stat().st_size > 0:
+            return path
+    return None
+
+
+def numeric(value: Any) -> float | None:
+    """A real number (bools excluded) as float, else None."""
+    if isinstance(value, (int, float)) and not isinstance(value, bool):
+        return float(value)
+    return None
+
+
+def rewards_from_mapping(data: Any) -> dict[str, float] | None:
+    """The full dict of numeric scores (no clamping). Accepts ``{"rewards": {...}}`` or flat."""
+    if not isinstance(data, dict):
+        return None
+    rewards = data.get("rewards") if isinstance(data.get("rewards"), dict) else data
+    scores = {key: numeric(val) for key, val in rewards.items() if numeric(val) is not None}
+    return scores or None
+
+
+def primary_score(rewards: dict[str, float] | None) -> float | None:
+    """The scalar used for routing: the ``reward`` key, else the first numeric value."""
+    if not rewards:
+        return None
+    primary = numeric(rewards.get("reward"))
+    if primary is not None:
+        return primary
+    for value in rewards.values():
+        primary = numeric(value)
+        if primary is not None:
+            return primary
+    return None
+
+
+def route_split(primary: float | None) -> str:
+    """passed = score 1, failed = score 0 / unscored, borderline = any other value."""
+    if primary is None or primary == 0:
+        return "failed"
+    if primary == 1:
+        return "passed"
+    return "borderline"
+
+
+def trace_metadata(native_dir: Path | None) -> dict[str, Any]:
+    """Metadata teich's converter recovers from a native trace (model, session, cwd, ...)."""
+    if native_dir is None:
+        return {}
+    try:
+        rows = convert_traces_to_training_data(native_dir)
+    except Exception:  # metadata is best-effort; never fail the harvest over it
+        return {}
+    return rows[0].get("metadata", {}) if rows else {}
+
+
+def harvest(cfg: Config, source: BenchSource, task: BenchTask, run: BenchRun) -> tuple[list[Path], str]:
+    """Write a backend's native trace (routed by score) + a per-task metadata sidecar.
+
+    Returns (written paths, split). The trace is written verbatim (plain agent-trace data);
+    scores + provenance go to ``output/metadata/<stem>.json``.
+    """
+    primary = primary_score(run.rewards)
+    split = route_split(primary)
+    stem = bench_stem(source, task.id)
+
+    trace_path = cfg.output.traces_dir / split / f"{stem}.jsonl"
+    # A re-run (without --resume) whose score crosses a routing boundary would otherwise leave a
+    # stale copy in the old split; the dataset scanners read every split, so a task would appear
+    # twice with contradictory labels. Drop any sibling copy before writing the new one.
+    for other in BENCH_SPLITS:
+        if other != split:
+            (cfg.output.traces_dir / other / f"{stem}.jsonl").unlink(missing_ok=True)
+    trace_path.parent.mkdir(parents=True, exist_ok=True)
+    trace_path.write_text("\n".join(run.native_lines) + "\n", encoding="utf-8")
+
+    metadata: dict[str, Any] = {
+        "task": task.id,
+        "source": source_id(source),
+        "type": source.type,
+        "split": split,
+        "reward": primary,
+        "rewards": run.rewards or {},
+        "agent": cfg.get_agent_provider(),
+        "trace_file": f"{split}/{stem}.jsonl",
+        **run.metadata,
+    }
+    meta_path = cfg.output.traces_dir / "metadata" / f"{stem}.json"
+    meta_path.parent.mkdir(parents=True, exist_ok=True)
+    meta_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
+    return [trace_path], split