jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-31 22:41:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

bench: leaderboard submission harness for terminal-bench 2.1

Browse Source 
The 89x5 leaderboard cell needs a job config whose agent identity, effort
label and dataset ref all agree with what actually ran, plus an ATIF
trajectory for every trial. Adds the manifest, the job-config deriver and
the trajectory converter, and threads them through the runner.

Signed-off-by: Atish Patel <atish@squareup.com>
Co-authored-by: Claude Code <noreply@anthropic.com>
Signed-off-by: Atish Patel <atish@squareup.com>
This commit is contained in:
Atish Patel
2026-08-05 11:03:30 -05:00
parent 8964ba9cb6
commit e25cbad425
8 changed files with 2839 additions and 1630 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
benchmarks/harbor-buzz-orchestra/manifests/lb-tb-solo-deepseek-v4-flash-max.yaml Normal file
View File

@@ -0,0 +1,84 @@
# LEADERBOARD SUBMISSION CELL -- solo deepseek/deepseek-v4-flash-0731 at
# reasoning effort `max`, via OpenRouter, for the Terminal-Bench 2.1 board.
#
# This is NOT a study cell and must not be pooled with one. Every study
# manifest raises the container's resources and the wall clock; this one
# cannot, because the leaderboard rejects both. The `lb-` prefix and the
# distinct `condition` keep the two from ever hashing together.
#
# WHAT MAKES A SUBMISSION VALID (leaderboard/src/leaderboard/ci/static_analysis.py):
# * timeout_multiplier unset or 1.0 -- run with `--timeout-multiplier 1.0`
# * no agent/verifier timeout overrides
# * no environment resource overrides -- hence NO `environment:` block below
# * all 89 tasks, >= 5 trials each -- `-k 5`
# * every rewarded trial carries an ATIF trajectory
# The dataset ref our runs already resolve
# (sha256:7d7bdc1cbedad549fc1140404bd4dc45e5fd0ea7c4186773687d177ad3a0699a)
# is exactly the DATASET_REF the board pins, so the tasks need nothing done to
# them.
#
# THE MISSING `environment:` BLOCK IS THE WHOLE DIFFERENCE, AND IT IS EXPENSIVE.
# Study cells give every task 4 CPUs and 8 GB. Here Harbor honours each task's
# own declaration, and across the 89 task.toml files that is 1 vCPU on 83 of
# them and 2048 MB on 68 -- shared with the task's own work, because the Buzz
# stack (buzz-acp + buzz-agent + buzz-dev-mcp + the relay forwarder) runs
# *inside* the task container. The agent clock shrinks the same way: at 1.0x,
# 48 tasks get 900s where the study gave them 2700s. The `high` cell already
# timed out on 16 of 89 at 3x. Expect this row to score materially below the
# 0.618 in tools/table.py, and read a drop as the settings rather than as the
# model.
#
# THE UPSTREAM IS PINNED, AND THE CHOICE IS NOT THE OBVIOUS ONE. OpenRouter
# serves this model id from nine upstreams spanning fp4 to fp8 quantization,
# 262K to 1M context, and $0.09-$0.14 per Mtok. Cheapest is deepinfra/fp4 --
# and it caches well -- but fp4 is a materially more degraded set of weights
# than the fp8 every other endpoint serves, so benchmarking it and labelling
# the row "deepseek-v4-flash-0731" would overstate what ran. This pins
# `gmicloud/fp8`: fp8, the full 1M context, 98.9% uptime, and the cheapest of
# the fp8 endpoints at $0.133/$0.266. A leaderboard row names a model, so the
# pin is what makes the name true.
#
# EFFORT IS `max`, WHICH IS NOT THIS MODEL'S DEFAULT. deepseek-v4-flash-0731
# advertises supported_efforts ["max","high","low"] and defaults to "high", so
# the study's `high` cell measured the provider default and this one does not.
# The effort also has to reach the leaderboard as a *label*: `lb filter` reads
# it from the job config's `agents[0].kwargs.reasoning_effort`, and a job that
# omits it publishes a row saying "none". scripts/lb_job_config.py derives that
# label from the field below rather than taking one of its own, so the label and
# the run cannot disagree; `--expect-effort max` asserts the intent against this
# file for an operator who wants it checked rather than assumed.
schema_version: "1"
condition: lb-tb-solo-deepseek-v4-flash-max
roster:
- id: solo
kind: orchestrator
role: solo
count: 1
endpoint: deepseek/deepseek-v4-flash-0731
model_revision: deepseek/deepseek-v4-flash-20260731
prompt:
# Byte-identical to every other solo cell in the study.
path: personas/bench/solo.md
sha256: 61c9ac9dc47cbd3c987a7ee6ceb7397496834df8d3fc284c8fe6980e1d5b4ebe
generation:
thinking_effort: max
prices:
deepseek/deepseek-v4-flash-0731:
# The gmicloud/fp8 endpoint's own rates -- NOT the model id's headline
# $0.14/$0.28, which belongs to a different set of upstreams. Pinning is
# what makes this row the correct one.
input_per_million_usd: 0.133
cached_input_per_million_usd: 0.0266
output_per_million_usd: 0.266
# Fallback only; the pinned route reports the real split and the harness
# prefers it. See tb-solo-kimi-k3-high.yaml.
cache_read_rate: 0.0
trial_budget:
# The harness's own ceiling, which must clear Harbor's longest task deadline
# or check_budget_clears_clock refuses the run. At 1.0x that is the largest
# `[agent] timeout_sec` in the dataset, 12000s, unscaled. Harbor's per-task
# deadline is always the shorter of the two here (900s on 48 tasks), so it is
# Harbor that ends every trial -- which is exactly what the board requires.
timeout_seconds: 12000
# No `environment:` block. See the header: adding one disqualifies the run.

131
benchmarks/harbor-buzz-orchestra/scripts/benchmark.py
View File

@@ -280,6 +280,33 @@ def container_scoring_probe(timeout: float = 180.0) -> tuple[str, str] | None:
return lines[0], lines[1]
# Escape hatch for the container-egress guard below. Deliberately narrow: it
# only applies to a run that is already too small to be a measurement, so it
# cannot be used to push a corrupted sweep through. Exists because the guard is
# also the thing standing between you and testing the harness at all when the
# proxy is having a bad day, and "score 0 everywhere" is a perfectly good result
# when what you are checking is whether trajectory.json gets written.
UNSCOREABLE_BYPASS = "BUZZ_BENCHMARK_ALLOW_UNSCOREABLE"
# A bypassed run may name at most this many tasks. Four is enough to cover the
# resource shapes worth smoke-testing and far too few to look like a sweep.
UNSCOREABLE_BYPASS_MAX_TASKS = 4
def unscoreable_bypass_allowed(args: argparse.Namespace) -> bool:
"""Whether this run may proceed with verifiers that cannot install anything.
Three conditions, all required: the variable is set, the run names a short
explicit task list, and it runs one attempt each. Any of them missing means
the run is or could become a real measurement, and a measurement taken
through a broken verifier path is worse than no measurement -- it looks like
a model that failed.
"""
if os.environ.get(UNSCOREABLE_BYPASS) != "1":
return False
tasks = list(args.include_task)
return 0 < len(tasks) <= UNSCOREABLE_BYPASS_MAX_TASKS and args.attempts == 1
def check_container_can_be_scored() -> None:
"""Refuse a sweep whose verifiers will not be able to install pytest.
@@ -416,12 +443,12 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
)
parser.add_argument("--n-concurrent", "-n", type=int, default=4, help="Concurrent trials")
parser.add_argument(
"--timeout-multiplier", type=float, default=DEFAULT_TIMEOUT_MULTIPLIER,
"--timeout-multiplier", type=float, default=None,
metavar="N",
help="Scale every Harbor phase deadline (agent, verifier, setup, build) "
f"by N (default: {DEFAULT_TIMEOUT_MULTIPLIER}). Pass 1.0 for a run "
"that can be submitted to the leaderboard; anything else makes the "
"job local-only.",
f"by N (default: {DEFAULT_TIMEOUT_MULTIPLIER}, or 1.0 with "
"--submission). Pass 1.0 for a run that can be submitted to the "
"leaderboard; anything else makes the job local-only.",
)
parser.add_argument(
"--jobs-dir", type=Path, default=PACKAGE_ROOT / "jobs", help="Job output root"
@@ -433,9 +460,29 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"Only trials whose verifier actually ran count as done. Use the "
"same manifest, or the two halves will not group together.",
)
parser.add_argument(
"--submission", action="store_true",
help="Produce a job submittable to the public Terminal-Bench "
"leaderboard. Forces --timeout-multiplier 1.0, drops the manifest's "
"environment resource overrides, and names the agent/model/effort "
"through a job config so the published row matches the run. "
"Requires a solo manifest that pins its reasoning effort.",
)
parser.add_argument(
"--upload", action="store_true", help="Upload to Harbor Hub when the job finishes"
)
parser.add_argument(
"--public", action="store_true",
help="Make the uploaded job public (requires --upload). Leaderboard CI "
"reads the job through the hub, so a submission has to be public.",
)
parser.add_argument(
"--share-org", action="append", default=[], metavar="ORG",
help="Share the uploaded job with a Harbor Hub organization "
"(requires --upload, repeatable). Harbor has no upload-as-org: an "
"upload belongs to the authenticated user, and sharing is what "
"makes it appear under hub.harborframework.com/organizations/<ORG>.",
)
parser.add_argument(
"--gui", action="store_true",
help="Open the Buzz desktop app as the benchmark user to watch the run live",
@@ -449,7 +496,21 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"--dry-run", action="store_true",
help="Print the underlying harbor command and exit (no stack bring-up)",
)
return parser.parse_args(argv)
args = parser.parse_args(argv)
# Resolved here rather than as an argparse default so an explicit value and
# an inherited one are distinguishable: --submission needs 1.0, and silently
# overriding a multiplier someone typed would be worse than refusing it.
if args.timeout_multiplier is None:
args.timeout_multiplier = 1.0 if args.submission else DEFAULT_TIMEOUT_MULTIPLIER
elif args.submission and args.timeout_multiplier != 1.0:
parser.error(
f"--submission requires --timeout-multiplier 1.0 (got "
f"{args.timeout_multiplier}). Terminal-Bench rejects a job whose "
"phase deadlines were scaled, so the two cannot both hold."
)
if (args.public or args.share_org) and not args.upload:
parser.error("--public / --share-org requires --upload")
return args
# -- state: secrets and identities, generated once --------------------------
@@ -1087,6 +1148,24 @@ def qualify_task_pattern(pattern: str, dataset: str) -> str:
return f"*/{pattern}"
def check_submission_manifest(manifest: Path) -> None:
"""Refuse a manifest whose own settings would disqualify the submission.
Only the resource block, because that is the one thing the manifest can
assert unilaterally. Everything else a submission depends on -- one model, a
pinned reasoning effort -- is checked by ``lb_job_config``, which builds the
config that publishes them.
"""
declared = environment_override_argv(manifest)
if declared:
raise DatasetError(
f"--submission: {manifest} declares an `environment:` block "
f"({' '.join(declared)}), and Terminal-Bench rejects a job carrying "
"resource overrides. Use a manifest without one — see "
"manifests/lb-*.yaml, whose header explains what that costs."
)
def leaderboard_argv(
args: argparse.Namespace, provisioner_config: Path, agent_bin_dir: Path
) -> list[str]:
@@ -1117,7 +1196,13 @@ def leaderboard_argv(
"--n-concurrent", str(args.n_concurrent),
"--jobs-dir", str(args.jobs_dir),
]
argv += environment_override_argv(args.manifest)
if args.submission:
# The manifest's resource block is deliberately not forwarded here;
# check_submission_manifest() has already refused a manifest that
# declares one, so there is nothing to drop silently.
argv.append("--submission")
else:
argv += environment_override_argv(args.manifest)
if args.timeout_multiplier != 1.0:
# Forwarded only when it changes something: Harbor's validator rejects
# the flag being *set*, not just being non-unity, so a submittable run
@@ -1127,6 +1212,10 @@ def leaderboard_argv(
argv += ["--job-name", args.job_name]
if args.upload:
argv.append("--upload")
if args.public:
argv.append("--public")
for org in args.share_org:
argv += ["--share-org", org]
if args.dry_run:
argv.append("--dry-run")
return argv
@@ -1134,6 +1223,12 @@ def leaderboard_argv(
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
if args.submission:
try:
check_submission_manifest(args.manifest)
except DatasetError as error:
print(f"error: {error}", file=sys.stderr)
return 2
# Before anything expensive: a local --path needs no registry, a registry
# dataset must resolve.
if not args.path:
@@ -1184,8 +1279,28 @@ def main(argv: list[str] | None = None) -> int:
try:
check_container_can_be_scored()
except ContainerEgressError as error:
print(f"error: {error}", file=sys.stderr)
return 2
if not unscoreable_bypass_allowed(args):
print(f"error: {error}", file=sys.stderr)
if os.environ.get(UNSCOREABLE_BYPASS) == "1":
print(
f" {UNSCOREABLE_BYPASS}=1 is set but this run is not a "
"filtered smoke test, so it does not apply. See the "
"variable's definition for why.",
file=sys.stderr,
)
return 2
print(
"\n"
" !! CONTAINER EGRESS IS BROKEN AND THIS RUN IS PROCEEDING ANYWAY.\n"
f" !! {error}\n"
" !! Every reward from this job is meaningless: the verifiers "
"cannot install their own\n"
" !! test dependencies, so tasks the agent solved will still "
"record 0. Use this job to\n"
" !! check plumbing (artifacts, config, trajectories) and "
"nothing else.\n",
file=sys.stderr,
)
state = load_state()
print_user_identity(state, show_secret=args.gui)
write_env_file(state)

279
benchmarks/harbor-buzz-orchestra/scripts/lb_job_config.py Normal file
View File

@@ -0,0 +1,279 @@
#!/usr/bin/env python3
"""Emit the ``harbor run -c`` job config that gives a leaderboard row its name.
WHY A CONFIG FILE INSTEAD OF ``--agent``. ``harbor run --agent
harbor_buzz_orchestra:BuzzOrchestraAgent`` records the *import path* as the
agent config's ``name``, while the finished trial records
``BuzzOrchestraAgent.name()`` -- ``buzz-orchestra``. ``lb filter`` joins those
two strings to find the run's reasoning effort::
efforts = {(a["name"], a["model_name"]): a["kwargs"]["reasoning_effort"]
for a in config["agents"]}
...
efforts.get((trial["agent_name"], trial_model(trial)))
(leaderboard/src/leaderboard/flows/filter.py). When they disagree the lookup
misses and the submission publishes ``reasoning_effort: "none"`` -- a wrong
label on a public row, not an error anyone catches. The CLI cannot make them
agree, because ``harbor/cli/jobs.py`` resolves
``resolved_import_path = agent_import_path if agent_name is None else None``:
you get one or the other. A job config can carry both, and
``AgentFactory.create_agent_from_config`` falls through to the import path for
any ``name`` that is not a built-in ``AgentName`` -- which ``buzz-orchestra``
is not, and cannot become, since ``AgentName`` is a closed enum with no
third-party registration.
The model name has the same shape of problem: ``-m/--model`` is read *only*
inside the ``--agent`` branch, so a config-supplied agent that omits
``model_name`` reports ``model_info: null`` and the row names no model at all.
Everything below is derived from the experiment manifest -- the same file that
decides which endpoint and which effort actually run -- so the published label
and the run cannot disagree by construction. There is no flag to set the model
or the effort by hand; ``--expect-effort`` only asserts what the manifest
already says, for an operator who wants the intent checked rather than assumed.
The emitted config also carries the adapter's runtime kwargs (binaries, relay
gateway, provisioner and endpoint configs). Those could be left to
``--agent-kwarg``, which merges into a config-supplied agent, but keeping them
here makes the config a complete record of the run: one file to read, and one
file the checks below can cover.
Standalone use, to inspect what a run would submit under::
uv run --project benchmarks/harbor-buzz-orchestra/testbed \
benchmarks/harbor-buzz-orchestra/scripts/lb_job_config.py \
--manifest benchmarks/harbor-buzz-orchestra/manifests/<CELL>.yaml \
--expect-effort max
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any
PACKAGE_ROOT = Path(__file__).resolve().parent.parent
AGENT_IMPORT = "harbor_buzz_orchestra:BuzzOrchestraAgent"
# BuzzOrchestraAgent.name(). Hard-coded rather than imported so this script
# runs without the adapter installed; check_agent_name() below proves the two
# still agree whenever it *is* importable.
AGENT_NAME = "buzz-orchestra"
# Job-config fields Terminal-Bench rejects outright
# (leaderboard/src/leaderboard/ci/static_analysis.py). Checked here so a
# disqualifying setting fails while it is still a file on disk, rather than
# after 445 trials have been paid for.
FORBIDDEN_JOB_FIELDS = (
"agent_timeout_multiplier",
"verifier_timeout_multiplier",
"agent_setup_timeout_multiplier",
"environment_build_timeout_multiplier",
)
FORBIDDEN_AGENT_FIELDS = (
"override_timeout_sec",
"override_setup_timeout_sec",
"max_timeout_sec",
)
FORBIDDEN_VERIFIER_FIELDS = ("override_timeout_sec", "max_timeout_sec")
FORBIDDEN_ENV_FIELDS = (
"override_cpus",
"override_gpus",
"override_memory_mb",
"override_storage_mb",
)
class ConfigError(ValueError):
"""Raised when a manifest cannot produce a valid submission config."""
def _roster_entry(manifest: Any) -> Any:
"""The single seat whose model the leaderboard row will name.
A team manifest is refused rather than guessed at: the row carries one
``model_name``, and picking the lead's would describe a run that two other
models also did the work for.
"""
roster = list(manifest.roster)
seats = sum(entry.count for entry in roster)
if len(roster) != 1 or seats != 1:
raise ConfigError(
f"{manifest.condition}: a leaderboard row names one model, but this "
f"manifest rosters {seats} seat(s) across {len(roster)} entr(ies). "
"Submit a solo cell, or add a deliberate model-naming rule first."
)
return roster[0]
def model_name(manifest: Any) -> str:
"""The model id the leaderboard row displays.
The manifest's ``endpoint`` -- the provider-facing id, e.g.
``deepseek/deepseek-v4-flash-0731`` -- rather than ``model_revision``, which
is this study's internal dated alias and would name nothing a reader could
look up. Harbor splits it on the first ``/`` into provider + name, and the
leaderboard rejoins them, so the round trip is exact.
"""
return _roster_entry(manifest).endpoint
def reasoning_effort(manifest: Any) -> str:
"""The effort the row displays, taken from the manifest that sets it.
Refuses an unset effort. Unset is legal for a study cell -- it means the
harness constant applies -- but a public row must not label a run with an
effort no file records.
"""
entry = _roster_entry(manifest)
effort = entry.generation.thinking_effort if entry.generation else None
if effort is None:
raise ConfigError(
f"{manifest.condition}: roster seat '{entry.id}' does not pin "
"generation.thinking_effort. A submission publishes the effort as a "
"label, so it has to be stated rather than inherited."
)
return effort
def check_agent_name() -> None:
"""Prove AGENT_NAME still matches what a trial will actually report."""
try:
from harbor_buzz_orchestra.agent import BuzzOrchestraAgent
except ImportError:
return # Inspecting the config outside the testbed; nothing to check.
actual = BuzzOrchestraAgent.name()
if actual != AGENT_NAME:
raise ConfigError(
f"agent name drift: this script writes '{AGENT_NAME}' into the job "
f"config but BuzzOrchestraAgent.name() returns '{actual}'. The "
"leaderboard joins those two, so the row would lose its effort label."
)
def build(
manifest: Any,
*,
kwargs: dict[str, Any] | None = None,
import_path: str = AGENT_IMPORT,
) -> dict[str, Any]:
"""The job config: one agent, named twice so both readers resolve it.
Only ``agents`` is set. Everything else about the run -- dataset, attempts,
concurrency, job name -- stays on the ``harbor run`` command line, where
``harbor/cli/jobs.py`` overwrites the corresponding config field. Nothing a
submission depends on is left to a default this file could shadow.
"""
check_agent_name()
agent: dict[str, Any] = {
# Both, deliberately: `name` is what the leaderboard joins on and what
# the trial reports; `import_path` is what actually gets constructed.
"name": AGENT_NAME,
"import_path": import_path,
"model_name": model_name(manifest),
"kwargs": {
**{key: str(value) for key, value in (kwargs or {}).items()},
# Read by `lb filter` off the job config, never by the agent -- it
# lands in **kwargs on BaseAgent.__init__ and is dropped. The effort
# that runs comes from the manifest; this is the label for it, which
# is why both are derived from the same field.
"reasoning_effort": reasoning_effort(manifest),
},
}
return {"agents": [agent]}
def check_submittable(config: dict[str, Any]) -> list[str]:
"""Return every reason Terminal-Bench would reject a job run from this
config. Empty means the config carries none of the disqualifying knobs --
it does not mean the *run* stays clean, since the command line can still
add ``--timeout-multiplier`` or an override flag.
"""
problems: list[str] = []
multiplier = config.get("timeout_multiplier")
if multiplier is not None and float(multiplier) != 1.0:
problems.append(f"timeout_multiplier is {multiplier} (must be 1.0 or unset)")
for field in FORBIDDEN_JOB_FIELDS:
if config.get(field) is not None:
problems.append(f"{field} is set ({config[field]})")
for field in FORBIDDEN_ENV_FIELDS:
value = (config.get("environment") or {}).get(field)
if value is not None:
problems.append(f"environment.{field} is set ({value})")
for field in FORBIDDEN_VERIFIER_FIELDS:
value = (config.get("verifier") or {}).get(field)
if value is not None:
problems.append(f"verifier.{field} is set ({value})")
for index, agent in enumerate(config.get("agents") or []):
for field in FORBIDDEN_AGENT_FIELDS:
if agent.get(field) is not None:
problems.append(f"agents[{index}].{field} is set ({agent[field]})")
if not agent.get("model_name"):
problems.append(f"agents[{index}].model_name is unset (row names no model)")
if not (agent.get("kwargs") or {}).get("reasoning_effort"):
problems.append(
f"agents[{index}].kwargs.reasoning_effort is unset "
"(row publishes effort 'none')"
)
return problems
def write(config: dict[str, Any], path: Path) -> Path:
"""Write the config, refusing to leave a disqualifying one on disk."""
problems = check_submittable(config)
if problems:
raise ConfigError(
"refusing to write a job config the leaderboard would reject:\n - "
+ "\n - ".join(problems)
)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8")
return path
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("--manifest", type=Path, required=True, help="Team manifest YAML")
parser.add_argument(
"--out", type=Path, default=None, help="Write here instead of stdout"
)
parser.add_argument(
"--expect-effort",
default=None,
metavar="EFFORT",
help="Fail unless the manifest pins this reasoning effort. Asserts the "
"operator's intent against the file that decides it.",
)
args = parser.parse_args(argv)
sys.path.insert(0, str(PACKAGE_ROOT / "src"))
from harbor_buzz_orchestra.manifest import ExperimentManifest
try:
manifest = ExperimentManifest.load(args.manifest)
effort = reasoning_effort(manifest)
if args.expect_effort is not None and effort != args.expect_effort:
raise ConfigError(
f"{args.manifest} pins thinking_effort '{effort}', not "
f"'{args.expect_effort}'. The manifest decides what runs; edit it "
"rather than the label, or the row will describe a different run."
)
config = build(manifest)
except ConfigError as error:
print(f"error: {error}", file=sys.stderr)
return 2
if args.out:
print(write(config, args.out))
else:
problems = check_submittable(config)
print(json.dumps(config, indent=2))
for problem in problems:
print(f"warning: {problem}", file=sys.stderr)
return 0
if __name__ == "__main__":
sys.exit(main())

224
benchmarks/harbor-buzz-orchestra/scripts/run_leaderboard.py
View File

@@ -17,10 +17,17 @@ Per-phase timeout knobs and ``--override-gpus`` are still never accepted:
nothing in Terminal-Bench 2.1 asks for a GPU, so that one could only add
capability rather than remove an artificial constraint.
Harbor rejects a submission carrying any of these
(``_check_no_job_overrides`` in ``harbor/leaderboard/static_validation.py``), so
a job produced with them is a local measurement, not a leaderboard entry, and
this script says so when it runs.
Terminal-Bench rejects a submission carrying any of these
(``leaderboard/src/leaderboard/ci/static_analysis.py``), so a job produced with
them is a local measurement, not a leaderboard entry, and this script says so
when it runs.
``--submission`` goes one step further: it refuses those opt-outs outright, and
routes the agent through a job config so the run publishes the model and effort
it actually used. See ``lb_job_config.py`` for why the command line alone cannot
express that. Without ``--submission`` the run still uses ``--agent``, which is
what every study cell has always done and what their recorded configs show.
After the run it writes a ``metadata.yaml`` template derived from the manifest
and prints the exact upload/submit commands.
@@ -49,6 +56,14 @@ from pathlib import Path
import yaml
# benchmark.py loads this file by explicit spec rather than by import, so the
# scripts directory is not necessarily on the path when a sibling is needed.
_SCRIPTS_DIR = Path(__file__).resolve().parent
if str(_SCRIPTS_DIR) not in sys.path:
sys.path.insert(0, str(_SCRIPTS_DIR))
import lb_job_config # noqa: E402 -- needs the path entry above
PACKAGE_ROOT = Path(__file__).resolve().parent.parent
AGENT_IMPORT = "harbor_buzz_orchestra:BuzzOrchestraAgent"
PROVISIONER_FACTORY = "harbor_buzz_testbed:provisioner_from_dict"
@@ -64,6 +79,10 @@ FORWARDER_BINARY = "relay-forwarder"
PROVIDER_ORGS = {"anthropic": "Anthropic", "openai": "OpenAI", "databricks": "Databricks"}
# leaderboard/src/leaderboard/ci/static_analysis.py: MIN_TRIALS_PER_TASK.
LEADERBOARD_MIN_ATTEMPTS = 5
LEADERBOARD_REPO = "harbor-framework/terminal-bench-2-1"
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
@@ -142,9 +161,27 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
)
parser.add_argument("--jobs-dir", type=Path, default=Path("jobs"), help="Job output root")
parser.add_argument("--job-name", default=None, help="Job name (default: lb-<condition>-<UTC>)")
parser.add_argument(
"--submission", action="store_true",
help="Produce a job that can be submitted to the public Terminal-Bench "
"leaderboard: refuse every override flag, and name the agent, model "
"and reasoning effort through a job config so the published row "
"matches the run. Requires a solo manifest that pins its effort.",
)
parser.add_argument(
"--upload", action="store_true", help="Upload to Harbor Hub when the job finishes"
)
parser.add_argument(
"--public", action="store_true",
help="Make the uploaded job public (requires --upload). A leaderboard "
"submission has to be public — CI reads the job through the hub.",
)
parser.add_argument(
"--share-org", action="append", default=[], metavar="ORG",
help="Share the uploaded job with a Harbor Hub organization "
"(requires --upload, repeatable). An upload is owned by the "
"authenticated user; sharing is how it also appears under an org.",
)
parser.add_argument("--dry-run", action="store_true", help="Print the harbor command and exit")
return parser.parse_args(argv)
@@ -180,10 +217,63 @@ def find_agent_binaries(bin_dir: Path, with_forwarder: bool = False) -> dict[str
return found
def agent_kwargs(
args: argparse.Namespace,
binaries: dict[str, Path],
agent_binaries: dict[str, Path],
) -> dict[str, object]:
"""Everything BuzzOrchestraAgent needs to construct itself.
Shared by both paths: ``--agent-kwarg`` on a study run, the job config's
``kwargs`` on a submission run. One definition, so the two cannot drift into
running different things.
"""
kwargs: dict[str, object] = {
"manifest": args.manifest,
"provisioner_factory": PROVISIONER_FACTORY,
"provisioner_config": args.provisioner_config,
"artifact_root": PACKAGE_ROOT,
"endpoint_config": args.endpoint_config,
"buzz_acp_binary": agent_binaries["buzz-acp"],
"buzz_agent_binary": agent_binaries["buzz-agent"],
"buzz_dev_mcp_binary": agent_binaries["buzz-dev-mcp"],
"buzz_cli_binary": binaries["buzz"],
"run_id": args.job_name,
}
if args.relay_gateway:
kwargs["relay_gateway"] = args.relay_gateway
kwargs["forwarder_binary"] = agent_binaries[FORWARDER_BINARY]
return kwargs
def write_job_config(
args: argparse.Namespace,
binaries: dict[str, Path],
agent_binaries: dict[str, Path],
) -> Path:
"""Emit the submission job config beside the job directory.
Raises ``lb_job_config.ConfigError`` when the manifest cannot name a single
model at a stated effort, or when the config would carry a disqualifying
field. Both are cheaper to hit here than after the run.
"""
from harbor_buzz_orchestra.manifest import ExperimentManifest
config = lb_job_config.build(
ExperimentManifest.load(args.manifest),
kwargs=agent_kwargs(args, binaries, agent_binaries),
import_path=AGENT_IMPORT,
)
return lb_job_config.write(
config, args.jobs_dir / f"{args.job_name}.jobconfig.json"
)
def build_command(
args: argparse.Namespace,
binaries: dict[str, Path],
agent_binaries: dict[str, Path],
job_config: Path | None = None,
) -> list[str]:
"""Compose the harbor invocation.
@@ -195,6 +285,12 @@ def build_command(
``--override-gpus`` is still never accepted. Nothing in Terminal-Bench 2.1
requests a GPU, so a flag for it could only ever change what the machine can
do rather than remove an artificial constraint.
With ``job_config`` the agent comes from that file instead of ``--agent`` +
``--agent-kwarg``. The two are mutually exclusive by Harbor's own rule:
passing ``--agent`` clears ``config.agents`` outright
(``harbor/cli/jobs.py``), which would discard the identity the config exists
to carry.
"""
command = [
"harbor", "run", "--yes",
@@ -203,6 +299,8 @@ def build_command(
"-k", str(args.attempts),
"--n-concurrent", str(args.n_concurrent),
]
if job_config is not None:
command += ["-c", str(job_config)]
if args.timeout_multiplier is not None:
command += ["--timeout-multiplier", str(args.timeout_multiplier)]
for flag, value in (
@@ -222,24 +320,14 @@ def build_command(
command += ["--exclude-task-name", pattern]
if args.upload:
command.append("--upload")
command += ["--agent", AGENT_IMPORT]
kwargs = {
"manifest": args.manifest,
"provisioner_factory": PROVISIONER_FACTORY,
"provisioner_config": args.provisioner_config,
"artifact_root": PACKAGE_ROOT,
"endpoint_config": args.endpoint_config,
"buzz_acp_binary": agent_binaries["buzz-acp"],
"buzz_agent_binary": agent_binaries["buzz-agent"],
"buzz_dev_mcp_binary": agent_binaries["buzz-dev-mcp"],
"buzz_cli_binary": binaries["buzz"],
"run_id": args.job_name,
}
if args.relay_gateway:
kwargs["relay_gateway"] = args.relay_gateway
kwargs["forwarder_binary"] = agent_binaries[FORWARDER_BINARY]
for key, value in kwargs.items():
command += ["--agent-kwarg", f"{key}={value}"]
if args.public:
command.append("--public")
for org in args.share_org:
command += ["--share-org", org]
if job_config is None:
command += ["--agent", AGENT_IMPORT]
for key, value in agent_kwargs(args, binaries, agent_binaries).items():
command += ["--agent-kwarg", f"{key}={value}"]
return command
@@ -284,6 +372,36 @@ def main(argv: list[str] | None = None) -> int:
):
if not path.is_file():
raise SystemExit(f"{label} not found: {path}")
if (args.public or args.share_org) and not args.upload:
raise SystemExit("--public / --share-org requires --upload")
if args.submission:
# Refused rather than warned about: --submission states an intent that
# any of these silently defeats, and the cost of finding out is a
# finished 445-trial run that CI rejects.
opted_out = [
f"--{name.replace('_', '-')} {getattr(args, name)}"
for name in (
"timeout_multiplier",
"override_cpus",
"override_memory_mb",
"override_storage_mb",
)
if getattr(args, name) is not None
]
if opted_out:
raise SystemExit(
"--submission cannot be combined with "
f"{', '.join(opted_out)}: Terminal-Bench rejects a job carrying "
"any of them. Drop the flag for a submittable run, or drop "
"--submission and report the score with the settings stated."
)
if args.attempts < LEADERBOARD_MIN_ATTEMPTS:
print(
f"note: --attempts {args.attempts} is below the leaderboard's "
f"minimum of {LEADERBOARD_MIN_ATTEMPTS} per task. The job will be "
"shaped correctly but too small to submit — fine for a smoke "
"test, not for the real run."
)
if args.job_name is None:
condition = yaml.safe_load(args.manifest.read_text()).get("condition", "team")
stamp = dt.datetime.now(dt.UTC).strftime("%Y%m%dT%H%M%SZ")
@@ -297,13 +415,29 @@ def main(argv: list[str] | None = None) -> int:
name: args.agent_bin_dir / name
for name in AGENT_BINARIES + (FORWARDER_BINARY,)
}
print(" ".join(build_command(args, binaries, agent_binaries)))
job_config = None
if args.submission:
# Written even on a dry run: it is the artifact worth reviewing
# before committing the machine time, and writing it is the check.
try:
job_config = write_job_config(args, binaries, agent_binaries)
except lb_job_config.ConfigError as error:
raise SystemExit(f"error: {error}") from error
print(f"# job config: {job_config}")
print(" ".join(build_command(args, binaries, agent_binaries, job_config)))
return 0
binaries = find_binaries(args.buzz_bin_dir)
agent_binaries = find_agent_binaries(
args.agent_bin_dir, with_forwarder=bool(args.relay_gateway)
)
command = build_command(args, binaries, agent_binaries)
job_config = None
if args.submission:
try:
job_config = write_job_config(args, binaries, agent_binaries)
except lb_job_config.ConfigError as error:
raise SystemExit(f"error: {error}") from error
print(f"submission job config: {job_config}")
command = build_command(args, binaries, agent_binaries, job_config)
if shutil.which("harbor") is None:
raise SystemExit(
"harbor not on PATH — run via: uv run --project "
@@ -316,7 +450,10 @@ def main(argv: list[str] | None = None) -> int:
print(f"harbor run failed (exit {result.returncode}); job dir: {job_dir}")
return result.returncode
metadata_path = write_metadata_template(args, job_dir)
# The submission flow builds its own metadata (`lb metadata` fills the
# display fields from display-names.json), so the template is only useful to
# a study run keeping a record of what it ran.
metadata_path = None if args.submission else write_metadata_template(args, job_dir)
overrides = [
f"{name}={value}"
for name, value in (
@@ -341,13 +478,40 @@ def main(argv: list[str] | None = None) -> int:
"with those settings stated alongside it."
)
return 0
print("\nLeaderboard-ready job complete.")
print(f" 1. Review submitter details in {metadata_path}")
print(f" 2. harbor upload {job_dir}")
print(f"\nLeaderboard-shaped job complete: {job_dir}")
if not args.submission:
# A study run has the right *settings* but not the right agent identity:
# --agent records the import path where the leaderboard expects
# 'buzz-orchestra', and reports no model at all. See lb_job_config.py.
print(
f" Review submitter details in {metadata_path}.\n"
" Re-run with --submission to submit: this job names the agent by "
"import path and reports no model, so a submission built from it "
"would publish a row with no model and reasoning effort 'none'."
)
return 0
if args.attempts < LEADERBOARD_MIN_ATTEMPTS:
print(
f" Correctly shaped, but {args.attempts} attempt(s) per task is "
f"below the required {LEADERBOARD_MIN_ATTEMPTS}. Check "
"<trial>/agent/trajectory.json exists on every trial, then run the "
"full set."
)
return 0
print(
" 3. harbor leaderboard submit -l terminal-bench/terminal-bench-2-1 "
f"-j <job UUID from upload> -m {metadata_path}"
" Submit from a clone of the leaderboard repo "
f"({LEADERBOARD_REPO}; you need a fork, since submissions arrive as "
"PRs):\n"
" uv run lb submit <job link printed by --upload>\n"
" That runs `lb filter` -> `lb metadata` -> `lb open-prs`. Fill the "
"null entries `lb filter` scaffolds into display-names.json for "
"buzz-orchestra and the model, or it will prompt for them."
)
if not (args.upload and args.public):
print(
" This job was not uploaded public, and CI reads the job through "
f"the hub: `harbor upload --public {job_dir}` first."
)
return 0

8
benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/agent.py
View File

@@ -18,8 +18,12 @@ from .runtime import OrchestraRuntime
class BuzzOrchestraAgent(BaseAgent):
"""Coordinate an arbitrary manifest-defined team through a Buzz trial."""
# Set True only once the runtime writes a validated agent/trajectory.json.
SUPPORTS_ATIF = False
# The runtime writes agent/trajectory.json and validates it against Harbor's
# own ATIF model before writing, so a document the hub would reject fails in
# the trial rather than in leaderboard CI. See `trajectory.py` — notably that
# tool *results* are not recorded anywhere in the stack, so every step's
# `observation` is absent and the trajectory says so in its `notes`.
SUPPORTS_ATIF = True
def __init__(
self,

26
benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py
View File

@@ -12,6 +12,7 @@ from __future__ import annotations
import asyncio
import json
import logging
import os
import shlex
from dataclasses import dataclass, field
@@ -21,12 +22,14 @@ from typing import Any
import certifi
from harbor.environments.base import BaseEnvironment
from . import accounting, bundle
from . import accounting, bundle, trajectory
from .accounting import TURN_ENDED_MARKERS, USAGE_MARKER
from .manifest import AgentClass, ExperimentManifest, GenerationConfig
from .provisioning import AgentCredential, TrialHandle
from .runtime import RuntimeResult
logger = logging.getLogger(__name__)
# Tool/LLM rounds allowed per agent turn (BUZZ_AGENT_MAX_ROUNDS).
#
@@ -437,6 +440,27 @@ class BuzzContainerRuntime:
accounting=trial_accounting,
endpoints=self.endpoints,
)
# Last, because it reads what everything above just wrote. Goes
# beside `buzz/` rather than inside it: Harbor's uploader publishes
# exactly `<trial>/agent/trajectory.json` as the trial's
# trajectory_path, and the Terminal-Bench leaderboard rejects a
# submission whose rewarded trials lack one.
written = trajectory.write(
logs_dir=self.logs_dir,
trial_dir=trial_dir,
manifest=manifest,
instruction=instruction,
session_id=trial.trial_id,
accounting=trial_accounting,
)
if written is None:
# Not fatal to the trial, fatal to submitting it. Say so here
# rather than letting CI discover it 445 trials later.
logger.warning(
"no ATIF trajectory written for trial %s; this trial cannot "
"be part of a leaderboard submission",
trial.trial_id,
)
return RuntimeResult(
input_tokens=trial_accounting.input_tokens,

537
benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/trajectory.py Normal file
View File

@@ -0,0 +1,537 @@
"""Write the trial's ATIF trajectory, assembled from artifacts we already keep.
Harbor's uploader publishes ``<trial>/agent/trajectory.json`` as the trial's
``trajectory_path``, and the Terminal-Bench leaderboard requires one on every
*rewarded* trial: its ``/judge`` step audits successful trajectories for reward
hacking, and a pass it cannot read is a pass it cannot clear. Without this file
`Rewarded trials have trajectories` fails for every solve and the submission is
invalid, whatever the score says.
WHAT THIS READS, AND WHY IT IS A SCRAPE RATHER THAN A FEED. The Buzz stack keeps
no session file. Three artifacts between them hold the run, and this module is
the seam that joins them:
* ``<agent_id>.stdout.log`` -- ``buzz-acp``'s tracing output. Every tool call
the provider asked for is logged at target ``acp::tool`` as a
``tool_call:``/``tool_call_detail:`` pair, the second carrying the tool's
``rawInput`` JSON: the shell command, or the path and contents of a file
edit. This is the substance of a reward-hacking audit.
* ``transcript.json`` -- the channel conversation, which is what each agent
actually *said*, including the harness's opening instruction.
* ``receipts.jsonl``, via the ``TrialAccounting`` we already computed -- the
token and dollar totals.
TWO LIMITS, BOTH STATED IN THE DOCUMENT'S OWN ``notes`` RATHER THAN LEFT FOR A
READER TO INFER:
1. Tool outputs are not recorded anywhere, so ATIF ``observation`` is always
absent. The judge sees what each command asked for but not what came back.
2. ``tool_call_detail`` truncates ``rawInput`` at 400 bytes
(``MAX_RAW_INPUT`` in ``buzz-acp``'s ``acp.rs``), which is a readability
cap on a human-facing log. A long shell command is exactly what it cuts, so
those calls arrive here as unparseable JSON and are kept as
``{"_raw": "<first 400 bytes>"}``. Measured across a full 89-trial
Terminal-Bench job: 583 of 1550 calls, 38%. ``notes`` carries the per-trial
count so nobody mistakes a prefix for a command.
Both close the same way -- emit a structured session log from ``buzz-acp``
(``acp.rs``, where the ``acp::tool`` lines are already produced) instead of
parsing its human-readable one. Worth doing; not required for a valid
submission, which gates on a trajectory being present, not on its depth.
BEST EFFORT, BUT LOUD. A trial whose trajectory cannot be written is a trial the
leaderboard will reject, so ``write`` returns the path it wrote and ``None`` when
it could not, and never raises: losing the trajectory must not turn a completed
trial into a failed one. The caller logs the ``None``.
"""
from __future__ import annotations
import json
import logging
import re
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING: # pragma: no cover -- import cycle at runtime, types only
from .accounting import TrialAccounting
from .manifest import ExperimentManifest
logger = logging.getLogger(__name__)
SCHEMA_VERSION = "ATIF-v1.7"
AGENT_NAME = "buzz-orchestra"
AGENT_VERSION = "0.1.0"
# `tracing` colourises its output even when redirected, so every line arrives
# wrapped in SGR escapes. Strip them before matching anything.
_ANSI = re.compile(r"\x1b\[[0-9;]*m")
# 2026-07-29T21:40:54.050009Z INFO acp::tool: tool_call: buzz-dev-mcp__shell (other)
_TOOL_LINE = re.compile(
r"^(?P<ts>\d{4}-\d{2}-\d{2}T[\d:.]+Z)\s+INFO\s+acp::tool:\s+"
r"(?P<event>tool_call|tool_call_detail|tool_call_update):\s*(?P<rest>.*)$"
)
# `buzz-dev-mcp__shell (other)` -- the trailing parenthesised word is ACP's tool
# *kind*, not part of the name.
_TOOL_TITLE = re.compile(r"^(?P<name>.*?)\s*\((?P<kind>[^()]*)\)\s*$")
# `call_WPn3YK0x8PaZtkYEC2LXzDBo -> completed` (the arrow is U+2192 in the log)
_TOOL_UPDATE = re.compile(r"^(?P<call_id>\S+)\s*(?:→|->)\s*(?P<status>\S+)\s*$")
@dataclass
class _ToolCall:
"""One tool invocation recovered from a seat's log."""
seat: str
epoch: float
timestamp: str
function_name: str
kind: str
arguments: dict[str, Any] = field(default_factory=dict)
# Present only when the log's own ordering let us pair the call with an id;
# see _pair_call_ids. Otherwise the writer synthesises one.
call_id: str | None = None
status: str | None = None
# Set when `rawInput` was absent or unparseable and we kept the raw line.
raw_detail: str | None = None
def _iso(epoch: float) -> str:
return datetime.fromtimestamp(epoch, UTC).isoformat().replace("+00:00", "Z")
def _epoch(iso: str) -> float | None:
try:
return datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp()
except ValueError:
return None
def parse_tool_calls(log_path: Path, seat: str) -> list[_ToolCall]:
"""Recover one seat's tool calls from its ``buzz-acp`` stdout log.
The three ``acp::tool`` events arrive as a stream, not as records: a
``tool_call`` names the tool, an optional ``tool_call_detail`` immediately
after it carries the arguments, and ``tool_call_update`` lines report status
transitions -- carrying the provider's call id, which the first two do not.
So a detail is attached to the most recent call, and ids are paired
positionally afterwards (see _pair_call_ids), which is why that pairing is
conditional rather than assumed.
"""
try:
text = log_path.read_text(encoding="utf-8", errors="replace")
except OSError:
return []
calls: list[_ToolCall] = []
# call_id -> last status seen, in first-appearance order.
statuses: dict[str, str] = {}
for line in text.splitlines():
match = _TOOL_LINE.match(_ANSI.sub("", line))
if match is None:
continue
event, rest = match["event"], match["rest"].strip()
epoch = _epoch(match["ts"])
if epoch is None:
continue
if event == "tool_call":
title = _TOOL_TITLE.match(rest)
name = (title["name"] if title else rest).strip() or "unknown"
calls.append(
_ToolCall(
seat=seat,
epoch=epoch,
timestamp=_iso(epoch),
function_name=name,
kind=(title["kind"].strip() if title else ""),
)
)
elif event == "tool_call_detail":
if not calls:
continue
arguments, raw = _parse_detail(rest)
calls[-1].arguments = arguments
calls[-1].raw_detail = raw
else: # tool_call_update
update = _TOOL_UPDATE.match(rest)
if update is not None:
statuses[update["call_id"]] = update["status"]
_pair_call_ids(calls, statuses)
return calls
def _parse_detail(rest: str) -> tuple[dict[str, Any], str | None]:
"""``rawInput={...}`` -> the parsed arguments; anything else -> a raw note.
``tool_call_detail`` also carries a ``locations=`` summary for tools that
declare the files they touch, and that form is not JSON. Keeping the line
verbatim under ``_raw`` beats dropping it: for a file edit it still names the
path, which is exactly what an auditor is looking for.
"""
if rest.startswith("rawInput="):
payload = rest[len("rawInput=") :]
try:
parsed = json.loads(payload)
except ValueError:
return {"_raw": payload}, payload
if isinstance(parsed, dict):
return parsed, None
# A non-object rawInput is legal JSON but not an ATIF arguments object.
return {"_value": parsed}, None
return {"_raw": rest}, rest
def _pair_call_ids(calls: list[_ToolCall], statuses: dict[str, str]) -> None:
"""Attach the provider's real call ids when the log's ordering proves them.
``buzz-acp`` emits its ``tool_call`` lines in the order the provider
returned the calls and its first ``tool_call_update`` per id in that same
order, so position identifies them -- but only if nothing was dropped. A log
truncated mid-turn, or a call that never reached ``in_progress``, breaks the
correspondence, and a *wrong* id is worse than no id: it would attribute a
command to a different call. So the pairing applies only on an exact count
match, and the writer synthesises ids otherwise.
"""
if len(statuses) != len(calls):
return
for call, (call_id, status) in zip(calls, statuses.items(), strict=True):
call.call_id = call_id
call.status = status
def _load_transcript(path: Path) -> tuple[list[dict[str, Any]], bool]:
"""``(messages, truncated)`` from the saved channel transcript.
``truncated`` is the relay's own cap flag: it means the earliest messages are
missing, which a reader of the trajectory has to be told rather than left to
deduce from a conversation that starts mid-sentence.
"""
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
return [], False
if not isinstance(payload, dict):
return [], False
messages = [m for m in payload.get("messages") or [] if isinstance(m, dict)]
return messages, bool(payload.get("truncated"))
def _seat_models(manifest: ExperimentManifest) -> dict[str, tuple[str, str | None]]:
"""``agent_id -> (model_revision, thinking_effort)`` for every seat.
Seat ids are ``<class id>-<n>`` (``solo-1``, ``worker-2``), which is how they
appear as transcript authors and as log filenames, so the map is built per
instance rather than per class.
"""
out: dict[str, tuple[str, str | None]] = {}
for entry in manifest.roster:
for index in range(1, entry.count + 1):
out[f"{entry.id}-{index}"] = (
entry.model_revision,
entry.generation.thinking_effort,
)
return out
def build(
*,
trial_dir: Path,
manifest: ExperimentManifest,
instruction: str,
session_id: str,
accounting: TrialAccounting | None = None,
) -> dict[str, Any]:
"""Assemble the ATIF document. Pure: reads ``trial_dir``, writes nothing.
Steps are the merged timeline of what was said (transcript) and what was
done (tool calls). A seat's tool calls are attached to the next message that
seat posts, because that is the turn they belong to -- an agent works and
then reports. Calls with no message after them still get a step of their own,
whether the seat was cut off by the trial deadline or simply acted once more
after its last post: they happened, and a trajectory that dropped them would
understate the run to the one reader who most needs it complete.
"""
seats = _seat_models(manifest)
messages, truncated = _load_transcript(trial_dir / "transcript.json")
# rank 0 sorts tool calls ahead of a message sharing their timestamp: the
# transcript's `created_at` has one-second resolution while the logs have
# microseconds, so a call made during a turn routinely ties with the message
# that ends it.
timeline: list[tuple[float, int, str, Any]] = []
for seat in sorted(seats):
for call in parse_tool_calls(trial_dir / f"{seat}.stdout.log", seat):
timeline.append((call.epoch, 0, "tool", call))
for message in messages:
try:
epoch = float(message.get("created_at") or 0)
except (TypeError, ValueError):
continue
timeline.append((epoch, 1, "message", message))
timeline.sort(key=lambda item: (item[0], item[1]))
steps: list[dict[str, Any]] = []
pending: dict[str, list[_ToolCall]] = {}
synthetic = 0
def _tool_payload(calls: list[_ToolCall]) -> list[dict[str, Any]]:
nonlocal synthetic
payload = []
for call in calls:
if call.call_id is None:
synthetic += 1
call_id = f"{call.seat}-{synthetic}"
else:
call_id = call.call_id
entry: dict[str, Any] = {
"tool_call_id": call_id,
"function_name": call.function_name,
"arguments": call.arguments,
}
extra = {
k: v
for k, v in (("kind", call.kind), ("status", call.status))
if v
}
if extra:
entry["extra"] = extra
payload.append(entry)
return payload
# The instruction, when the transcript did not survive to carry it. Harbor
# hands it to us verbatim, so this is the one step that never depends on a
# scrape.
if not messages:
steps.append(
{
"step_id": 1,
"timestamp": None,
"source": "user",
"message": instruction,
}
)
for _epoch_value, _rank, kind, item in timeline:
if kind == "tool":
pending.setdefault(item.seat, []).append(item)
continue
author = str(item.get("author") or "unknown")
content = str(item.get("content") or "")
stamp = _iso(float(item.get("created_at") or 0))
if author in seats:
model, effort = seats[author]
step: dict[str, Any] = {
"step_id": len(steps) + 1,
"timestamp": stamp,
"source": "agent",
"model_name": model,
"message": content,
}
if effort:
step["reasoning_effort"] = effort
calls = pending.pop(author, [])
if calls:
step["tool_calls"] = _tool_payload(calls)
steps.append(step)
else:
# The harness user (and anything else that is not a roster seat).
# ATIF forbids model_name/tool_calls/metrics on a non-agent step.
steps.append(
{
"step_id": len(steps) + 1,
"timestamp": stamp,
"source": "user",
"message": content,
}
)
for seat in sorted(pending):
calls = pending[seat]
model, effort = seats[seat]
step = {
"step_id": len(steps) + 1,
"timestamp": calls[-1].timestamp,
"source": "agent",
"model_name": model,
"message": (
"(no channel message reported these calls: the seat was still "
"working when the trial ended, or acted after its last post)"
),
"tool_calls": _tool_payload(calls),
}
if effort:
step["reasoning_effort"] = effort
steps.append(step)
if not steps: # ATIF requires at least one step.
steps.append(
{
"step_id": 1,
"timestamp": None,
"source": "user",
"message": instruction,
}
)
# The primary model is the one that did the talking, so a team's row is
# labelled by its lead rather than by whichever seat sorts first.
turns_by_seat: dict[str, int] = {}
for step in steps:
if step["source"] == "agent":
turns_by_seat[step["model_name"]] = turns_by_seat.get(step["model_name"], 0) + 1
primary = (
max(turns_by_seat.items(), key=lambda kv: kv[1])[0]
if turns_by_seat
else manifest.roster[0].model_revision
)
document: dict[str, Any] = {
"schema_version": SCHEMA_VERSION,
"session_id": session_id,
"agent": {
"name": AGENT_NAME,
"version": AGENT_VERSION,
"model_name": primary,
"extra": {
"condition": manifest.condition,
"manifest_sha256": manifest.sha256,
"roster": [
{
"id": entry.id,
"role": entry.role,
"count": entry.count,
"model_revision": entry.model_revision,
"thinking_effort": entry.generation.thinking_effort,
}
for entry in manifest.roster
],
},
},
"steps": steps,
"notes": _notes(
truncated=truncated,
had_transcript=bool(messages),
total_calls=sum(len(step.get("tool_calls") or []) for step in steps),
clipped_calls=sum(
1
for step in steps
for call in (step.get("tool_calls") or [])
if "_raw" in (call.get("arguments") or {})
),
),
}
if accounting is not None:
document["final_metrics"] = {
"total_prompt_tokens": accounting.input_tokens,
"total_completion_tokens": accounting.output_tokens,
"total_cached_tokens": accounting.estimated_cache_read_tokens,
"total_cost_usd": round(accounting.cost_usd, 6),
"total_steps": len(steps),
}
return document
def _notes(
*,
truncated: bool,
had_transcript: bool,
total_calls: int = 0,
clipped_calls: int = 0,
) -> str:
"""The document's own account of what it is and is not.
A trajectory assembled from logs can mislead by omission, so the omissions
travel with it, counted rather than described. The judge reads this.
"""
parts = [
"Assembled by harbor_buzz_orchestra.trajectory from the trial's own "
"artifacts: agent messages from the Buzz channel transcript, tool calls "
"from buzz-acp's acp::tool trace, token and cost totals from the "
"per-seat receipts.",
"TOOL RESULTS ARE NOT AVAILABLE: buzz-acp logs what each tool was asked "
"to do but not what it returned, so every step's `observation` is "
"absent. What a command did has to be inferred from the next message.",
"Per-step `metrics` are absent for the same reason -- usage is reported "
"per seat per provider round, not per channel message -- so the totals "
"live in `final_metrics` only.",
]
if clipped_calls:
# Not a rounding detail: buzz-acp's MAX_RAW_INPUT caps the logged
# rawInput at 400 bytes (acp.rs), and a long shell command is exactly
# what gets cut. An auditor reading `_raw` must know it is a prefix and
# how much of this trial's evidence is in that shape.
parts.append(
f"{clipped_calls} of {total_calls} tool calls have an `arguments` "
"object of the form {\"_raw\": \"...\"}: buzz-acp truncates the "
"logged rawInput at 400 bytes, so those are the first 400 bytes of "
"the request and not the whole of it. The remaining "
f"{total_calls - clipped_calls} are complete, parsed objects."
)
if not had_transcript:
parts.append(
"The channel transcript was not recovered for this trial, so the "
"steps below carry the task instruction and any tool calls found in "
"the seat logs, without the agents' messages."
)
if truncated:
parts.append(
"The relay's message query hit its cap, so the earliest channel "
"messages are missing from this trajectory."
)
return " ".join(parts)
def write(
*,
logs_dir: Path,
trial_dir: Path,
manifest: ExperimentManifest,
instruction: str,
session_id: str,
accounting: TrialAccounting | None = None,
) -> Path | None:
"""Build, validate and write ``<logs_dir>/trajectory.json``.
``logs_dir`` is Harbor's agent log directory -- ``<trial>/agent`` -- because
that exact path is what the uploader looks for. ``trial_dir`` is the
``buzz/`` subdirectory the stack's own artifacts land in.
Validation runs through Harbor's own pydantic model when it is importable,
so a document this writer would emit but the hub would reject fails here,
next to the code that produced it, rather than in CI on a finished 445-trial
run.
"""
try:
document = build(
trial_dir=trial_dir,
manifest=manifest,
instruction=instruction,
session_id=session_id,
accounting=accounting,
)
except Exception: # noqa: BLE001 -- a lost trajectory is not a failed trial
logger.exception("could not assemble the ATIF trajectory")
return None
try:
from harbor.models.trajectories.trajectory import Trajectory
document = Trajectory.model_validate(document).to_json_dict()
except ImportError: # pragma: no cover -- harbor is always present in a run
logger.warning("harbor trajectory model unavailable; writing unvalidated")
except Exception: # noqa: BLE001
logger.exception("assembled trajectory failed ATIF validation")
return None
path = logs_dir / "trajectory.json"
try:
path.write_text(json.dumps(document, indent=2), encoding="utf-8")
except OSError:
logger.exception("could not write %s", path)
return None
return path

3180
benchmarks/harbor-buzz-orchestra/uv.lock generated
View File

File diff suppressed because it is too large Load Diff