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) <noreply@anthropic.com>

Author: alexandrujircan

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:

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

src/coder_eval/cli_recorder.py src/coder_eval/invocation_log.pysrc/coder_eval/cli_recorder.py renamed to 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,21 +105,31 @@ 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:
             continue
         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

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=(

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,22 +494,26 @@ 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
         # 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")
+        log_path.write_text("", encoding="utf-8")
 
         for spec in self.config.record_cli:
             shim = recorder_dir / spec.tool

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