Add Buzz-native collaboration benchmarks (#6264)

## Why
Terminal-Bench can show task correctness while missing Buzz-specific
collaboration failures, such as posting a final answer outside the
user's thread. Add a small Buzz-native suite that scores relay behavior
directly.

## What
- Add Harbor tasks for direct thread replies, event-level callback
mentions to a three-word user identity, targeted reads of explicitly
named paths outside the workspace, and exact temporary-channel
membership against a seeded directory of 50 users and 10 bots
- Export sanitized post-agent relay evidence and extend solo
runtime/provisioning for deterministic verification
- Expose channel TTL in CLI search output and cover positive and
negative transcript/state fixtures

## Risk Assessment
Low to medium — this primarily adds benchmark and testbed machinery. The
CLI change is an additive `ttl_seconds` field in channel-search JSON;
relay behavior is unchanged.

## References
- Motivated by the Buzz-specific behavior observed while benchmarking
https://github.com/block/buzz/pull/6128
- The named-path regression case covers the issue being fixed by
https://github.com/block/buzz/pull/6261
- Local live Sonnet runs: `reply-to-thread`, `user-mention`, and
`create-channel-invite-users` each scored reward 1.0 with zero
exceptions
- The current branch prompt scored 3/5 on a concurrent
`read-named-path-outside-workspace` batch: one explicit
workspace-boundary refusal and one missing final response

Generated with Codex

---------

Signed-off-by: Salman Mohammed <smohammed@squareup.com>

benchmarks/buzz-dataset/reply-to-thread/README.md

@@ -0,0 +1,71 @@
+# reply-to-thread
+
+## What the agent does
+
+Answers a six-month financial projection posted by the trial user
+([instruction.md](instruction.md)). The arithmetic is incidental — this task
+measures **where** the answer lands: in the user's thread, not as a new
+top-level channel message.
+
+> **The instruction deliberately says nothing about threading.** Threading is
+> the behavior under test, and it must come from `buzz-acp`'s production base
+> prompt rather than from the task prompt. Do not "fix" the instruction by
+> telling the agent to reply in-thread — that would make the task measure
+> instruction-following instead of product behavior.
+
+## Environment
+
+`python:3.12-slim-bookworm`, no extra packages: the agent never runs in this
+container's shell. `BuzzOrchestraAgent` launches the real `buzz-acp` /
+`buzz-agent` stack against a dedicated relay, and the agent works entirely
+through Buzz. Agent timeout 300s; the manifest's `trial_budget` is the
+effective clock.
+
+## Verifier
+
+Reads the post-agent `/logs/artifacts/buzz-evidence.json` snapshot (written by
+`BuzzContainerRuntime._collect_evidence` after the agent stops, so the agent
+cannot influence it). Every dimension is programmatic; `reward` is the
+conjunction of all of them.
+
+| Dimension | Type | Measures |
+| --- | --- | --- |
+| `evidence_complete` | programmatic | Snapshot is v1, untruncated, has one orchestrator, and resolves the task event and a candidate reply. Harness health, not agent skill — a 0 here means investigate the run |
+| `expected_author` | programmatic | The scored message was published by the orchestrator |
+| `same_channel` | programmatic | Reply carries the trial channel's `h` tag |
+| `reply_to_thread` | programmatic | Reply carries `["e", <task event>, "", "reply"]` — the behavior under test |
+| `answer_correct` | programmatic | Month-6 revenue (160,811), month-6 expenses (84,462), and cumulative profit (374,470), each ±1 and each on a line naming it |
+
+`answer_correct` requires the label and the value on the same line so a
+work-showing table with a wrong stated answer cannot pass on its intermediate
+rows. `instruction.md` asks for that formatting explicitly.
+
+## Layout
+
+```
+reply-to-thread/
+├── instruction.md          # Prompt posted to the agent as the trial user
+├── task.toml               # Metadata, timeouts, 1 CPU / 1 GiB environment
+├── environment/Dockerfile  # Bare python image; the relay stack is uploaded
+└── tests/
+    ├── test.sh             # Runs verify.py against the evidence snapshot
+    └── verify.py           # Deterministic scorer (see table above)
+```
+
+## Running
+
+```bash
+just benchmark \
+  --path benchmarks/buzz-dataset/reply-to-thread \
+  --attempts 1 \
+  --manifest benchmarks/harbor-buzz-orchestra/manifests/buzz-native-solo-luna.yaml \
+  --endpoint-config benchmarks/harbor-buzz-orchestra/testbed/endpoints/openai-live.json \
+  --n-concurrent 1
+```
+
+`harbor run -a oracle` does **not** work here, and no `solution/solve.sh` is
+shipped: the Oracle agent replaces `BuzzOrchestraAgent`, so no relay trial is
+provisioned and no evidence snapshot is exported. The verifier is covered
+instead by positive and negative fixture tests in
+`../harbor-buzz-orchestra/tests/test_reply_to_thread_verifier.py` (run from
+`benchmarks/harbor-buzz-orchestra`: `uv run --extra dev pytest -q`).

benchmarks/buzz-dataset/reply-to-thread/environment/Dockerfile

@@ -0,0 +1,3 @@
+FROM python:3.12-slim-bookworm
+
+WORKDIR /app

benchmarks/buzz-dataset/reply-to-thread/instruction.md

@@ -0,0 +1,10 @@
+Prepare a six-month financial projection using these assumptions:
+
+- Month 0 revenue is $120,000 and expenses are $75,000.
+- Revenue grows by 5% each month.
+- Expenses grow by 2% each month.
+- Project months 1 through 6, compounding each month from the prior month.
+
+Reply with the month 6 revenue, month 6 expenses, and cumulative operating
+profit across months 1 through 6. Round each result to the nearest dollar, and
+label each figure on the same line as its value.

benchmarks/buzz-dataset/reply-to-thread/task.toml

@@ -0,0 +1,24 @@
+schema_version = "1.3"
+
+[task]
+name = "buzz-native/reply-to-thread"
+description = "Answer a financial projection in the thread started by the user."
+authors = [{ name = "Buzz" }]
+keywords = ["buzz-native", "messaging", "threading"]
+
+[metadata]
+difficulty = "easy"
+category = "collaboration"
+tags = ["messaging", "threading", "implicit-behavior"]
+
+[agent]
+timeout_sec = 300.0
+
+[verifier]
+timeout_sec = 30.0
+
+[environment]
+network_mode = "public"
+cpus = 1
+memory_mb = 1024
+storage_mb = 1024

benchmarks/buzz-dataset/reply-to-thread/tests/test.sh

@@ -0,0 +1,8 @@
+#!/bin/sh
+set -eu
+
+mkdir -p /logs/verifier
+python3 /tests/verify.py \
+  --evidence /logs/artifacts/buzz-evidence.json \
+  --reward /logs/verifier/reward.json \
+  --details /logs/verifier/details.json

benchmarks/buzz-dataset/reply-to-thread/tests/verify.py

@@ -0,0 +1,174 @@
+#!/usr/bin/env python3
+"""Deterministic verifier for the Buzz reply-to-thread task."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from pathlib import Path
+from typing import Any
+
+# Each amount must appear on a line that also names what it is. Scanning the
+# whole reply for bare numbers passes a work-showing table whose month-6 rows
+# are right but whose stated answer is wrong.
+EXPECTED_ANSWERS = (
+    ("revenue", 160_811.0),
+    ("expense", 84_462.0),
+    ("profit", 374_470.0),
+)
+EXPECTED_AMOUNTS = tuple(amount for _, amount in EXPECTED_ANSWERS)
+NUMBER = re.compile(r"(?<![A-Za-z0-9_])-?\$?\d[\d,]*(?:\.\d+)?")
+
+
+def _numbers(content: str) -> list[float]:
+    values: list[float] = []
+    for token in NUMBER.findall(content):
+        try:
+            values.append(float(token.replace("$", "").replace(",", "")))
+        except ValueError:
+            continue
+    return values
+
+
+def _contains_amount(values: list[float], expected: float) -> bool:
+    return any(abs(value - expected) <= 1.0 for value in values)
+
+
+def _labelled_amount(content: str, label: str, expected: float) -> bool:
+    """Whether some line names ``label`` and carries ``expected`` on it."""
+    return any(
+        label in line.casefold() and _contains_amount(_numbers(line), expected)
+        for line in content.splitlines()
+    )
+
+
+def _has_tag(message: dict[str, Any], expected: list[str]) -> bool:
+    return any(tag == expected for tag in message.get("tags", []))
+
+
+def score_evidence(evidence: object) -> tuple[dict[str, float], dict[str, Any]]:
+    if not isinstance(evidence, dict):
+        return _zero_metrics(), {"error": "evidence root is not an object"}
+
+    task_event_id = evidence.get("task_event_id")
+    trial = evidence.get("trial") if isinstance(evidence.get("trial"), dict) else {}
+    channel_id = trial.get("channel_id")
+    identities = (
+        evidence.get("identities")
+        if isinstance(evidence.get("identities"), dict)
+        else {}
+    )
+    agents = [
+        identity
+        for identity in identities.values()
+        if isinstance(identity, dict) and identity.get("role") == "orchestrator"
+    ]
+    agent_pubkey = agents[0].get("pubkey") if len(agents) == 1 else None
+    messages = [
+        message for message in evidence.get("messages", []) if isinstance(message, dict)
+    ]
+    root_indexes = [
+        index
+        for index, message in enumerate(messages)
+        if message.get("id") == task_event_id
+    ]
+    root_index = root_indexes[0] if len(root_indexes) == 1 else -1
+    candidates = [
+        message
+        for message in messages[root_index + 1 :]
+        if agent_pubkey and message.get("pubkey") == agent_pubkey
+    ]
+    final = candidates[-1] if candidates else None
+
+    evidence_complete = float(
+        evidence.get("schema_version") == 1
+        and evidence.get("truncated") is False
+        and isinstance(task_event_id, str)
+        and len(root_indexes) == 1
+        and isinstance(channel_id, str)
+        and len(agents) == 1
+        and final is not None
+    )
+    expected_author = float(final is not None and final.get("pubkey") == agent_pubkey)
+    same_channel = float(
+        final is not None
+        and final.get("channel_id") == channel_id
+        and _has_tag(final, ["h", channel_id])
+    )
+    reply_to_thread = float(
+        final is not None
+        and final.get("reply_to_event_id") == task_event_id
+        and _has_tag(final, ["e", task_event_id, "", "reply"])
+    )
+    content = str(final.get("content", "")) if final is not None else ""
+    values = _numbers(content)
+    answer_correct = float(
+        all(
+            _labelled_amount(content, label, expected)
+            for label, expected in EXPECTED_ANSWERS
+        )
+    )
+    reward = float(
+        all(
+            metric == 1.0
+            for metric in (
+                evidence_complete,
+                expected_author,
+                same_channel,
+                reply_to_thread,
+                answer_correct,
+            )
+        )
+    )
+    metrics = {
+        "reward": reward,
+        "answer_correct": answer_correct,
+        "reply_to_thread": reply_to_thread,
+        "same_channel": same_channel,
+        "expected_author": expected_author,
+        "evidence_complete": evidence_complete,
+    }
+    details = {
+        "task_event_id": task_event_id,
+        "selected_message_id": final.get("id") if final is not None else None,
+        "selected_message_content": final.get("content") if final is not None else None,
+        "parsed_numbers": values,
+        "expected_amounts": list(EXPECTED_AMOUNTS),
+    }
+    return metrics, details
+
+
+def _zero_metrics() -> dict[str, float]:
+    return {
+        "reward": 0.0,
+        "answer_correct": 0.0,
+        "reply_to_thread": 0.0,
+        "same_channel": 0.0,
+        "expected_author": 0.0,
+        "evidence_complete": 0.0,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--evidence", type=Path, required=True)
+    parser.add_argument("--reward", type=Path, required=True)
+    parser.add_argument("--details", type=Path, required=True)
+    args = parser.parse_args()
+
+    try:
+        evidence = json.loads(args.evidence.read_text(encoding="utf-8"))
+        metrics, details = score_evidence(evidence)
+    except (OSError, json.JSONDecodeError) as error:
+        metrics, details = _zero_metrics(), {"error": str(error)}
+
+    args.reward.write_text(json.dumps(metrics, sort_keys=True) + "\n", encoding="utf-8")
+    args.details.write_text(
+        json.dumps(details, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())