feat(bench): multi-turn agent harness — read-only tools, loop, runner wiring (#102/#103)

bench/config.toml

@@ -4,6 +4,11 @@ trials = 10           # completions per (scenario x condition x model)
 temperature = 1.0     # realistic sampling; stochasticity is part of what we measure
 max_tokens = 1024
 
+# "single" = one-shot completion (v1). "multi" = multi-turn tool-using agent (v2 / #102): the model
+# gets read-only tools and chooses whether to verify the doc against the hidden dependency.
+mode = "single"
+max_turns = 8         # agent turn budget in multi mode
+
 # Models are referenced by short name. provider="mock" needs no API key (offline pipeline test).
 [models.mock]
 provider = "mock"

bench/pyproject.toml

@@ -13,3 +13,7 @@ dev = ["pytest>=8"]
 
 [tool.setuptools.packages.find]
 include = ["surface_bench*"]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]
+testpaths = ["tests"]

bench/surface_bench/agent.py

@@ -0,0 +1,122 @@
+"""The multi-turn agent loop (v2 / milestone #12).
+
+Drives a `ToolModel` over the read-only tool surface (`tools_runtime`) until it submits a final
+answer or runs out of turns. The loop is provider-agnostic: it only ever sees neutral `Step`/
+`ToolCall` objects and a neutral message history; each model adapter owns the translation to/from
+its wire format (and stashes the raw assistant message in `Step.provider_msg` so history round-trips
+losslessly).
+
+Neutral message schema (what `messages` holds, oldest first) — the contract each adapter reads:
+
+    {"role": "user",      "content": str}                         # the initial task, or a nudge
+    {"role": "assistant", "step": Step}                           # one model turn (carries provider_msg)
+    {"role": "tool",      "results": [{"id": str, "content": str}]}  # results for the prior turn's calls
+
+Termination, in priority order:
+  1. the model calls `final_answer`        -> stop_reason "final_answer"
+  2. the model emits text and no tool calls -> stop_reason "text_answer" (trailing text is the answer)
+  3. `max_turns` reached mid-tool-use       -> one forced "answer now" nudge -> stop_reason "forced_answer"
+
+The final answer is graded by the *existing* deterministic graders exactly as in single-shot, so the
+two modes stay comparable: it must still carry the scenario's `VERDICT:` line or `FILE:` blocks.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from .models import Step, ToolModel
+from .tools_runtime import TOOL_SPECS, ToolContext, dispatch
+
+NUDGE = (
+    "You have reached the step limit. Call final_answer now with your complete answer "
+    "(including any required VERDICT line or FILE: blocks)."
+)
+
+
+@dataclass
+class Trajectory:
+    final_text: str
+    turns: int
+    stop_reason: str
+    tool_calls: list[dict] = field(default_factory=list)  # [{turn, name, args}]
+    accessed: list[str] = field(default_factory=list)  # workspace-relative paths read/grepped
+    input_tokens: int = 0
+    output_tokens: int = 0
+    per_turn_tokens: list[tuple[int, int]] = field(default_factory=list)
+
+
+def run_agent(
+    model: ToolModel,
+    system: str,
+    user: str,
+    ctx: ToolContext,
+    *,
+    tools: list[dict] = TOOL_SPECS,
+    max_turns: int = 8,
+) -> Trajectory:
+    messages: list[dict] = [{"role": "user", "content": user}]
+    tool_log: list[dict] = []
+    per_turn: list[tuple[int, int]] = []
+    in_tok = out_tok = 0
+    final: str | None = None
+    stop = ""
+    turns = 0
+
+    def _account(step: Step) -> None:
+        nonlocal in_tok, out_tok
+        in_tok += step.input_tokens
+        out_tok += step.output_tokens
+        per_turn.append((step.input_tokens, step.output_tokens))
+
+    while turns < max_turns:
+        turns += 1
+        step = model.step(system, messages, tools)
+        _account(step)
+        messages.append({"role": "assistant", "step": step})
+
+        if not step.tool_calls:
+            # Text-only turn: providers vary on whether they reliably call a terminal tool, so we
+            # accept trailing prose as the answer.
+            final, stop = step.text, "text_answer"
+            break
+
+        results = []
+        ended = False
+        for tc in step.tool_calls:
+            tool_log.append({"turn": turns, "name": tc.name, "args": tc.args})
+            out, is_final = dispatch(tc.name, tc.args, ctx)
+            results.append({"id": tc.id, "content": out})
+            ended = ended or is_final
+        messages.append({"role": "tool", "results": results})
+        if ended:
+            final, stop = ctx.final_answer or "", "final_answer"
+            break
+
+    if final is None:
+        # Exhausted the budget still mid-tool-use: one forced answer-now turn.
+        messages.append({"role": "user", "content": NUDGE})
+        turns += 1
+        step = model.step(system, messages, tools)
+        _account(step)
+        messages.append({"role": "assistant", "step": step})
+        final = ""
+        for tc in step.tool_calls:
+            tool_log.append({"turn": turns, "name": tc.name, "args": tc.args})
+            out, is_final = dispatch(tc.name, tc.args, ctx)
+            if is_final:
+                final = ctx.final_answer or ""
+        if not final:
+            final = step.text
+        stop = "forced_answer"
+
+    return Trajectory(
+        final_text=final,
+        turns=turns,
+        stop_reason=stop,
+        tool_calls=tool_log,
+        accessed=list(ctx.accessed),
+        input_tokens=in_tok,
+        output_tokens=out_tok,
+        per_turn_tokens=per_turn,
+    )

bench/surface_bench/models.py

@@ -13,7 +13,7 @@
 
 import os
 from dataclasses import dataclass, field
-from typing import Protocol
+from typing import Any, Protocol
 
 
 @dataclass
@@ -30,6 +30,38 @@ class Model(Protocol):
     def complete(self, system: str, user: str) -> Completion: ...
 
 
+# ---- Multi-turn (agentic) types ------------------------------------------------------------
+# A provider-neutral one-turn result. Each adapter (Anthropic/OpenAI/Gemini) translates its wire
+# response into a `Step` and stashes the raw assistant message in `provider_msg`, so the agent loop
+# can echo it back verbatim on the next turn without the loop ever learning the provider's shape.
+
+
+@dataclass
+class ToolCall:
+    id: str
+    name: str
+    args: dict
+
+
+@dataclass
+class Step:
+    text: str = ""
+    tool_calls: list[ToolCall] = field(default_factory=list)
+    input_tokens: int = 0
+    output_tokens: int = 0
+    stop_reason: str = ""
+    provider_msg: Any = None  # raw assistant message, re-sent verbatim into history
+
+
+class ToolModel(Protocol):
+    """A model that can run the multi-turn loop. Separate from `Model` so single-shot-only adapters
+    don't have to implement `step`. `messages` is the neutral history built by `agent.run_agent`."""
+
+    name: str
+
+    def step(self, system: str, messages: list[dict], tools: list[dict]) -> Step: ...
+
+
 class MockModel:
     """Offline model for pipeline tests. Returns a canned reply (optionally per-condition).
 
@@ -52,6 +84,49 @@ def complete(self, system: str, user: str) -> Completion:
         return Completion(text=text, input_tokens=len(user.split()), output_tokens=len(text.split()))
 
 
+class MockToolModel:
+    """Offline tool-using model for loop tests: returns a fixed `script` of `Step`s, one per call.
+
+    Once the script is exhausted it falls back to a text-only `Step` (no tool calls), which the loop
+    treats as a final answer — so a script that never calls `final_answer` still terminates cleanly
+    (exercising the max-turns / forced-answer path). No network, no key.
+    """
+
+    def __init__(
+        self,
+        name: str = "mock-tool",
+        script: list[Step] | None = None,
+        fallback: str = "",
+        default: str = "",
+        replies: dict | None = None,
+    ):
+        self.name = name
+        self._script = list(script or [])
+        self._fallback = fallback
+        self._default = default
+        self._replies = replies or {}
+        self._condition: str | None = None
+        self._i = 0
+
+    def set_condition(self, condition: str) -> None:
+        self._condition = condition
+
+    def step(self, system: str, messages: list[dict], tools: list[dict]) -> Step:
+        if self._i < len(self._script):
+            step = self._script[self._i]
+            self._i += 1
+            return step
+        if self._fallback:
+            # Text-only turn -> the loop accepts it as the answer (exercises the forced-answer path).
+            return Step(text=self._fallback, output_tokens=len(self._fallback.split()))
+        # Canned mode (run.py offline smoke): answer immediately with the condition's reply.
+        reply = self._replies.get(self._condition, self._default)
+        return Step(
+            tool_calls=[ToolCall(id="final", name="final_answer", args={"answer": reply})],
+            output_tokens=len(reply.split()),
+        )
+
+
 class AnthropicModel:
     def __init__(self, name: str, model_id: str, temperature: float, max_tokens: int):
         try:
@@ -86,9 +161,15 @@ def complete(self, system: str, user: str) -> Completion:
         )
 
 
-def build_model(name: str, spec: dict, *, temperature: float, max_tokens: int) -> Model:
+def build_model(
+    name: str, spec: dict, *, temperature: float, max_tokens: int, mode: str = "single"
+) -> Model:
     provider = spec.get("provider")
     if provider == "mock":
+        if mode == "multi":
+            return MockToolModel(
+                name=name, default=spec.get("default", ""), replies=spec.get("replies")
+            )
         return MockModel(name=name, default=spec.get("default", ""), replies=spec.get("replies"))
     if provider == "anthropic":
         return AnthropicModel(

bench/surface_bench/run.py

@@ -19,9 +19,11 @@
 from pathlib import Path
 
 from . import grade_code, grade_qa
-from .models import MockModel, build_model
+from .agent import run_agent
+from .models import build_model
 from .prompts import CONDITIONS, build_prompt
 from .scenarios import discover
+from .tools_runtime import ToolContext, scenario_sandbox, touched_hidden
 
 BENCH_ROOT = Path(__file__).resolve().parent.parent
 
@@ -48,13 +50,21 @@ def main() -> None:
     ap.add_argument("--scenarios", nargs="*", help="subset of scenario ids")
     ap.add_argument("--conditions", nargs="*", default=list(CONDITIONS))
     ap.add_argument("--trials", type=int, help="override trials from config")
+    ap.add_argument(
+        "--mode",
+        choices=("single", "multi"),
+        help="single-shot completion (v1) or multi-turn tool-using agent (v2)",
+    )
+    ap.add_argument("--max-turns", type=int, help="agent turn budget in multi mode")
     ap.add_argument("--out", help="results dir (default results/<timestamp>)")
     args = ap.parse_args()
 
     cfg = tomllib.loads(Path(args.config).read_text())
     trials = args.trials or cfg.get("trials", 10)
     temperature = cfg.get("temperature", 1.0)
     max_tokens = cfg.get("max_tokens", 1024)
+    mode = args.mode or cfg.get("mode", "single")
+    max_turns = args.max_turns or cfg.get("max_turns", 8)
     model_specs = cfg.get("models", {})
     model_names = args.models or list(model_specs)
 
@@ -63,9 +73,16 @@ def main() -> None:
         sys.exit("no scenarios matched")
 
     models = {
-        n: build_model(n, model_specs[n], temperature=temperature, max_tokens=max_tokens)
+        n: build_model(
+            n, model_specs[n], temperature=temperature, max_tokens=max_tokens, mode=mode
+        )
         for n in model_names
     }
+    mock_names = {n for n in model_names if model_specs[n].get("provider") == "mock"}
+    if mode == "multi":
+        missing = [n for n, m in models.items() if not hasattr(m, "step")]
+        if missing:
+            sys.exit(f"multi mode needs a tool-using model; {missing} have no step() yet")
     # USD per token, per model (from config; 0 if unpriced, e.g. the mock).
     pricing = {
         n: (
@@ -85,6 +102,8 @@ def main() -> None:
         "trials": trials,
         "temperature": temperature,
         "max_tokens": max_tokens,
+        "mode": mode,
+        "max_turns": max_turns if mode == "multi" else None,
         "conditions": args.conditions,
         "models": {n: model_specs[n] for n in model_names},
         "scenarios": [s.id for s in scenarios],
@@ -99,7 +118,7 @@ def main() -> None:
             for condition in args.conditions:
                 system, user = build_prompt(scenario, condition)
                 for model_name, model in models.items():
-                    if isinstance(model, MockModel):
+                    if hasattr(model, "set_condition"):
                         model.set_condition(condition)
                     for trial in range(trials):
                         row = {
@@ -109,20 +128,50 @@ def main() -> None:
                             "condition": condition,
                             "model": model_name,
                             "trial": trial,
+                            "mode": mode,
                         }
                         try:
-                            comp = model.complete(system, user)
-                            grade = _grade(scenario, comp.text)
+                            if mode == "multi":
+                                # Fresh per-trial sandbox: the agent's tools can reach the hidden
+                                # dependency that the prompt withholds.
+                                with scenario_sandbox(scenario) as ws:
+                                    traj = run_agent(
+                                        model, system, user, ToolContext(ws), max_turns=max_turns
+                                    )
+                                text, in_tok, out_tok = (
+                                    traj.final_text,
+                                    traj.input_tokens,
+                                    traj.output_tokens,
+                                )
+                                agent_fields = dict(
+                                    turns=traj.turns,
+                                    stop_reason=traj.stop_reason,
+                                    tool_calls=traj.tool_calls,
+                                    verified_hidden=touched_hidden(
+                                        traj.accessed, scenario.hidden_paths
+                                    ),
+                                    per_turn_tokens=traj.per_turn_tokens,
+                                )
+                            else:
+                                comp = model.complete(system, user)
+                                text, in_tok, out_tok = (
+                                    comp.text,
+                                    comp.input_tokens,
+                                    comp.output_tokens,
+                                )
+                                agent_fields = {}
+                            grade = _grade(scenario, text)
                             in_price, out_price = pricing[model_name]
                             row.update(
-                                output=comp.text,
-                                input_tokens=comp.input_tokens,
-                                output_tokens=comp.output_tokens,
-                                cost_usd=comp.input_tokens * in_price + comp.output_tokens * out_price,
+                                output=text,
+                                input_tokens=in_tok,
+                                output_tokens=out_tok,
+                                cost_usd=in_tok * in_price + out_tok * out_price,
                                 ok=grade["ok"],
                                 misled=grade["misled"],
                                 detail=grade["detail"],
                                 parsed=grade["parsed"],
+                                **agent_fields,
                             )
                         except Exception as e:  # keep the matrix going; record the failure
                             row.update(output=None, ok=False, misled=False, error=repr(e))
@@ -134,7 +183,7 @@ def main() -> None:
                             end="",
                             file=sys.stderr,
                         )
-                        if not isinstance(model, MockModel):
+                        if model_name not in mock_names:
                             time.sleep(0)  # placeholder for rate-limit backoff hook
     print(f"\nwrote {raw_path}", file=sys.stderr)