diff --git a/bench/config.toml b/bench/config.toml index d7f5d19..28d4a7b 100644 --- a/bench/config.toml +++ b/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" diff --git a/bench/pyproject.toml b/bench/pyproject.toml index 4f5a999..afce9ed 100644 --- a/bench/pyproject.toml +++ b/bench/pyproject.toml @@ -13,3 +13,7 @@ dev = ["pytest>=8"] [tool.setuptools.packages.find] include = ["surface_bench*"] + +[tool.pytest.ini_options] +pythonpath = ["."] +testpaths = ["tests"] diff --git a/bench/surface_bench/agent.py b/bench/surface_bench/agent.py new file mode 100644 index 0000000..f85be0a --- /dev/null +++ b/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, + ) diff --git a/bench/surface_bench/models.py b/bench/surface_bench/models.py index 1654b77..83e205d 100644 --- a/bench/surface_bench/models.py +++ b/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( diff --git a/bench/surface_bench/run.py b/bench/surface_bench/run.py index 382d088..50a8506 100644 --- a/bench/surface_bench/run.py +++ b/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,6 +50,12 @@ 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/)") args = ap.parse_args() @@ -55,6 +63,8 @@ def main() -> None: 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) diff --git a/bench/surface_bench/tools_runtime.py b/bench/surface_bench/tools_runtime.py new file mode 100644 index 0000000..6a2269e --- /dev/null +++ b/bench/surface_bench/tools_runtime.py @@ -0,0 +1,227 @@ +"""Read-only tool surface for the multi-turn agent loop (v2 / milestone #12). + +The agentic track gives the model tools so it can *choose* to read the dependency a doc describes, +instead of being forced to trust the doc (the single-shot cascade's structural limit). The whole +point is to measure whether a confident *stale* doc suppresses that verification — so the surface is +deliberately **read-only**: `read_file`, `grep`, `list_dir`, and a `final_answer` terminator. There is +no `run_tests`/shell: a test runner would let the agent brute-force ground truth and wash out the +doc-trust signal we are trying to measure. + +Tools are scoped to a per-trial sandbox (see `scenario_sandbox`): a fresh copy of the scenario's +`code/` tree that **includes the hidden dependency**. The initial prompt still omits `hidden_paths` +(`prompts._render_code`), but the file is on disk, so `read_file("code/limiter/window.py")` succeeds +— the hidden truth is reachable *by choice, not absent*. Every path is resolved inside the sandbox; +escapes (`..`, absolute paths) are rejected. + +`TOOL_SPECS` is provider-neutral: a list of `{name, description, parameters}` (parameters is a JSON +Schema). Each model adapter translates it into its own tool format (Anthropic `input_schema`, OpenAI +`function.parameters`, Gemini `FunctionDeclaration`), so `agent.py` and the graders never learn which +provider ran. +""" + +from __future__ import annotations + +import re +import shutil +import tempfile +from collections.abc import Iterator +from contextlib import contextmanager +from fnmatch import fnmatch +from pathlib import Path + +from .scenarios import Scenario + +# Bound a single tool result so one huge file can't blow the turn's token budget. Fixtures are tiny; +# this only guards against pathological inputs. +MAX_READ_BYTES = 64 * 1024 +MAX_GREP_MATCHES = 100 + + +class ToolError(Exception): + """A tool was called with bad arguments (e.g. a path escape). The message is fed back to the + model as the tool result so it can recover, rather than crashing the run.""" + + +# Provider-neutral tool schemas. `parameters` is a JSON Schema object; adapters translate it. +TOOL_SPECS: list[dict] = [ + { + "name": "list_dir", + "description": "List the files and directories under a workspace-relative path.", + "parameters": { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "Workspace-relative directory (default the workspace root).", + } + }, + "required": [], + }, + }, + { + "name": "read_file", + "description": "Read the full contents of a workspace-relative file.", + "parameters": { + "type": "object", + "properties": { + "path": {"type": "string", "description": "Workspace-relative file path."} + }, + "required": ["path"], + }, + }, + { + "name": "grep", + "description": "Search files for a regular expression, returning matching lines as " + "`path:lineno: line`.", + "parameters": { + "type": "object", + "properties": { + "pattern": {"type": "string", "description": "A Python regular expression."}, + "path": { + "type": "string", + "description": "Workspace-relative file or directory to search " + "(default the workspace root).", + }, + }, + "required": ["pattern"], + }, + }, + { + "name": "final_answer", + "description": "Submit your final answer and end the task. Provide the complete answer " + "(including any required VERDICT line or FILE: blocks) in `answer`.", + "parameters": { + "type": "object", + "properties": { + "answer": {"type": "string", "description": "The complete final answer."} + }, + "required": ["answer"], + }, + }, +] + +TOOL_NAMES = frozenset(spec["name"] for spec in TOOL_SPECS) + + +class ToolContext: + """Executes the read-only tools against a sandbox root and records what was accessed. + + `accessed` is the list of workspace-relative paths the agent actually read or grepped — the raw + material for the *verification rate* metric (did it read a load-bearing hidden dependency before + answering?). `final_answer` is set once the agent terminates. + """ + + def __init__(self, root: Path): + self.root = root.resolve() + self.accessed: list[str] = [] + self.final_answer: str | None = None + + # -- path safety ------------------------------------------------------------------------- + def _resolve(self, rel: str) -> Path: + # `root / "/abs"` collapses to "/abs" in pathlib, and `..` walks up — resolve() then makes + # any escape visible, and we reject anything not inside the sandbox. + p = (self.root / rel).resolve() + if p != self.root and self.root not in p.parents: + raise ToolError(f"path {rel!r} escapes the workspace") + return p + + def _rel(self, p: Path) -> str: + return p.relative_to(self.root).as_posix() + + # -- tools ------------------------------------------------------------------------------- + def list_dir(self, path: str = ".") -> str: + p = self._resolve(path) + if not p.exists(): + raise ToolError(f"no such path: {path!r}") + if not p.is_dir(): + raise ToolError(f"not a directory: {path!r}") + names = sorted(c.name + ("/" if c.is_dir() else "") for c in p.iterdir()) + return "\n".join(names) if names else "(empty)" + + def read_file(self, path: str) -> str: + p = self._resolve(path) + if not p.is_file(): + raise ToolError(f"no such file: {path!r}") + self.accessed.append(self._rel(p)) + data = p.read_bytes() + truncated = len(data) > MAX_READ_BYTES + text = data[:MAX_READ_BYTES].decode("utf-8", errors="replace") + if truncated: + text += f"\n... [truncated at {MAX_READ_BYTES} bytes]" + return text + + def grep(self, pattern: str, path: str = ".") -> str: + try: + rx = re.compile(pattern) + except re.error as e: + raise ToolError(f"invalid regex: {e}") from e + p = self._resolve(path) + if not p.exists(): + raise ToolError(f"no such path: {path!r}") + files = [p] if p.is_file() else [f for f in sorted(p.rglob("*")) if f.is_file()] + matches: list[str] = [] + for f in files: + rel = self._rel(f) + try: + lines = f.read_text(errors="replace").splitlines() + except OSError: + continue + hit = False + for i, line in enumerate(lines, 1): + if rx.search(line): + matches.append(f"{rel}:{i}: {line}") + hit = True + if len(matches) >= MAX_GREP_MATCHES: + break + if hit: + self.accessed.append(rel) + if len(matches) >= MAX_GREP_MATCHES: + matches.append("... [more matches truncated]") + break + return "\n".join(matches) if matches else "(no matches)" + + def submit(self, answer: str) -> str: + self.final_answer = answer + return "Final answer recorded." + + # -- verification metric helper ---------------------------------------------------------- + def verified(self, hidden_paths: list[str]) -> bool: + """True iff the agent read/grepped a file matching one of the scenario's hidden globs — + i.e. it went and checked the dependency the (possibly stale) doc describes.""" + return touched_hidden(self.accessed, hidden_paths) + + +def touched_hidden(accessed: list[str], hidden_paths: list[str]) -> bool: + """Did any accessed path match a hidden glob? The basis of the verification-rate metric; lives + at module scope so the runner can compute it from a `Trajectory.accessed` list without a ctx.""" + return any(fnmatch(a, pat) for a in accessed for pat in hidden_paths) + + +def dispatch(name: str, args: dict, ctx: ToolContext) -> tuple[str, bool]: + """Run one tool call. Returns `(result_text, is_final)`. Bad calls return an error string (not + an exception) so the loop can feed it back and let the model recover.""" + try: + if name == "list_dir": + return ctx.list_dir(args.get("path", ".")), False + if name == "read_file": + return ctx.read_file(args["path"]), False + if name == "grep": + return ctx.grep(args["pattern"], args.get("path", ".")), False + if name == "final_answer": + return ctx.submit(args["answer"]), True + return f"error: unknown tool {name!r}", False + except KeyError as e: + return f"error: missing required argument {e}", False + except ToolError as e: + return f"error: {e}", False + + +@contextmanager +def scenario_sandbox(scenario: Scenario) -> Iterator[Path]: + """Yield a fresh per-trial workspace root containing a copy of the scenario's `code/` tree, + **including the hidden dependency**. A `ToolContext(root)` scoped here can read any file the + grader runs against, while the prompt still omits `hidden_paths`.""" + with tempfile.TemporaryDirectory(prefix=f"surfbench-{scenario.id}-") as td: + ws = Path(td) + shutil.copytree(scenario.root / "code", ws / "code") + yield ws diff --git a/bench/tests/test_agent.py b/bench/tests/test_agent.py new file mode 100644 index 0000000..ebb1db6 --- /dev/null +++ b/bench/tests/test_agent.py @@ -0,0 +1,97 @@ +"""Offline tests for the multi-turn agent loop (no API, no spend).""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from surface_bench.agent import run_agent +from surface_bench.models import Step, ToolCall, MockToolModel +from surface_bench.scenarios import load_scenario +from surface_bench.tools_runtime import ToolContext, scenario_sandbox + +BENCH_ROOT = Path(__file__).resolve().parents[1] + + +def _read(path: str, tid: str = "t") -> Step: + return Step(tool_calls=[ToolCall(id=tid, name="read_file", args={"path": path})], output_tokens=4) + + +def _final(answer: str, tid: str = "f") -> Step: + return Step( + tool_calls=[ToolCall(id=tid, name="final_answer", args={"answer": answer})], output_tokens=6 + ) + + +@pytest.fixture +def workspace(tmp_path: Path) -> Path: + code = tmp_path / "code" + (code / "limiter").mkdir(parents=True) + (code / "throttle.py").write_text("def plan_batches(total):\n return total\n") + (code / "limiter" / "window.py").write_text("WINDOW_LIMIT = 10 # admits 11 (<=)\n") + return tmp_path + + +def test_read_then_final(workspace: Path) -> None: + ctx = ToolContext(workspace) + model = MockToolModel(script=[_read("code/limiter/window.py"), _final("VERDICT: capacity=11")]) + traj = run_agent(model, "sys", "task", ctx, max_turns=8) + + assert traj.stop_reason == "final_answer" + assert traj.final_text == "VERDICT: capacity=11" + assert traj.turns == 2 + assert ctx.accessed == ["code/limiter/window.py"] + assert ctx.verified(["code/limiter/*.py"]) is True + assert [c["name"] for c in traj.tool_calls] == ["read_file", "final_answer"] + assert traj.output_tokens == 10 # 4 + 6 + + +def test_text_only_answer_terminates(workspace: Path) -> None: + model = MockToolModel(script=[Step(text="VERDICT: capacity=10", output_tokens=5)]) + traj = run_agent(model, "sys", "task", ToolContext(workspace), max_turns=8) + assert traj.stop_reason == "text_answer" + assert traj.final_text == "VERDICT: capacity=10" + assert traj.turns == 1 + + +def test_bad_call_is_recoverable(workspace: Path) -> None: + ctx = ToolContext(workspace) + model = MockToolModel( + script=[_read("code/missing.py", "1"), _read("code/limiter/window.py", "2"), _final("ok", "3")] + ) + traj = run_agent(model, "sys", "task", ctx, max_turns=8) + assert traj.stop_reason == "final_answer" + assert traj.final_text == "ok" + assert traj.turns == 3 + # The bad read didn't count as access; the good one did. + assert ctx.accessed == ["code/limiter/window.py"] + + +def test_max_turns_forces_an_answer(workspace: Path) -> None: + # A model that never stops calling tools. With max_turns=2 and an exhausted script, the loop + # falls back to a forced answer-now turn (MockToolModel returns its text fallback). + ctx = ToolContext(workspace) + model = MockToolModel( + script=[_read("code/throttle.py", "1"), _read("code/throttle.py", "2")], + fallback="VERDICT: forced", + ) + traj = run_agent(model, "sys", "task", ctx, max_turns=2) + assert traj.stop_reason == "forced_answer" + assert traj.final_text == "VERDICT: forced" + assert traj.turns == 3 # 2 capped turns + 1 forced + + +def test_loop_against_real_scenario_sandbox() -> None: + """End-to-end on a real cascade scenario: the agent reads the hidden dependency it would only + otherwise know by doc, and that read registers as verification.""" + scenario = load_scenario(BENCH_ROOT / "scenarios" / "cascade-quota-batcher-code") + with scenario_sandbox(scenario) as ws: + ctx = ToolContext(ws) + model = MockToolModel( + script=[_read("code/limiter/window.py"), _final("FILE: code/throttle.py\n```python\nx=1\n```")] + ) + traj = run_agent(model, "sys", "task", ctx, max_turns=8) + assert traj.stop_reason == "final_answer" + assert ctx.verified(scenario.hidden_paths) is True + assert any(c["name"] == "read_file" for c in traj.tool_calls) diff --git a/bench/tests/test_run_multi.py b/bench/tests/test_run_multi.py new file mode 100644 index 0000000..085e7ef --- /dev/null +++ b/bench/tests/test_run_multi.py @@ -0,0 +1,44 @@ +"""Integration test for the runner's --mode multi wiring (offline, mock model, no spend).""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +from surface_bench import run + +_AGENT_FIELDS = {"turns", "stop_reason", "tool_calls", "verified_hidden", "per_turn_tokens"} + + +def _run(monkeypatch, tmp_path: Path, mode: str) -> tuple[dict, list[dict]]: + argv = [ + "run", "--models", "mock", "--mode", mode, + "--scenarios", "cascade-quota-batcher-code", "--trials", "1", + "--out", str(tmp_path), + ] + monkeypatch.setattr(sys, "argv", argv) + run.main() + meta = json.loads((tmp_path / "run.json").read_text()) + rows = [json.loads(line) for line in (tmp_path / "raw.jsonl").read_text().splitlines()] + return meta, rows + + +def test_multi_mode_writes_agent_fields(monkeypatch, tmp_path: Path) -> None: + meta, rows = _run(monkeypatch, tmp_path, "multi") + assert meta["mode"] == "multi" and meta["max_turns"] == 8 + assert rows, "expected rows" + for r in rows: + assert r["mode"] == "multi" + assert _AGENT_FIELDS <= r.keys() + assert r["stop_reason"] in {"final_answer", "text_answer", "forced_answer"} + assert isinstance(r["verified_hidden"], bool) + + +def test_single_mode_omits_agent_fields(monkeypatch, tmp_path: Path) -> None: + meta, rows = _run(monkeypatch, tmp_path, "single") + assert meta["mode"] == "single" and meta["max_turns"] is None + assert rows + for r in rows: + assert r["mode"] == "single" + assert not (_AGENT_FIELDS & r.keys()) diff --git a/bench/tests/test_tools_runtime.py b/bench/tests/test_tools_runtime.py new file mode 100644 index 0000000..321141c --- /dev/null +++ b/bench/tests/test_tools_runtime.py @@ -0,0 +1,123 @@ +"""Offline tests for the read-only agent tool surface and per-trial sandbox (no API, no spend).""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from surface_bench.prompts import build_prompt +from surface_bench.scenarios import load_scenario +from surface_bench.tools_runtime import ( + TOOL_NAMES, + TOOL_SPECS, + ToolContext, + dispatch, + scenario_sandbox, +) + +BENCH_ROOT = Path(__file__).resolve().parents[1] + + +@pytest.fixture +def workspace(tmp_path: Path) -> Path: + """A minimal sandbox: a code/ tree with a visible file and a 'hidden' dependency.""" + code = tmp_path / "code" + (code / "limiter").mkdir(parents=True) + (code / "throttle.py").write_text("def plan_batches(total):\n return total\n") + (code / "limiter" / "window.py").write_text( + "WINDOW_LIMIT = 10\n\nclass FixedWindowLimiter:\n def allow(self, n):\n return n <= self.limit\n" + ) + return tmp_path + + +def test_list_dir(workspace: Path) -> None: + ctx = ToolContext(workspace) + assert ctx.list_dir(".") == "code/" + assert set(ctx.list_dir("code").split("\n")) == {"limiter/", "throttle.py"} + + +def test_read_file_and_access_tracking(workspace: Path) -> None: + ctx = ToolContext(workspace) + assert ctx.accessed == [] + body = ctx.read_file("code/limiter/window.py") + assert "WINDOW_LIMIT = 10" in body + assert ctx.accessed == ["code/limiter/window.py"] + + +def test_grep(workspace: Path) -> None: + ctx = ToolContext(workspace) + out = ctx.grep("WINDOW_LIMIT", "code") + assert "code/limiter/window.py:1: WINDOW_LIMIT = 10" in out + assert "code/limiter/window.py" in ctx.accessed + assert ctx.grep("nomatchhere", "code") == "(no matches)" + + +def test_grep_invalid_regex_is_recoverable(workspace: Path) -> None: + out, is_final = dispatch("grep", {"pattern": "("}, ToolContext(workspace)) + assert out.startswith("error: invalid regex") + assert is_final is False + + +@pytest.mark.parametrize("bad", ["../etc/passwd", "/etc/passwd", "code/../../secret"]) +def test_path_escape_rejected(workspace: Path, bad: str) -> None: + out, is_final = dispatch("read_file", {"path": bad}, ToolContext(workspace)) + assert out.startswith("error:") and "escapes the workspace" in out + assert is_final is False + + +def test_missing_file_is_recoverable(workspace: Path) -> None: + out, _ = dispatch("read_file", {"path": "code/nope.py"}, ToolContext(workspace)) + assert out.startswith("error: no such file") + + +def test_final_answer_terminates(workspace: Path) -> None: + ctx = ToolContext(workspace) + out, is_final = dispatch("final_answer", {"answer": "VERDICT: x=1"}, ctx) + assert is_final is True + assert ctx.final_answer == "VERDICT: x=1" + + +def test_unknown_tool_and_missing_arg(workspace: Path) -> None: + ctx = ToolContext(workspace) + out, is_final = dispatch("delete_everything", {}, ctx) + assert out == "error: unknown tool 'delete_everything'" and is_final is False + out, _ = dispatch("read_file", {}, ctx) # missing required 'path' + assert out.startswith("error: missing required argument") + + +def test_verified_helper(workspace: Path) -> None: + ctx = ToolContext(workspace) + hidden = ["code/limiter/*.py"] + assert ctx.verified(hidden) is False + ctx.read_file("code/throttle.py") # visible file — not verification + assert ctx.verified(hidden) is False + ctx.read_file("code/limiter/window.py") # the hidden dependency + assert ctx.verified(hidden) is True + + +def test_tool_specs_contract() -> None: + # The neutral schema each provider adapter will translate. Guard its shape so an adapter can + # rely on every spec having a name + a JSON-Schema `parameters` object. + assert {"list_dir", "read_file", "grep", "final_answer"} == set(TOOL_NAMES) + for spec in TOOL_SPECS: + assert spec.keys() >= {"name", "description", "parameters"} + assert spec["parameters"]["type"] == "object" + assert "properties" in spec["parameters"] + + +def test_sandbox_exposes_hidden_dependency_that_prompt_hides() -> None: + """The keystone of the multi-turn design: a real cascade scenario's hidden dependency is absent + from the prompt but reachable via tools in the sandbox.""" + scenario = load_scenario(BENCH_ROOT / "scenarios" / "cascade-quota-batcher-code") + assert scenario.hidden_paths, "expected a cascade scenario with hidden_paths" + + _, user = build_prompt(scenario, "C1") + assert "### code/limiter/window.py" not in user # the dependency is withheld from the prompt + + with scenario_sandbox(scenario) as ws: + ctx = ToolContext(ws) + assert (ws / "code" / "limiter" / "window.py").is_file() # but present on disk + body = ctx.read_file("code/limiter/window.py") # and reachable by choice + assert body.strip() != "" + assert ctx.verified(scenario.hidden_paths) is True