diff --git a/docs/TASK_DEFINITION_GUIDE.md b/docs/TASK_DEFINITION_GUIDE.md index f10a7ccb..ef97b474 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,33 @@ 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` — 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. + +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). @@ -870,7 +898,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: @@ -881,6 +909,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 @@ -925,7 +955,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 6e196b4c..74532c68 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 @@ -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. @@ -100,19 +132,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: @@ -126,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 @@ -205,27 +226,24 @@ 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: 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/invocation_log.py b/src/coder_eval/invocation_log.py new file mode 100644 index 00000000..40228f57 --- /dev/null +++ b/src/coder_eval/invocation_log.py @@ -0,0 +1,151 @@ +"""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 +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 +import sys + +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 = '''\ +#!{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 +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}) +LOG_ERROR_PATH = LOG_PATH + ".error" + + +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 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): + """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, 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, + stderr=spec.stderr, + log_filename=LOG_FILENAME, + ) + + +def parse_log(text: str) -> tuple[list[tuple[list[str], dict[str, object]]], int]: + """Parse recorder-log text into ``(usable, unusable_count)``. + + 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. + """ + usable: list[tuple[list[str], dict[str, object]]] = [] + unusable = 0 + for line in text.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 = 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/__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 dadc2237..9795a74e 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..d2083d09 100644 --- a/src/coder_eval/models/sandbox.py +++ b/src/coder_eval/models/sandbox.py @@ -307,6 +307,97 @@ 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_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): + """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( + 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, + 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" + ), + ) + 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") + 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 + + class SandboxConfig(BaseModel): """Configuration for the sandboxed execution environment. @@ -360,6 +451,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/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 07217061..2748443f 100644 --- a/src/coder_eval/sandbox.py +++ b/src/coder_eval/sandbox.py @@ -6,10 +6,14 @@ import os import shutil import subprocess +import sys import tempfile from pathlib import Path +from .invocation_log import render_recorder from .models import ( + RECORD_CLI_DIR, + RECORD_CLI_LOG, RepoSource, SandboxConfig, StarterFilesSource, @@ -203,6 +207,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() @@ -430,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. @@ -442,15 +451,114 @@ 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: + # 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}/{clash.name}). " + "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 + # 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 + 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 + 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'"{interpreter}" "%~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. @@ -714,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_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( 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 new file mode 100644 index 00000000..ef8a9bac --- /dev/null +++ b/tests/test_sandbox_record_cli.py @@ -0,0 +1,506 @@ +"""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 os +import subprocess +import sys + +import pytest +from pydantic import ValidationError + +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, + 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, + ) + + +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")]) + 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 + # 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_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: + 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 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_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.""" + 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") + 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 = _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 + 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 = _records((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 = _records((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 = _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) + + 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 = _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) + + +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", + # 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 + 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_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