From 8c51957f0120815653c32f5ccc81d245d268e9d8 Mon Sep 17 00:00:00 2001 From: Alexandru Jircan Date: Mon, 3 Aug 2026 15:18:49 +0300 Subject: [PATCH 1/5] feat(sandbox): generate CLI recording shims via record_cli MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cli_called reads a JSON Lines invocation log, but nothing produced one: every suite had to hand-write a recording mock and get the record shape right, making the format a contract between the harness and each consumer repository. That is how contracts drift — and it drifted inside a single downstream suite, where two of five mock templates were copies that never gained the log. record_cli closes the loop. Declaring a tool generates a self-contained shim into cli_mocks/, PATH-prepended through the existing mock_path_dirs machinery, appending records to cli_mocks/calls.jsonl — which cli_called now reads by default, so a task sets neither mock_path_dirs nor log:. sandbox: record_cli: - {tool: uip, exit_code: 1, stderr: "not connected\n"} - {tool: curl} The shim records the invocation, writes the configured output, and exits. Nothing is executed: no network, no auth, no side effects. It stubs a tool; it does not proxy one, and it serves no per-invocation responses. Recording a REAL executable on the way through depends on the tool being installed, on PATH order, and usually on live credentials — state the harness cannot guarantee — so that stays a hand-written wrapper under mock_path_dirs, as does anything needing a fixture set. Keeping the generated shim to the case that is always well-defined is what lets every record carry a real exit code and keeps the shim free of platform-conditional code. Decisions worth noting: - A .cmd twin ships beside each shim so a bare `uip` also resolves through Windows PATHEXT lookup. - The log is seeded empty: a correct run that calls nothing must satisfy max_count: 0, while a MISSING log (mock never ran) must still fail. - stdin is never read — it would block whenever the sandbox leaves it on an open pipe, hanging the task. - A name collision with a mock_path_dirs entry raises instead of letting directory order silently decide which executable runs. - The rendered shim imports nothing from coder_eval and is pure ASCII: it runs inside a sandbox where this package is not installed. 21 new tests, including the round trip that matters — generate, execute, then grade the produced log with cli_called and no log: configured. Full suite, ruff, pyright and all 166 custom lint rules pass. Co-Authored-By: Claude Opus 5 (1M context) --- docs/TASK_DEFINITION_GUIDE.md | 28 +++ src/coder_eval/cli_recorder.py | 121 ++++++++++++ src/coder_eval/models/__init__.py | 6 + src/coder_eval/models/criteria.py | 10 +- src/coder_eval/models/sandbox.py | 73 +++++++ src/coder_eval/sandbox.py | 86 ++++++++- tests/test_sandbox_record_cli.py | 310 ++++++++++++++++++++++++++++++ 7 files changed, 631 insertions(+), 3 deletions(-) create mode 100644 src/coder_eval/cli_recorder.py create mode 100644 tests/test_sandbox_record_cli.py diff --git a/docs/TASK_DEFINITION_GUIDE.md b/docs/TASK_DEFINITION_GUIDE.md index 1e988237..55f4a120 100644 --- a/docs/TASK_DEFINITION_GUIDE.md +++ b/docs/TASK_DEFINITION_GUIDE.md @@ -17,6 +17,7 @@ Complete reference for defining evaluation tasks in Coder Eval. - [Agent Configuration](#agent-configuration) - [Run Limits](#run-limits) - [Sandbox Configuration](#sandbox-configuration) + - [Recording CLI Invocations](#recording-cli-invocations) - [Template Sources](#template-sources) - [Success Criteria](#success-criteria) - [Continuous Scoring](#continuous-scoring) @@ -513,6 +514,31 @@ Under `driver: tempdir` only `timeout` is enforced — the agent can consume arbitrary host memory, CPU, and PIDs. Use `driver: docker` when you need the container limits above to actually bind. +### Recording CLI Invocations + +`record_cli` shadows executables with generated recording shims, so a task can assert on **what the agent actually ran** without hand-writing a mock: + +```yaml +sandbox: + record_cli: + - tool: uip + exit_code: 1 + stderr: "uip: not connected to a tenant in this sandbox.\n" + - tool: curl # so a disobedient agent cannot reach the network +``` + +Each shim records the invocation, writes the configured `stdout`/`stderr`, and exits with `exit_code`. **Nothing is executed** — no network, no auth, no side effects. + +The sandbox writes the shims into `cli_mocks/` and PATH-prepends that directory, then appends one JSON record per invocation to `cli_mocks/calls.jsonl` — the log [`cli_called`](#cli_called) reads by default. Nothing else to wire: no `mock_path_dirs`, no `template_sources`, no `log:` on the criterion. + +Notes: + +- **A `.cmd` twin** is generated beside each shim so a bare `uip` also resolves through Windows PATHEXT lookup. +- **The log is seeded empty**, so a correct run that legitimately calls nothing still satisfies a `max_count: 0` guard — while a *missing* log (mock never ran, or wrote elsewhere) still fails. +- **stdin is never read** by the shim: reading it would block whenever the sandbox leaves stdin attached to an open pipe, hanging the task. +- **Collisions are rejected.** If a `mock_path_dirs` entry already provides an executable of the same name, setup raises rather than letting directory order decide which one runs. +- **It stubs a tool; it does not proxy one, and it does not serve per-invocation responses.** Recording a *real* executable on the way through, or returning different output per invocation, stays a hand-written mock under `mock_path_dirs` — both depend on state the harness cannot guarantee (the tool being installed, PATH order, live credentials, a fixture set). + ## Template Sources Tasks can start with preset files instead of an empty sandbox. Multiple sources are applied sequentially (last wins for conflicts). @@ -863,6 +889,8 @@ Use this instead of `command_executed` or `file_matches_regex` when a test shado ignore_flags: ["output"] # Flags dropped before matching (default: ["output"]) ``` +`log` defaults to `cli_mocks/calls.jsonl`, where [`sandbox.record_cli`](#recording-cli-invocations) writes — so a task using generated recorders never sets it. Point it elsewhere only when supplying your own mock. + **Log format.** One JSON object per line. Only `argv` is required; `tool` lets one log serve several shadowed executables, and `exit`/`ts` are recorded for reporting rather than matched. Unknown keys are ignored, so a mock may record more. ```json diff --git a/src/coder_eval/cli_recorder.py b/src/coder_eval/cli_recorder.py new file mode 100644 index 00000000..cbfa56ae --- /dev/null +++ b/src/coder_eval/cli_recorder.py @@ -0,0 +1,121 @@ +"""Source template for the CLI recording shims that ``SandboxConfig.record_cli`` generates. + +The rendered script runs INSIDE the sandbox, where ``coder_eval`` is not +installed, so it imports nothing from this package: its configuration arrives as +embedded literals and everything else comes from the standard library. + +Keeping the template here rather than inline in :mod:`coder_eval.sandbox` lets +:func:`render_recorder` be exercised directly (render, execute, read the log) +without standing up a sandbox. +""" + +import json + +from coder_eval.models import RecordedCli + + +# Written beside the shims, inside the generated recorder directory, so the log +# travels with them if the sandbox root moves. +LOG_FILENAME = "calls.jsonl" + +_TEMPLATE = '''\ +#!/usr/bin/env python3 +"""Recording shim for `{tool}` - generated by coder_eval SandboxConfig.record_cli. + +Appends one JSON record per invocation to {log_filename} beside this script, in +the format the `cli_called` success criterion reads. Do not edit: regenerated on +every sandbox setup. +""" + +import json +import os +import sys +import time + +TOOL = {tool!r} +EXIT_CODE = {exit_code!r} +STDOUT_TEXT = {stdout!r} +STDERR_TEXT = {stderr!r} + +SHIM_DIR = os.path.dirname(os.path.abspath(__file__)) +LOG_PATH = os.path.join(SHIM_DIR, {log_filename!r}) + + +def record(argv, exit_code): + """Append this invocation to the log. + + Best-effort: a logging failure must never break the command the agent ran, + which would turn an evidence problem into a behaviour problem. + + argv is stored as a LIST. A space-joined string cannot distinguish + `--flag "two words"` from two arguments, which is the whole reason this log + exists instead of a flattened command line. stdin is deliberately never + read: it would block whenever the sandbox leaves it on an open pipe, and in + passthrough mode it would consume the payload the real tool needs. + """ + entry = {{ + "ts": round(time.time(), 3), + "tool": TOOL, + "argv": list(argv), + "exit": exit_code, + }} + try: + # ensure_ascii escapes non-ASCII and any stray surrogate from + # undecodable argv bytes, so an exotic argument cannot make this write + # raise and silently drop the record. + with open(LOG_PATH, "a", encoding="utf-8", newline="\\n") as handle: + handle.write(json.dumps(entry) + "\\n") + except OSError: + pass + + +def main(argv): + """Record the invocation, then fail like the tool would with nothing behind it. + + Nothing is executed: no network, no auth, no side effects. A test that needs + the real tool's behavior recorded instead should supply its own wrapper under + mock_path_dirs -- proxying a live executable is a different job from stubbing + one, and this shim deliberately does only the second. + """ + record(argv[1:], EXIT_CODE) + if STDOUT_TEXT: + sys.stdout.write(STDOUT_TEXT) + if STDERR_TEXT: + sys.stderr.write(STDERR_TEXT) + return EXIT_CODE + + +if __name__ == "__main__": + sys.exit(main(sys.argv)) +''' + + +def render_recorder(spec: RecordedCli) -> str: + """Render the shim source for one ``record_cli`` entry.""" + return _TEMPLATE.format( + tool=spec.tool, + exit_code=spec.exit_code, + stdout=spec.stdout, + stderr=spec.stderr, + log_filename=LOG_FILENAME, + ) + + +def parse_log(text: str) -> list[dict[str, object]]: + """Parse recorder-log text into records, skipping unparseable lines. + + Shared with tests and any caller that wants the log without duplicating the + JSON-Lines handling in :mod:`coder_eval.criteria.cli_called`. + """ + records: list[dict[str, object]] = [] + for line in text.splitlines(): + stripped = line.strip() + if not stripped: + continue + try: + parsed = json.loads(stripped) + except ValueError: + continue + if isinstance(parsed, dict): + records.append(parsed) + return records diff --git a/src/coder_eval/models/__init__.py b/src/coder_eval/models/__init__.py index ad33fdfd..51504e92 100644 --- a/src/coder_eval/models/__init__.py +++ b/src/coder_eval/models/__init__.py @@ -155,10 +155,13 @@ # Sandbox from coder_eval.models.sandbox import ( + RECORD_CLI_DIR, + RECORD_CLI_LOG, DockerBuildConfig, DockerDriverConfig, NodeEnvConfig, PythonEnvConfig, + RecordedCli, ResourceLimits, SandboxConfig, validate_template_sources_list, @@ -271,6 +274,9 @@ "NodeEnvConfig", "PythonEnvConfig", "SandboxConfig", + "RecordedCli", + "RECORD_CLI_DIR", + "RECORD_CLI_LOG", "ResourceLimits", "validate_template_sources_list", # Telemetry diff --git a/src/coder_eval/models/criteria.py b/src/coder_eval/models/criteria.py index c368ca3b..db980419 100644 --- a/src/coder_eval/models/criteria.py +++ b/src/coder_eval/models/criteria.py @@ -16,6 +16,7 @@ from coder_eval.models.agent_config import AgentConfig, ClaudeCodeAgentConfig, parse_agent_config from coder_eval.models.enums import AgentKind from coder_eval.models.judge_defaults import DEFAULT_JUDGE_MODEL +from coder_eval.models.sandbox import RECORD_CLI_LOG # SECURITY: ignore_patterns floor. The judge's working directory is a copy of @@ -565,7 +566,14 @@ class CliCalledCriterion(BaseSuccessCriterion): """ type: Literal["cli_called"] = "cli_called" - log: str = Field(description="Path to the JSON Lines invocation log, relative to the sandbox working directory") + log: str = Field( + default=RECORD_CLI_LOG, + description=( + "Path to the JSON Lines invocation log, relative to the sandbox working directory. " + f"Defaults to '{RECORD_CLI_LOG}', where SandboxConfig.record_cli writes, so a task using " + "generated recorders never repeats it" + ), + ) verb: str | None = Field( default=None, min_length=1, diff --git a/src/coder_eval/models/sandbox.py b/src/coder_eval/models/sandbox.py index eb7b6dc1..ae72ceaa 100644 --- a/src/coder_eval/models/sandbox.py +++ b/src/coder_eval/models/sandbox.py @@ -307,6 +307,65 @@ def _validate_working_dir(cls, v: str | None) -> str | None: return v +# Sandbox-relative location of the generated CLI recorders and their shared log. +# Not dot-prefixed on purpose: CI artifact upload (actions/upload-artifact) skips +# hidden files, and the log is primary evidence for every `cli_called` criterion, +# so it must survive into the run artifact. +RECORD_CLI_DIR = "cli_mocks" +RECORD_CLI_LOG = f"{RECORD_CLI_DIR}/calls.jsonl" + + +class RecordedCli(BaseModel): + """One executable to shadow with a generated recording shim. + + The shim records the invocation, writes the configured output, and exits — + nothing is executed, so there is no network, no auth, and no side effect. Each + invocation becomes a JSON Lines record in :data:`RECORD_CLI_LOG`, the log the + ``cli_called`` criterion reads by default, so a task asserts on what actually + ran without hand-rolling a mock and without the record shape being a contract + between two repositories. + + It stubs a tool; it does not proxy one. A test that needs a REAL executable's + behavior recorded on the way through still supplies its own wrapper under + ``mock_path_dirs`` — that depends on the tool being installed, on PATH order, + and usually on live credentials, which is a different problem with different + failure modes. + """ + + model_config = ConfigDict(extra="forbid") + + tool: str = Field(description="Executable name to shadow on PATH (e.g. 'uip', 'curl', 'git')") + exit_code: int = Field( + default=1, + description=( + "Exit status the shim returns. Defaults to 1 so an unconfigured tool looks like a failing " + "one rather than silently succeeding" + ), + ) + stdout: str = Field(default="", description="Text the shim writes to stdout") + stderr: str = Field( + default="", + description=( + "Text the shim writes to stderr. Use it to explain the failure the way the real tool " + "would, so an agent reads a plausible error rather than silence" + ), + ) + + @field_validator("tool") + @classmethod + def validate_tool_name(cls, v: str) -> str: + """Reject names that are not a bare filename. + + The shim is written as ``/``; a separator or a + traversal segment would place it outside the managed directory. + """ + if not v or v != v.strip(): + raise ValueError("record_cli tool must be a non-empty name without surrounding whitespace") + if "/" in v or "\\" in v or v in {".", ".."}: + raise ValueError(f"record_cli tool {v!r} must be a bare executable name, not a path") + return v + + class SandboxConfig(BaseModel): """Configuration for the sandboxed execution environment. @@ -360,6 +419,20 @@ class SandboxConfig(BaseModel): ), ) + record_cli: list[RecordedCli] | None = MergeField( + strategy="replace", + default=None, + description=( + "Executables to shadow with a generated recording shim. The sandbox writes each shim " + f"into '{RECORD_CLI_DIR}/' and PATH-prepends that directory, so the agent's calls are " + f"recorded as JSON Lines in '{RECORD_CLI_LOG}' — the log a 'cli_called' criterion reads " + "by default. Use instead of hand-writing a mock under mock_path_dirs when all the test " + "needs is a faithful record of what ran plus a canned exit status and message. It does " + "NOT serve per-invocation responses and does NOT proxy the real executable; supply your " + "own mock for either. Replaced (not merged) across config layers, like mock_path_dirs." + ), + ) + # Customizable ignore patterns ignore_patterns: list[str] = MergeField( strategy="replace", diff --git a/src/coder_eval/sandbox.py b/src/coder_eval/sandbox.py index 07217061..ca459fe9 100644 --- a/src/coder_eval/sandbox.py +++ b/src/coder_eval/sandbox.py @@ -9,7 +9,10 @@ import tempfile from pathlib import Path +from .cli_recorder import render_recorder from .models import ( + RECORD_CLI_DIR, + RECORD_CLI_LOG, RepoSource, SandboxConfig, StarterFilesSource, @@ -203,6 +206,10 @@ def _setup_tempdir(self, target_dir: Path | None = None) -> Path: # Setup template content (repo, directory, or inline files) self._setup_template() + # Generate recording shims for `record_cli` tools (before the +x pass + # below, which also covers them) + self._generate_cli_recorders() + # Mark mock binaries executable so the agent's PATH can shadow real CLIs self._prepare_mock_path_dirs() @@ -442,15 +449,90 @@ def resolved_mock_path_dirs(self) -> list[Path]: Mirrors the ``mount_point`` containment check in :meth:`_apply_template_dir_source`. """ - if self.sandbox_dir is None or not self.config.mock_path_dirs: + if self.sandbox_dir is None: return [] resolved: list[Path] = [] - for rel in self.config.mock_path_dirs: + # Generated recorders go FIRST: `_generate_cli_recorders` refuses to + # generate a shim whose name a user mock dir already provides, so this + # order can never silently shadow a task's own mock — it only fixes which + # directory wins for names the harness itself owns. + if self.config.record_cli: + generated = self._resolve_within_sandbox(RECORD_CLI_DIR, field="record_cli directory") + if generated.is_dir(): + resolved.append(generated) + for rel in self.config.mock_path_dirs or []: candidate = self._resolve_within_sandbox(rel, field="mock_path_dirs entry") if candidate.is_dir(): resolved.append(candidate) return resolved + def _generate_cli_recorders(self) -> None: + """Write a recording shim for every ``SandboxConfig.record_cli`` entry. + + Each shim is a self-contained Python script — it must run inside the + sandbox, where ``coder_eval`` is not installed, so it imports nothing + from this package and carries its configuration as embedded literals. + A ``.cmd`` twin is written beside it so a bare ``uip`` also resolves + through Windows PATHEXT lookup on the tempdir driver. + + Raises: + RuntimeError: a task's own ``mock_path_dirs`` already provides an + executable with the same name. Generating ours anyway would make + which one runs depend on directory order — a silent, confusing + override — so the collision is surfaced instead. + """ + assert self.sandbox_dir is not None, "Sandbox directory not initialized" + if not self.config.record_cli: + return + + for rel in self.config.mock_path_dirs or []: + user_dir = self._resolve_within_sandbox(rel, field="mock_path_dirs entry") + if not user_dir.is_dir(): + continue + for spec in self.config.record_cli: + clash = user_dir / spec.tool + if clash.exists(): + msg = ( + f"record_cli would generate a '{spec.tool}' shim, but mock_path_dirs entry " + f"'{rel}' already provides one ({clash.relative_to(self.sandbox_dir)}). " + "Remove the record_cli entry to keep your own mock, or drop the file to use " + "the generated recorder." + ) + raise RuntimeError(msg) + + recorder_dir = self._resolve_within_sandbox(RECORD_CLI_DIR, field="record_cli directory") + recorder_dir.mkdir(parents=True, exist_ok=True) + + # Seed the log so it always exists: `cli_called` treats a MISSING log as a + # harness fault (score 0 even for a negative guard), which is right when a + # mock never ran, but wrong for a correct run that legitimately called + # nothing. An empty file distinguishes the two. + log_path = self.sandbox_dir / RECORD_CLI_LOG + if not log_path.exists(): + log_path.write_text("", encoding="utf-8") + + for spec in self.config.record_cli: + shim = recorder_dir / spec.tool + shim.write_text(render_recorder(spec), encoding="utf-8", newline="\n") + shim.chmod(shim.stat().st_mode | 0o111) + # `python "%~dp0" %*` — the extensionless script beside this file. + cmd_lines = [ + "@echo off", + "REM Generated by coder_eval SandboxConfig.record_cli.", + "REM Windows PATHEXT lookup resolves this; POSIX uses the extensionless twin.", + f'python "%~dp0{spec.tool}" %*', + ] + (recorder_dir / f"{spec.tool}.cmd").write_text( + "\r\n".join(cmd_lines) + "\r\n", + encoding="utf-8", + newline="", + ) + + logger.info( + f"Generated {len(self.config.record_cli)} CLI recorder(s) in {RECORD_CLI_DIR}/: " + + ", ".join(f"{s.tool}(exit {s.exit_code})" for s in self.config.record_cli) + ) + def _apply_starter_files_source(self, source: StarterFilesSource) -> None: """Create inline starter files in sandbox with overwrite tracking. diff --git a/tests/test_sandbox_record_cli.py b/tests/test_sandbox_record_cli.py new file mode 100644 index 00000000..f5337561 --- /dev/null +++ b/tests/test_sandbox_record_cli.py @@ -0,0 +1,310 @@ +"""Tests for SandboxConfig.record_cli — generated CLI recording shims. + +The load-bearing test here is the ROUND TRIP: generate a shim, actually run it, +then grade the log it wrote with the `cli_called` criterion. Writer and reader +ship in the same package precisely so that contract can be tested, rather than +asserted in prose across two repositories. +""" + +import json +import subprocess +import sys + +import pytest +from pydantic import ValidationError + +from coder_eval.cli_recorder import parse_log, render_recorder +from coder_eval.evaluation.checker import SuccessChecker +from coder_eval.models import ( + RECORD_CLI_DIR, + RECORD_CLI_LOG, + CliCalledCriterion, + RecordedCli, + SandboxConfig, + StarterFile, + StarterFilesSource, +) +from coder_eval.sandbox import Sandbox + + +def _sandbox(task_id: str, **kwargs) -> Sandbox: + config = SandboxConfig(driver="tempdir", python=None, **kwargs) + return Sandbox(config, task_id=task_id) + + +def _run_shim(sandbox_dir, tool: str, args: list[str]) -> subprocess.CompletedProcess: + """Invoke a generated shim the way the agent's shell would.""" + shim = sandbox_dir / RECORD_CLI_DIR / tool + return subprocess.run( + [sys.executable, str(shim), *args], + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + + +class TestGeneration: + def test_generates_shim_cmd_twin_and_seeded_log(self): + sandbox = _sandbox("record_gen", record_cli=[RecordedCli(tool="uip")]) + try: + sandbox_dir = sandbox.setup() + recorder_dir = sandbox_dir / RECORD_CLI_DIR + assert (recorder_dir / "uip").is_file() + # Windows PATHEXT lookup needs the .cmd; POSIX uses the extensionless twin. + assert (recorder_dir / "uip.cmd").is_file() + # Seeded empty: distinguishes "mock never ran" from "correct run made no calls". + log = sandbox_dir / RECORD_CLI_LOG + assert log.is_file() + assert log.read_text(encoding="utf-8") == "" + finally: + sandbox.cleanup(preserve=False) + + def test_recorder_dir_is_path_prepended_before_user_mocks(self): + sandbox = _sandbox("record_path", record_cli=[RecordedCli(tool="uip")], mock_path_dirs=["mocks"]) + try: + sandbox_dir = sandbox.setup() + (sandbox_dir / "mocks").mkdir(exist_ok=True) + resolved = sandbox.resolved_mock_path_dirs + assert resolved[0] == sandbox_dir / RECORD_CLI_DIR + finally: + sandbox.cleanup(preserve=False) + + def test_no_record_cli_leaves_no_directory(self): + sandbox = _sandbox("record_absent") + try: + sandbox_dir = sandbox.setup() + assert not (sandbox_dir / RECORD_CLI_DIR).exists() + assert sandbox.resolved_mock_path_dirs == [] + finally: + sandbox.cleanup(preserve=False) + + def test_collision_with_user_mock_raises(self): + """Silently shadowing a task's own mock would make PATH order load-bearing.""" + sandbox = _sandbox( + "record_clash", + record_cli=[RecordedCli(tool="uip")], + mock_path_dirs=["mocks"], + template_sources=[ + StarterFilesSource( + type="starter_files", + files=[StarterFile(path="mocks/uip", content="#!/bin/sh\nexit 0\n")], + ) + ], + ) + try: + with pytest.raises(RuntimeError, match="already provides one"): + sandbox.setup() + finally: + if sandbox.sandbox_dir is not None: + sandbox.cleanup(preserve=False) + + +class TestRecording: + def test_records_argv_and_fails_without_running_anything(self): + spec = RecordedCli(tool="uip", exit_code=1, stderr="uip: not connected\n") + sandbox = _sandbox("record_offline", record_cli=[spec]) + try: + sandbox_dir = sandbox.setup() + proc = _run_shim( + sandbox_dir, + "uip", + ["ixp", "projects", "configure-model", "proj-1", "--model", "gemini_2_5_pro"], + ) + assert proc.returncode == 1 + assert proc.stderr == "uip: not connected\n" + + records = parse_log((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + assert len(records) == 1 + assert records[0]["tool"] == "uip" + assert records[0]["exit"] == 1 + assert records[0]["argv"] == [ + "ixp", + "projects", + "configure-model", + "proj-1", + "--model", + "gemini_2_5_pro", + ] + finally: + sandbox.cleanup(preserve=False) + + def test_stdout_text_is_emitted(self): + spec = RecordedCli(tool="fake", exit_code=0, stdout='{"Result":"Success"}') + sandbox = _sandbox("record_stdout", record_cli=[spec]) + try: + sandbox_dir = sandbox.setup() + proc = _run_shim(sandbox_dir, "fake", ["anything"]) + assert proc.returncode == 0 + assert proc.stdout == '{"Result":"Success"}' + finally: + sandbox.cleanup(preserve=False) + + def test_quoted_argument_with_spaces_stays_one_element(self): + """The defect a flattened command line cannot represent.""" + sandbox = _sandbox("record_quoted", record_cli=[RecordedCli(tool="uip")]) + try: + sandbox_dir = sandbox.setup() + _run_shim(sandbox_dir, "uip", ["fields", "rename", "--group", "Invoice Header"]) + records = parse_log((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + assert records[0]["argv"][-1] == "Invoice Header" + finally: + sandbox.cleanup(preserve=False) + + def test_multiline_argument_survives_as_one_element(self): + """A heredoc-expanded JSON payload must not split into several records.""" + payload = '[\n {"name": "Invoice Number"}\n]' + sandbox = _sandbox("record_multiline", record_cli=[RecordedCli(tool="uip")]) + try: + sandbox_dir = sandbox.setup() + _run_shim(sandbox_dir, "uip", ["fields", "update-prompts", "--updates", payload]) + records = parse_log((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + assert len(records) == 1 + assert records[0]["argv"][-1] == payload + finally: + sandbox.cleanup(preserve=False) + + def test_repeated_invocations_append_in_order(self): + sandbox = _sandbox("record_append", record_cli=[RecordedCli(tool="uip")]) + try: + sandbox_dir = sandbox.setup() + for n in range(3): + _run_shim(sandbox_dir, "uip", ["documents", "upload", f"doc{n}.pdf"]) + records = parse_log((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + assert [r["argv"][-1] for r in records] == ["doc0.pdf", "doc1.pdf", "doc2.pdf"] + finally: + sandbox.cleanup(preserve=False) + + def test_several_tools_share_one_log_tagged_by_tool(self): + sandbox = _sandbox( + "record_multi", + record_cli=[RecordedCli(tool="uip"), RecordedCli(tool="curl")], + ) + try: + sandbox_dir = sandbox.setup() + _run_shim(sandbox_dir, "uip", ["projects", "list"]) + _run_shim(sandbox_dir, "curl", ["-s", "https://example.invalid"]) + records = parse_log((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + assert [r["tool"] for r in records] == ["uip", "curl"] + finally: + sandbox.cleanup(preserve=False) + + +class TestRoundTripWithCliCalled: + """Generate → run → grade. The contract this feature exists to guarantee.""" + + def test_cli_called_grades_the_generated_log_with_no_log_path_configured(self): + sandbox = _sandbox("record_roundtrip", record_cli=[RecordedCli(tool="uip")]) + try: + sandbox_dir = sandbox.setup() + _run_shim( + sandbox_dir, + "uip", + ["ixp", "projects", "configure-model", "proj-1", "--model", "gemini_2_5_pro", "--output", "json"], + ) + # No `log:` — the default points at where record_cli writes. + criterion = CliCalledCriterion( + description="switched to the capable model", + verb="ixp projects configure-model", + positional=["proj-1"], + flags={"model": "gemini_2_5_pro"}, + ) + result = SuccessChecker(sandbox).check(criterion) + assert result.score == 1.0, result.details + assert result.error is None + finally: + sandbox.cleanup(preserve=False) + + def test_negative_guard_passes_on_a_seeded_empty_log(self): + """A correct run that calls nothing must satisfy max_count: 0 — the seeded + empty log is what separates that from a mock that never ran.""" + sandbox = _sandbox("record_roundtrip_neg", record_cli=[RecordedCli(tool="uip")]) + try: + sandbox.setup() + criterion = CliCalledCriterion( + description="did not delete anything", + verb="ixp projects delete", + min_count=0, + max_count=0, + ) + result = SuccessChecker(sandbox).check(criterion) + assert result.score == 1.0 + assert result.error is None + finally: + sandbox.cleanup(preserve=False) + + def test_negative_guard_catches_the_forbidden_call(self): + sandbox = _sandbox("record_roundtrip_neg2", record_cli=[RecordedCli(tool="uip")]) + try: + sandbox_dir = sandbox.setup() + _run_shim(sandbox_dir, "uip", ["ixp", "projects", "delete", "proj-1", "-y"]) + criterion = CliCalledCriterion( + description="did not delete anything", + verb="ixp projects delete", + min_count=0, + max_count=0, + ) + assert SuccessChecker(sandbox).check(criterion).score == 0.0 + finally: + sandbox.cleanup(preserve=False) + + +class TestModelValidation: + @pytest.mark.parametrize("bad", ["../evil", "a/b", "a\\b", ".", "..", "", " uip"]) + def test_tool_must_be_a_bare_name(self, bad): + with pytest.raises(ValidationError): + RecordedCli(tool=bad) + + @pytest.mark.parametrize("field", ["mode", "response", "passthrough"]) + def test_unknown_field_rejected(self, field): + """extra='forbid' catches a typo — and a config written against a shape + this model does not (yet) have, such as a mode or a canned response.""" + with pytest.raises(ValidationError, match="Extra inputs are not permitted"): + RecordedCli(tool="uip", **{field: "x"}) + + def test_defaults(self): + spec = RecordedCli(tool="uip") + # exit_code 1: an unconfigured tool should look like a failing one rather + # than silently succeeding. + assert spec.exit_code == 1 + assert (spec.stdout, spec.stderr) == ("", "") + + +class TestRenderedSource: + def test_rendered_shim_is_valid_python_and_embeds_config(self): + spec = RecordedCli(tool="uip", exit_code=3, stderr="boom\n") + source = render_recorder(spec) + compile(source, "uip", "exec") + # Config arrives as literals; exec the module to read them back rather + # than pattern-matching the rendered text. + # __file__ must be present: the shim derives its log path from it. + namespace: dict = {"__name__": "shim", "__file__": "uip"} + exec(compile(source, "uip", "exec"), namespace) + assert namespace["TOOL"] == "uip" + assert namespace["EXIT_CODE"] == 3 + assert namespace["STDERR_TEXT"] == "boom\n" + + def test_rendered_shim_does_not_execute_anything(self): + """It stubs a tool rather than proxying one: no subprocess, no exec.""" + source = render_recorder(RecordedCli(tool="uip")) + for forbidden in ("subprocess", "execv", "execvp", "popen", "system("): + assert forbidden not in source + + def test_rendered_shim_imports_nothing_from_coder_eval(self): + """It runs inside the sandbox, where this package is not installed.""" + source = render_recorder(RecordedCli(tool="uip")) + imports = [ + line.strip() + for line in source.splitlines() + if line.strip().startswith(("import ", "from ")) and "coder_eval" in line + ] + assert imports == [] + + def test_rendered_shim_is_pure_ascii(self): + """Written into arbitrary sandboxes and read by whatever python3 is there.""" + source = render_recorder(RecordedCli(tool="uip")) + source.encode("ascii") + + def test_parse_log_skips_unparseable_lines(self): + text = json.dumps({"tool": "uip", "argv": []}) + "\ngarbage\n\n" + assert len(parse_log(text)) == 1 From 27d81fb0ded46527dcf595568bb9ffbfcf17f9ca Mon Sep 17 00:00:00 2001 From: Alexandru Jircan Date: Wed, 5 Aug 2026 12:33:44 +0300 Subject: [PATCH 2/5] fix(sandbox): repair record_cli defects found reviewing #73 Three blockers, all reproduced against the branch before changing anything. 1. `tool` was interpolated unescaped into generated shim source. The validator only rejected path separators, so `tool: 'a"""b'` validated and produced a shim that fails to compile; a crafted name reached executable position. Constrained the field to `^[A-Za-z0-9._+-]+$` and covered quote/newline/space cases in the tests, which had only probed path-shaped names. 2. The recorder log was seeded only `if not log_path.exists()`, so its sole effect was PRESERVING a previous run's log. Under DIRECT_WRITE (the docker default, which deliberately does not clear the target dir) a stale record scored the current run: a `min_count: 1` criterion returned 1.0 with zero agent activity. The log is now truncated unconditionally and the recorder directory wiped before regeneration, so a shim for a tool no longer declared cannot linger on PATH. 3. The collision guard built its message with `clash.relative_to(sandbox_dir)`, comparing a resolved path against an unresolved root. Wherever the sandbox traverses a symlink (macOS /var, a symlinked --run-dir on Linux) that raised ValueError instead of the intended RuntimeError, so the friendly error never existed and the branch was red on macOS. The message no longer computes a relative path, and the path-prepend test compares resolved to resolved -- it had passed only because Windows and Linux tempdirs are not symlinked. Also: parse_log had no production callers while the checker re-implemented the same JSON-Lines loop, so it now returns (usable, unusable_count) and is the single reader. The module is renamed cli_recorder -> invocation_log: it owns both halves now, and CE004's prefix match reads `coder_eval.cli_recorder` as the cli layer. Fixed the guide's `log:` comment, which said "(required)" a line above the paragraph documenting its default. Co-Authored-By: Claude Opus 5 (1M context) --- docs/TASK_DEFINITION_GUIDE.md | 2 +- src/coder_eval/criteria/cli_called.py | 35 +---------- .../{cli_recorder.py => invocation_log.py} | 32 +++++++--- src/coder_eval/models/sandbox.py | 9 ++- src/coder_eval/sandbox.py | 12 ++-- tests/test_sandbox_record_cli.py | 62 ++++++++++++++++--- 6 files changed, 94 insertions(+), 58 deletions(-) rename src/coder_eval/{cli_recorder.py => invocation_log.py} (75%) diff --git a/docs/TASK_DEFINITION_GUIDE.md b/docs/TASK_DEFINITION_GUIDE.md index 55f4a120..523ea51d 100644 --- a/docs/TASK_DEFINITION_GUIDE.md +++ b/docs/TASK_DEFINITION_GUIDE.md @@ -878,7 +878,7 @@ Use this instead of `command_executed` or `file_matches_regex` when a test shado ```yaml - type: "cli_called" description: "Switched the project to the capable model" - log: "mocks/calls.jsonl" # Path to the JSON Lines invocation log (required) + log: "mocks/calls.jsonl" # Invocation log; omit it to use the record_cli default verb: "ixp projects configure-model" # Ordered prefix of the non-flag arguments positional: ["my_invoices-ixp"] # Non-flag arguments following the verb, in order flags: diff --git a/src/coder_eval/criteria/cli_called.py b/src/coder_eval/criteria/cli_called.py index 6e196b4c..a64a17ac 100644 --- a/src/coder_eval/criteria/cli_called.py +++ b/src/coder_eval/criteria/cli_called.py @@ -1,12 +1,12 @@ """CLI-called criterion checker — structured matching over an invocation log.""" -import json import logging import re import shlex from typing import TYPE_CHECKING, Any from coder_eval.criteria.base import BaseCriterion, CheckContext, register_criterion +from coder_eval.invocation_log import parse_log from coder_eval.models import CliCalledCriterion, CriterionResult, FlagMatch @@ -100,19 +100,6 @@ def _flag_matches(predicate: FlagMatch, values: list[str] | None) -> bool: raise AssertionError(f"FlagMatch has no matcher arm: {predicate!r}") -def _usable_argv(record: dict[str, Any]) -> list[str] | None: - """The record's ``argv`` when it is a list of strings, else None. - - None means the record cannot be evaluated at all — a different thing from - "evaluated and did not match", which is why the caller reports it rather than - quietly treating it as a non-match. - """ - argv = record.get("argv") - if isinstance(argv, list) and all(isinstance(item, str) for item in argv): - return argv - return None - - def _record_matches(criterion: CliCalledCriterion, argv: list[str], record: dict[str, Any]) -> bool: """Whether one log record satisfies every configured facet of the criterion.""" if criterion.tool is not None and record.get("tool") != criterion.tool: @@ -207,25 +194,7 @@ def _check_impl( content = sandbox.get_file_content(criterion.log) - usable: list[tuple[list[str], dict[str, Any]]] = [] - unusable = 0 - for line in content.splitlines(): - stripped = line.strip() - if not stripped: - continue - try: - parsed = json.loads(stripped) - except ValueError: - unusable += 1 - continue - if not isinstance(parsed, dict): - unusable += 1 - continue - argv = _usable_argv(parsed) - if argv is None: - unusable += 1 - continue - usable.append((argv, parsed)) + usable, unusable = parse_log(content) if unusable: # A record we cannot read might BE the call a max_count: 0 guard diff --git a/src/coder_eval/cli_recorder.py b/src/coder_eval/invocation_log.py similarity index 75% rename from src/coder_eval/cli_recorder.py rename to src/coder_eval/invocation_log.py index cbfa56ae..4739dd8f 100644 --- a/src/coder_eval/cli_recorder.py +++ b/src/coder_eval/invocation_log.py @@ -1,4 +1,8 @@ -"""Source template for the CLI recording shims that ``SandboxConfig.record_cli`` generates. +"""The structured invocation log: the recording shim that writes it, and the reader. + +Named for the artifact rather than the writer because both sides live here -- the +shim template `SandboxConfig.record_cli` renders, and `parse_log`, which the +`cli_called` criterion reads it back with. The rendered script runs INSIDE the sandbox, where ``coder_eval`` is not installed, so it imports nothing from this package: its configuration arrives as @@ -101,13 +105,16 @@ def render_recorder(spec: RecordedCli) -> str: ) -def parse_log(text: str) -> list[dict[str, object]]: - """Parse recorder-log text into records, skipping unparseable lines. +def parse_log(text: str) -> tuple[list[tuple[list[str], dict[str, object]]], int]: + """Parse recorder-log text into ``(usable, unusable_count)``. - Shared with tests and any caller that wants the log without duplicating the - JSON-Lines handling in :mod:`coder_eval.criteria.cli_called`. + A usable entry pairs the record's ``argv`` with the whole record. Unusable + means unparseable, not an object, or an ``argv`` that is not a list of + strings — counted rather than dropped, because a record that cannot be read + might be the very call a negative guard forbids. """ - records: list[dict[str, object]] = [] + usable: list[tuple[list[str], dict[str, object]]] = [] + unusable = 0 for line in text.splitlines(): stripped = line.strip() if not stripped: @@ -115,7 +122,14 @@ def parse_log(text: str) -> list[dict[str, object]]: try: parsed = json.loads(stripped) except ValueError: + unusable += 1 + continue + if not isinstance(parsed, dict): + unusable += 1 continue - if isinstance(parsed, dict): - records.append(parsed) - return records + argv = parsed.get("argv") + if isinstance(argv, list) and all(isinstance(item, str) for item in argv): + usable.append((argv, parsed)) + else: + unusable += 1 + return usable, unusable diff --git a/src/coder_eval/models/sandbox.py b/src/coder_eval/models/sandbox.py index ae72ceaa..10251ecb 100644 --- a/src/coder_eval/models/sandbox.py +++ b/src/coder_eval/models/sandbox.py @@ -334,7 +334,14 @@ class RecordedCli(BaseModel): model_config = ConfigDict(extra="forbid") - tool: str = Field(description="Executable name to shadow on PATH (e.g. 'uip', 'curl', 'git')") + tool: str = Field( + pattern=r"^[A-Za-z0-9._+-]+$", + description=( + "Executable name to shadow on PATH (e.g. 'uip', 'curl', 'git'). Constrained to " + "executable-name characters: the value is interpolated into generated shim source, so a " + "quote or newline would emit a broken script" + ), + ) exit_code: int = Field( default=1, description=( diff --git a/src/coder_eval/sandbox.py b/src/coder_eval/sandbox.py index ca459fe9..1e65c81f 100644 --- a/src/coder_eval/sandbox.py +++ b/src/coder_eval/sandbox.py @@ -9,7 +9,7 @@ import tempfile from pathlib import Path -from .cli_recorder import render_recorder +from .invocation_log import render_recorder from .models import ( RECORD_CLI_DIR, RECORD_CLI_LOG, @@ -494,13 +494,18 @@ def _generate_cli_recorders(self) -> None: if clash.exists(): msg = ( f"record_cli would generate a '{spec.tool}' shim, but mock_path_dirs entry " - f"'{rel}' already provides one ({clash.relative_to(self.sandbox_dir)}). " + f"'{rel}' already provides one ({rel}/{spec.tool}). " "Remove the record_cli entry to keep your own mock, or drop the file to use " "the generated recorder." ) raise RuntimeError(msg) recorder_dir = self._resolve_within_sandbox(RECORD_CLI_DIR, field="record_cli directory") + # Wipe rather than reuse: DIRECT_WRITE (the docker default) does not clear the + # target dir, so a reused --run-dir would leave a previous run's log to be + # scored as this run's, and stale shims for tools no longer declared on PATH. + if recorder_dir.exists(): + shutil.rmtree(recorder_dir, ignore_errors=True) recorder_dir.mkdir(parents=True, exist_ok=True) # Seed the log so it always exists: `cli_called` treats a MISSING log as a @@ -508,8 +513,7 @@ def _generate_cli_recorders(self) -> None: # mock never ran, but wrong for a correct run that legitimately called # nothing. An empty file distinguishes the two. log_path = self.sandbox_dir / RECORD_CLI_LOG - if not log_path.exists(): - log_path.write_text("", encoding="utf-8") + log_path.write_text("", encoding="utf-8") for spec in self.config.record_cli: shim = recorder_dir / spec.tool diff --git a/tests/test_sandbox_record_cli.py b/tests/test_sandbox_record_cli.py index f5337561..8f9ccdd5 100644 --- a/tests/test_sandbox_record_cli.py +++ b/tests/test_sandbox_record_cli.py @@ -13,8 +13,8 @@ import pytest from pydantic import ValidationError -from coder_eval.cli_recorder import parse_log, render_recorder from coder_eval.evaluation.checker import SuccessChecker +from coder_eval.invocation_log import parse_log, render_recorder from coder_eval.models import ( RECORD_CLI_DIR, RECORD_CLI_LOG, @@ -44,6 +44,12 @@ def _run_shim(sandbox_dir, tool: str, args: list[str]) -> subprocess.CompletedPr ) +def _records(text: str) -> list[dict]: + """Just the records; parse_log also returns the unusable count.""" + usable, _ = parse_log(text) + return [record for _, record in usable] + + class TestGeneration: def test_generates_shim_cmd_twin_and_seeded_log(self): sandbox = _sandbox("record_gen", record_cli=[RecordedCli(tool="uip")]) @@ -66,10 +72,38 @@ def test_recorder_dir_is_path_prepended_before_user_mocks(self): sandbox_dir = sandbox.setup() (sandbox_dir / "mocks").mkdir(exist_ok=True) resolved = sandbox.resolved_mock_path_dirs - assert resolved[0] == sandbox_dir / RECORD_CLI_DIR + # The property resolves symlinks; comparing an unresolved path passes on + # Linux/Windows and fails wherever the tempdir traverses one (macOS /var). + assert resolved[0] == (sandbox_dir / RECORD_CLI_DIR).resolve() finally: sandbox.cleanup(preserve=False) + def test_reused_target_dir_does_not_carry_a_prior_runs_log(self, tmp_path): + """DIRECT_WRITE does not clear the target dir, so a preserved log let a + previous run's invocations score this one with zero agent activity.""" + target = tmp_path / "artifacts" + stale = target / RECORD_CLI_LOG + stale.parent.mkdir(parents=True, exist_ok=True) + stale.write_text( + json.dumps({"tool": "uip", "argv": ["ixp", "projects", "delete", "proj-1"]}) + "\n", + encoding="utf-8", + ) + sandbox = _sandbox("record_reuse", record_cli=[RecordedCli(tool="uip")]) + sandbox.setup(target_dir=target) + assert (target / RECORD_CLI_LOG).read_text(encoding="utf-8") == "" + criterion = CliCalledCriterion(description="deleted the project", verb="ixp projects delete", min_count=1) + assert SuccessChecker(sandbox).check(criterion).score == 0.0 + + def test_stale_shim_for_an_undeclared_tool_is_removed(self, tmp_path): + """A shim left by a previous run would stay on PATH shadowing the real tool.""" + target = tmp_path / "artifacts" + (target / RECORD_CLI_DIR).mkdir(parents=True, exist_ok=True) + (target / RECORD_CLI_DIR / "curl").write_text("stale", encoding="utf-8") + sandbox = _sandbox("record_stale_shim", record_cli=[RecordedCli(tool="uip")]) + sandbox.setup(target_dir=target) + assert not (target / RECORD_CLI_DIR / "curl").exists() + assert (target / RECORD_CLI_DIR / "uip").is_file() + def test_no_record_cli_leaves_no_directory(self): sandbox = _sandbox("record_absent") try: @@ -114,7 +148,7 @@ def test_records_argv_and_fails_without_running_anything(self): assert proc.returncode == 1 assert proc.stderr == "uip: not connected\n" - records = parse_log((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + records = _records((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) assert len(records) == 1 assert records[0]["tool"] == "uip" assert records[0]["exit"] == 1 @@ -146,7 +180,7 @@ def test_quoted_argument_with_spaces_stays_one_element(self): try: sandbox_dir = sandbox.setup() _run_shim(sandbox_dir, "uip", ["fields", "rename", "--group", "Invoice Header"]) - records = parse_log((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + records = _records((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) assert records[0]["argv"][-1] == "Invoice Header" finally: sandbox.cleanup(preserve=False) @@ -158,7 +192,7 @@ def test_multiline_argument_survives_as_one_element(self): try: sandbox_dir = sandbox.setup() _run_shim(sandbox_dir, "uip", ["fields", "update-prompts", "--updates", payload]) - records = parse_log((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + records = _records((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) assert len(records) == 1 assert records[0]["argv"][-1] == payload finally: @@ -170,7 +204,7 @@ def test_repeated_invocations_append_in_order(self): sandbox_dir = sandbox.setup() for n in range(3): _run_shim(sandbox_dir, "uip", ["documents", "upload", f"doc{n}.pdf"]) - records = parse_log((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + records = _records((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) assert [r["argv"][-1] for r in records] == ["doc0.pdf", "doc1.pdf", "doc2.pdf"] finally: sandbox.cleanup(preserve=False) @@ -184,7 +218,7 @@ def test_several_tools_share_one_log_tagged_by_tool(self): sandbox_dir = sandbox.setup() _run_shim(sandbox_dir, "uip", ["projects", "list"]) _run_shim(sandbox_dir, "curl", ["-s", "https://example.invalid"]) - records = parse_log((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + records = _records((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) assert [r["tool"] for r in records] == ["uip", "curl"] finally: sandbox.cleanup(preserve=False) @@ -305,6 +339,14 @@ def test_rendered_shim_is_pure_ascii(self): source = render_recorder(RecordedCli(tool="uip")) source.encode("ascii") - def test_parse_log_skips_unparseable_lines(self): - text = json.dumps({"tool": "uip", "argv": []}) + "\ngarbage\n\n" - assert len(parse_log(text)) == 1 + def test_parse_log_separates_usable_from_unusable(self): + text = ( + json.dumps({"tool": "uip", "argv": ["a"]}) + + "\ngarbage\n\n" + + json.dumps({"tool": "uip", "argv": "not-a-list"}) + + "\n" + ) + usable, unusable = parse_log(text) + assert [argv for argv, _ in usable] == [["a"]] + # An argv that is not list[str] is unusable, not a non-match. + assert unusable == 2 From 8a6cf9aeb86ab200a5e53916075a3e5b4187994d Mon Sep 17 00:00:00 2001 From: Alexandru Jircan Date: Wed, 5 Aug 2026 12:39:10 +0300 Subject: [PATCH 3/5] fix(criteria): split clustered short flags and keep negative numbers positional Two shapes the review flagged as leaving the guard hole one keystroke away. `-yf` parsed as a single flag named `yf`, so an `aliases: ["y"]` predicate -- added precisely to close the `-y`/`--yes` gap -- missed it, and an `absent` guard passed on a confirmed delete. Clustered short flags are now split per character, so `-rf` matches predicates on `r` and `f`. A bare `-1` became a flag named `1` and vanished from the positionals: the same silent disappearance that let `--yes proj-1` slip a delete past a guard. Numeric tokens stay positional. Both stay declaration-driven, consistent with value binding: a name the criterion mentions is taken whole, so a genuine multi-char short flag still matches (`-rf` declared) and `head -1` still parses as a flag when declared. `-fvalue` binds when `f` is value-bearing. Co-Authored-By: Claude Opus 5 (1M context) --- docs/TASK_DEFINITION_GUIDE.md | 2 +- src/coder_eval/criteria/cli_called.py | 38 ++++++++++++++++++++-- tests/test_cli_called_criterion.py | 45 +++++++++++++++++++++++++++ 3 files changed, 82 insertions(+), 3 deletions(-) diff --git a/docs/TASK_DEFINITION_GUIDE.md b/docs/TASK_DEFINITION_GUIDE.md index 523ea51d..a38c4e02 100644 --- a/docs/TASK_DEFINITION_GUIDE.md +++ b/docs/TASK_DEFINITION_GUIDE.md @@ -935,7 +935,7 @@ Defaulting to "switch" is deliberate: `--yes` / `--force` / `-y` before the targ `ignore_flags` drops a flag from matching but does **not** make it value-bearing — an ignored flag that takes a value must also appear in `value_flags` (as `output` does by default). Otherwise `ignore_flags: ["verbose"]` on `delete --verbose proj-1` would let `--verbose` eat `proj-1`. -**Limitation: bundled short flags are not split.** `-rf` parses as one flag named `rf`, so a predicate on `f` will not see it — including `absent: true`, which passes despite `-rf` being present. Assert on the long spelling, or add the bundled form via `aliases`. Likewise a bare negative number in flag position (`seek -1`) is read as a flag named `1`. +**Clustered short flags are split, and declarations win.** `-rf` matches predicates on `r` and `f` — so a `-yf` cannot escape an `aliases: ["y"]` guard. If your CLI has a genuine multi-character short flag, naming it (in `flags`, `value_flags`, or `ignore_flags`) keeps it whole; and `-fvalue` binds when `f` is value-bearing. A bare negative number stays positional (`seek -1`), unless you declare a flag by that name (`head -1`). **Negative guards want the FEWEST facets that capture the forbidden act.** This is the opposite of a positive assertion, and it is easy to get backwards. `max_count: 0` passes when *nothing matches*, so every facet you add is another way for the real invocation to slip past the pattern and report a false PASS. diff --git a/src/coder_eval/criteria/cli_called.py b/src/coder_eval/criteria/cli_called.py index a64a17ac..73852586 100644 --- a/src/coder_eval/criteria/cli_called.py +++ b/src/coder_eval/criteria/cli_called.py @@ -21,6 +21,7 @@ def _split_flags( argv: list[str], ignore: frozenset[str], value_flags: frozenset[str], + known_names: frozenset[str] = frozenset(), ) -> tuple[list[str], dict[str, list[str]]]: """Split ``argv`` into non-flag arguments and a flag map. @@ -33,8 +34,9 @@ def _split_flags( ``ignore`` names are dropped with their values. ``--`` ends flag parsing and is itself dropped; a lone ``-`` is positional. - Known limitation: bundled short flags are not split, so ``-rf`` is one flag - named ``rf`` and a predicate on ``f`` will not see it. + ``known_names`` are the flag names the criterion mentions at all (including + presence predicates and aliases). A declared name is always taken whole, so a + genuine multi-char short flag still matches; undeclared ones are split. """ positional: list[str] = [] flags: dict[str, list[str]] = {} @@ -63,6 +65,28 @@ def record(name: str, value: str) -> None: continue name = token.lstrip("-") + known = name in value_flags or name in known_names + + # A bare negative number is a value, not a flag. Reading `-1` as a flag + # named `1` drops it from the positionals -- the same silent-disappearance + # that let `--yes proj-1` slip a delete past a guard. + if not known and _is_number(name): + positional.append(token) + continue + + # Clustered short flags: `-rf` is `-r -f`. Declared names win, so a real + # multi-char short flag still matches, and `-fvalue` binds when `f` takes + # a value; otherwise each character is its own switch, which is what stops + # `-yf` escaping an `aliases: [y]` predicate. + if not known and not token.startswith("--") and len(name) > 1: + head, rest = name[0], name[1:] + if head in value_flags: + record(head, rest) + else: + for char in name: + record(char, "") + continue + if name in value_flags and index < len(argv): record(name, argv[index]) index += 1 @@ -73,6 +97,14 @@ def record(name: str, value: str) -> None: return positional, flags +def _is_number(text: str) -> bool: + try: + float(text) + except ValueError: + return False + return True + + def _flag_matches(predicate: FlagMatch, values: list[str] | None) -> bool: """Whether a recorded flag satisfies one :class:`FlagMatch` predicate. @@ -113,6 +145,8 @@ def _record_matches(criterion: CliCalledCriterion, argv: list[str], record: dict frozenset(criterion.ignore_flags), frozenset(n for name, p in (criterion.flags or {}).items() if p.needs_value for n in (name, *p.aliases)) | frozenset(criterion.value_flags), + frozenset(n for name, p in (criterion.flags or {}).items() for n in (name, *p.aliases)) + | frozenset(criterion.ignore_flags), ) offset = 0 diff --git a/tests/test_cli_called_criterion.py b/tests/test_cli_called_criterion.py index 1ba57207..bda64b92 100644 --- a/tests/test_cli_called_criterion.py +++ b/tests/test_cli_called_criterion.py @@ -443,6 +443,51 @@ def test_failure_details_show_what_was_actually_recorded(self, sandbox_with_log) assert "ixp projects get proj-1" in details assert "(+1 more)" in details + def test_clustered_short_flags_are_split(self, sandbox_with_log): + """`-yf` used to parse as one flag named `yf`, so an aliases: [y] predicate + missed it -- leaving the `-y` escape one keystroke away from the hole + `aliases` exists to close.""" + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "fields", "delete", "-yf", "proj-1"])]) + guard = CliCalledCriterion( + description="never deleted without confirming", + log=LOG, + verb="ixp fields delete", + flags={"yes": {"absent": True, "aliases": ["y"]}}, + min_count=0, + max_count=0, + ) + assert SuccessChecker(sandbox).check(guard).score == 1.0 + + def test_declared_multi_char_short_flag_is_taken_whole(self): + """Declaring the name wins over splitting, for CLIs with real -ab flags.""" + assert _split_flags(["rm", "-rf", "p"], frozenset(), frozenset(), frozenset({"rf"})) == ( + ["rm", "p"], + {"rf": [""]}, + ) + + def test_attached_value_on_a_short_flag(self): + assert _split_flags(["g", "-ff-002"], frozenset(), frozenset({"f"}), frozenset({"f"})) == ( + ["g"], + {"f": ["f-002"]}, + ) + + def test_bare_negative_number_stays_positional(self): + """`-1` as a flag named `1` dropped it from the positionals -- the same + silent disappearance as the --yes bug.""" + assert _split_flags(["seek", "-1"], frozenset(), frozenset(), frozenset()) == ( + ["seek", "-1"], + {}, + ) + assert _split_flags(["seek", "-1.5"], frozenset(), frozenset(), frozenset())[0] == ["seek", "-1.5"] + + def test_declared_numeric_flag_still_parses_as_a_flag(self): + """`head -1 file` -- declaring it wins over the numeric rule.""" + assert _split_flags(["head", "-1", "f.txt"], frozenset(), frozenset(), frozenset({"1"})) == ( + ["head", "f.txt"], + {"1": [""]}, + ) + def test_declared_value_flag_consumes_a_dash_leading_value(self): """`--limit -1 proj-1`: declared value flags bind even a dash-leading value.""" positional, flags = _split_flags( From a5660bb12e36de5b4104c8d04864beeb80ab4637 Mon Sep 17 00:00:00 2001 From: Alexandru Jircan Date: Wed, 5 Aug 2026 14:18:27 +0300 Subject: [PATCH 4/5] fix(sandbox): close the remaining record_cli findings from #73 The severe one first: nothing stopped `tool: python3`, and the shim's `#!/usr/bin/env python3` resolved through the very PATH this feature prepends, so the shim re-execed itself until the task timed out (tempdir enforces no pid cap). `git`/`uv`/`curl` were also shadowed for run_command criteria, since the orchestrator reuses resolved_mock_path_dirs as the command base PATH. The interpreter is now baked in as an absolute path in both the shebang and the .cmd line, and a reserved set is rejected at load time. Other findings: - The collision guard only checked the bare name, so on Windows a task's own `mocks/uip.cmd` was silently shadowed by the generated `uip.cmd` that PATHEXT resolves first -- the exact outcome the comment above the PATH ordering claims is impossible. It now covers .cmd/.bat/.exe, and the generator refuses to overwrite a file it already wrote this setup. - `tool: calls.jsonl` overwrote the log every criterion reads; `uip.cmd` as a tool name clobbered the generated twin. Both rejected. - `exit_code` accepted any int while sys.exit truncates mod 256, so `exit_code: 256` produced a *succeeding* tool with 256 in the log. Bounded 0-255. - The shim swallowed a failed log write with a bare `pass`, making an unwritable log score identically to "the agent never ran the command". It now writes a `.error` sentinel plus stderr, and cli_called surfaces that as an error rather than scoring an incomplete log. - record_cli was the only SandboxConfig merge field missing from the merge-strategy parametrize list. - Docs: exit_code's default of 1 was never stated, so a bare `- tool: curl` silently made the tool fail. Test gap worth naming: every existing test invoked shims as `sys.executable `, so the shebang, the +x bit and the PATH prepend were exercised by nothing -- removing any of them kept the suite green. Added a POSIX test that runs a bare `uip` through the prepended PATH and grades the log, plus assertions on the .cmd body and the absolute shebang. Also extended the tool-name reject cases with the injection-shaped names (a triple quote, newline, `;`, `$`) that an earlier patch claimed to add but silently missed. Co-Authored-By: Claude Opus 5 (1M context) --- docs/TASK_DEFINITION_GUIDE.md | 4 +- src/coder_eval/criteria/cli_called.py | 15 +++ src/coder_eval/invocation_log.py | 26 +++++- src/coder_eval/models/sandbox.py | 27 +++++- src/coder_eval/sandbox.py | 36 +++++-- tests/test_merge_strategy_annotations.py | 1 + tests/test_sandbox_record_cli.py | 114 ++++++++++++++++++++++- 7 files changed, 207 insertions(+), 16 deletions(-) diff --git a/docs/TASK_DEFINITION_GUIDE.md b/docs/TASK_DEFINITION_GUIDE.md index a38c4e02..62455662 100644 --- a/docs/TASK_DEFINITION_GUIDE.md +++ b/docs/TASK_DEFINITION_GUIDE.md @@ -527,7 +527,9 @@ sandbox: - tool: curl # so a disobedient agent cannot reach the network ``` -Each shim records the invocation, writes the configured `stdout`/`stderr`, and exits with `exit_code`. **Nothing is executed** — no network, no auth, no side effects. +Each shim records the invocation, writes the configured `stdout`/`stderr`, and exits with `exit_code` — which **defaults to 1**, so a bare `- tool: curl` makes the shadowed tool look like it failed. Set `exit_code: 0` when the agent should see success. Values outside 0-255 are rejected, since `sys.exit` truncates mod 256. + +`tool` must be a bare executable name, and a small reserved set (`python`, `python3`, `env`, `sh`, `bash`, `node`, `git`, `uv`, `cmd`) is refused: shadowing those breaks the harness itself rather than the tool under test — the shim's own interpreter, or the shell that `run_command` criteria use. The sandbox writes the shims into `cli_mocks/` and PATH-prepends that directory, then appends one JSON record per invocation to `cli_mocks/calls.jsonl` — the log [`cli_called`](#cli_called) reads by default. Nothing else to wire: no `mock_path_dirs`, no `template_sources`, no `log:` on the criterion. diff --git a/src/coder_eval/criteria/cli_called.py b/src/coder_eval/criteria/cli_called.py index 73852586..74532c68 100644 --- a/src/coder_eval/criteria/cli_called.py +++ b/src/coder_eval/criteria/cli_called.py @@ -226,6 +226,21 @@ def _check_impl( error=f"Invocation log '{criterion.log}' does not exist", ) + # The recorder leaves this beside the log when a write failed, so a record + # it could not append does not read as "the agent never ran the command". + sentinel = f"{criterion.log}.error" + if sandbox.file_exists(sentinel): + detail = sandbox.get_file_content(sentinel).strip().splitlines() + return CriterionResult( + criterion_type=criterion.type, + description=criterion.description, + score=0.0, + error=( + f"Recorder could not write to '{criterion.log}' ({len(detail)} dropped record(s)); " + f"the log is incomplete so the verdict cannot be trusted. First: {detail[0] if detail else '?'}" + ), + ) + content = sandbox.get_file_content(criterion.log) usable, unusable = parse_log(content) diff --git a/src/coder_eval/invocation_log.py b/src/coder_eval/invocation_log.py index 4739dd8f..40228f57 100644 --- a/src/coder_eval/invocation_log.py +++ b/src/coder_eval/invocation_log.py @@ -14,6 +14,7 @@ """ import json +import sys from coder_eval.models import RecordedCli @@ -23,7 +24,7 @@ LOG_FILENAME = "calls.jsonl" _TEMPLATE = '''\ -#!/usr/bin/env python3 +#!{interpreter} """Recording shim for `{tool}` - generated by coder_eval SandboxConfig.record_cli. Appends one JSON record per invocation to {log_filename} beside this script, in @@ -43,6 +44,7 @@ SHIM_DIR = os.path.dirname(os.path.abspath(__file__)) LOG_PATH = os.path.join(SHIM_DIR, {log_filename!r}) +LOG_ERROR_PATH = LOG_PATH + ".error" def record(argv, exit_code): @@ -69,8 +71,15 @@ def record(argv, exit_code): # raise and silently drop the record. with open(LOG_PATH, "a", encoding="utf-8", newline="\\n") as handle: handle.write(json.dumps(entry) + "\\n") - except OSError: - pass + except OSError as exc: + # Keep the agent's command working, but never lose a record silently: a + # dropped record reads exactly like "the agent never ran it". + sys.stderr.write("coder_eval recorder: log write failed: %r\\n" % (exc,)) + try: + with open(LOG_ERROR_PATH, "a", encoding="utf-8") as sentinel: + sentinel.write("%r %r\\n" % (exc, argv)) + except OSError: + pass def main(argv): @@ -94,9 +103,16 @@ def main(argv): ''' -def render_recorder(spec: RecordedCli) -> str: - """Render the shim source for one ``record_cli`` entry.""" +def render_recorder(spec: RecordedCli, interpreter: str | None = None) -> str: + """Render the shim source for one ``record_cli`` entry. + + ``interpreter`` is baked into the shebang as an ABSOLUTE path (defaulting to + the running interpreter). A ``#!/usr/bin/env python3`` shebang resolves + through the same PATH the recorder dir is prepended to, so `tool: python3` + made the shim re-exec itself forever. + """ return _TEMPLATE.format( + interpreter=interpreter or sys.executable, tool=spec.tool, exit_code=spec.exit_code, stdout=spec.stdout, diff --git a/src/coder_eval/models/sandbox.py b/src/coder_eval/models/sandbox.py index 10251ecb..d2083d09 100644 --- a/src/coder_eval/models/sandbox.py +++ b/src/coder_eval/models/sandbox.py @@ -312,7 +312,17 @@ def _validate_working_dir(cls, v: str | None) -> str | None: # hidden files, and the log is primary evidence for every `cli_called` criterion, # so it must survive into the run artifact. RECORD_CLI_DIR = "cli_mocks" -RECORD_CLI_LOG = f"{RECORD_CLI_DIR}/calls.jsonl" +RECORD_CLI_LOG_NAME = "calls.jsonl" +RECORD_CLI_LOG = f"{RECORD_CLI_DIR}/{RECORD_CLI_LOG_NAME}" + +# Shadowing any of these breaks the harness rather than the tool under test: the +# shim is a script run by an interpreter, and its directory goes FIRST on a PATH +# the orchestrator also reuses for run_command criteria. `tool: python3` made the +# shim re-resolve its own interpreter to itself -- an exec loop that spins to the +# task timeout, since tempdir enforces no pid cap. +RECORD_CLI_RESERVED_TOOLS = frozenset( + {"python", "python3", "py", "env", "sh", "bash", "zsh", "cmd", "node", "uv", "git"} +) class RecordedCli(BaseModel): @@ -344,6 +354,8 @@ class RecordedCli(BaseModel): ) exit_code: int = Field( default=1, + ge=0, + le=255, description=( "Exit status the shim returns. Defaults to 1 so an unconfigured tool looks like a failing " "one rather than silently succeeding" @@ -370,6 +382,19 @@ def validate_tool_name(cls, v: str) -> str: raise ValueError("record_cli tool must be a non-empty name without surrounding whitespace") if "/" in v or "\\" in v or v in {".", ".."}: raise ValueError(f"record_cli tool {v!r} must be a bare executable name, not a path") + stem = v.lower().removesuffix(".exe") + if stem in RECORD_CLI_RESERVED_TOOLS: + reserved = ", ".join(sorted(RECORD_CLI_RESERVED_TOOLS)) + msg = ( + f"record_cli tool {v!r} is reserved: shadowing it breaks the harness itself (the " + + "shim's own interpreter, or the shell run_command criteria use). " + + f"Reserved: {reserved}" + ) + raise ValueError(msg) + if v == RECORD_CLI_LOG_NAME: + raise ValueError(f"record_cli tool {v!r} would overwrite the invocation log criteria read") + if v.lower().endswith((".cmd", ".bat")): + raise ValueError(f"record_cli tool {v!r} collides with the generated Windows twin; declare the bare name") return v diff --git a/src/coder_eval/sandbox.py b/src/coder_eval/sandbox.py index 1e65c81f..9c8854e0 100644 --- a/src/coder_eval/sandbox.py +++ b/src/coder_eval/sandbox.py @@ -6,6 +6,7 @@ import os import shutil import subprocess +import sys import tempfile from pathlib import Path @@ -437,9 +438,10 @@ def _prepare_mock_path_dirs(self) -> None: @property def resolved_mock_path_dirs(self) -> list[Path]: - """Absolute paths of configured mock dirs that exist on disk. + """Absolute paths of mock dirs that exist on disk, in PATH-prepend order. - Returned in the order they appear in ``SandboxConfig.mock_path_dirs``; + The generated ``record_cli`` directory comes first when configured, then the + entries in ``SandboxConfig.mock_path_dirs`` in order; non-existent and non-directory entries are filtered out so the caller can pass the result straight to PATH-prepend logic. @@ -490,11 +492,21 @@ def _generate_cli_recorders(self) -> None: if not user_dir.is_dir(): continue for spec in self.config.record_cli: - clash = user_dir / spec.tool - if clash.exists(): + # Every name this feature generates, not just the bare one: on + # Windows PATHEXT resolves `uip` to the generated `uip.cmd` ahead of + # the task's own `mocks/uip.cmd`, silently changing what runs. + clash = next( + ( + user_dir / name + for name in (spec.tool, f"{spec.tool}.cmd", f"{spec.tool}.bat", f"{spec.tool}.exe") + if (user_dir / name).exists() + ), + None, + ) + if clash is not None: msg = ( f"record_cli would generate a '{spec.tool}' shim, but mock_path_dirs entry " - f"'{rel}' already provides one ({rel}/{spec.tool}). " + f"'{rel}' already provides one ({rel}/{clash.name}). " "Remove the record_cli entry to keep your own mock, or drop the file to use " "the generated recorder." ) @@ -515,16 +527,26 @@ def _generate_cli_recorders(self) -> None: log_path = self.sandbox_dir / RECORD_CLI_LOG log_path.write_text("", encoding="utf-8") + interpreter = os.path.realpath(sys.executable) for spec in self.config.record_cli: shim = recorder_dir / spec.tool - shim.write_text(render_recorder(spec), encoding="utf-8", newline="\n") + if shim.exists(): + msg = ( + f"record_cli would overwrite '{RECORD_CLI_DIR}/{spec.tool}', already written this " + "setup. Two entries generating the same filename?" + ) + raise RuntimeError(msg) + shim.write_text(render_recorder(spec, interpreter), encoding="utf-8", newline="\n") + # +x here rather than relying on _prepare_mock_path_dirs: that pass is + # what makes the bit real for PATH lookup, but the shim must be + # executable even if the recorder dir is consumed some other way. shim.chmod(shim.stat().st_mode | 0o111) # `python "%~dp0" %*` — the extensionless script beside this file. cmd_lines = [ "@echo off", "REM Generated by coder_eval SandboxConfig.record_cli.", "REM Windows PATHEXT lookup resolves this; POSIX uses the extensionless twin.", - f'python "%~dp0{spec.tool}" %*', + f'"{interpreter}" "%~dp0{spec.tool}" %*', ] (recorder_dir / f"{spec.tool}.cmd").write_text( "\r\n".join(cmd_lines) + "\r\n", diff --git a/tests/test_merge_strategy_annotations.py b/tests/test_merge_strategy_annotations.py index 407e06bc..4d66ec43 100644 --- a/tests/test_merge_strategy_annotations.py +++ b/tests/test_merge_strategy_annotations.py @@ -32,6 +32,7 @@ class TestSandboxStrategies: [ (SandboxConfig, "template_sources", "append"), (SandboxConfig, "mock_path_dirs", "replace"), + (SandboxConfig, "record_cli", "replace"), (SandboxConfig, "ignore_patterns", "replace"), (SandboxConfig, "driver", "replace"), # nested models / dicts take the type-aware deep default (no annotation): diff --git a/tests/test_sandbox_record_cli.py b/tests/test_sandbox_record_cli.py index 8f9ccdd5..18185cbf 100644 --- a/tests/test_sandbox_record_cli.py +++ b/tests/test_sandbox_record_cli.py @@ -7,6 +7,7 @@ """ import json +import os import subprocess import sys @@ -134,6 +135,61 @@ def test_collision_with_user_mock_raises(self): sandbox.cleanup(preserve=False) +class TestInvokedThroughPath: + """Exercise the shim the way the agent does -- bare name, resolved via PATH. + + Every other test runs it as `sys.executable `, which bypasses the + three mechanisms the agent actually depends on: the baked shebang, the +x bit, + and the PATH prepend. Without this, removing any of them kept the suite green. + """ + + @pytest.mark.skipif(os.name == "nt", reason="POSIX shebang + exec bit path") + def test_bare_name_through_path_records_and_is_gradeable(self): + sandbox = _sandbox("record_path_exec", record_cli=[RecordedCli(tool="uip", exit_code=3)]) + try: + sandbox_dir = sandbox.setup() + recorder_dir = sandbox_dir / RECORD_CLI_DIR + env = {**os.environ, "PATH": f"{recorder_dir}{os.pathsep}{os.environ['PATH']}"} + proc = subprocess.run( + ["uip", "ixp", "projects", "list", "--output", "json"], + capture_output=True, + text=True, + encoding="utf-8", + env=env, + check=False, + ) + assert proc.returncode == 3, proc.stderr + criterion = CliCalledCriterion(description="listed", verb="ixp projects list") + assert SuccessChecker(sandbox).check(criterion).score == 1.0 + finally: + sandbox.cleanup(preserve=False) + + def test_shebang_is_an_absolute_interpreter(self): + """`#!/usr/bin/env python3` resolved through the PATH this feature prepends, + so `tool: python3` made the shim re-exec itself until the task timed out.""" + source = render_recorder(RecordedCli(tool="uip")) + shebang = source.splitlines()[0] + assert shebang.startswith("#!") + interpreter = shebang[2:] + assert os.path.isabs(interpreter), shebang + assert "env " not in shebang + + def test_cmd_twin_body_uses_the_absolute_interpreter(self): + sandbox = _sandbox("record_cmd_body", record_cli=[RecordedCli(tool="uip")]) + try: + sandbox_dir = sandbox.setup() + # newline="" so the CRLF survives -- read_text() would translate it away + # on Windows and the assertion below would pass vacuously. + body = (sandbox_dir / RECORD_CLI_DIR / "uip.cmd").read_text(encoding="utf-8", newline="") + assert "%~dp0uip" in body + assert "\r\n" in body, "cmd needs CRLF" + # A bare `python` would resolve through the prepended dir too. + assert '"python"' not in body and "\npython " not in body + assert os.path.isabs(body.splitlines()[-1].split('"')[1]) + finally: + sandbox.cleanup(preserve=False) + + class TestRecording: def test_records_argv_and_fails_without_running_anything(self): spec = RecordedCli(tool="uip", exit_code=1, stderr="uip: not connected\n") @@ -284,11 +340,65 @@ def test_negative_guard_catches_the_forbidden_call(self): class TestModelValidation: - @pytest.mark.parametrize("bad", ["../evil", "a/b", "a\\b", ".", "..", "", " uip"]) - def test_tool_must_be_a_bare_name(self, bad): + @pytest.mark.parametrize( + "bad", + [ + "../evil", + "a/b", + "a\\b", + ".", + "..", + "", + " uip", + # Not path-shaped, but interpolated into generated source: these emitted + # an unparseable shim, which the path-only cases never caught. + 'a"""b', + 'x") or 1 or ("', + "a\nb", + "a b", + "a;b", + "a$b", + "a`b", + ], + ) + def test_tool_must_be_an_executable_name(self, bad): with pytest.raises(ValidationError): RecordedCli(tool=bad) + @pytest.mark.parametrize("reserved", ["python", "python3", "env", "sh", "bash", "node", "git", "uv", "cmd"]) + def test_reserved_tool_names_rejected(self, reserved): + """Shadowing these breaks the harness, not the tool under test: the shim's own + interpreter, or the shell run_command criteria use. `tool: python3` hung the + task outright by re-execing itself.""" + with pytest.raises(ValidationError, match="reserved"): + RecordedCli(tool=reserved) + + @pytest.mark.parametrize("name", ["PYTHON3", "Python.exe"]) + def test_reserved_names_are_case_and_exe_aware(self, name): + with pytest.raises(ValidationError, match="reserved"): + RecordedCli(tool=name) + + def test_log_filename_as_tool_rejected(self): + """It overwrote the log every cli_called criterion reads by default.""" + with pytest.raises(ValidationError, match="invocation log"): + RecordedCli(tool="calls.jsonl") + + @pytest.mark.parametrize("name", ["uip.cmd", "uip.bat"]) + def test_windows_twin_name_as_tool_rejected(self, name): + with pytest.raises(ValidationError, match="Windows twin"): + RecordedCli(tool=name) + + @pytest.mark.parametrize("bad_exit", [256, -1, 300]) + def test_exit_code_outside_posix_range_rejected(self, bad_exit): + """sys.exit truncates mod 256, so exit_code: 256 made a 'failing' tool exit 0 + while the log still recorded 256.""" + with pytest.raises(ValidationError): + RecordedCli(tool="uip", exit_code=bad_exit) + + def test_exit_code_bounds_are_inclusive(self): + assert RecordedCli(tool="uip", exit_code=0).exit_code == 0 + assert RecordedCli(tool="uip", exit_code=255).exit_code == 255 + @pytest.mark.parametrize("field", ["mode", "response", "passthrough"]) def test_unknown_field_rejected(self, field): """extra='forbid' catches a typo — and a config written against a shape From 1165c8383982253f578852ebe2596faafff67e1a Mon Sep 17 00:00:00 2001 From: Alexandru Jircan Date: Wed, 5 Aug 2026 16:23:44 +0300 Subject: [PATCH 5/5] fix(sandbox): stop the recorder dir defeating the PLUGIN_TOOLS_DIR pin The recorder dir goes first on a PATH the orchestrator syncs back and reuses for run_command criteria, so `record_cli: [{tool: uip}]` -- the example in every doc here -- made `shutil.which("uip")` resolve to the shim. A shim is not inside a node_modules/@uipath tree, so resolve_uipath_plugin_dir returned None and PLUGIN_TOOLS_DIR silently stopped being exported (the MST-9795 pin). Verified: which('uip') on uip_search_path -> /cli_mocks/uip.CMD which('uip') on discovery path -> C:\Users\...\.bun\bin\uip.EXE Plugin discovery now filters the generated dir out. A hand-written mock under mock_path_dirs shadows the lookup the same way, but that predates this feature and narrowing it would change existing tasks, so it stays as-is. Also from the review's "What's Missing": - cli_mocks/ was not in default_ignore_patterns.yaml, so the agent_judge workspace copy handed a Bash-enabled judge the generated shim source and calls.jsonl as if the agent had authored them. Artifact capture uses a separate list, so the log is still preserved as evidence. - The shim's stdin invariant -- never read, because an open pipe would hang the task -- was the design's stated reason for a choice and nothing asserted it. Added a test that leaves stdin=PIPE unwritten and unclosed and asserts the shim still returns. Co-Authored-By: Claude Opus 5 (1M context) --- .../resources/default_ignore_patterns.yaml | 7 +++ src/coder_eval/sandbox.py | 20 ++++++++- tests/test_sandbox_record_cli.py | 44 +++++++++++++++++++ 3 files changed, 70 insertions(+), 1 deletion(-) diff --git a/src/coder_eval/resources/default_ignore_patterns.yaml b/src/coder_eval/resources/default_ignore_patterns.yaml index 009bf62d..f2935d90 100644 --- a/src/coder_eval/resources/default_ignore_patterns.yaml +++ b/src/coder_eval/resources/default_ignore_patterns.yaml @@ -58,3 +58,10 @@ compiled: - "*.so" - "*.dylib" - "*.dll" + +# Harness-generated content, not agent work: SandboxConfig.record_cli writes +# recording shims and the invocation log here. Excluded from the agent_judge +# workspace copy so a Bash-enabled judge does not read them as authored files. +# Artifact capture uses a separate list, so the log is still preserved as evidence. +harness_generated: + - "cli_mocks" diff --git a/src/coder_eval/sandbox.py b/src/coder_eval/sandbox.py index 9c8854e0..2748443f 100644 --- a/src/coder_eval/sandbox.py +++ b/src/coder_eval/sandbox.py @@ -822,9 +822,27 @@ def _refresh_plugin_tools_dir(self) -> None: """ from .utils import resolve_uipath_plugin_dir - resolved = resolve_uipath_plugin_dir(self.uip_search_path) + resolved = resolve_uipath_plugin_dir(self._plugin_discovery_path()) self._plugin_tools_dir = str(resolved) if resolved is not None else None + def _plugin_discovery_path(self) -> str: + """``uip_search_path`` minus the generated recorder dir. + + A recording shim is not the real CLI, so letting it win the `uip` lookup + made resolve_uipath_plugin_dir return None (a shim is not inside a + node_modules/@uipath tree) and silently drop the PLUGIN_TOOLS_DIR pin for + every run_command criterion -- for `record_cli: [{tool: uip}]`, the + documented example. A hand-written mock under mock_path_dirs shadows the + lookup the same way, but that predates this feature and changing it would + alter existing tasks. + """ + search_path = self.uip_search_path + if not self.config.record_cli or self.sandbox_dir is None: + return search_path + recorder_dir = str((self.sandbox_dir / RECORD_CLI_DIR).resolve()) + kept = [entry for entry in search_path.split(os.pathsep) if entry and os.path.realpath(entry) != recorder_dir] + return os.pathsep.join(kept) + def _maybe_remediate_home_plugins_pollution(self) -> Path | None: """Optionally delete ``$HOME/node_modules/@uipath`` before the task runs. diff --git a/tests/test_sandbox_record_cli.py b/tests/test_sandbox_record_cli.py index 18185cbf..ef8a9bac 100644 --- a/tests/test_sandbox_record_cli.py +++ b/tests/test_sandbox_record_cli.py @@ -105,6 +105,23 @@ def test_stale_shim_for_an_undeclared_tool_is_removed(self, tmp_path): assert not (target / RECORD_CLI_DIR / "curl").exists() assert (target / RECORD_CLI_DIR / "uip").is_file() + def test_recorder_dir_is_excluded_from_plugin_discovery(self): + """A shim must not win the `uip` lookup that pins PLUGIN_TOOLS_DIR. + + It is not inside a node_modules/@uipath tree, so letting it win made + resolve_uipath_plugin_dir return None and silently stop exporting the pin + to every run_command criterion -- for the documented `tool: uip` example. + """ + sandbox = _sandbox("record_plugin_dir", record_cli=[RecordedCli(tool="uip")]) + try: + sandbox_dir = sandbox.setup() + recorder = str((sandbox_dir / RECORD_CLI_DIR).resolve()) + sandbox.set_command_base_path(f"{recorder}{os.pathsep}{os.environ.get('PATH', '')}") + assert recorder in sandbox.uip_search_path.split(os.pathsep) + assert recorder not in sandbox._plugin_discovery_path().split(os.pathsep) + finally: + sandbox.cleanup(preserve=False) + def test_no_record_cli_leaves_no_directory(self): sandbox = _sandbox("record_absent") try: @@ -164,6 +181,33 @@ def test_bare_name_through_path_records_and_is_gradeable(self): finally: sandbox.cleanup(preserve=False) + def test_shim_returns_with_stdin_left_open(self): + """The invariant the docstring claims: stdin is never read, so an open pipe + cannot hang the task. Nothing asserted it before.""" + sandbox = _sandbox("record_stdin", record_cli=[RecordedCli(tool="uip", exit_code=2)]) + try: + sandbox_dir = sandbox.setup() + shim = sandbox_dir / RECORD_CLI_DIR / "uip" + proc = subprocess.Popen( + [sys.executable, str(shim), "ixp", "projects", "list"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + try: + # Deliberately never write or close stdin before waiting. + assert proc.wait(timeout=20) == 2 + finally: + if proc.poll() is None: + proc.kill() + proc.stdin.close() + proc.stdout.close() + proc.stderr.close() + records = _records((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + assert records[0]["argv"] == ["ixp", "projects", "list"] + finally: + sandbox.cleanup(preserve=False) + def test_shebang_is_an_absolute_interpreter(self): """`#!/usr/bin/env python3` resolved through the PATH this feature prepends, so `tool: python3` made the shim re-exec itself until the task timed out."""