"""Deterministic tool-calling / code pack ("tools pack").
The qualitative ``eval`` scores free text with an LLM judge. Structured output — tool choice,
JSON arguments, schema conformance, code that has to run — can be checked mechanically, and a
judge would only add noise there. This module is that mechanical half:
ToolsPack (YAML) → items: prompt (+ pre-read fixture files) + deterministic checks
collect_turn() → folds the client's raw ToolStreamEvents into one ToolTurn (pure)
run_checks() → ToolTurn × item → CheckResults (pure, except code execution in a subprocess)
run_tools() → matrix model × item × repeat → responses.jsonl (incremental, resumable)
compare_bundles() → paired A/B per item: discordant counts + exact McNemar p (pure)
Tool schemas live in the pack (data), mirrored from opencode so a failure here is a failure
there. The only engine-aware code stays in ``client.py`` (``stream_tools``).
"""
from __future__ import annotations
import contextlib
import csv
import html.parser
import json
import math
import os
import re
import shutil
import signal
import subprocess
import sys
import tempfile
import time
from collections.abc import Callable, Iterable, Iterator
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Literal
import yaml
from pydantic import BaseModel, Field, model_validator
# --------------------------------------------------------------------------- pack schema
CheckType = Literal[
"calls",
"no_tools",
"schema_valid",
"no_empty_args",
"finish_reason",
"arg_equals",
"arg_regex",
"write_regex",
"write_json_equals",
"write_python",
"write_html",
"write_js_syntax",
"write_check",
"write_node",
"call_paths",
"edit_applies",
"edited_python",
"edited_regex",
]
class Check(BaseModel):
type: CheckType
tool: str | None = None # calls/arg_*: restrict to this tool
count: int | None = None # calls: exact
min: int | None = None # calls: at least
max: int | None = None # calls: at most
key: str | None = None # arg_*: argument key
value: Any = None # arg_equals / write_json_equals / finish_reason / call_paths (list)
pattern: str | None = None # *_regex
absent: bool = False # *_regex: pattern must NOT match
path: str | None = None # write_* / edit*: target filePath
test: str | None = None # *_python/_node: asserts run after the file's code;
# write_check: python asserts with the written file bound to CONTENT
tags: dict[str, int] = Field(default_factory=dict) # write_html: tag → min count
label: str = "" # optional human label for the report
def name(self) -> str:
if self.label:
return self.label
bits: list[str] = [self.type]
for v in (self.tool, self.key, self.path):
if v:
bits.append(str(v))
return ":".join(bits)
class TurnReply(BaseModel):
"""A canned tool result for a call the item anticipates (data, not code): the first reply
whose tool and pattern match is returned instead of the neutral receipt."""
tool: str | None = None # None → any tool
pattern: str | None = None # searched in the call's leading argument (bash: command)
output: str
class ToolItem(BaseModel):
id: str
title: str
category: str
prompt: str
tests: str = "" # what the item probes (documentation)
fixtures: dict[str, str] = Field(default_factory=dict) # filePath → original content
preread: list[str] = Field(default_factory=list) # fixtures fed as prior read tool results
# Long-context items: files (names inside the pack's context_dir) read in an EARLIER exchange,
# before the task — reproduces a running opencode session with a large, realistic context.
context: list[str] = Field(default_factory=list)
context_fixtures: dict[str, str] = Field(default_factory=dict) # filled by load_tools_pack
max_tokens: int | None = None # None → pack default
repeats: int = 1
# Anticipated exploration: `ls`/`find`/… get this output instead of the neutral receipt.
turn_results: list[TurnReply] = Field(default_factory=list)
checks: list[Check]
@model_validator(mode="after")
def _preread_known(self) -> ToolItem:
missing = [p for p in self.preread if p not in self.fixtures]
if missing:
raise ValueError(f"item {self.id}: preread paths without fixture: {missing}")
for c in self.checks:
if c.type in ("edit_applies", "edited_python", "edited_regex") and (
c.path not in self.fixtures
):
raise ValueError(f"item {self.id}: {c.type} path {c.path!r} has no fixture")
return self
class ToolsSampling(BaseModel):
# None omits the field from the request instead of sending it — needed to replicate a
# client (e.g. opencode without `temperature: true` on the model entry) that sends neither,
# so the server falls back to its own default sampling instead of our forced 0.0/42.
temperature: float | None = 0.0
seed: int | None = 42
class ToolsPack(BaseModel):
id: str
title: str
version: int = 1
description: str = ""
source: str = "" # where the tool schemas were taken from
context_dir: str | None = None # relative to the pack file; holds long-context snapshots
# Agent loop: how many model turns one item may take. Scoring is over the SUM of all turns,
# so a preparatory call (ls/mkdir/read) costs a turn but is not a failure — opencode does
# the same. The loop never stops because the checks are satisfied (that would leak the
# scoring into the run); it stops at a turn without tool calls, or at this budget.
# 5 because a real opencode session needed 4–5 steps for a comparable task (measured by the
# opencode session 2026-09-20: build 4 steps, its old thinking config 5); with 3 the
# multi-file items were cut off mid-exploration. Every turn re-sends the whole exchange, so
# the budget costs prefill — on the long-context items (L*) that is the dominant cost.
max_turns: int = 5
context_root: str = "/work/proj/src/touchstone" # virtual dir the context files appear under
context_intro: str = (
"Verschaff dir zuerst einen Überblick über die Kernmodule unter /work/proj/src/touchstone/ "
"— lies sie. Die eigentliche Aufgabe kommt danach."
)
context_ack: str = (
"Ich habe die Kernmodule gelesen und habe den Überblick. Was ist die Aufgabe?"
)
system_prompt: str
max_tokens: int | None = None # default per-item cap (None = server decides)
sampling: ToolsSampling = Field(default_factory=ToolsSampling)
tools: list[dict[str, Any]]
items: list[ToolItem]
@model_validator(mode="after")
def _unique_ids(self) -> ToolsPack:
ids = [i.id for i in self.items]
dup = {i for i in ids if ids.count(i) > 1}
if dup:
raise ValueError(f"duplicate item ids: {sorted(dup)}")
return self
def tool_schemas(self) -> dict[str, dict[str, Any]]:
return {t["function"]["name"]: t["function"].get("parameters", {}) for t in self.tools}
def load_tools_pack(path: str | Path) -> ToolsPack:
raw = yaml.safe_load(Path(path).read_text(encoding="utf-8"))
if not isinstance(raw, dict):
raise ValueError(f"tools pack {path} did not parse to a mapping")
pack = ToolsPack.model_validate(raw)
needs = [it for it in pack.items if it.context]
if needs:
if not pack.context_dir:
raise ValueError("items use `context` but the pack has no context_dir")
base = Path(path).resolve().parent / pack.context_dir
for it in needs:
fx: dict[str, str] = {}
for name in it.context:
src = base / f"{name}.txt"
if not src.is_file():
raise ValueError(f"item {it.id}: context file missing: {src}")
fx[f"{pack.context_root}/{name}"] = src.read_text(encoding="utf-8")
it.context_fixtures = fx
return pack
def pack_fingerprint(path: str | Path, pack: ToolsPack) -> str:
"""sha256 over the pack YAML and every long-context file it pulls in — resume refuses a
bundle whose pack changed in between (a mixed run would not be comparable)."""
import hashlib
h = hashlib.sha256(Path(path).read_bytes())
for it in pack.items:
for vpath, text in sorted(it.context_fixtures.items()):
h.update(vpath.encode())
h.update(text.encode("utf-8"))
return h.hexdigest()
def checks_fingerprint(pack: ToolsPack) -> str:
"""sha256 over everything that defines WHAT is measured: items, checks, fixtures, tool
schemas, the framing prompts, the token budget and the turn budget — but **not** the
sampling. Two effort levels run with their own recommended sampling (thinking vs instruct)
are different runs of the SAME test bench; a compare must be allowed to pair them and name
the difference. ``pack_fingerprint`` stays strict for resume, where a parameter change
mid-bundle would mix cells.
"""
import hashlib
payload = {
"tools": pack.tools,
"system_prompt": pack.system_prompt,
"context_intro": pack.context_intro,
"context_ack": pack.context_ack,
"max_tokens": pack.max_tokens,
"max_turns": pack.max_turns,
"items": [it.model_dump(mode="json") for it in pack.items],
}
blob = json.dumps(payload, sort_keys=True, ensure_ascii=True).encode()
return hashlib.sha256(blob).hexdigest()
def needs_node(pack: ToolsPack) -> bool:
return any(c.type in ("write_js_syntax", "write_node") for it in pack.items for c in it.checks)
# --------------------------------------------------------------------------- messages
def read_tool_output(path: str, content: str) -> str:
"""Render a file the way opencode's read tool returns it (1.18.x: ``N: line``)."""
lines = content.split("\n")
if lines and lines[-1] == "":
lines = lines[:-1]
body = "\n".join(f"{i + 1}: {ln}" for i, ln in enumerate(lines))
return (
f"<path>{path}</path>\n<type>file</type>\n<content>\n{body}\n\n"
f"(End of file - total {len(lines)} lines)\n</content>"
)
def build_messages(pack: ToolsPack, item: ToolItem) -> list[dict[str, Any]]:
msgs: list[dict[str, Any]] = [{"role": "system", "content": pack.system_prompt}]
if item.context_fixtures:
msgs.append({"role": "user", "content": pack.context_intro})
msgs += _read_exchange(item.context_fixtures, "call_ctx")
msgs.append({"role": "assistant", "content": pack.context_ack})
msgs.append({"role": "user", "content": item.prompt})
if item.preread:
msgs += _read_exchange({p: item.fixtures[p] for p in item.preread}, "call_read")
return msgs
def _read_exchange(files: dict[str, str], prefix: str) -> list[dict[str, Any]]:
"""One assistant turn with parallel read calls + one tool result per file (opencode shape)."""
calls = [
{
"id": f"{prefix}_{n}",
"type": "function",
"function": {"name": "read", "arguments": json.dumps({"filePath": p})},
}
for n, p in enumerate(files)
]
out: list[dict[str, Any]] = [{"role": "assistant", "content": "", "tool_calls": calls}]
for n, (p, content) in enumerate(files.items()):
out.append(
{
"role": "tool",
"tool_call_id": f"{prefix}_{n}",
"content": read_tool_output(p, content),
}
)
return out
# --------------------------------------------------------------------------- stream → turn
@dataclass
class ToolStreamEvent:
"""One raw streaming fragment, engine-agnostic (produced by ``client.stream_tools``)."""
content: str | None = None
reasoning: str | None = None
is_tool: bool = False # a tool-call fragment (tc_index may still be None on some servers)
tc_index: int | None = None
tc_id: str | None = None
tc_name: str | None = None
tc_args: str | None = None
finish_reason: str | None = None
prompt_tokens: int | None = None
completion_tokens: int | None = None
reasoning_tokens: int | None = None # usage.completion_tokens_details, if the server sends it
@dataclass
class ToolCall:
index: int
id: str
name: str
arguments: str # raw string as streamed — may be "" (truncated) or invalid JSON
@dataclass
class ToolTurn:
content: str = ""
reasoning: str = ""
tool_calls: list[ToolCall] = field(default_factory=list)
finish_reason: str | None = None
prompt_tokens: int | None = None
completion_tokens: int | None = None
reasoning_tokens: int | None = None
ttft_s: float | None = None # first generated token of any kind
t_first_tool_s: float | None = None
e2e_s: float = 0.0
error: str = "" # transport failure (dead/evicted server) — not a model result, re-run
rejected: str = "" # deterministic 4xx (e.g. context overflow) — a result, counted as fail
# Agent loop: how EVERY turn ended. ``finish_reason`` stays the last turn's (the episode's
# ending), but a check asks about the turn that did the work — in the loop the last turn is
# the closing one without calls, i.e. "stop".
finishes: list[str | None] = field(default_factory=list)
def classify_exception(e: Exception) -> tuple[str, str]:
"""(error, rejected). A 4xx other than 408/429 is deterministic for this model+load (e.g.
context overflow) → a result that counts as a fail; everything else is transport."""
msg = f"{type(e).__name__}: {e}"[:500]
status = getattr(e, "status_code", None)
if isinstance(status, int) and 400 <= status < 500 and status not in (408, 429):
return "", msg
return msg, ""
def collect_turn(events: Iterable[ToolStreamEvent], clock: Callable[[], float]) -> ToolTurn:
"""Fold streamed fragments into one turn. Tool-call fragments are keyed by ``index``
(OpenAI streaming contract); without an index a new ``id`` starts a new call. The name is
taken once (like the AI SDK opencode uses — a server repeating it must not yield
"readread"); arguments are concatenated."""
t0 = clock()
turn = ToolTurn()
calls: dict[int, ToolCall] = {}
current: int | None = None
content: list[str] = []
reasoning: list[str] = []
try:
for ev in events:
now = clock() - t0
tool = ev.is_tool or ev.tc_index is not None
if (ev.content or ev.reasoning or tool) and turn.ttft_s is None:
turn.ttft_s = now
if ev.content:
content.append(ev.content)
if ev.reasoning:
reasoning.append(ev.reasoning)
if tool:
if turn.t_first_tool_s is None:
turn.t_first_tool_s = now
if ev.tc_index is not None:
key = ev.tc_index
elif (
current is None
or (ev.tc_id and calls[current].id not in ("", ev.tc_id))
or (not ev.tc_id and ev.tc_name and calls[current].name not in ("", ev.tc_name))
):
key = max(calls, default=-1) + 1 # index-less server: new id → new call
else:
key = current
current = key
tc = calls.setdefault(key, ToolCall(key, "", "", ""))
if ev.tc_id and not tc.id:
tc.id = ev.tc_id
if ev.tc_name and not tc.name:
tc.name = ev.tc_name
if ev.tc_args:
tc.arguments += ev.tc_args
if ev.finish_reason:
turn.finish_reason = ev.finish_reason
if ev.prompt_tokens is not None:
turn.prompt_tokens = ev.prompt_tokens
if ev.completion_tokens is not None:
turn.completion_tokens = ev.completion_tokens
if ev.reasoning_tokens is not None:
turn.reasoning_tokens = ev.reasoning_tokens
except Exception as e: # connection drop / server error mid-stream: keep what arrived
turn.error, turn.rejected = classify_exception(e)
turn.e2e_s = clock() - t0
turn.content = "".join(content)
turn.reasoning = "".join(reasoning)
turn.tool_calls = [calls[i] for i in sorted(calls)]
return turn
# --------------------------------------------------------------------------- schema check
_JSON_TYPES: dict[str, tuple[type, ...]] = {
"string": (str,),
"integer": (int,),
"number": (int, float),
"boolean": (bool,),
"object": (dict,),
"array": (list,),
}
def unknown_keys(args: Any, schema: dict[str, Any]) -> list[str]:
"""Argument keys the schema doesn't declare. Reported separately, not a schema failure:
opencode's schema decoding most likely ignores extra keys (not proven)."""
if not isinstance(args, dict):
return []
return [k for k in args if k not in schema.get("properties", {})]
def validate_args(args: Any, schema: dict[str, Any]) -> list[str]:
"""Minimal JSON-schema check for flat tool parameters: object, required keys, types,
enums. Unknown keys are skipped here (see ``unknown_keys``). Empty list = valid."""
if not isinstance(args, dict):
return [f"arguments not an object ({type(args).__name__})"]
problems: list[str] = []
props: dict[str, Any] = schema.get("properties", {})
for req in schema.get("required", []):
if req not in args:
problems.append(f"missing required {req!r}")
for k, v in args.items():
if k not in props:
continue
spec = props[k]
t = spec.get("type")
if isinstance(t, str) and t in _JSON_TYPES:
ok = isinstance(v, _JSON_TYPES[t])
if t in ("integer", "number") and isinstance(v, bool):
ok = False # bool is an int subclass in Python, not in JSON
if t == "integer" and isinstance(v, float) and v.is_integer():
ok = True # 120000.0 is a JSON integer value
if not ok:
problems.append(f"{k!r}: expected {t}, got {type(v).__name__}")
if "enum" in spec and v not in spec["enum"]:
problems.append(f"{k!r}: {v!r} not in enum")
return problems
def parse_args(call: ToolCall) -> tuple[Any, str]:
if not call.arguments.strip():
return None, "empty arguments"
try:
return json.loads(call.arguments), ""
except json.JSONDecodeError as e:
return None, f"invalid JSON: {e}"
# --------------------------------------------------------------------------- code execution
def _sandbox_profile(workdir: str) -> str:
# macOS seatbelt: no network, writes only inside the per-check temp dir (realpath — /var is
# /private/var). Reads stay allowed (interpreter + stdlib).
return (
"(version 1)(allow default)(deny network*)(deny file-write*)"
f'(allow file-write* (subpath "{workdir}"))'
)
def _exec_untrusted(argv: list[str], workdir: str, timeout_s: float) -> tuple[bool, str]:
"""Run model-written code unattended without letting it hurt the host: sandboxed where
``sandbox-exec`` exists, minimal env, no stdin, own process group killed on timeout."""
real = str(Path(workdir).resolve())
bin_dirs = sorted({str(Path(argv[0]).parent), "/usr/bin", "/bin"})
env = {"HOME": real, "PATH": ":".join(bin_dirs), "LANG": "en_US.UTF-8", "TMPDIR": real}
sbx = shutil.which("sandbox-exec")
cmd = [sbx, "-p", _sandbox_profile(real), *argv] if sbx else argv
proc = subprocess.Popen(
cmd, cwd=real, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, text=True, errors="replace", start_new_session=True,
) # fmt: skip
try:
out, err = proc.communicate(timeout=timeout_s)
except subprocess.TimeoutExpired:
with contextlib.suppress(ProcessLookupError):
os.killpg(proc.pid, signal.SIGKILL)
# A grandchild that escaped the group (setsid) may still hold the pipes — don't wait on it.
with contextlib.suppress(subprocess.TimeoutExpired):
proc.communicate(timeout=5)
for pipe in (proc.stdout, proc.stderr):
if pipe is not None:
pipe.close()
return False, f"timeout after {timeout_s:.0f}s"
if proc.returncode == 0:
return True, ""
tail = (err or out).strip().splitlines()[-3:]
return False, " | ".join(tail)[:400]
def _write(path: Path, text: str) -> None:
# errors="replace": a lone surrogate from a broken \\ud800 escape must not crash the run
path.write_text(text, encoding="utf-8", errors="replace")
def run_python(module_code: str, test: str, timeout_s: float = 20.0) -> tuple[bool, str]:
"""Load model code as a module NOT named __main__ (a demo ``if __name__ == "__main__"``
block must not run), then run the asserts against its namespace."""
with tempfile.TemporaryDirectory(prefix="touchstone-tools-") as d:
_write(Path(d) / "candidate.py", module_code)
_write(
Path(d) / "check.py",
"import runpy\n"
'globals().update(runpy.run_path("candidate.py", run_name="candidate"))\n\n' + test,
)
return _exec_untrusted([sys.executable, "-I", "-B", "check.py"], d, timeout_s)
def node_path() -> str | None:
return shutil.which("node")
def js_syntax_ok(code: str, timeout_s: float = 20.0) -> tuple[bool | None, str]:
node = node_path()
if node is None:
return None, "node not found — not measured"
with tempfile.TemporaryDirectory(prefix="touchstone-tools-") as d:
_write(Path(d) / "main.js", code)
return _exec_untrusted([node, "--check", "main.js"], d, timeout_s)
def run_node(code: str, timeout_s: float = 20.0) -> tuple[bool | None, str]:
node = node_path()
if node is None:
return None, "node not found — not measured"
with tempfile.TemporaryDirectory(prefix="touchstone-tools-") as d:
_write(Path(d) / "check.js", code)
return _exec_untrusted([node, "check.js"], d, timeout_s)
class _TagCounter(html.parser.HTMLParser):
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.counts: dict[str, int] = {}
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
self.counts[tag] = self.counts.get(tag, 0) + 1
def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
self.counts[tag] = self.counts.get(tag, 0) + 1
def html_tag_counts(doc: str) -> dict[str, int]:
p = _TagCounter()
p.feed(doc)
p.close()
return p.counts
# --------------------------------------------------------------------------- checks
@dataclass
class CheckResult:
name: str
ok: bool | None # None = not measured (e.g. node missing) — never counted as pass
detail: str = ""
def _calls_of(turn: ToolTurn, tool: str | None) -> list[ToolCall]:
return [c for c in turn.tool_calls if tool is None or c.name == tool]
def _writes(turn: ToolTurn) -> dict[str, str]:
"""filePath → content of the LAST valid write call per path."""
out: dict[str, str] = {}
for c in _calls_of(turn, "write"):
a, _ = parse_args(c)
if isinstance(a, dict) and isinstance(a.get("filePath"), str):
content = a.get("content")
if isinstance(content, str):
out[a["filePath"]] = content
return out
def _trimmed_match(text: str, old: str) -> bool:
"""Would a line-trimmed comparison (one of opencode's fallback strategies) find oldString?"""
want = [ln.strip() for ln in old.strip("\n").split("\n")]
have = [ln.strip() for ln in text.split("\n")]
n = len(want)
return n > 0 and any(have[i : i + n] == want for i in range(len(have) - n + 1))
def apply_edits(turn: ToolTurn, path: str, original: str) -> tuple[str | None, str]:
"""Apply every edit call on ``path`` in order, with opencode's semantics: oldString must
occur; more than once is an error unless replaceAll. Returns (new_text | None, problem)."""
text = original
edits = []
for c in _calls_of(turn, "edit"):
a, err = parse_args(c)
if isinstance(a, dict) and a.get("filePath") == path:
edits.append(a)
elif err and c.name == "edit":
return None, f"edit call unparseable: {err}"
if not edits:
return None, "no edit call on this path"
for n, a in enumerate(edits, 1):
old, new = a.get("oldString"), a.get("newString")
if not isinstance(old, str) or not isinstance(new, str):
return None, f"edit {n}: oldString/newString missing or not strings"
if old == new:
return None, f"edit {n}: oldString == newString"
if old == "":
return None, f"edit {n}: empty oldString"
k = text.count(old)
if k == 0:
hint = " (whitespace-tolerant match exists — opencode would likely accept)"
return None, f"edit {n}: oldString not found" + (
hint if _trimmed_match(text, old) else ""
)
if k > 1 and a.get("replaceAll") is not True:
return None, f"edit {n}: oldString ambiguous ({k} matches) without replaceAll"
text = text.replace(old, new) if a.get("replaceAll") is True else text.replace(old, new, 1)
return text, ""
def _among(cs: list[ToolCall]) -> str:
return f" (of {len(cs)} call(s))" if len(cs) > 1 else ""
def run_check(check: Check, turn: ToolTurn, item: ToolItem, schemas: dict[str, Any]) -> CheckResult:
name = check.name()
t = check.type
if t == "calls":
# Only COMPLETE calls count (arguments parse to an object) — a call truncated at the
# budget edge arrives with a name but empty arguments and must not fill the quota.
all_calls = _calls_of(turn, check.tool)
n = sum(1 for c in all_calls if isinstance(parse_args(c)[0], dict))
ok = (
(check.count is None or n == check.count)
and (check.min is None or n >= check.min)
and (check.max is None or n <= check.max)
)
return CheckResult(name, ok, f"{n} complete of {len(all_calls)} call(s)")
if t == "no_tools":
ok = not turn.tool_calls and bool(turn.content.strip())
return CheckResult(
name, ok, f"{len(turn.tool_calls)} call(s), content={bool(turn.content)}"
)
if t == "schema_valid":
if not turn.tool_calls:
return CheckResult(name, False, "no tool calls")
problems = []
for c in turn.tool_calls:
if c.name not in schemas:
problems.append(f"#{c.index} unknown tool {c.name!r}")
continue
a, err = parse_args(c)
if err:
problems.append(f"#{c.index} {c.name}: {err[:80]}")
continue
problems += [f"#{c.index} {c.name}: {p}" for p in validate_args(a, schemas[c.name])]
return CheckResult(name, not problems, "; ".join(problems)[:400])
if t == "no_empty_args":
empty = [c.index for c in turn.tool_calls if not c.arguments.strip()]
return CheckResult(name, not empty, f"empty: {empty}" if empty else "")
if t == "finish_reason":
finishes = turn.finishes or [turn.finish_reason]
return CheckResult(name, check.value in finishes, ", ".join(str(x) for x in finishes))
if t == "call_paths":
paths = []
for c in _calls_of(turn, check.tool):
a, _ = parse_args(c)
if isinstance(a, dict):
paths.append(a.get("filePath"))
want = sorted(check.value or [])
return CheckResult(
name, sorted(p for p in paths if p) == want and len(paths) == len(want), f"got {paths}"
)
if t in ("arg_equals", "arg_regex"):
cs = _calls_of(turn, check.tool)
if not cs:
return CheckResult(name, False, f"no {check.tool} call")
# ANY of the tool's calls may satisfy it: a model that looks around first (ls/find/read)
# and THEN does the real thing is doing what opencode's loop expects, not failing.
# A forbidden pattern (absent=True) is the mirror image: it must hold for EVERY call,
# otherwise a harmless `ls` would excuse the forbidden flag in the call next to it.
if t == "arg_regex" and check.absent:
bad = [
str(a.get(check.key))
for a in (parse_args(c)[0] for c in cs)
if isinstance(a, dict)
and check.key in a
and re.search(check.pattern or "", str(a[check.key]), re.S)
]
return CheckResult(name, not bad, ("forbidden in: " + " | ".join(bad))[:300])
seen: list[str] = []
for c in cs:
a, err = parse_args(c)
if not isinstance(a, dict):
seen.append(err or "arguments not an object")
continue
if check.key not in a:
seen.append(f"key {check.key!r} missing")
continue
v = a[check.key]
if t == "arg_equals":
hit = v == check.value and type(v) is type(check.value)
if isinstance(check.value, int) and isinstance(v, float) and v.is_integer():
hit = int(v) == check.value
else:
hit = bool(re.search(check.pattern or "", str(v), re.S)) != check.absent
if hit:
return CheckResult(name, True, f"got {str(v)[:120]!r}" + _among(cs))
seen.append(f"{str(v)[:120]!r}")
return CheckResult(name, False, "got " + " | ".join(seen)[:300])
if t.startswith("write_"):
writes = _writes(turn)
path = check.path or ""
if path not in writes:
return CheckResult(name, False, f"no valid write to {path} (have: {sorted(writes)})")
body = writes[path]
if t == "write_regex":
ok = bool(re.search(check.pattern or "", body, re.S | re.M))
return CheckResult(name, ok != check.absent, "")
if t == "write_json_equals":
try:
parsed = json.loads(body)
except json.JSONDecodeError as e:
return CheckResult(name, False, f"file is not JSON: {e}")
# Typed comparison: Python's == would accept 0 for false and 8080.0 for 8080.
same = json.dumps(parsed, sort_keys=True) == json.dumps(check.value, sort_keys=True)
return CheckResult(name, same, "" if same else "differs")
if t == "write_python":
ok, det = run_python(body, check.test or "")
return CheckResult(name, ok, det)
if t == "write_html":
counts = html_tag_counts(body)
short = {
k: (counts.get(k, 0), v) for k, v in check.tags.items() if counts.get(k, 0) < v
}
return CheckResult(name, not short, f"too few (have, need): {short}" if short else "")
if t == "write_js_syntax":
ok_js, det = js_syntax_ok(body)
return CheckResult(name, ok_js, det)
if t == "write_check":
ok, det = run_python(f"CONTENT = {body!r}\n", check.test or "")
return CheckResult(name, ok, det)
if t == "write_node":
ok_n, det = run_node(body + "\n\n" + (check.test or ""))
return CheckResult(name, ok_n, det)
if t in ("edit_applies", "edited_python", "edited_regex"):
path = check.path or ""
new, problem = apply_edits(turn, path, item.fixtures[path])
if new is None:
return CheckResult(name, False, problem)
if t == "edit_applies":
return CheckResult(name, True, "")
if t == "edited_python":
ok, det = run_python(new, check.test or "")
return CheckResult(name, ok, det)
ok = bool(re.search(check.pattern or "", new, re.S | re.M))
return CheckResult(name, ok != check.absent, "")
raise ValueError(f"unhandled check type {t}") # pragma: no cover — Literal guards this
def run_checks(item: ToolItem, turn: ToolTurn, schemas: dict[str, Any]) -> list[CheckResult]:
out = []
for c in item.checks:
try:
out.append(run_check(c, turn, item, schemas))
except Exception as e: # odd model output must never crash an unattended run
out.append(
CheckResult(c.name(), False, f"check crashed: {type(e).__name__}: {e}"[:300])
)
return out
# --------------------------------------------------------------------------- result row
@dataclass
class ToolResponse:
pack_id: str
pack_version: int
model: str
quant: str
item_id: str
category: str
repeat: int
passed: bool # every measured check ok and at least one measured
checks_ok: int
checks_measured: int
checks: list[dict[str, Any]]
finish_reason: str | None
truncated: bool # finish_reason == "length"
n_calls: int
n_empty_args: int
tool_calls: list[dict[str, Any]]
content: str
reasoning_chars: int
prompt_tokens: int | None
completion_tokens: int | None
ttft_s: float | None
t_first_tool_s: float | None
e2e_s: float
error: str
t_start: float
rejected: str = "" # deterministic 4xx — counted as a fail, not re-run, not dropped
reasoning_tokens: int | None = None # from usage; None if the server doesn't report it
reasoning_text: str = "" # persisted only when there is no tool call and no content
unknown_keys: list[str] = field(default_factory=list) # "tool.key" not in the schema (info)
n_turns: int = 1 # model turns the item took (agent loop); 1 = answered right away
turn_calls: list[list[str]] = field(default_factory=list) # tool names per turn
def make_response(
pack: ToolsPack,
item: ToolItem,
model: str,
quant: str,
repeat: int,
turn: ToolTurn,
results: list[CheckResult],
t_start: float,
schemas: dict[str, Any] | None = None,
turns: list[ToolTurn] | None = None,
) -> ToolResponse:
extra: list[str] = []
for c in turn.tool_calls:
a, _ = parse_args(c)
extra += [f"{c.name}.{k}" for k in unknown_keys(a, (schemas or {}).get(c.name, {}))]
measured = [r for r in results if r.ok is not None]
ok = sum(1 for r in measured if r.ok)
bare = not turn.tool_calls and not turn.content.strip()
return ToolResponse(
pack_id=pack.id,
pack_version=pack.version,
model=model,
quant=quant,
item_id=item.id,
category=item.category,
repeat=repeat,
passed=bool(measured) and ok == len(measured) and not turn.error and not turn.rejected,
checks_ok=ok,
checks_measured=len(measured),
checks=[asdict(r) for r in results],
finish_reason=turn.finish_reason,
truncated=turn.finish_reason == "length",
n_calls=len(turn.tool_calls),
n_empty_args=sum(1 for c in turn.tool_calls if not c.arguments.strip()),
tool_calls=[asdict(c) for c in turn.tool_calls],
content=turn.content,
reasoning_chars=len(turn.reasoning),
prompt_tokens=turn.prompt_tokens,
completion_tokens=turn.completion_tokens,
ttft_s=turn.ttft_s,
t_first_tool_s=turn.t_first_tool_s,
e2e_s=turn.e2e_s,
error=turn.error,
rejected=turn.rejected,
reasoning_tokens=turn.reasoning_tokens,
t_start=t_start,
reasoning_text=turn.reasoning if bare else "",
unknown_keys=extra,
n_turns=len(turns) if turns else 1,
turn_calls=[[c.name for c in t.tool_calls] for t in (turns or [turn])],
)
def load_tool_responses(path: str | Path) -> list[ToolResponse]:
"""Tolerates a half-written final line (interrupted run). A cell re-run on resume appends
a new line — the LAST line per (model, item, repeat) wins."""
latest: dict[tuple[str, str, int], ToolResponse] = {}
p = Path(path)
if not p.exists():
return []
for line in p.read_text(encoding="utf-8").splitlines():
if not line.strip():
continue
try:
r = ToolResponse(**json.loads(line))
except (json.JSONDecodeError, TypeError):
continue
latest[(r.model, r.item_id, r.repeat)] = r
return list(latest.values())
# --------------------------------------------------------------------------- orchestration
StreamFn = Callable[..., Iterator[ToolStreamEvent]]
PREFLIGHT_PROMPT = "Antworte nur mit OK. Rufe kein Tool auf."
def preflight(stream_fn: StreamFn, model: str, extra_body: dict[str, Any], pack: ToolsPack) -> str:
"""One tiny request with the model's exact extra_body (e.g. reasoning_effort) before the
matrix. Returns "" or the error — a bad knob must fail loudly up front, not per item."""
try:
turn = collect_turn(
stream_fn(
model=model,
messages=[
{"role": "system", "content": pack.system_prompt},
{"role": "user", "content": PREFLIGHT_PROMPT},
],
tools=pack.tools,
max_tokens=pack.max_tokens,
temperature=pack.sampling.temperature,
seed=pack.sampling.seed,
extra_body=extra_body or None,
),
time.perf_counter,
)
except Exception as e:
return f"{type(e).__name__}: {e}"[:500]
return turn.error or turn.rejected
# --------------------------------------------------------------------------- agent loop
_LEAD_ARG = {"bash": "command", "read": "filePath", "write": "filePath", "edit": "filePath"}
NEUTRAL_RESULT = "(exit 0)"
def known_paths(item: ToolItem) -> set[str]:
"""Every path the stand-in world knows: the item's fixtures, its long-context files and the
paths its checks name as write/edit targets. Data the item already carries — nothing invented."""
out = set(item.fixtures) | set(item.context_fixtures)
for ch in item.checks:
if ch.path:
out.add(ch.path)
if ch.type == "call_paths" and isinstance(ch.value, list):
out |= {v for v in ch.value if isinstance(v, str)}
return out
def known_dirs(item: ToolItem) -> set[str]:
dirs: set[str] = set()
for p in known_paths(item):
parts = p.split("/")
for n in range(2, len(parts)):
dirs.add("/".join(parts[:n]))
return dirs
def _listing(item: ToolItem, path: str, recursive: bool) -> str | None:
"""`ls`/`find` answered from the item's own paths. None → the directory is unknown."""
base = (path or "/work/proj").rstrip("/") or "/"
files = sorted(known_paths(item))
if base not in known_dirs(item) and base not in files:
return None
if recursive:
return "\n".join(p for p in files if p.startswith(base + "/") or p == base)
entries: set[str] = set()
for p in files:
if not p.startswith(base + "/"):
continue
rest = p[len(base) + 1 :]
entries.add(rest.split("/")[0] + ("/" if "/" in rest else ""))
return "\n".join(sorted(entries))
def tool_result_for(item: ToolItem, c: ToolCall) -> str:
"""The tool result the item hands back for one call. ``read`` on a fixture returns the real
file (opencode's format); ``read`` on anything else says so. Everything else gets the item's
canned reply if one matches, otherwise a neutral receipt — we never invent file contents."""
args, _ = parse_args(c)
a = args if isinstance(args, dict) else {}
lead = a.get(_LEAD_ARG.get(c.name, ""), "")
for r in item.turn_results:
if r.tool not in (None, c.name):
continue
if r.pattern and not re.search(r.pattern, str(lead), re.S):
continue
return r.output
known = {**item.fixtures, **item.context_fixtures}
if c.name == "read":
path = str(lead)
if path in known:
return read_tool_output(path, known[path])
if path.rstrip("/") in known_dirs(item):
return f"Error: {path} is a directory, not a file"
return f"Error: file not found: {path}"
if c.name == "bash":
# A plain read of a single file is answered from the fixtures. Not a shell simulator:
# only `cat/head/tail <path>` with nothing else on the line qualifies. Measured
# 2026-09-20: with a silent receipt the model keeps looking around and never reaches
# the task — an artefact of the stand-in world, not model behaviour.
m = re.fullmatch(r"\s*(cat|head|tail)(?:\s+-n\s*\d+)?\s+(\S+)\s*", str(lead))
if m:
path = m.group(2).strip("\"'")
if path in known:
return known[path]
return f"{m.group(1)}: {path}: No such file or directory"
cmd = str(lead).strip()
m = re.fullmatch(r"(?:ls|find|tree)((?:\s+-{1,2}[\w-]+)*)(?:\s+(\S+))?(?:\s+-.*)?", cmd)
if m:
path = (m.group(2) or "/work/proj").strip("\"'")
listing = _listing(item, path, recursive=cmd.startswith("find"))
if listing is None:
return f"ls: {path}: No such file or directory"
return listing or "(leer)"
m = re.fullmatch(r"echo\s+(?:-n\s+)?(.*)", cmd)
if m:
return m.group(1).strip("\"'")
if cmd == "pwd":
return "/work/proj"
return NEUTRAL_RESULT
def follow_up_messages(item: ToolItem, turn: ToolTurn) -> list[dict[str, Any]]:
"""The model's own turn echoed back plus one tool result per call — the shape an
OpenAI-compatible server expects before it will continue the conversation."""
calls = [
{
"id": c.id,
"type": "function",
"function": {"name": c.name, "arguments": c.arguments or "{}"},
}
for c in turn.tool_calls
]
assistant: dict[str, Any] = {"role": "assistant", "content": turn.content, "tool_calls": calls}
if turn.reasoning:
# The previous turn's thinking has to come back, or the model plans without memory of
# its own last step — not what the agent under test does. Two field names are in the
# wild and servers are picky about which one they read, so send both:
# `reasoning_content` — read by Hetzner (measured 2026-09-20: 668 vs 343 prompt
# tokens, and the model knew a codeword that existed only in the reasoning).
# `reasoning` — what opencode 1.18.31 actually sends (measured 2026-09-21 on the wire)
# and the only field the LGS-KI chain (all-llama-proxy) reads. With only
# `reasoning_content` every chain there dies silently after two turns — HTTP 200,
# finish_reason stop, empty content — and scores as model failure. Depth pack:
# 1/4 items before, 4/4 after.
assistant["reasoning"] = turn.reasoning
assistant["reasoning_content"] = turn.reasoning
out: list[dict[str, Any]] = [assistant]
for c in turn.tool_calls:
out.append({"role": "tool", "tool_call_id": c.id, "content": tool_result_for(item, c)})
return out
def merge_turns(turns: list[ToolTurn]) -> ToolTurn:
"""Fold an item's turns into the one turn the checks are run against. Calls are concatenated
in order and re-indexed; timings are the first turn's start and the sum of the durations;
``prompt_tokens`` is the LAST turn's (it already covers the whole exchange)."""
if not turns: # pragma: no cover — run_tools always has at least one turn
return ToolTurn()
calls: list[ToolCall] = []
for t in turns:
for c in t.tool_calls:
calls.append(ToolCall(index=len(calls), id=c.id, name=c.name, arguments=c.arguments))
last = turns[-1]
comp = [t.completion_tokens for t in turns if t.completion_tokens is not None]
reas = [t.reasoning_tokens for t in turns if t.reasoning_tokens is not None]
return ToolTurn(
content=last.content,
reasoning="".join(t.reasoning for t in turns),
tool_calls=calls,
finish_reason=last.finish_reason,
prompt_tokens=last.prompt_tokens,
completion_tokens=sum(comp) if comp else None,
reasoning_tokens=sum(reas) if reas else None,
ttft_s=turns[0].ttft_s,
t_first_tool_s=next(
(t.t_first_tool_s for t in turns if t.t_first_tool_s is not None), None
),
e2e_s=sum(t.e2e_s for t in turns),
error=last.error,
rejected=last.rejected,
finishes=[t.finish_reason for t in turns],
)
class ToolsAborted(RuntimeError):
"""Too many consecutive transport errors — the server is gone; stop instead of recording
the rest of the matrix as instant failures (resume redoes the error cells)."""
def run_tools(
pack: ToolsPack,
models: list[tuple[str, str, dict[str, Any]]], # (id, quant, extra_body)
stream_fn: StreamFn,
out_dir: Path,
*,
resume: bool = False,
max_consecutive_errors: int = 3,
clock: Callable[[], float] = time.perf_counter,
wall: Callable[[], float] = time.time,
log: Callable[[str], None] = print,
) -> list[ToolResponse]:
"""Drive model × item × repeat; append each response as it finishes. Resume skips cells
that got an answer; cells that ended in a transport error are re-run (a dead or evicted
server is not a model result). ``max_consecutive_errors`` (0 = off) aborts a run whose
server stopped answering."""
out_dir.mkdir(parents=True, exist_ok=True)
path = out_dir / "responses.jsonl"
prior = load_tool_responses(path) if resume else []
if not resume and path.exists():
path.unlink()
kept = {(r.model, r.item_id, r.repeat): r for r in prior if not r.error}
schemas = pack.tool_schemas()
cells = [(m, it, rep) for m in models for it in pack.items for rep in range(it.repeats)]
todo = [c for c in cells if (c[0][0], c[1].id, c[2]) not in kept]
results = list(kept.values())
streak = 0
for n, ((model, quant, extra_body), item, rep) in enumerate(todo, 1):
t_start = wall()
cap = item.max_tokens if item.max_tokens is not None else pack.max_tokens
msgs = build_messages(pack, item)
turns: list[ToolTurn] = []
for _turn_no in range(max(1, pack.max_turns)):
try:
events = stream_fn(
model=model,
messages=msgs,
tools=pack.tools,
max_tokens=cap,
temperature=pack.sampling.temperature,
seed=pack.sampling.seed,
extra_body=extra_body or None,
)
one = collect_turn(events, clock)
except Exception as e: # request rejected before streaming began
err, rej = classify_exception(e)
one = ToolTurn(error=err, rejected=rej)
turns.append(one)
# Stop on a turn without tool calls (the model is done talking) and on a broken
# exchange — feeding results into a dead stream would only pile up errors.
if not one.tool_calls or one.error or one.rejected:
break
msgs = msgs + follow_up_messages(item, one)
turn = merge_turns(turns)
res = run_checks(item, turn, schemas)
resp = make_response(
pack, item, model, quant, rep, turn, res, t_start, schemas, turns=turns
)
with path.open("a", encoding="utf-8") as f:
f.write(json.dumps(asdict(resp), ensure_ascii=True) + "\n")
results.append(resp)
mark = "✓" if resp.passed else ("✗ ERR" if resp.error else "✗")
log(
f"[{len(cells) - len(todo) + n}/{len(cells)}] {model} {item.id}#{rep} {mark} "
f"{resp.checks_ok}/{resp.checks_measured} calls={resp.n_calls} "
f"finish={resp.finish_reason} tok={resp.completion_tokens} {resp.e2e_s:.0f}s"
+ (f" — {(resp.error or resp.rejected)[:120]}" if resp.error or resp.rejected else "")
)
streak = streak + 1 if resp.error else 0
if max_consecutive_errors and streak >= max_consecutive_errors:
raise ToolsAborted(
f"{streak} consecutive transport errors (last: {resp.error[:200]}) — aborted; "
"resume re-runs the error cells"
)
return results
# --------------------------------------------------------------------------- report / compare
CSV_COLUMNS = [
"model", "quant", "item_id", "category", "repeat", "passed", "checks_ok", "checks_measured",
"finish_reason", "truncated", "n_calls", "n_turns", "n_empty_args", "reasoning_chars", "prompt_tokens",
"completion_tokens", "ttft_s", "t_first_tool_s", "e2e_s", "error", "failed_checks",
] # fmt: skip
def write_results_csv(rows: list[ToolResponse], path: Path) -> None:
with path.open("w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(CSV_COLUMNS)
for r in rows:
failed = "; ".join(c["name"] for c in r.checks if c["ok"] is False)
d = asdict(r)
d["failed_checks"] = failed
w.writerow([d[c] for c in CSV_COLUMNS])
def _summary(rows: list[ToolResponse]) -> dict[str, Any]:
n = len(rows)
return {
"items": n,
"passed": sum(r.passed for r in rows),
"checks_ok": sum(r.checks_ok for r in rows),
"checks_measured": sum(r.checks_measured for r in rows),
"not_measured": sum(1 for r in rows for c in r.checks if c["ok"] is None),
"truncated": sum(r.truncated for r in rows),
"empty_args_calls": sum(r.n_empty_args for r in rows),
"calls": sum(r.n_calls for r in rows),
"errors": sum(1 for r in rows if r.error),
"rejected": sum(1 for r in rows if r.rejected),
"unknown_keys": sum(len(r.unknown_keys) for r in rows),
"completion_tokens": sum(r.completion_tokens or 0 for r in rows),
"e2e_s": sum(r.e2e_s for r in rows),
}
def render_report_md(pack: ToolsPack, rows: list[ToolResponse], meta: dict[str, Any]) -> str:
lines = [f"# Tools-Pack `{pack.id}` v{pack.version} — {pack.title}", ""]
for k, v in meta.items():
lines.append(f"- **{k}:** {v}")
lines.append("")
by_model: dict[str, list[ToolResponse]] = {}
for r in rows:
by_model.setdefault(r.model, []).append(r)
lines += [
"| Modell | Quant | Items bestanden | Checks ok | nicht gemessen | length-Abbrüche "
"| leere arguments | Transportfehler | 4xx abgelehnt | unbek. Keys | Tokens | Laufzeit |",
"|---|---|---|---|---|---|---|---|---|---|---|---|",
]
for m, rs in by_model.items():
s = _summary(rs)
lines.append(
f"| {m} | {rs[0].quant} | {s['passed']}/{s['items']} | "
f"{s['checks_ok']}/{s['checks_measured']} | {s['not_measured']} | {s['truncated']} | "
f"{s['empty_args_calls']}/{s['calls']} | {s['errors']} | {s['rejected']} | "
f"{s['unknown_keys']} | "
f"{s['completion_tokens']} | "
f"{s['e2e_s'] / 60:.0f} min |"
)
lines += ["", "## Je Item", "", "| Modell | Item | Kat. | ✓ | Checks | finish | Calls | "
"Züge | fehlgeschlagen |", "|---|---|---|---|---|---|---|---|---|"] # fmt: skip
for r in rows:
failed = "; ".join(
f"{c['name']} ({c['detail'][:60]})" for c in r.checks if c["ok"] is False
)
lines.append(
f"| {r.model} | {r.item_id}#{r.repeat} | {r.category} | {'✓' if r.passed else '✗'} | "
f"{r.checks_ok}/{r.checks_measured} | {r.finish_reason} | {r.n_calls} | "
f"{'·'.join(','.join(t) or '—' for t in r.turn_calls) or r.n_turns} | "
f"{failed.replace('|', '/')} |"
)
return "\n".join(lines) + "\n"
def _median(xs: list[float]) -> float | None:
xs = sorted(xs)
if not xs:
return None
m = len(xs) // 2
return xs[m] if len(xs) % 2 else (xs[m - 1] + xs[m]) / 2
def side_stats(rows: list[ToolResponse]) -> dict[str, Any]:
"""Per-bundle counters the effort comparison asks for (hypothesis: less effort → fewer
truncated calls on L1). Reasoning tokens come from usage when reported, else ≈ chars/4."""
rt = [r.reasoning_tokens for r in rows if r.reasoning_tokens is not None]
est = not rt
vals = rt if rt else [r.reasoning_chars / 4 for r in rows]
return {
"items": len(rows),
"passed": sum(r.passed for r in rows),
"truncated_length": sum(r.truncated for r in rows),
"empty_args_calls": sum(r.n_empty_args for r in rows),
"calls": sum(r.n_calls for r in rows),
# Züge sind der Preis der Umsicht: in opencode kostet jeder einen weiteren Request.
"median_turns": _median([float(r.n_turns) for r in rows]),
"items_multi_turn": sum(1 for r in rows if r.n_turns > 1),
"rejected": sum(1 for r in rows if r.rejected),
"transport_errors": sum(1 for r in rows if r.error),
"median_reasoning_tokens": _median([float(v) for v in vals]),
"reasoning_tokens_estimated": est,
"median_completion_tokens": _median(
[float(r.completion_tokens) for r in rows if r.completion_tokens is not None]
),
"l_items": {
r.item_id: {"passed": r.passed, "finish": r.finish_reason, "empty_args": r.n_empty_args}
for r in rows
if r.category == "longctx"
},
}
def mcnemar_exact_p(b: int, c: int) -> float:
"""Two-sided exact McNemar = binomial sign test on the discordant pairs."""
n = b + c
if n == 0:
return 1.0
tail: float = sum(math.comb(n, i) for i in range(0, min(b, c) + 1)) / 2**n
return min(1.0, 2 * tail)
@dataclass
class PairedComparison:
label_a: str
label_b: str
pairs: int
both_pass: int
only_a: int
only_b: int
neither: int
p_mcnemar: float
checks_a: tuple[int, int]
checks_b: tuple[int, int]
per_item: list[tuple[str, bool, bool]]
dropped_errors: list[str] = field(default_factory=list) # pairs with a transport error
stats_a: dict[str, Any] = field(default_factory=dict)
stats_b: dict[str, Any] = field(default_factory=dict)
def compare_bundles(a: list[ToolResponse], b: list[ToolResponse]) -> PairedComparison:
"""Pair by (item_id, repeat). Item pass is binary → discordant pairs + exact McNemar.
Each side must hold exactly one model (else pairs would silently mix). Pairs where either
side hit a transport error are dropped and listed — a dead server is not a model result."""
for side, rows in (("A", a), ("B", b)):
models = {r.model for r in rows}
if len(models) != 1:
raise ValueError(f"bundle {side} must hold exactly one model, has {sorted(models)}")
packs = {(r.pack_id, r.pack_version) for r in rows}
if len(packs) != 1:
raise ValueError(f"bundle {side} mixes pack versions: {sorted(packs)}")
if {(r.pack_id, r.pack_version) for r in a} != {(r.pack_id, r.pack_version) for r in b}:
raise ValueError(
"bundles were run with different pack_id/pack_version — v2 (first turn only) "
"and v3 (agent loop) are not comparable"
)
ka = {(r.item_id, r.repeat): r for r in a}
kb = {(r.item_id, r.repeat): r for r in b}
both = sorted(set(ka) & set(kb))
dropped = [f"{k[0]}#{k[1]}" for k in both if ka[k].error or kb[k].error]
keys = [k for k in both if not (ka[k].error or kb[k].error)]
per = [(f"{k[0]}#{k[1]}", ka[k].passed, kb[k].passed) for k in keys]
only_a = sum(1 for _, x, y in per if x and not y)
only_b = sum(1 for _, x, y in per if y and not x)
ca = (sum(ka[k].checks_ok for k in keys), sum(ka[k].checks_measured for k in keys))
cb = (sum(kb[k].checks_ok for k in keys), sum(kb[k].checks_measured for k in keys))
return PairedComparison(
label_a=a[0].quant or a[0].model if a else "A",
label_b=b[0].quant or b[0].model if b else "B",
pairs=len(keys),
both_pass=sum(1 for _, x, y in per if x and y),
only_a=only_a,
only_b=only_b,
neither=sum(1 for _, x, y in per if not x and not y),
p_mcnemar=mcnemar_exact_p(only_a, only_b),
checks_a=ca,
checks_b=cb,
per_item=per,
dropped_errors=dropped,
stats_a=side_stats(a),
stats_b=side_stats(b),
)
def render_compare_md(cmp: PairedComparison) -> str:
a, b = cmp.label_a, cmp.label_b
lines = [
f"# Paarvergleich Tools-Pack: {a} vs {b}",
"",
f"- Paare (Item × Wiederholung): {cmp.pairs}",
f"- beide bestanden: {cmp.both_pass} · nur {a}: {cmp.only_a} · nur {b}: {cmp.only_b} · "
f"keiner: {cmp.neither}",
f"- exakter McNemar-Test (zweiseitig) auf die {cmp.only_a + cmp.only_b} diskordanten "
f"Paare: p = {cmp.p_mcnemar:.3f}",
f"- Einzel-Checks: {a} {cmp.checks_a[0]}/{cmp.checks_a[1]} · "
f"{b} {cmp.checks_b[0]}/{cmp.checks_b[1]}",
f"- wegen Transportfehler nicht gewertet: {', '.join(cmp.dropped_errors) or 'keine'}",
"",
"## Je Stufe",
"",
"| | Items ✓ | Züge (Median · >1 Zug) | finish=length | leere arguments | 4xx "
"| Transportfehler | Median Reasoning-Token | Median completion |",
"|---|---|---|---|---|---|---|---|---|",
*[
f"| {lab} | {st['passed']}/{st['items']} | "
f"{st['median_turns']} · {st['items_multi_turn']}/{st['items']} | "
f"{st['truncated_length']} | "
f"{st['empty_args_calls']}/{st['calls']} | {st['rejected']} | {st['transport_errors']} | "
f"{st['median_reasoning_tokens']}{' (≈ aus Zeichen/4)' if st['reasoning_tokens_estimated'] else ''} | "
f"{st['median_completion_tokens']} |"
for lab, st in ((a, cmp.stats_a), (b, cmp.stats_b))
],
"",
f"- Langkontext {a}: {cmp.stats_a.get('l_items') or 'n. v.'}",
f"- Langkontext {b}: {cmp.stats_b.get('l_items') or 'n. v.'}",
"",
f"| Item | {a} | {b} |",
"|---|---|---|",
]
for k, x, y in cmp.per_item:
lines.append(f"| {k} | {'✓' if x else '✗'} | {'✓' if y else '✗'} |")
return "\n".join(lines) + "\n"