llama-cpp/examples/llama-eval/agentic_eval.py

#!/usr/bin/env python3
"""Multi-turn agentic coding eval: tools over a real working tree.

The existing code suites are single-shot -- one prompt, one reply, execute,
grade. This one is not. The model is given a small but real repository and a
deliberately narrow tool set, and has to find its way to the defect before it
can fix it. That shape is what puts a run in the tens-of-thousands-of-tokens
range without any artificial padding: the context grows because the model
chose to read things.

Three deliberate constraints:

* No shell, and no way to run the tests. The only feedback channel is `lint`.
  A syntax or type error can be driven out mechanically; a logic error has to
  be reasoned about. Suites that let an agent iterate against the test suite
  measure something closer to search than to understanding.
* Two edit tools, one line-addressed and one content-addressed. Which one a
  model reaches for, and whether it keeps line numbers straight after its own
  earlier edit, is itself a signal.
* Tests never touch the working tree. They are held outside it and copied in
  after the agent has stopped, so they can be neither read nor edited.

Everything here is imported lazily by llama-eval.py, and every dependency it
needs is scoped to the language actually under test, so none of it is
installed for anyone running the other suites.
"""

import html
import json
import os
import re
import shutil
import tempfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Tuple

MAX_READ_LINES = 400
MAX_TOOL_CHARS = 12000
MAX_SEARCH_HITS = 60


@dataclass
class LangSpec:
    """Everything language-specific: how to lint, how to test, what to install.

    Kept as data so adding a language is a corpus concern rather than a harness
    change, and so a run only ever provisions the toolchain for the language it
    is actually evaluating.
    """
    name: str
    exts: Tuple[str, ...]
    packages: Tuple[str, ...] = ()
    provision: Tuple[str, ...] = ()
    env: Dict[str, str] = field(default_factory=dict)
    address_space_limit: Optional[int] = 4 * 1024 * 1024 * 1024

    def source_files(self, root: Path) -> List[Path]:
        return sorted(p for p in root.rglob("*")
                      if p.is_file() and p.suffix in self.exts)


PYTHON = LangSpec(
    name="python",
    exts=(".py",),
    packages=("ruff==0.14.2", "mypy==1.18.2"),
)

TYPESCRIPT = LangSpec(
    name="typescript",
    exts=(".ts",),
    address_space_limit=16 * 1024 * 1024 * 1024,
    provision=(
        "import os, subprocess, shutil, sys\n"
        "npm = shutil.which('npm')\n"
        "if not npm: sys.exit('npm not found on PATH; needed for the typescript suite')\n"
        "dest = os.path.join(os.environ['SANDBOX_VENV'], 'node')\n"
        "os.makedirs(dest, exist_ok=True)\n"
        "r = subprocess.run([npm, 'install', '--silent', '--no-audit', '--no-fund',\n"
        "                    '--prefix', dest, 'typescript@5.9.3'],\n"
        "                   capture_output=True, text=True)\n"
        "sys.exit(0 if r.returncode == 0 else (r.stderr or r.stdout)[-2000:])\n",
    ),
)

LANGS: Dict[str, LangSpec] = {"python": PYTHON, "typescript": TYPESCRIPT}


class PathEscape(Exception):
    """A tool was handed a path outside the repository."""


class Workspace:
    """A private, writable copy of one task's repository.

    Every tool path is resolved through here. Confinement is checked on the
    fully-resolved path, so a symlink planted inside the tree cannot be used to
    read or write outside it.
    """

    def __init__(self, root: Path):
        self.root = root.resolve()
        self.escape_attempts = 0

    def resolve(self, rel: str) -> Path:
        if rel is None or not str(rel).strip():
            raise PathEscape("no path given")
        raw = Path(str(rel))
        base = raw if raw.is_absolute() else (self.root / raw)
        candidate = base.resolve()
        if candidate != self.root and self.root not in candidate.parents:
            self.escape_attempts += 1
            raise PathEscape(f"path {rel!r} is outside the repository")
        return candidate

    def rel(self, path: Path) -> str:
        return str(path.relative_to(self.root))

    def files(self) -> List[Path]:
        return sorted(p for p in self.root.rglob("*")
                      if p.is_file() and "__pycache__" not in p.parts)

    def snapshot(self) -> Dict[str, str]:
        out = {}
        for p in self.files():
            try:
                out[self.rel(p)] = p.read_text()
            except (UnicodeDecodeError, OSError):
                pass
        return out


def _clip(text: str, limit: int = MAX_TOOL_CHARS) -> str:
    if len(text) <= limit:
        return text
    return text[:limit] + f"\n... [truncated, {len(text) - limit} more characters]"


def _numbered(lines: Sequence[str], start: int) -> str:
    width = len(str(start + len(lines) - 1))
    return "\n".join(f"{start + i:>{width}}\t{ln}" for i, ln in enumerate(lines))


def tool_schemas() -> List[Dict[str, Any]]:
    """OpenAI-format function schemas for the whole tool set."""
    def fn(name, desc, props, required=()):
        return {"type": "function", "function": {
            "name": name, "description": desc,
            "parameters": {"type": "object", "properties": props,
                           "required": list(required)}}}

    return [
        fn("list_files", "List files in the repository.",
           {"path": {"type": "string",
                     "description": "Directory relative to the repository root. "
                                    "Omit for the whole tree."}}),
        fn("read_file",
           "Read a file. Output is prefixed with 1-based line numbers. Give "
           "start_line/end_line to read only part of a large file.",
           {"path": {"type": "string", "description": "File path relative to the repository root."},
            "start_line": {"type": "integer", "description": "First line to read, 1-based, inclusive."},
            "end_line": {"type": "integer", "description": "Last line to read, 1-based, inclusive."}},
           ["path"]),
        fn("search",
           "Search the repository with a regular expression. Returns matching "
           "lines with their file and line number.",
           {"pattern": {"type": "string", "description": "Python regular expression."},
            "path": {"type": "string", "description": "Limit the search to this subdirectory."},
            "glob": {"type": "string", "description": "Limit to files matching this glob, e.g. '*.py'."}},
           ["pattern"]),
        fn("edit_lines",
           "Replace an inclusive 1-based line range with new text. Line numbers "
           "refer to the file as it is now, so re-read it after any edit that "
           "changed the number of lines.",
           {"path": {"type": "string", "description": "File path relative to the repository root."},
            "start_line": {"type": "integer", "description": "First line to replace, 1-based, inclusive."},
            "end_line": {"type": "integer", "description": "Last line to replace, 1-based, inclusive."},
            "new_text": {"type": "string", "description": "Replacement text. Use an empty string to delete the range."}},
           ["path", "start_line", "end_line", "new_text"]),
        fn("edit_replace",
           "Replace an exact snippet of text. The snippet must occur exactly "
           "once in the file unless replace_all is set.",
           {"path": {"type": "string", "description": "File path relative to the repository root."},
            "old_text": {"type": "string", "description": "Exact text to find, including indentation."},
            "new_text": {"type": "string", "description": "Replacement text."},
            "replace_all": {"type": "boolean", "description": "Replace every occurrence instead of requiring exactly one."}},
           ["path", "old_text", "new_text"]),
        fn("write_file",
           "Write a whole file, creating or overwriting it. Prefer the edit "
           "tools for changes to existing files.",
           {"path": {"type": "string", "description": "File path relative to the repository root."},
            "content": {"type": "string", "description": "Full new file contents."}},
           ["path", "content"]),
        fn("lint",
           "Run the language's linter and type checker over the repository and "
           "return the diagnostics. This is the only feedback available; the "
           "test suite cannot be run.",
           {"path": {"type": "string", "description": "Limit to this file or directory."}}),
        fn("finish",
           "Declare the task complete and end the session. This is the only "
           "way to end it; call this once the change is made.",
           {"summary": {"type": "string", "description": "Brief description of the change made."}}),
    ]


class ToolBox:
    """Executes tool calls against one workspace. Returns plain text results.

    Failures are reported as ordinary text rather than raised, because how a
    model recovers from a rejected edit is part of what is being measured.
    """

    def __init__(self, ws: Workspace, linter=None):
        self.ws = ws
        self.linter = linter
        self.calls = 0
        self.errors = 0
        self.edits = 0
        self.finished = False
        self.finish_summary: Optional[str] = None
        self.per_tool: Dict[str, int] = {}

    def dispatch(self, name: str, args: Dict[str, Any]) -> str:
        self.calls += 1
        self.per_tool[name] = self.per_tool.get(name, 0) + 1
        handler = getattr(self, f"_t_{name}", None)
        if handler is None:
            result = f"Error: unknown tool {name!r}."
        else:
            try:
                result = handler(args)
            except PathEscape as e:
                result = f"Error: {e}"
            except KeyError as e:
                result = f"Error: missing required argument {e}."
            except Exception as e:
                result = f"Error: {type(e).__name__}: {e}"
        if result.startswith("Error:"):
            self.errors += 1
        return _clip(result)

    def _t_list_files(self, a):
        base = self.ws.resolve(a.get("path") or ".")
        if not base.exists():
            return f"Error: {a.get('path')!r} does not exist."
        paths = [p for p in self.ws.files() if p == base or base in p.parents]
        if not paths:
            return "(no files)"
        return "\n".join(f"{self.ws.rel(p)}\t{p.stat().st_size} bytes" for p in paths)

    def _t_read_file(self, a):
        path = self.ws.resolve(a["path"])
        if not path.is_file():
            return f"Error: {a['path']!r} is not a file."
        lines = path.read_text().splitlines()
        start = max(1, int(a.get("start_line") or 1))
        end = int(a.get("end_line") or len(lines))
        end = min(end, len(lines))
        if start > len(lines):
            return f"Error: {a['path']!r} has only {len(lines)} lines."
        window = lines[start - 1:end]
        note = ""
        if len(window) > MAX_READ_LINES:
            window = window[:MAX_READ_LINES]
            note = (f"\n... [showing {MAX_READ_LINES} of {end - start + 1} requested "
                    f"lines; re-read with start_line/end_line for the rest]")
        header = f"{a['path']} ({len(lines)} lines total)\n"
        return header + _numbered(window, start) + note

    def _t_search(self, a):
        try:
            rx = re.compile(a["pattern"])
        except re.error as e:
            return f"Error: bad regular expression: {e}"
        base = self.ws.resolve(a.get("path") or ".")
        glob = a.get("glob")
        hits, scanned = [], 0
        for p in self.ws.files():
            if not (p == base or base in p.parents):
                continue
            if glob and not p.match(glob):
                continue
            try:
                text = p.read_text()
            except (UnicodeDecodeError, OSError):
                continue
            scanned += 1
            for i, line in enumerate(text.splitlines(), 1):
                if rx.search(line):
                    hits.append(f"{self.ws.rel(p)}:{i}:{line.strip()[:200]}")
                    if len(hits) >= MAX_SEARCH_HITS:
                        return ("\n".join(hits) +
                                f"\n... [stopped at {MAX_SEARCH_HITS} matches; narrow the pattern]")
        if not hits:
            return f"No matches for {a['pattern']!r} in {scanned} file(s)."
        return "\n".join(hits)

    def _t_edit_lines(self, a):
        path = self.ws.resolve(a["path"])
        if not path.is_file():
            return f"Error: {a['path']!r} is not a file."
        lines = path.read_text().splitlines()
        start, end = int(a["start_line"]), int(a["end_line"])
        if start < 1 or start > len(lines):
            return (f"Error: start_line {start} out of range; "
                    f"{a['path']!r} has {len(lines)} lines.")
        if end < start or end > len(lines):
            return (f"Error: end_line {end} out of range; must be between "
                    f"{start} and {len(lines)}.")
        new = str(a["new_text"]).splitlines()
        path.write_text("\n".join(lines[:start - 1] + new + lines[end:]) + "\n")
        self.edits += 1
        delta = len(new) - (end - start + 1)
        return (f"Replaced lines {start}-{end} of {a['path']} "
                f"({end - start + 1} -> {len(new)} lines, {delta:+d}). "
                f"Later line numbers have shifted." if delta else
                f"Replaced lines {start}-{end} of {a['path']}.")

    def _t_edit_replace(self, a):
        path = self.ws.resolve(a["path"])
        if not path.is_file():
            return f"Error: {a['path']!r} is not a file."
        text = path.read_text()
        old, new = str(a["old_text"]), str(a["new_text"])
        if not old:
            return "Error: old_text must not be empty."
        n = text.count(old)
        if n == 0:
            return (f"Error: old_text not found in {a['path']}. It must match "
                    f"exactly, including indentation.")
        if n > 1 and not a.get("replace_all"):
            return (f"Error: old_text occurs {n} times in {a['path']}. Include "
                    f"more surrounding context to make it unique, or set "
                    f"replace_all.")
        path.write_text(text.replace(old, new))
        self.edits += 1
        return f"Replaced {n} occurrence(s) in {a['path']}."

    def _t_write_file(self, a):
        path = self.ws.resolve(a["path"])
        path.parent.mkdir(parents=True, exist_ok=True)
        existed = path.is_file()
        path.write_text(str(a["content"]))
        self.edits += 1
        return f"{'Overwrote' if existed else 'Created'} {a['path']}."

    def _t_lint(self, a):
        if self.linter is None:
            return "Error: no linter is configured for this task."
        target = a.get("path")
        if target:
            self.ws.resolve(target)
        return self.linter(target)

    def _t_finish(self, a):
        self.finished = True
        self.finish_summary = a.get("summary")
        return "Task marked complete."


PY_TEST_RUNNER = r'''
import json, os, random, sys, unittest
sys.path.insert(0, os.getcwd())
random.seed(0)

class Collect(unittest.TestResult):
    def __init__(self):
        super().__init__()
        self.outcomes = {}
    def addSuccess(self, t): self.outcomes[t.id()] = "pass"
    def addFailure(self, t, e): self.outcomes[t.id()] = "fail"
    def addError(self, t, e): self.outcomes[t.id()] = "error"
    def addSkip(self, t, r): self.outcomes[t.id()] = "skip"
    def addExpectedFailure(self, t, e): self.outcomes[t.id()] = "pass"
    def addUnexpectedSuccess(self, t): self.outcomes[t.id()] = "fail"

def flatten(s):
    for t in s:
        if isinstance(t, unittest.TestSuite):
            yield from flatten(t)
        else:
            yield t

out = {"tests": {}}
try:
    suite = unittest.TestLoader().discover(os.environ.get("TEST_DIR", "tests"),
                                           top_level_dir=".")
    res = Collect()
    for t in flatten(suite):
        tid = t.id()
        try:
            t.run(res)
        except Exception:
            pass
        res.outcomes.setdefault(tid, "error")
    out["tests"] = res.outcomes
except Exception as e:
    out["error"] = f"{type(e).__name__}: {e}"

print("###RESULT###")
print(json.dumps(out))
'''


def _node_roots() -> List[Path]:
    """Directories that must stay visible for node to start under bwrap."""
    node = shutil.which("node")
    if not node:
        return []
    real = Path(node).resolve()
    return [Path(node).parent.parent, real.parent.parent]


class Runner:
    """Runs the linter and the hidden tests for one language, in the sandbox."""

    def __init__(self, sandbox, lang: LangSpec, timeout: float = 90.0):
        self.sandbox = sandbox
        self.lang = lang
        self.timeout = timeout

    def _binds(self) -> List[Path]:
        return _node_roots() if self.lang.name == "typescript" else []

    def _env(self) -> Dict[str, str]:
        if self.lang.name != "typescript":
            return {}
        node = shutil.which("node")
        return {"PATH": f"{Path(node).parent}:/usr/bin:/bin"} if node else {}

    def lint(self, workdir: Path, target: Optional[str] = None) -> str:
        if self.lang.name == "python":
            return self._lint_python(workdir, target)
        return self._lint_typescript(workdir, target)

    def _lint_python(self, workdir: Path, target: Optional[str]) -> str:
        venv = self.sandbox.venv_dir
        tgt = target or "."
        chunks = []
        r = self.sandbox.run_argv(
            [str(venv / "bin" / "ruff"), "check", "--no-cache",
             "--output-format", "concise", tgt],
            workdir, timeout=self.timeout)
        text = (r.stdout or r.stderr).strip()
        if text and "All checks passed" not in text:
            chunks.append(text)
        r = self.sandbox.run_argv(
            [str(venv / "bin" / "mypy"), "--no-color-output", "--no-error-summary",
             "--cache-dir", "/dev/null", "--ignore-missing-imports", tgt],
            workdir, timeout=self.timeout)
        text = (r.stdout or r.stderr).strip()
        if text and not text.startswith("Success:"):
            chunks.append(text)
        return "\n".join(chunks) if chunks else "No problems found."

    def _lint_typescript(self, workdir: Path, target: Optional[str]) -> str:
        tsc = self.sandbox.venv_dir / "node" / "node_modules" / "typescript" / "bin" / "tsc"
        files = [str(p.relative_to(workdir)) for p in sorted(workdir.rglob("*.ts"))
                 if "node_modules" not in p.parts]
        if target:
            t = (workdir / target).resolve()
            files = [f for f in files if (workdir / f).resolve() == t
                     or t in (workdir / f).resolve().parents]
        if not files:
            return "No TypeScript files to check."
        r = self.sandbox.run_argv(
            ["node", str(tsc), "--noEmit", "--strict", "--target", "es2022",
             "--module", "esnext", "--moduleResolution", "bundler",
             "--allowImportingTsExtensions", *files],
            workdir, timeout=self.timeout, env=self._env(), ro_binds=self._binds())
        out = (r.stdout or "") + (r.stderr or "")
        return out.strip() or "No problems found."

    def test(self, workdir: Path) -> Dict[str, Any]:
        if self.lang.name == "python":
            return self._test_python(workdir)
        return self._test_typescript(workdir)

    def _test_python(self, workdir: Path) -> Dict[str, Any]:
        runner = workdir / "__run_tests.py"
        runner.write_text(PY_TEST_RUNNER)
        r = self.sandbox.run_argv(
            [str(self.sandbox.python), "-I", "-B", str(runner)],
            workdir, timeout=self.timeout, env={"TEST_DIR": "tests"})
        return self._parse_json_result(r)

    def _parse_json_result(self, r) -> Dict[str, Any]:
        if r.status == "timeout":
            return {"tests": {}, "error": "timeout"}
        marker = "###RESULT###"
        if marker not in (r.stdout or ""):
            return {"tests": {},
                    "error": f"no result: {(r.stderr or r.stdout or '')[-400:]}"}
        try:
            return json.loads(r.stdout.split(marker, 1)[1].strip())
        except Exception as e:
            return {"tests": {}, "error": f"unparseable result: {e}"}

    def _test_typescript(self, workdir: Path) -> Dict[str, Any]:
        tests = sorted(p for p in (workdir / "tests").rglob("*.ts")) \
            if (workdir / "tests").is_dir() else []
        if not tests:
            return {"tests": {}, "error": "no test files"}
        r = self.sandbox.run_argv(
            ["node", "--test", "--experimental-strip-types",
             "--test-reporter=tap",
             *[str(p.relative_to(workdir)) for p in tests]],
            workdir, timeout=self.timeout, env=self._env(), ro_binds=self._binds())
        return self._parse_tap(r)

    @staticmethod
    def _parse_tap(r) -> Dict[str, Any]:
        if r.status == "timeout":
            return {"tests": {}, "error": "timeout"}
        text = (r.stdout or "") + "\n" + (r.stderr or "")
        outcomes: Dict[str, str] = {}
        for line in text.splitlines():
            m = re.match(r"\s*(not )?ok\s+\d+\s*-\s*(.+?)\s*$", line)
            if not m:
                continue
            name = m.group(2).strip()
            if name.endswith(" # SKIP") or " # SKIP" in name:
                outcomes[name.split(" # ")[0]] = "skip"
                continue
            verdict = "fail" if m.group(1) else "pass"
            if outcomes.get(name) != "fail":
                outcomes[name] = verdict
        if not outcomes:
            return {"tests": {},
                    "error": f"no TAP output: {text.strip()[-400:]}"}
        return {"tests": outcomes}


SYSTEM_PROMPT = """You are an expert {lang} engineer working in an existing repository.

Resolve the user's request by editing files with the tools provided.

How this session works:
- You are working autonomously. There is no human watching; nobody will answer
  questions, approve a plan, or read progress reports.
- Every reply you send must contain at least one tool call until the task is
  done. A reply that contains only text does nothing: no work happens and the
  turn is wasted. Do not stop to describe what you are about to do; do it.
- The session ends only when you call `finish`. Call it once the change is
  made, and not before. Only the files as you leave them are evaluated.

Important constraints:
- You cannot run the code, the tests, or any shell command. The only feedback
  available is the `lint` tool, which reports syntax and type problems.
- The repository already has a test suite that will be run against your work
  after you finish, but you cannot see it or run it. Do not create test files.
- Read enough of the code to understand it before editing. A change that only
  makes the symptom go away will usually fail the tests.
- Make the smallest correct change. Do not reformat or refactor unrelated code.

Keep calling tools until the change is complete, then call `finish`."""


@dataclass
class AgenticTask:
    task_id: str
    lang: str
    category: str
    difficulty: int
    instruction: str
    files: Dict[str, str]
    tests: Dict[str, str]
    gold: Dict[str, str] = field(default_factory=dict)
    fail_to_pass: List[str] = field(default_factory=list)
    pass_to_pass: List[str] = field(default_factory=list)

    @classmethod
    def from_record(cls, rec: Dict[str, Any]) -> "AgenticTask":
        def as_map(v):
            return v if isinstance(v, dict) else json.loads(v or "{}")
        def as_list(v):
            return v if isinstance(v, list) else json.loads(v or "[]")
        return cls(
            task_id=rec["task_id"], lang=rec["lang"], category=rec["category"],
            difficulty=int(rec.get("difficulty", 3)),
            instruction=rec["instruction"],
            files=as_map(rec["files"]), tests=as_map(rec["tests"]),
            gold=as_map(rec.get("gold")),
            fail_to_pass=as_list(rec.get("fail_to_pass")),
            pass_to_pass=as_list(rec.get("pass_to_pass")),
        )

    @property
    def spec(self) -> LangSpec:
        return LANGS[self.lang]


def materialise(files: Dict[str, str], dest: Path) -> Path:
    """Write a file map out as a real directory tree."""
    dest.mkdir(parents=True, exist_ok=True)
    for rel, content in files.items():
        p = dest / rel
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content)
    return dest


def repo_overview(files: Dict[str, str]) -> str:
    """A compact tree, so a run is not lost merely to not finding the files."""
    lines = []
    for rel in sorted(files):
        n = files[rel].count("\n") + 1
        lines.append(f"  {rel} ({n} lines)")
    return "\n".join(lines)


def user_prompt(task: AgenticTask) -> str:
    return (f"Repository files:\n{repo_overview(task.files)}\n\n"
            f"Task:\n{task.instruction}")


@dataclass
class EpisodeResult:
    stop_reason: str = "unknown"
    turns: int = 0
    tool_calls: int = 0
    tool_errors: int = 0
    edits: int = 0
    escape_attempts: int = 0
    finished: bool = False
    bad_tool_json: int = 0
    nudges: int = 0
    prompt_tokens: int = 0
    completion_tokens: int = 0
    peak_context: int = 0
    truncated_turns: int = 0
    per_tool: Dict[str, int] = field(default_factory=dict)
    error: Optional[str] = None


NUDGE = ("Your last reply contained no tool call, so nothing happened. There is "
         "no human in this session to respond to text. If your change is "
         "complete, call `finish`. Otherwise continue working by calling the "
         "tools; do not reply with text alone.")


def run_episode(task: AgenticTask, toolbox: ToolBox, chat,
                max_turns: int = 50,
                context_limit: Optional[int] = None,
                max_nudges: int = 3) -> Tuple[EpisodeResult, List[Dict]]:
    """Drive one multi-turn tool-calling episode.

    `chat(messages, tools) -> response dict` is supplied by the caller so this
    module stays free of any HTTP concerns.
    """
    tools = tool_schemas()
    messages: List[Dict[str, Any]] = [
        {"role": "system", "content": SYSTEM_PROMPT.format(lang=task.spec.name)},
        {"role": "user", "content": user_prompt(task)},
    ]
    res = EpisodeResult()

    for turn in range(max_turns):
        res.turns = turn + 1
        try:
            reply = chat(messages, tools)
        except Exception as e:
            res.stop_reason, res.error = "request_failed", f"{type(e).__name__}: {e}"
            break

        usage = reply.get("usage") or {}
        pt = int(usage.get("prompt_tokens") or 0)
        ct = int(usage.get("completion_tokens") or 0)
        res.prompt_tokens += pt
        res.completion_tokens += ct
        res.peak_context = max(res.peak_context, pt + ct)

        choices = reply.get("choices") or []
        if not choices:
            res.stop_reason, res.error = "no_choices", json.dumps(reply)[:300]
            break
        msg = choices[0].get("message") or {}
        if choices[0].get("finish_reason") == "length":
            res.truncated_turns += 1
        calls = msg.get("tool_calls") or []

        reasoning = msg.get("reasoning_content")
        messages.append({
            "role": "assistant",
            "content": msg.get("content") or "",
            **({"reasoning_content": reasoning} if reasoning else {}),
            **({"tool_calls": calls} if calls else {}),
        })

        if not calls:
            if toolbox.finished:
                res.stop_reason = "finished"
                break
            if res.nudges >= max_nudges:
                res.stop_reason = "no_tool_calls"
                break
            res.nudges += 1
            messages.append({"role": "user", "content": NUDGE})
            continue

        for call in calls:
            fn = call.get("function") or {}
            name = fn.get("name") or ""
            raw = fn.get("arguments")
            if isinstance(raw, dict):
                args = raw
            else:
                try:
                    args = json.loads(raw or "{}")
                except (json.JSONDecodeError, TypeError):
                    res.bad_tool_json += 1
                    args = None
            output = (toolbox.dispatch(name, args) if args is not None else
                      f"Error: arguments for {name!r} were not valid JSON.")
            messages.append({"role": "tool",
                             "tool_call_id": call.get("id") or f"call_{res.tool_calls}",
                             "content": output})

        if toolbox.finished:
            res.stop_reason = "finished"
            break
        if context_limit and res.peak_context > context_limit:
            res.stop_reason = "context_limit"
            break
    else:
        res.stop_reason = "max_turns"

    res.tool_calls = toolbox.calls
    res.tool_errors = toolbox.errors
    res.edits = toolbox.edits
    res.escape_attempts = toolbox.ws.escape_attempts
    res.finished = toolbox.finished
    res.per_tool = dict(toolbox.per_tool)
    return res, messages


def run_tests_against(files: Dict[str, str], task: AgenticTask, runner: Runner,
                      workroot: Path, label: str) -> Dict[str, str]:
    """Materialise a tree, drop the hidden tests in, and run them.

    The tests are only ever written into this throwaway copy, never into the
    tree the agent worked in, so there is no window in which they could have
    been read or edited.
    """
    dest = workroot / f"grade-{label}"
    shutil.rmtree(dest, ignore_errors=True)
    materialise(files, dest)
    materialise(task.tests, dest)
    return runner.test(dest).get("tests", {})


def score(task: AgenticTask, outcomes: Dict[str, str]) -> Dict[str, Any]:
    """Resolved means every target test passes and nothing previously passing broke."""
    def passing(tid):
        return outcomes.get(tid) == "pass"

    f2p_ok = [t for t in task.fail_to_pass if passing(t)]
    p2p_ok = [t for t in task.pass_to_pass if passing(t)]
    regressions = [t for t in task.pass_to_pass if not passing(t)]
    resolved = (len(f2p_ok) == len(task.fail_to_pass)
                and not regressions and bool(task.fail_to_pass))
    return {
        "resolved": resolved,
        "f2p_passed": len(f2p_ok), "f2p_total": len(task.fail_to_pass),
        "p2p_passed": len(p2p_ok), "p2p_total": len(task.pass_to_pass),
        "regressions": regressions[:10],
        "n_regressions": len(regressions),
        "partial": (len(f2p_ok) / len(task.fail_to_pass)) if task.fail_to_pass else 0.0,
    }


def make_sandbox(lang_name: str, sandbox_cls, cache_root=None, base_python=None):
    """A sandbox carrying only the toolchain for `lang_name`.

    Keyed per language rather than per suite: running the Python tasks must
    never provision node, and vice versa.
    """
    lang = LANGS[lang_name]
    return sandbox_cls(
        name=f"agentic-{lang_name}",
        packages=list(lang.packages),
        provision=list(lang.provision),
        env=dict(lang.env),
        address_space_limit=lang.address_space_limit,
        cache_root=cache_root,
        base_python=base_python,
    )


def run_task(task: AgenticTask, chat, sandbox, workroot: Optional[Path] = None,
             max_turns: int = 50, context_limit: Optional[int] = None,
             test_timeout: float = 90.0) -> Dict[str, Any]:
    """One full attempt: fresh workspace, agent episode, then hidden tests."""
    root = Path(tempfile.mkdtemp(prefix=f"agentic-{task.task_id}-",
                                 dir=str(workroot) if workroot else None))
    try:
        ws_root = root / "repo"
        materialise(task.files, ws_root)
        ws = Workspace(ws_root)
        runner = Runner(sandbox, task.spec, timeout=test_timeout)
        toolbox = ToolBox(ws, linter=lambda t: runner.lint(ws_root, t))

        episode, messages = run_episode(task, toolbox, chat,
                                        max_turns=max_turns,
                                        context_limit=context_limit)
        final = ws.snapshot()
        changed = sorted(k for k in final
                         if final[k] != task.files.get(k)) + \
                  sorted(k for k in task.files if k not in final)

        outcomes = run_tests_against(final, task, runner, root, "final")
        result = score(task, outcomes)
        result.update({
            "task_id": task.task_id, "lang": task.lang,
            "category": task.category, "difficulty": task.difficulty,
            "changed_files": changed,
            "episode": {k: v for k, v in vars(episode).items()},
            "messages": messages,
        })
        return result
    finally:
        shutil.rmtree(root, ignore_errors=True)


TRANSCRIPT_CSS = """
body { font-family: system-ui, sans-serif; margin: 0; padding: 16px; background: #fff; color: #222; max-width: 1100px; }
h1 { font-size: 16px; margin: 0 0 4px; }
.meta { font-size: 13px; color: #555; font-family: 'SF Mono', 'Menlo', 'Consolas', monospace; margin-bottom: 12px; white-space: pre-wrap; }
.msg { border: 1px solid #e1e4e8; border-left-width: 4px; margin: 8px 0; padding: 6px 10px; background: #fafbfc; }
.msg .role { font-size: 12px; font-weight: 600; color: #555; margin-bottom: 4px; }
.system { border-left-color: #888; }
.user { border-left-color: #0969da; }
.nudge { border-left-color: #9a6700; }
.assistant { border-left-color: #1a7f37; background: #fff; }
.tool { border-left-color: #8250df; }
.tool.err { border-left-color: #cf222e; }
pre { font-family: 'SF Mono', 'Menlo', 'Consolas', monospace; font-size: 12px; background: #fff; border: 1px solid #e1e4e8; padding: 6px 8px; margin: 4px 0; white-space: pre-wrap; word-wrap: break-word; }
.reasoning pre { color: #57606a; background: #f6f8fa; }
.call { font-size: 12px; font-weight: 600; color: #8250df; margin-top: 6px; }
details > summary { cursor: pointer; font-size: 12px; color: #555; }
.empty { font-size: 12px; color: #888; font-style: italic; }
"""


def _pre(text: str) -> str:
    return f"<pre>{html.escape(text)}</pre>"


def _render_call(call: Dict[str, Any]) -> str:
    fn = call.get("function") or {}
    raw = fn.get("arguments")
    try:
        args = raw if isinstance(raw, dict) else json.loads(raw or "{}")
        shown = json.dumps(args, indent=2, ensure_ascii=False)
    except (json.JSONDecodeError, TypeError):
        shown = f"(invalid JSON)\n{raw}"
    return (f'<div class="call">&rarr; {html.escape(fn.get("name") or "?")} '
            f'<span style="font-weight:400;color:#888">{html.escape(call.get("id") or "")}</span></div>'
            + _pre(shown))


def render_transcript_html(messages: List[Dict[str, Any]], title: str,
                           meta: str = "") -> str:
    """One page per episode: every turn, with reasoning, tool calls and results."""
    names = {}
    for m in messages:
        for c in m.get("tool_calls") or []:
            names[c.get("id")] = (c.get("function") or {}).get("name") or "?"

    parts = []
    turn = 0
    for m in messages:
        role = m.get("role") or "?"
        content = m.get("content") or ""
        if role == "assistant":
            turn += 1
            body = ""
            if m.get("reasoning_content"):
                body += ('<details class="reasoning" open><summary>reasoning '
                         f'({len(m["reasoning_content"])} chars)</summary>'
                         + _pre(m["reasoning_content"]) + "</details>")
            body += _pre(content) if content.strip() else ""
            body += "".join(_render_call(c) for c in m.get("tool_calls") or [])
            if not body:
                body = '<div class="empty">(empty reply)</div>'
            parts.append(f'<div class="msg assistant"><div class="role">assistant '
                         f'&middot; turn {turn}</div>{body}</div>')
        elif role == "tool":
            name = names.get(m.get("tool_call_id"), "?")
            err = " err" if content.startswith("Error") else ""
            parts.append(f'<div class="msg tool{err}"><div class="role">tool result '
                         f'&middot; {html.escape(name)}</div>{_pre(content)}</div>')
        elif role == "system":
            parts.append('<div class="msg system"><details><summary>system prompt'
                         f'</summary>{_pre(content)}</details></div>')
        else:
            cls = "nudge" if content == NUDGE else role
            label = "user (nudge)" if content == NUDGE else role
            parts.append(f'<div class="msg {html.escape(cls)}"><div class="role">'
                         f'{html.escape(label)}</div>{_pre(content)}</div>')

    return (f'<!DOCTYPE html>\n<html>\n<head>\n<meta charset="UTF-8">\n'
            f'<title>{html.escape(title)}</title>\n<style>{TRANSCRIPT_CSS}</style>\n'
            f'</head>\n<body>\n<h1>{html.escape(title)}</h1>\n'
            f'<div class="meta">{html.escape(meta)}</div>\n'
            + "\n".join(parts) + "\n</body>\n</html>\n")


def write_transcript(messages: List[Dict[str, Any]], dest: Path, title: str,
                     meta: str = "") -> Path:
    """Write `<dest>.json` (the raw message history) and `<dest>.html`."""
    dest.parent.mkdir(parents=True, exist_ok=True)
    json_path = dest.with_name(dest.name + ".json")
    with open(json_path, "w") as f:
        json.dump(messages, f, indent=2, ensure_ascii=False)
    with open(dest.with_name(dest.name + ".html"), "w") as f:
        f.write(render_transcript_html(messages, title, meta))
    return json_path