From 2067d2bf95d7725164340be90aeeb6c542b44848 Mon Sep 17 00:00:00 2001 From: Alexandru Jircan Date: Wed, 2 Sep 2026 16:00:43 +0300 Subject: [PATCH 1/3] feat(record_cli): serve a different canned response per invocation A `record_cli` shim answered every invocation with one fixed exit_code/stdout/stderr, so an agent whose next step depends on what the tool just told it could not be evaluated: `uip ixp dummy1` and `uip ixp dummy2` got the same reply. Each entry may now declare `responses`, a list of rules tried in declaration order, first match wins, falling back to the entry's own three fields for anything no rule claims. `exit_code` defaults to 0 on a rule (the opposite of the entry default of 1): a rule exists because the author described that invocation. `when` is not a second pattern language. The criterion's matcher moved to `argv_match.py` -- stdlib-only, plain dicts -- and both surfaces lower to one spec dict, so the pattern that serves a response is the pattern that grades it. `render_recorder` embeds that module's SOURCE into the shim, which runs where coder_eval is not installed; CE047 keeps its imports stdlib-only, since one package import there would make every shadowed CLI die with an ImportError the agent reads as "the tool is broken". `FlagMatch` moved to the new cycle-free leaf `models/cli_match.py` alongside `CliMatch` and the shared verb/flag validators: models/sandbox.py cannot import from models/criteria.py, which already takes RECORD_CLI_LOG from it. Two deliberate divergences from the criterion, both tested: `ignore_flags` is empty on a rule (grading must not depend on --output; dispatch may), and `tool` stays criterion-only, addressing a log record rather than argv. Also fixes a pre-existing silent no-match: a flag written into a verb (`verb: "ixp projects get --output json"`) validated and then matched nothing, because a verb is compared against the non-flag arguments -- the criterion scored 0 against a log holding that exact call. Now rejected on every surface, reusing the splitter's own is_number rule so `head -1` stays legal. The shim records `"rule": ` when a rule answered, and omits the key when none did. Co-Authored-By: Claude Opus 5 (1M context) --- CLAUDE.md | 8 +- docs/TASK_DEFINITION_GUIDE.md | 36 +- src/coder_eval/argv_match.py | 226 ++++++++++ src/coder_eval/criteria/cli_called.py | 164 +------- src/coder_eval/invocation_log.py | 99 ++++- src/coder_eval/models/__init__.py | 10 +- src/coder_eval/models/cli_match.py | 385 ++++++++++++++++++ src/coder_eval/models/criteria.py | 203 ++------- src/coder_eval/models/sandbox.py | 60 ++- src/coder_eval/sandbox.py | 7 +- .../rules/ce047_embedded_shim_stdlib_only.py | 63 +++ tests/lint/runner.py | 2 + tests/test_cli_called_criterion.py | 30 +- tests/test_cli_match_parity.py | 102 +++++ tests/test_custom_lint.py | 32 ++ tests/test_merge_strategy_annotations.py | 2 + tests/test_sandbox_record_cli.py | 164 ++++++++ 17 files changed, 1227 insertions(+), 366 deletions(-) create mode 100644 src/coder_eval/argv_match.py create mode 100644 src/coder_eval/models/cli_match.py create mode 100644 tests/lint/rules/ce047_embedded_shim_stdlib_only.py create mode 100644 tests/test_cli_match_parity.py diff --git a/CLAUDE.md b/CLAUDE.md index ac00d5ef..0fdb1b31 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -27,6 +27,8 @@ coder_eval/ ├── fs_permissions.py # set_permissions: stacked chmod window (via Sandbox.set_permissions) ├── pricing.py # Model pricing / cost calculation (ModelPricing, calculate_cost, register_pricing) ├── litellm_cost.py # Join proxy-captured ACTUAL per-call cost/cache onto turns (LiteLLM backend; apply_actual_cost) +├── invocation_log.py # record_cli recording shim: renders it (embedding argv_match), and parse_log reads its JSON Lines back +├── argv_match.py # Structured argv matcher. STDLIB-ONLY (CE047): its source is embedded into every response-serving shim, so `cli_called` and a `record_cli` response rule dispatch on ONE semantic ├── utils.py # Version info helpers │ ├── agents/ @@ -38,10 +40,11 @@ coder_eval/ │ ├── criteria.py # 15 success criterion types + base + union │ ├── experiment.py # ExperimentDefinition, ExperimentVariant, ResolvedTask, result models │ ├── judge_defaults.py # DEFAULT_JUDGE_MODEL constant (cycle-free leaf) +│ ├── cli_match.py # FlagMatch + CliMatch (a `when:` pattern) + the shared verb/flag validators (cycle-free leaf: criteria.py and sandbox.py both import it) │ ├── mutations.py # PromptMutation variants (prefix/suffix/replace/template) │ ├── results.py # CriterionResult (+ ClassificationCriterionResult), TurnRecord, EvaluationResult, EarlyStopInfo/EarlyStopReason, CriterionAggregate, ThresholdCheck, SuiteRollup │ ├── routing.py # ApiRoute (DirectRoute/BedrockRoute) -│ ├── sandbox.py # SandboxConfig, ResourceLimits +│ ├── sandbox.py # SandboxConfig, ResourceLimits, RecordedCli + CliResponse (per-invocation stub responses) │ ├── tasks.py # TaskDefinition, AgentConfig, Dataset (dataset fan-out + sample) │ ├── telemetry.py # CommandTelemetry, CommandStatistics, TokenUsage, ProviderCallCost, ReconciliationMessage, TranscriptMessage │ └── templates.py # RepoSource, TemplateDirSource, StarterFilesSource @@ -51,6 +54,7 @@ coder_eval/ │ ├── base.py # BaseCriterion (async _check_impl_async is primary; sync _check_impl derives from it, or vice versa) + @handle_criterion_errors(_async) │ ├── _classification_aggregate.py # Shared overlay: accuracy / P/R/F1 / confusion matrix │ ├── classification_match.py # File-based label matcher +│ ├── cli_called.py # Structured match over the record_cli invocation log (matching engine: argv_match.py) │ ├── command_executed.py │ ├── commands_efficiency.py │ ├── file_check.py @@ -216,7 +220,7 @@ make plugin-reference # the plugin's bundled criteria reference from the models Editing `src/coder_eval/pricing.py` means editing `evalboard/lib/pricing.ts` too — it is a hand-copied mirror, and `evalboard/lib/__tests__/pricing-parity.test.ts` fails the build on drift in either direction. -Recent additions, each traceable to a shipped defect: **CE037** (no unreferenced module-level private helper in `src/` — a helper whose docstring documents a bug the live code still has is worse than none), **CE038** (in an `@asynccontextmanager`, the acquire must sit INSIDE the `try` whose `finally` releases it — `asyncio.shield` protects the inner task, NOT the await, so a cancel on `__aenter__` skips the unwind while the work completes), **CE039** (a criterion checker must not return a gating `score=0.0` from an `except OSError` over a path the *task author* named — that books an eval-config error as an agent failure; raise `CheckerMisuseError` instead, and `# noqa: CE039` the cases that really are the agent's). +Recent additions, each traceable to a shipped defect: **CE037** (no unreferenced module-level private helper in `src/` — a helper whose docstring documents a bug the live code still has is worse than none), **CE038** (in an `@asynccontextmanager`, the acquire must sit INSIDE the `try` whose `finally` releases it — `asyncio.shield` protects the inner task, NOT the await, so a cancel on `__aenter__` skips the unwind while the work completes), **CE039** (a criterion checker must not return a gating `score=0.0` from an `except OSError` over a path the *task author* named — that books an eval-config error as an agent failure; raise `CheckerMisuseError` instead, and `# noqa: CE039` the cases that really are the agent's)), **CE047** (a module whose SOURCE is embedded into a generated sandbox shim — `argv_match.py` — may import stdlib only: the shim runs where `coder_eval` and its dependencies are not installed, so one package import makes every shadowed CLI die with an ImportError the agent reads as "the tool is broken"). When fixing a bug, ask: *could a custom lint rule have prevented this?* If the root cause is a mechanically detectable pattern (e.g., "always import from `coder_eval.models`", "never call blocking IO in async"), add a rule to `tests/lint/rules/` following the CE001+ pattern and wire it up in `tests/lint/runner.py`. This turns a one-time fix into permanent enforcement. See `tests/test_custom_lint.py` for how rules are tested. (Doc-surface / whole-tree rules that reason over Markdown/YAML or the entire `src/` tree rather than one `.py` AST at a time — CE026–CE031, CE033–CE036 — are not `BaseRule`s in the runner; they are wired as dedicated `@pytest.mark.lint` test classes. CE036 enforces the `live_verdict` determinism + monotonicity contract (`criteria/base.py`) that `EarlyStopWatcher`'s latching, deferred fail-stop, and flip-attribution silently depend on: monotonicity over arbitrary Python is undecidable, so instead of a static check it REPLAYS each live criterion against every prefix of recorded trajectories (`tests/lint/live_verdict_contract.py::CASES`) — on the authored ordering AND under seeded shuffles (`permuted_violations`, which catch order-sensitive bugs the authored walk misses) — and asserts the property directly, plus registry-derived coverage — every `LiveSuccessCriterion` in the union must have cases, and every polarity its instances claim via `live_decidable_polarities()` must actually be reached by one (otherwise a single always-`undecided` fixture would "cover" a type while proving nothing). Adding a live criterion therefore means adding `ContractCase`s in the same change. CE035 resolves every `steps..outputs.` / `needs..outputs.` reference in `.github/workflows/**` to a writer that actually produces that key — GitHub expands an unwritten output to the empty string, so a typo degrades a gate silently and actionlint models `steps.*.outputs` as an open string map. CE034 scans `tasks/` and forces an armed, live-*passable* `command_executed` to set `require_success` — a crashed invocation would otherwise latch a live PASS, fire `on_pass: stop`, and let FIRED-ONLY armed gating report SUCCESS without ever consulting the unarmed criteria (negative assertions are fail-only and are exempt). CE033 keeps the plugin's bundled `reference/criteria.md` in parity with the `SuccessCriterion` union that generates it (`make plugin-reference` writes it; the rule re-renders and diffs — never hand-edit the file). CE031 guards against dead config: a behavior-driving field on `SimulationConfig`/`RunLimits`/`Dataset` that no code reads by name. CE026 keeps the GitHub Action's onboarding surfaces honest — `README.md`, `docs/CI_GATE.md`, `docs/tutorials/02-ci-pipeline.md`, and the plugin's `ci` skill, whose emitted workflow users copy into their own repos: a page's *first* Action snippet must show the agent-runtime prerequisite steps (pinned to the `action-dogfood` job that proves them in CI), a zero-install absolute next to such a snippet must name the channel it means, every `github.com/marketplace/actions/` link plus the shields badge label must match `action.yml`'s `name:`, and every `with:` key on a snippet's action step must be a real `action.yml` input (GitHub ignores unknown inputs, so a rename would silently degrade every copied workflow). Renaming an action input or changing its runtime prerequisites therefore means updating the skill too.) diff --git a/docs/TASK_DEFINITION_GUIDE.md b/docs/TASK_DEFINITION_GUIDE.md index 57151805..c1b2fbe0 100644 --- a/docs/TASK_DEFINITION_GUIDE.md +++ b/docs/TASK_DEFINITION_GUIDE.md @@ -567,7 +567,39 @@ Notes: - **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). +- **It stubs a tool; it does not proxy one.** Recording a *real* executable on the way through stays a hand-written mock under `mock_path_dirs` — that depends on state the harness cannot guarantee (the tool being installed, PATH order, live credentials). + +#### Answering each invocation differently + +An agent whose next step depends on what the tool just told it cannot be evaluated by a stub that replies the same way to everything it types. `responses` gives one shadowed executable a reply per invocation: + +```yaml +sandbox: + record_cli: + - tool: uip + exit_code: 1 # fallback: anything no rule claims + stderr: "uip: unknown command\n" + responses: + - when: {verb: "ixp dummy1"} + stdout: "response1\n" + - when: {verb: "ixp dummy2"} + stdout: "response2\n" + - when: # any cli_called facet, ANDed + verb: "ixp projects get" + positional: ["proj-1"] + flags: {output: json} + stdout: '{"id": "proj-1", "name": "Invoices"}' + - when: {verb: "ixp projects get missing"} + exit_code: 4 + stderr: "project not found\n" +``` + +- **`when` takes the same facets as [`cli_called`](#cli_called)** — `verb`, `verb_any_of`, `positional`, `flags`, `value_flags`, `ignore_flags` — evaluated by the same matcher, so the pattern that *serves* a response is the pattern that *grades* it. Always a mapping: a bare `when: "ixp dummy1"` is rejected (with the `{verb: ...}` spelling in the message), since a pattern has six facets and a lone string leaves which one you meant to inference. +- **First match wins**, in declaration order: put the specific rule above the general one. An invocation no rule claims gets the entry's own `exit_code` / `stdout` / `stderr`. +- **`exit_code` defaults to 0 on a rule** — the opposite of the entry default of 1. A rule exists because you described that invocation, so the natural reading is "and this is what it answers"; an undescribed one should still look like a tool that failed. +- **`ignore_flags` is empty on a rule**, unlike the criterion's `[output]`: grading must not depend on a flag that changes nothing about the outcome, but a rule may legitimately answer differently for `--output json`. +- **The log names the rule that answered** (`"rule": 1`), and omits the key when none did — the first thing you want to know when an expected canned response does not arrive. +- **Still stateless.** A rule answers the same way however many times it matches; a counter would have to survive concurrent agent commands. For a tool whose reply must change over a run, hand-write a mock under `mock_path_dirs`. ## Template Sources @@ -973,6 +1005,8 @@ Use this instead of `command_executed` or `file_matches_regex` when a test shado Do **not** shorten the verb instead. `verb: "ixp projects"` matches all of its subcommands, so a positive assertion that the agent *read* a project is equally satisfied by `ixp projects delete`. Two entries are rejected when one prefixes the other, since the shorter already accepts everything the longer does. +**A verb holds subcommands only.** `verb: "ixp projects get --output json"` is rejected: the verb is compared against the *non-flag* arguments, so a flag inside it could never match — the criterion would score 0 against a log holding that exact call. Put it in `flags:` instead. (`head -1` still validates: a bare negative number is a value to the argument splitter, not a flag.) + **The argument tail stays open.** `positional` is a prefix too, so `verb: "ixp projects list"` with `positional: ["proj-1"]` also matches `ixp projects list proj-1 dummy`. To require a specific tail, name every argument in it. `positional: []` is rejected — it would assert nothing. **Declare value-bearing flags when you use `positional`.** An undeclared flag is treated as a switch, so its value stays among the non-flag arguments and shifts the ones you named. `get proj-1 --folder Finance` matches `positional: ["proj-1"]`, but `get --folder Finance proj-1` does **not** — `Finance` takes the first slot. Add `folder` to `value_flags` (or name it in `flags`) to fix it. Resolving the ambiguity this way is deliberate: guessing that an unknown flag consumes the next token let `--yes proj-1` bind `yes=proj-1` and swallow the project name, which made a `max_count: 0` delete guard pass on the delete it forbade. diff --git a/src/coder_eval/argv_match.py b/src/coder_eval/argv_match.py new file mode 100644 index 00000000..136333b2 --- /dev/null +++ b/src/coder_eval/argv_match.py @@ -0,0 +1,226 @@ +"""Structured argv matching — the one engine both CLI surfaces share. + +Two places ask the same question about one invocation. The ``cli_called`` +criterion reads a recorded ``argv`` back afterwards and asks *did this happen*; +a ``record_cli`` response rule asks it live, inside the sandbox, to choose which +canned response to serve. An author who writes ``verb: "ixp projects get"`` in a +rule and again in the criterion that grades it must get one semantic, not two +that drift. + +Everything here takes PLAIN DICTS rather than pydantic models, and imports +nothing beyond the standard library: :func:`coder_eval.invocation_log.render_recorder` +embeds this module's SOURCE into every generated shim, and that shim runs inside +the sandbox, where ``coder_eval`` is not installed. Lint rule CE047 keeps the +imports stdlib-only. + +The spec dict is what ``CliMatch.match_spec`` emits:: + + {"verb_spellings": [["ixp", "projects", "get"]], + "positional": ["proj-1"], + "flags": {"model": {"equals": "pro", "aliases": [], "flags": 0, ...}}, + "value_flags": ["output"], + "ignore_flags": []} + +``verb_spellings`` is empty when there is no verb constraint; ``positional`` and +``flags`` are absent or None when unconstrained. +""" + +import re +from typing import Any + + +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. + + Only flags in ``value_flags`` consume a following token; everything else is a + switch. Guessing instead (``--yes proj-1`` binding ``yes=proj-1``) let a + ``max_count: 0`` guard pass on the delete it forbade, so ambiguity resolves + toward keeping the token positional. + + ``--flag=value`` binds directly, being unambiguous. Repeated flags accumulate. + ``ignore`` names are dropped with their values. ``--`` ends flag parsing and is + itself dropped; a lone ``-`` is positional. + + ``known_names`` are the flag names the spec 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]] = {} + + def record(name: str, value: str) -> None: + if name not in ignore: + flags.setdefault(name, []).append(value) + + index = 0 + end_of_flags = False + while index < len(argv): + token = argv[index] + index += 1 + + if end_of_flags or not token.startswith("-") or token == "-": + positional.append(token) + continue + if token == "--": + end_of_flags = True + continue + + # Equals form: unambiguous, bind it and move on. + if "=" in token: + name, _, value = token.partition("=") + record(name.lstrip("-"), value) + 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 + else: + # Switch: empty value, and the next token is left for the positionals. + record(name, "") + + return positional, flags + + +def is_number(text: str) -> bool: + """Whether ``text`` parses as a number, so ``-1`` reads as a value not a flag.""" + try: + float(text) + except ValueError: + return False + return True + + +def predicate_needs_value(predicate: dict[str, Any]) -> bool: + """Whether evaluating this flag predicate requires the flag's VALUE. + + Presence predicates (``present`` / ``absent``) do not, so they must not make + a flag value-bearing: asserting a boolean switch would otherwise make it + consume the following token, dropping that token from the positionals. + Mirrors ``FlagMatch.needs_value``; both sides of the spec boundary must agree + or a rule and the criterion grading it would parse the same argv differently. + """ + return not (predicate.get("present") or predicate.get("absent")) + + +def flag_matches(predicate: dict[str, Any], values: list[str] | None) -> bool: + """Whether a recorded flag satisfies one flag predicate. + + ``values`` is None when the flag was not passed at all. Every non-``absent`` + predicate is satisfied by ANY of a repeated flag's values. + """ + if predicate.get("absent"): + return values is None + if predicate.get("present"): + return values is not None + if values is None: + return False + if predicate.get("equals") is not None: + return any(value == predicate["equals"] for value in values) + if predicate.get("contains") is not None: + return any(predicate["contains"] in value for value in values) + if predicate.get("any_of") is not None: + allowed = set(predicate["any_of"]) + return any(value in allowed for value in values) + if predicate.get("matches_regex") is not None: + regex = re.compile(predicate["matches_regex"], predicate.get("flags", 0)) + return any(regex.search(value) is not None for value in values) + # Unreachable: the model guarantees exactly one predicate. Raise rather than + # return False so a predicate added without a matcher arm here fails loudly. + raise AssertionError(f"flag predicate has no matcher arm: {predicate!r}") + + +def argv_matches(spec: dict[str, Any], argv: list[str]) -> bool: + """Whether ``argv`` satisfies every configured facet of one match spec.""" + flag_specs: dict[str, Any] = spec.get("flags") or {} + + def names_of(flag: str, predicate: dict[str, Any]) -> tuple[str, ...]: + return (flag, *(predicate.get("aliases") or ())) + + # Declarations only. Folding `ignore_flags` into value_flags made ignored + # SWITCHES value-bearing, which swallowed the next positional and reopened a + # guard false-PASS; an ignored flag that takes a value declares it in + # value_flags. + ignore = frozenset(spec.get("ignore_flags") or ()) + value_flags = frozenset( + name + for flag, predicate in flag_specs.items() + if predicate_needs_value(predicate) + for name in names_of(flag, predicate) + ) | frozenset(spec.get("value_flags") or ()) + known_names = ( + frozenset(name for flag, predicate in flag_specs.items() for name in names_of(flag, predicate)) | ignore + ) + + positional, flags = split_flags(argv, ignore, value_flags, known_names) + + offset = 0 + spellings = spec.get("verb_spellings") or [] + if spellings: + # Token-wise, not a subset and not a string startswith: `labellings confirm` + # must never be satisfied by `labellings unconfirm`. Taking the first match is + # safe because validation rejects one spelling prefixing another, so no argv + # can match two. + matched = next((tokens for tokens in spellings if positional[: len(tokens)] == list(tokens)), None) + if matched is None: + return False + # Measured from the spelling that matched, since spellings can differ in length. + offset = len(matched) + + expected = spec.get("positional") + if expected is not None and positional[offset : offset + len(expected)] != list(expected): + return False + + for flag, predicate in flag_specs.items(): + # [] means absent under every spelling, which flag_matches distinguishes + # from a switch's "present with empty value" ([""]). + collected = [value for name in names_of(flag, predicate) for value in flags.get(name, [])] + if not flag_matches(predicate, collected or None): + return False + + return True + + +def select_rule(rules: list[dict[str, Any]], argv: list[str]) -> tuple[int, dict[str, Any]] | None: + """``(index, rule)`` of the first rule whose ``when`` spec matches ``argv``, or None. + + First match wins, so ordering is the author's disambiguation tool: the + specific rule goes above the general one. Stateless by design -- the same + argv gets the same answer every time, which keeps the shim free of on-disk + counters that two concurrent agent commands would race on. + + The index travels with the rule because the shim records it: "no rule + matched" and "a rule matched and looks like the default" are otherwise the + same line in the log. + """ + for index, rule in enumerate(rules): + if argv_matches(rule.get("when") or {}, argv): + return index, rule + return None diff --git a/src/coder_eval/criteria/cli_called.py b/src/coder_eval/criteria/cli_called.py index 55811e6d..ea970519 100644 --- a/src/coder_eval/criteria/cli_called.py +++ b/src/coder_eval/criteria/cli_called.py @@ -5,9 +5,10 @@ import shlex from typing import TYPE_CHECKING, Any +from coder_eval.argv_match import argv_matches 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 +from coder_eval.models import CliCalledCriterion, CriterionResult if TYPE_CHECKING: @@ -17,165 +18,16 @@ logger = logging.getLogger(__name__) -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. - - Only flags in ``value_flags`` consume a following token; everything else is a - switch. Guessing instead (``--yes proj-1`` binding ``yes=proj-1``) let a - ``max_count: 0`` guard pass on the delete it forbade, so ambiguity resolves - toward keeping the token positional. - - ``--flag=value`` binds directly, being unambiguous. Repeated flags accumulate. - ``ignore`` names are dropped with their values. ``--`` ends flag parsing and is - itself dropped; a lone ``-`` is positional. - - ``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]] = {} - - def record(name: str, value: str) -> None: - if name not in ignore: - flags.setdefault(name, []).append(value) - - index = 0 - end_of_flags = False - while index < len(argv): - token = argv[index] - index += 1 - - if end_of_flags or not token.startswith("-") or token == "-": - positional.append(token) - continue - if token == "--": - end_of_flags = True - continue - - # Equals form: unambiguous, bind it and move on. - if "=" in token: - name, _, value = token.partition("=") - record(name.lstrip("-"), value) - 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 - else: - # Switch: empty value, and the next token is left for the positionals. - record(name, "") - - 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. +def _record_matches(criterion: CliCalledCriterion, argv: list[str], record: dict[str, Any]) -> bool: + """Whether one log record satisfies every configured facet of the criterion. - ``values`` is None when the flag was not passed at all. Every non-``absent`` - predicate is satisfied by ANY of a repeated flag's values. + ``tool`` is checked here rather than in :func:`argv_matches` because it is a + property of the RECORD, not of the arguments -- the shim that serves a + response knows which tool it is before it looks at argv. """ - if predicate.absent: - return values is None - if predicate.present: - return values is not None - if values is None: - return False - if predicate.equals is not None: - return any(value == predicate.equals for value in values) - if predicate.contains is not None: - return any(predicate.contains in value for value in values) - if predicate.any_of is not None: - allowed = set(predicate.any_of) - return any(value in allowed for value in values) - if predicate.matches_regex is not None: - regex = re.compile(predicate.matches_regex, predicate.flags) - return any(regex.search(value) is not None for value in values) - # Unreachable: FlagMatch guarantees exactly one predicate. Raise rather than - # return False so a predicate added without a matcher arm here fails loudly. - raise AssertionError(f"FlagMatch has no matcher arm: {predicate!r}") - - -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: return False - - # Declarations only. Folding `ignore_flags` in here made ignored SWITCHES - # value-bearing, which swallowed the next positional and reopened the guard - # false-PASS; an ignored flag that takes a value declares it in value_flags. - positional, flags = _split_flags( - argv, - 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 - spellings = criterion.verb_spellings - if spellings: - # Token-wise, not a subset and not a string startswith: `labellings confirm` - # must never be satisfied by `labellings unconfirm`. Taking the first match is - # safe because validation rejects one spelling prefixing another, so no argv - # can match two. - matched = next((tokens for tokens in spellings if positional[: len(tokens)] == tokens), None) - if matched is None: - return False - # Measured from the spelling that matched, since spellings can differ in length. - offset = len(matched) - - if criterion.positional is not None: - expected = criterion.positional - if positional[offset : offset + len(expected)] != expected: - return False - - if criterion.flags: - for name, predicate in criterion.flags.items(): - # [] means absent under every spelling, which _flag_matches - # distinguishes from a switch's "present with empty value" ([""]). - collected = [v for n in (name, *predicate.aliases) for v in flags.get(n, [])] - if not _flag_matches(predicate, collected or None): - return False - - return True + return argv_matches(criterion.match_spec, argv) @register_criterion diff --git a/src/coder_eval/invocation_log.py b/src/coder_eval/invocation_log.py index 40228f57..f1efa0d7 100644 --- a/src/coder_eval/invocation_log.py +++ b/src/coder_eval/invocation_log.py @@ -6,7 +6,9 @@ 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. +embedded literals, the argv matcher that dispatches its per-invocation responses +arrives as embedded SOURCE (:mod:`coder_eval.argv_match`, stdlib-only for exactly +that reason), 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) @@ -15,6 +17,7 @@ import json import sys +from importlib import resources from coder_eval.models import RecordedCli @@ -41,13 +44,16 @@ EXIT_CODE = {exit_code!r} STDOUT_TEXT = {stdout!r} STDERR_TEXT = {stderr!r} - +# Per-invocation responses in declaration order, empty when the entry declared +# none -- in which case every invocation gets the three defaults above. +RULES = {rules!r} +{matcher_source} 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): +def record(argv, exit_code, rule): """Append this invocation to the log. Best-effort: a logging failure must never break the command the agent ran, @@ -58,6 +64,12 @@ def record(argv, exit_code): 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. + + `rule` is the index of the response rule that answered, recorded only when + one did. Without it, "no rule matched, so this is the default" and "rule 2 + answered, and happens to look like the default" are indistinguishable in the + log -- the first question asked when an expected canned response does not + arrive. """ entry = {{ "ts": round(time.time(), 3), @@ -65,6 +77,8 @@ def record(argv, exit_code): "argv": list(argv), "exit": exit_code, }} + if rule is not None: + entry["rule"] = rule try: # ensure_ascii escapes non-ASCII and any stray surrogate from # undecodable argv bytes, so an exotic argument cannot make this write @@ -82,20 +96,46 @@ def record(argv, exit_code): pass +def respond(argv): + """Pick this invocation's (exit code, stdout, stderr, matched rule index). + + First matching rule wins; whatever no rule claims gets the defaults. The + matcher above is embedded only when RULES is non-empty, so this guard is + what keeps `select_rule` from being named when it was not embedded. + """ + if not RULES: + return EXIT_CODE, STDOUT_TEXT, STDERR_TEXT, None + try: + selected = select_rule(RULES, list(argv)) + except Exception as exc: + # Best-effort, like the log write: a matcher fault must not turn the stub + # into a crashing executable, which the agent would read as the tool + # itself breaking in a way the task never described. + sys.stderr.write("coder_eval recorder: response matching failed: %r\\n" % (exc,)) + return EXIT_CODE, STDOUT_TEXT, STDERR_TEXT, None + if selected is None: + return EXIT_CODE, STDOUT_TEXT, STDERR_TEXT, None + index, rule = selected + return rule["exit"], rule["stdout"], rule["stderr"], index + + def main(argv): - """Record the invocation, then fail like the tool would with nothing behind it. + """Answer the invocation from the canned responses, and record what happened. - Nothing is executed: no network, no auth, no side effects. A test that needs + Nothing is executed: no network, no auth, no side effects -- a response is + text the task author wrote, never the real tool's output. 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 + 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 + args = argv[1:] + exit_code, stdout_text, stderr_text, rule = respond(args) + record(args, exit_code, rule) + if stdout_text: + sys.stdout.write(stdout_text) + if stderr_text: + sys.stderr.write(stderr_text) + return exit_code if __name__ == "__main__": @@ -103,6 +143,26 @@ def main(argv): ''' +# Marked off because the embedded copy is the only place this source exists at +# runtime: whoever reads a generated shim needs to see which half is generated +# glue and which half is a verbatim module they can go and look up. +_MATCHER_SECTION = """ +# --- begin embedded coder_eval/argv_match.py --------------------------------- +{source} +# --- end embedded coder_eval/argv_match.py ----------------------------------- +""" + + +def _matcher_source() -> str: + """The stdlib-only argv matcher, as source to embed in a shim. + + Read as a package resource rather than reconstructed or re-implemented: the + shim must dispatch on the SAME matcher the ``cli_called`` criterion grades + with, and every transformation in between is a place the two could diverge. + """ + return resources.files("coder_eval").joinpath("argv_match.py").read_text(encoding="utf-8") + + def render_recorder(spec: RecordedCli, interpreter: str | None = None) -> str: """Render the shim source for one ``record_cli`` entry. @@ -110,13 +170,28 @@ def render_recorder(spec: RecordedCli, interpreter: str | None = None) -> str: 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. + + The matcher is embedded only when the entry declares ``responses``: a shim + that answers every invocation the same way never consults it, and leaving it + out keeps the common shim as small as it was before rules existed. """ + rules = [ + { + "when": response.when.match_spec, + "exit": response.exit_code, + "stdout": response.stdout, + "stderr": response.stderr, + } + for response in spec.responses + ] return _TEMPLATE.format( interpreter=interpreter or sys.executable, tool=spec.tool, exit_code=spec.exit_code, stdout=spec.stdout, stderr=spec.stderr, + rules=rules, + matcher_source=_MATCHER_SECTION.format(source=_matcher_source()) if rules else "", log_filename=LOG_FILENAME, ) diff --git a/src/coder_eval/models/__init__.py b/src/coder_eval/models/__init__.py index 97d4f64c..ccee11ae 100644 --- a/src/coder_eval/models/__init__.py +++ b/src/coder_eval/models/__init__.py @@ -20,6 +20,12 @@ parse_agent_config, ) +# Argv matching (shared by cli_called and record_cli response rules) +from coder_eval.models.cli_match import ( + CliMatch, + FlagMatch, +) + # Container paths (leaf constants; re-exported so consumers obey CE001) from coder_eval.models.container_paths import ( CONTAINER_INPUT_DIR, @@ -47,7 +53,6 @@ FileContainsCriterion, FileExistsCriterion, FileMatchesRegexCriterion, - FlagMatch, JMESPathAssertion, JsonCheckCriterion, LivePolarity, @@ -165,6 +170,7 @@ from coder_eval.models.sandbox import ( RECORD_CLI_DIR, RECORD_CLI_LOG, + CliResponse, DockerBuildConfig, DockerDriverConfig, NodeEnvConfig, @@ -252,6 +258,8 @@ "ReferenceComparisonCriterion", "CommandExecutedCriterion", "CliCalledCriterion", + "CliMatch", + "CliResponse", "FlagMatch", "CommandsEfficiencyCriterion", "UiPathEvalCriterion", diff --git a/src/coder_eval/models/cli_match.py b/src/coder_eval/models/cli_match.py new file mode 100644 index 00000000..e60fb88d --- /dev/null +++ b/src/coder_eval/models/cli_match.py @@ -0,0 +1,385 @@ +"""Argv-matching models shared by ``cli_called`` and ``record_cli`` response rules. + +A cycle-free leaf, like :mod:`coder_eval.models.judge_defaults`: both +:mod:`coder_eval.models.criteria` and :mod:`coder_eval.models.sandbox` import it, +and sandbox.py could not import from criteria.py in any case (criteria.py already +takes ``RECORD_CLI_LOG`` from sandbox.py). + +The matching *semantics* live in :mod:`coder_eval.argv_match`, which is +stdlib-only because its source is embedded into generated shims. This module +holds the authoring surface — the pydantic models and the validators that reject +a pattern which cannot mean what it looks like — and lowers it to the plain spec +dict that engine consumes. +""" + +from __future__ import annotations + +import itertools +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from coder_eval.argv_match import is_number + + +class FlagMatch(BaseModel): + """Predicate for ONE flag value. + + Exactly one predicate field may be set. In YAML a bare scalar is accepted as + shorthand for ``equals`` (``model: gemini_2_5_pro`` == ``model: {equals: + gemini_2_5_pro}``), which keeps the common case unnested. + + ``absent: true`` asserts the flag was NOT passed — distinct from "passed with + a different value", and the reason this is a predicate rather than a bare + ``dict[str, str]`` on the criterion. + + The one-predicate rule means a conjunction on a single flag ("contains BOTH + A and B") is not expressible here. Either declare two ``cli_called`` criteria + over the same log, or use one ``matches_regex`` that spans both — the latter + is what a heredoc-built JSON payload usually wants, together with + ``flags: 16`` (``re.DOTALL``) so ``.`` crosses the payload's newlines. + """ + + model_config = ConfigDict(extra="forbid") + + equals: str | None = Field(default=None, description="Flag value must equal this string exactly") + contains: str | None = Field(default=None, description="Flag value must contain this substring") + matches_regex: str | None = Field( + default=None, + description="Flag value must match this regex. Scoped to ONE value, unlike a whole-line pattern", + ) + any_of: list[str] | None = Field( + default=None, + min_length=1, + description=( + "Flag value must equal one of these strings. Non-empty: an empty list would match " + "nothing, so a max_count: 0 guard built on it would pass vacuously" + ), + ) + absent: bool = Field(default=False, description="Flag must NOT be present in the invocation") + present: bool = Field( + default=False, + description=( + "Flag must be present, whatever its value -- the predicate for a boolean switch. Unlike " + '`equals: ""` it survives a CLI that spells the switch `--force true`, and it never makes ' + "the flag value-bearing, so asserting a switch cannot swallow the next positional" + ), + ) + aliases: list[str] = Field( + default_factory=list, + description=( + "Other names for the SAME flag, e.g. aliases: [y] on a `yes` predicate so `-y` and " + "`--yes` are one flag. Values are gathered across every name: `present` holds if any " + "appeared, `absent` only if none did, a value predicate matches if any value under any " + "name satisfies it" + ), + ) + flags: int = Field( + default=0, + description=( + "Regex flags for matches_regex (re.IGNORECASE=2, re.MULTILINE=8, re.DOTALL=16), " + "mirroring FileMatchesRegexCriterion.flags. DOTALL is the usual need, since a " + "heredoc-built flag value spans lines" + ), + ) + + @property + def needs_value(self) -> bool: + """Whether evaluating this predicate requires the flag's VALUE. + + Presence predicates (``present`` / ``absent``) do not, so they must not + make a flag value-bearing. Otherwise asserting a boolean switch would + make it consume the following token: adding ``flags: {yes: {present: + true}}`` to a guard on ``delete --yes proj-1`` would bind + ``yes=proj-1``, drop ``proj-1`` from the positionals, and hand the guard + a false PASS -- reintroducing the very defect declared value-binding + exists to prevent. + """ + return not (self.present or self.absent) + + @model_validator(mode="before") + @classmethod + def _coerce_scalar_shorthand(cls, value: Any) -> Any: + """Accept ``model: gemini_2_5_pro`` as ``model: {equals: ...}``.""" + if isinstance(value, str): + return {"equals": value} + return value + + @model_validator(mode="after") + def _exactly_one_predicate(self) -> FlagMatch: + set_predicates = [ + name for name in ("equals", "contains", "matches_regex", "any_of") if getattr(self, name) is not None + ] + if self.absent: + set_predicates.append("absent") + if self.present: + set_predicates.append("present") + if len(set_predicates) != 1: + msg = ( + "FlagMatch requires exactly one of equals / contains / matches_regex / any_of / absent / present, " + f"got {sorted(set_predicates) or 'none'}" + ) + raise ValueError(msg) + # `flags` only reaches re.compile via matches_regex; setting it beside any + # other predicate is a silent no-op, so reject it rather than mislead. + if self.flags and self.matches_regex is None: + msg = f"FlagMatch.flags applies only to matches_regex, but the predicate is {set_predicates[0]!r}" + raise ValueError(msg) + return self + + +# The argv facets every matching surface must offer. `cli_called` declares these +# fields itself (with grading-specific guidance in each description) rather than +# inheriting them, so a facet added to one surface and forgotten on the other is +# caught by the parity test in tests/test_cli_match_parity.py instead of shipping +# as a rule the criterion cannot express. +MATCH_FACET_FIELDS: tuple[str, ...] = ("verb", "verb_any_of", "positional", "flags", "value_flags", "ignore_flags") + + +def verb_spellings_of(verb: str | None, verb_any_of: list[str] | None) -> list[list[str]]: + """Each accepted verb as its token list; empty when there is no verb constraint. + + The only place either verb field is split, so the validators, the matcher and + the failure detail cannot disagree. + """ + if verb is not None: + return [verb.split()] + if verb_any_of is not None: + return [spelling.split() for spelling in verb_any_of] + return [] + + +def validate_verbs(verb: str | None, verb_any_of: list[str] | None, spellings: list[list[str]], label: str) -> None: + """Reject verb declarations that cannot mean what they look like. + + ``label`` names the surface (``cli_called``, ``record_cli response when``) so + the message points at the block the author actually wrote. + """ + if verb is not None and verb_any_of is not None: + msg = f"{label} accepts verb or verb_any_of, not both" + raise ValueError(msg) + # Falsy, so an at-least-one-facet check would read it as "no verb". + if verb_any_of is not None and not verb_any_of: + msg = f"{label} verb_any_of must not be empty: drop the field to match any verb" + raise ValueError(msg) + # A character count would pass " ", whose split() is an empty prefix. + if any(not tokens for tokens in spellings): + msg = f"{label} verb must not be blank: a blank verb is an empty prefix and matches every invocation" + raise ValueError(msg) + # A verb is compared against the NON-FLAG arguments, so a flag written into it + # can never match anything -- and the failure is silent: the criterion scores 0 + # against a log that holds the very call it describes, and a response rule falls + # through to the tool's default. Inviting, too, since a whole verb reads like a + # command line. `is_number` mirrors the splitter's own rule so this check cannot + # forbid a token (`-1`) that the matcher would in fact have seen. + for tokens in spellings: + for token in tokens: + if token.startswith("-") and token != "-" and not is_number(token.lstrip("-")): + msg = ( + f"{label} verb token {token!r} looks like a flag. A verb matches only the " + "non-flag arguments, so a flag inside it can never match. Put it in `flags:` " + f"instead, e.g. flags: {{{token.lstrip('-').split('=')[0]}: }}." + ) + raise ValueError(msg) + for first, second in itertools.combinations(spellings, 2): + if first == second: + msg = f"{label} verb_any_of lists {' '.join(first)!r} twice" + raise ValueError(msg) + # Sorting by length is total here: two DISTINCT entries of equal length + # cannot prefix each other, since an equal-length prefix is the same list. + shorter, longer = sorted((first, second), key=len) + if longer[: len(shorter)] == shorter: + msg = ( + f"{label} verb_any_of entry {' '.join(shorter)!r} is a prefix of " + f"{' '.join(longer)!r}; the shorter one already accepts every invocation the " + "longer one does, so drop the longer entry or list only the verbs you mean." + ) + raise ValueError(msg) + + +def validate_positional(positional: list[str] | None, label: str) -> None: + """Reject an empty positional list, which slices to itself and asserts nothing.""" + if positional is not None and not positional: + msg = ( + f"{label} positional must not be empty: an empty list asserts nothing. List the " + "arguments you expect, or drop the field." + ) + raise ValueError(msg) + + +def validate_flag_ownership(flags: dict[str, FlagMatch] | None, ignore_flags: list[str], label: str) -> None: + """Reject flag predicates that collide with each other or with ``ignore_flags``. + + An alias that is also a key, or shared between two predicates, would make + which predicate owns a recorded flag depend on dict order. A predicate on an + ignored flag can never be evaluated: ignore_flags drops the flag before any + predicate runs, so ``absent`` would pass vacuously and ``equals`` could never + match. + """ + seen: dict[str, str] = {} + for key, predicate in (flags or {}).items(): + for name in (key, *predicate.aliases): + if name in seen and seen[name] != key: + msg = ( + f"{label} flag name {name!r} is claimed by both {seen[name]!r} and {key!r} " + "(via aliases); a flag can belong to only one predicate" + ) + raise ValueError(msg) + seen[name] = key + if key in predicate.aliases: + msg = f"{label} flag {key!r} lists itself in aliases" + raise ValueError(msg) + + shadowed = sorted(set(seen) & set(ignore_flags)) + if shadowed: + names = ", ".join(repr(n) for n in shadowed) + msg = ( + f"{label} flag predicate(s) {names} are also listed in ignore_flags (directly or as " + "an alias), which drops them before matching. Remove them from ignore_flags, or drop " + "the predicate." + ) + raise ValueError(msg) + + +def build_match_spec( + *, + verb_spellings: list[list[str]], + positional: list[str] | None, + flags: dict[str, FlagMatch] | None, + value_flags: list[str], + ignore_flags: list[str], +) -> dict[str, Any]: + """Lower an authored match surface to the plain dict :mod:`coder_eval.argv_match` reads. + + JSON-serializable on purpose: the same dict is embedded verbatim into a + generated shim, so a spec the criterion evaluates in-process and a spec the + shim evaluates in the sandbox are the same bytes. + """ + return { + "verb_spellings": verb_spellings, + "positional": positional, + "flags": {name: predicate.model_dump() for name, predicate in flags.items()} if flags else None, + "value_flags": list(value_flags), + "ignore_flags": list(ignore_flags), + } + + +class CliMatch(BaseModel): + """A pattern over ONE invocation's arguments, used to dispatch a canned response. + + The ``when:`` block of a ``record_cli`` response rule. Facets are ANDed, and + an unmentioned facet is unconstrained — an extra ``--output json`` never + stops a rule from matching. Matching semantics are identical to the + ``cli_called`` criterion of the same shape, so the pattern that selects a + stub response is the pattern that grades it. + + Always a mapping, never a bare string: a pattern has six possible facets, so + a lone ``"ixp dummy1"`` would leave the reader to infer which one it sets, and + a quoted verb reads enough like a command line to invite the flags a verb + cannot hold. :class:`FlagMatch` one level down keeps its scalar shorthand + (``flags: {output: json}``) — a single-valued predicate has only one facet a + scalar could mean, so nothing is left to infer there. + """ + + model_config = ConfigDict(extra="forbid") + + verb: str | None = Field( + default=None, + description=( + "Whitespace-separated subcommand chain that must be an ORDERED PREFIX of the " + "invocation's non-flag arguments, compared token by token (so 'projects list' never " + "matches 'projects lists', and 'labellings confirm' never matches 'labellings " + "unconfirm'). Tokens after it are unconstrained, so a short verb claims every " + "invocation under it: 'projects' answers 'projects delete' as readily as 'projects get'" + ), + ) + verb_any_of: list[str] | None = Field( + default=None, + description=( + "Alternative whole verbs; matches if ANY of them does, e.g. ['projects list', " + "'projects get'] to serve one response for both spellings. Each entry is a complete " + "verb in the same form `verb` takes, NOT one token of a chain. Mutually exclusive with " + "`verb`" + ), + ) + positional: list[str] | None = Field( + default=None, + description=( + "Non-flag arguments that must follow the verb, in order. A PREFIX of what followed, so " + "anything past them is unconstrained. Use it to answer differently per project/id, e.g. " + "positional: ['proj-1']. Depends on value_flags being complete — an undeclared flag's " + "value stays non-flag and shifts these slots" + ), + ) + flags: dict[str, FlagMatch] | None = Field( + default=None, + description=( + "Flag name (without leading dashes) to predicate. A bare scalar means 'equals', e.g. " + "flags: {output: json} to serve JSON only when the agent asked for it. Flags not listed " + "are ignored, so an unrelated flag never stops the rule matching" + ), + ) + value_flags: list[str] = Field( + default_factory=lambda: ["output"], + description=( + "Flag names (no leading dashes) that consume a following token as their value. Keys of " + "`flags` are value-bearing already; everything else is a switch whose following token " + "stays positional. Declare a flag here when its value would otherwise be read as a " + "positional, e.g. [folder] for `--folder F proj-1`. Defaults to [output]" + ), + ) + ignore_flags: list[str] = Field( + default_factory=list, + description=( + "Flag names dropped before matching. Empty by default, unlike the cli_called criterion: " + "a response rule dispatches rather than grades, so nothing is outcome-invisible here and " + "a rule may key on any flag it declares" + ), + ) + + @property + def verb_spellings(self) -> list[list[str]]: + """Each accepted verb as its token list; empty when there is no verb constraint.""" + return verb_spellings_of(self.verb, self.verb_any_of) + + @property + def match_spec(self) -> dict[str, Any]: + """This pattern as the plain dict :func:`coder_eval.argv_match.argv_matches` reads.""" + return build_match_spec( + verb_spellings=self.verb_spellings, + positional=self.positional, + flags=self.flags, + value_flags=self.value_flags, + ignore_flags=self.ignore_flags, + ) + + @model_validator(mode="before") + @classmethod + def _reject_scalar_shorthand(cls, value: Any) -> Any: + """Name the fix, rather than let pydantic report a bare type error. + + ``when: "ixp dummy1"`` is the obvious thing to try, and the generic + "Input should be a valid dictionary" says nothing about which key was + meant. + """ + if isinstance(value, str): + msg = f'record_cli response `when` must be a mapping, not a bare string: use {{verb: "{value}"}}' + raise ValueError(msg) + return value + + @model_validator(mode="after") + def _validate_match(self) -> CliMatch: + label = "record_cli response `when`" + validate_verbs(self.verb, self.verb_any_of, self.verb_spellings, label) + validate_positional(self.positional, label) + validate_flag_ownership(self.flags, self.ignore_flags, label) + # Falsiness, not `is None`: `verb: ""` would otherwise match every + # invocation and shadow every rule below it. + if not self.verb and not self.verb_any_of and not self.positional and not self.flags: + msg = ( + "record_cli response `when` requires at least one of verb / verb_any_of / positional " + "/ flags. A rule that matches everything is the tool's default response: set the " + "entry's own exit_code / stdout / stderr instead." + ) + raise ValueError(msg) + return self diff --git a/src/coder_eval/models/criteria.py b/src/coder_eval/models/criteria.py index b0822638..ea0daaae 100644 --- a/src/coder_eval/models/criteria.py +++ b/src/coder_eval/models/criteria.py @@ -8,7 +8,6 @@ from __future__ import annotations -import itertools from abc import ABC, abstractmethod from pathlib import PurePosixPath from typing import Annotated, Any, ClassVar, Literal, Self @@ -16,6 +15,14 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from coder_eval.models.agent_config import AgentConfig, ClaudeCodeAgentConfig, parse_agent_config +from coder_eval.models.cli_match import ( + FlagMatch, + build_match_spec, + validate_flag_ownership, + validate_positional, + validate_verbs, + verb_spellings_of, +) 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 @@ -420,112 +427,6 @@ class FileMatchesRegexCriterion(BaseSuccessCriterion): flags: int = Field(default=0, description="Regex flags (e.g., re.IGNORECASE=2, re.MULTILINE=8, re.DOTALL=16)") -class FlagMatch(BaseModel): - """Predicate for ONE flag value within :class:`CliCalledCriterion`. - - Exactly one predicate field may be set. In YAML a bare scalar is accepted as - shorthand for ``equals`` (``model: gemini_2_5_pro`` == ``model: {equals: - gemini_2_5_pro}``), which keeps the common case unnested. - - ``absent: true`` asserts the flag was NOT passed — distinct from "passed with - a different value", and the reason this is a predicate rather than a bare - ``dict[str, str]`` on the criterion. - - The one-predicate rule means a conjunction on a single flag ("contains BOTH - A and B") is not expressible here. Either declare two ``cli_called`` criteria - over the same log, or use one ``matches_regex`` that spans both — the latter - is what a heredoc-built JSON payload usually wants, together with - ``flags: 16`` (``re.DOTALL``) so ``.`` crosses the payload's newlines. - """ - - model_config = ConfigDict(extra="forbid") - - equals: str | None = Field(default=None, description="Flag value must equal this string exactly") - contains: str | None = Field(default=None, description="Flag value must contain this substring") - matches_regex: str | None = Field( - default=None, - description="Flag value must match this regex. Scoped to ONE value, unlike a whole-line pattern", - ) - any_of: list[str] | None = Field( - default=None, - min_length=1, - description=( - "Flag value must equal one of these strings. Non-empty: an empty list would match " - "nothing, so a max_count: 0 guard built on it would pass vacuously" - ), - ) - absent: bool = Field(default=False, description="Flag must NOT be present in the invocation") - present: bool = Field( - default=False, - description=( - "Flag must be present, whatever its value -- the predicate for a boolean switch. Unlike " - '`equals: ""` it survives a CLI that spells the switch `--force true`, and it never makes ' - "the flag value-bearing, so asserting a switch cannot swallow the next positional" - ), - ) - aliases: list[str] = Field( - default_factory=list, - description=( - "Other names for the SAME flag, e.g. aliases: [y] on a `yes` predicate so `-y` and " - "`--yes` are one flag. Values are gathered across every name: `present` holds if any " - "appeared, `absent` only if none did, a value predicate matches if any value under any " - "name satisfies it" - ), - ) - flags: int = Field( - default=0, - description=( - "Regex flags for matches_regex (re.IGNORECASE=2, re.MULTILINE=8, re.DOTALL=16), " - "mirroring FileMatchesRegexCriterion.flags. DOTALL is the usual need, since a " - "heredoc-built flag value spans lines" - ), - ) - - @property - def needs_value(self) -> bool: - """Whether evaluating this predicate requires the flag's VALUE. - - Presence predicates (``present`` / ``absent``) do not, so they must not - make a flag value-bearing. Otherwise asserting a boolean switch would - make it consume the following token: adding ``flags: {yes: {present: - true}}`` to a guard on ``delete --yes proj-1`` would bind - ``yes=proj-1``, drop ``proj-1`` from the positionals, and hand the guard - a false PASS -- reintroducing the very defect declared value-binding - exists to prevent. - """ - return not (self.present or self.absent) - - @model_validator(mode="before") - @classmethod - def _coerce_scalar_shorthand(cls, value: Any) -> Any: - """Accept ``model: gemini_2_5_pro`` as ``model: {equals: ...}``.""" - if isinstance(value, str): - return {"equals": value} - return value - - @model_validator(mode="after") - def _exactly_one_predicate(self) -> FlagMatch: - set_predicates = [ - name for name in ("equals", "contains", "matches_regex", "any_of") if getattr(self, name) is not None - ] - if self.absent: - set_predicates.append("absent") - if self.present: - set_predicates.append("present") - if len(set_predicates) != 1: - msg = ( - "FlagMatch requires exactly one of equals / contains / matches_regex / any_of / absent / present, " - f"got {sorted(set_predicates) or 'none'}" - ) - raise ValueError(msg) - # `flags` only reaches re.compile via matches_regex; setting it beside any - # other predicate is a silent no-op, so reject it rather than mislead. - if self.flags and self.matches_regex is None: - msg = f"FlagMatch.flags applies only to matches_regex, but the predicate is {set_predicates[0]!r}" - raise ValueError(msg) - return self - - class CliCalledCriterion(BaseSuccessCriterion): """Check whether a CLI invocation matching a structured pattern was recorded. @@ -664,40 +565,28 @@ def verb_spellings(self) -> list[list[str]]: The only place either verb field is split, so the validators, the matcher and the failure detail cannot disagree. """ - if self.verb is not None: - return [self.verb.split()] - if self.verb_any_of is not None: - return [spelling.split() for spelling in self.verb_any_of] - return [] + return verb_spellings_of(self.verb, self.verb_any_of) + + @property + def match_spec(self) -> dict[str, Any]: + """This criterion's argv facets as the dict :mod:`coder_eval.argv_match` reads. + + The same lowering a ``record_cli`` response rule uses, so a rule that + serves a response and the criterion that grades it cannot read one argv + two ways. ``tool`` stays out: it matches a log record's field, not argv. + """ + return build_match_spec( + verb_spellings=self.verb_spellings, + positional=self.positional, + flags=self.flags, + value_flags=self.value_flags, + ignore_flags=self.ignore_flags, + ) @model_validator(mode="after") def _validate_verb(self) -> CliCalledCriterion: """Verb rules, kept off _validate_bounds so neither grows unreadable.""" - if self.verb is not None and self.verb_any_of is not None: - msg = "cli_called accepts verb or verb_any_of, not both" - raise ValueError(msg) - # Falsy, so the at-least-one-facet check below would read it as "no verb". - if self.verb_any_of is not None and not self.verb_any_of: - msg = "cli_called verb_any_of must not be empty: drop the field to match any verb" - raise ValueError(msg) - # A character count would pass " ", whose split() is an empty prefix. - if any(not tokens for tokens in self.verb_spellings): - msg = "cli_called verb must not be blank: a blank verb is an empty prefix and matches every record" - raise ValueError(msg) - for first, second in itertools.combinations(self.verb_spellings, 2): - if first == second: - msg = f"cli_called verb_any_of lists {' '.join(first)!r} twice" - raise ValueError(msg) - # Sorting by length is total here: two DISTINCT entries of equal length - # cannot prefix each other, since an equal-length prefix is the same list. - shorter, longer = sorted((first, second), key=len) - if longer[: len(shorter)] == shorter: - msg = ( - f"cli_called verb_any_of entry {' '.join(shorter)!r} is a prefix of " - f"{' '.join(longer)!r}; the shorter one already accepts every invocation the " - "longer one does, so drop the longer entry or list only the verbs you mean." - ) - raise ValueError(msg) + validate_verbs(self.verb, self.verb_any_of, self.verb_spellings, "cli_called") return self @model_validator(mode="after") @@ -715,45 +604,15 @@ def _validate_bounds(self) -> CliCalledCriterion: raise ValueError(msg) # Matching slices an empty expectation and compares it to itself, so this reads # as "took no arguments" while asserting nothing. - if self.positional is not None and not self.positional: - msg = ( - "cli_called positional must not be empty: an empty list asserts nothing. List the " - "arguments you expect, or drop the field." - ) - raise ValueError(msg) + validate_positional(self.positional, "cli_called") # Falsiness, not `is None`: `verb: ""` slipped past an `is None` check here and - # then matched every record, scoring 1.0. + # then matched every record, scoring 1.0. `tool` counts as a facet here (but not + # for a response rule), since a criterion may legitimately count every + # invocation of one shadowed executable. if not self.verb and not self.verb_any_of and not self.positional and not self.flags and not self.tool: msg = "cli_called requires at least one of verb / verb_any_of / positional / flags / tool to match on" raise ValueError(msg) - # A predicate on an ignored flag can never be evaluated: ignore_flags drops - # the flag before any predicate runs, so `absent` would pass vacuously and - # `equals` could never match. - # An alias that is also a key, or shared between two predicates, would make - # which predicate owns a recorded flag depend on dict order. - seen: dict[str, str] = {} - for key, predicate in (self.flags or {}).items(): - for name in (key, *predicate.aliases): - if name in seen and seen[name] != key: - msg = ( - f"cli_called flag name {name!r} is claimed by both {seen[name]!r} and {key!r} " - "(via aliases); a flag can belong to only one predicate" - ) - raise ValueError(msg) - seen[name] = key - if key in predicate.aliases: - msg = f"cli_called flag {key!r} lists itself in aliases" - raise ValueError(msg) - - shadowed = sorted(set(seen) & set(self.ignore_flags)) - if shadowed: - names = ", ".join(repr(n) for n in shadowed) - msg = ( - f"cli_called flag predicate(s) {names} are also listed in ignore_flags (directly or as " - "an alias), which drops them before matching. Remove them from ignore_flags, or drop " - "the predicate." - ) - raise ValueError(msg) + validate_flag_ownership(self.flags, self.ignore_flags, "cli_called") return self diff --git a/src/coder_eval/models/sandbox.py b/src/coder_eval/models/sandbox.py index d2083d09..afe9eb9b 100644 --- a/src/coder_eval/models/sandbox.py +++ b/src/coder_eval/models/sandbox.py @@ -7,6 +7,7 @@ from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator, model_validator +from coder_eval.models.cli_match import CliMatch from coder_eval.models.container_paths import CONTAINER_WORK_DIR, RESERVED_CONTAINER_DIRS from coder_eval.models.merge_strategy import MergeField from coder_eval.models.templates import TemplateSource @@ -325,6 +326,40 @@ def _validate_working_dir(cls, v: str | None) -> str | None: ) +class CliResponse(BaseModel): + """One canned response, served when an invocation matches ``when``. + + The reason a shadowed tool can answer `uip ixp dummy1` and `uip ixp dummy2` + differently instead of returning one fixed pair of streams for everything an + agent types. Rules are tried in declaration order and the FIRST match wins, + so the specific rule goes above the general one; an invocation matching no + rule falls back to the entry's own ``exit_code`` / ``stdout`` / ``stderr``. + + ``exit_code`` defaults to 0 here, the opposite of :class:`RecordedCli`: a rule + exists because the author described this exact invocation, so the natural + reading is "and this is what it answers", whereas an undescribed one should + look like a tool that failed rather than a silent success. + """ + + model_config = ConfigDict(extra="forbid") + + when: CliMatch = Field( + description=( + 'Pattern the invocation must match, e.g. {verb: "ixp dummy1"}. Always a mapping -- same ' + "facets and same matching semantics as the cli_called criterion, so the pattern that " + "serves a response is the pattern that grades it" + ) + ) + exit_code: int = Field( + default=0, + ge=0, + le=255, + description="Exit status the shim returns for a matching invocation. Defaults to 0 (success)", + ) + stdout: str = Field(default="", description="Text the shim writes to stdout for a matching invocation") + stderr: str = Field(default="", description="Text the shim writes to stderr for a matching invocation") + + class RecordedCli(BaseModel): """One executable to shadow with a generated recording shim. @@ -335,6 +370,11 @@ class RecordedCli(BaseModel): ran without hand-rolling a mock and without the record shape being a contract between two repositories. + The fields below are what every invocation gets; ``responses`` overrides them + per invocation, so one shadowed ``uip`` can answer ``ixp dummy1`` and + ``ixp dummy2`` differently — what an agent needs when its next step depends on + what the tool just told it. + 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, @@ -369,6 +409,17 @@ class RecordedCli(BaseModel): "would, so an agent reads a plausible error rather than silence" ), ) + responses: list[CliResponse] = MergeField( + strategy="replace", + default_factory=list, + description=( + "Per-invocation responses, tried in order until one matches; the fields above are the " + "fallback for an invocation none of them claim. Use it when the agent's next step " + "depends on what the tool answered -- `ixp projects list` returning a project the agent " + "then acts on, say -- instead of one fixed reply to everything. Replaced (not merged) " + "across config layers, like the enclosing record_cli list" + ), + ) @field_validator("tool") @classmethod @@ -458,10 +509,11 @@ class SandboxConfig(BaseModel): "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." + "by default. Use instead of hand-writing a mock under mock_path_dirs when the test needs " + "a faithful record of what ran plus canned output -- one reply per entry, or a different " + "one per invocation via that entry's 'responses'. It does NOT proxy the real executable " + "(nothing is run, so no network, auth, or side effect); supply your own mock for that. " + "Replaced (not merged) across config layers, like mock_path_dirs." ), ) diff --git a/src/coder_eval/sandbox.py b/src/coder_eval/sandbox.py index 55a24b5a..bc3ea8ac 100644 --- a/src/coder_eval/sandbox.py +++ b/src/coder_eval/sandbox.py @@ -645,10 +645,11 @@ def _generate_cli_recorders(self) -> None: 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) + summary = ", ".join( + f"{s.tool}(exit {s.exit_code}" + (f", {len(s.responses)} rule(s)" if s.responses else "") + ")" + for s in self.config.record_cli ) + logger.info(f"Generated {len(self.config.record_cli)} CLI recorder(s) in {RECORD_CLI_DIR}/: {summary}") def _apply_starter_files_source(self, source: StarterFilesSource) -> None: """Create inline starter files in sandbox with overwrite tracking. diff --git a/tests/lint/rules/ce047_embedded_shim_stdlib_only.py b/tests/lint/rules/ce047_embedded_shim_stdlib_only.py new file mode 100644 index 00000000..5e1e9c9f --- /dev/null +++ b/tests/lint/rules/ce047_embedded_shim_stdlib_only.py @@ -0,0 +1,63 @@ +"""CE047: modules embedded into a generated sandbox shim import stdlib only. + +``invocation_log.render_recorder`` embeds the SOURCE of ``coder_eval/argv_match.py`` +into every ``record_cli`` shim that declares response rules. That shim runs inside +the sandbox, where ``coder_eval`` is not installed and no project dependency is +guaranteed — so a single ``from coder_eval.models import ...`` or ``import +pydantic`` added to the embedded module makes every shadowed CLI die with an +ImportError the moment the agent runs it. The failure surfaces as "the tool is +broken", never as "the harness embedded an unimportable module", and it costs a +whole run to diagnose. + +Import-time enforcement (a test that renders and executes a shim) only catches it +when a test happens to declare a response rule; this rule catches the import the +moment it is written. + +A stdlib module that is genuinely needed is added to ``STDLIB_ALLOWED`` below — +deliberately an allowlist rather than a check against ``sys.stdlib_module_names``, +so growing the shim's surface is a decision someone makes on purpose. +""" + +import ast +import re + +from tests.lint.rules.base import BaseRule + + +class EmbeddedShimStdlibOnly(BaseRule): + id = "CE047" + + # Modules whose source is embedded into a generated shim. Keyed by path + # fragment so the rule fires on the file itself, wherever the tree is rooted. + _EMBEDDED = re.compile(r"[/\\]coder_eval[/\\]argv_match\.py$") + + # Small on purpose: everything here has to exist in whatever interpreter the + # sandbox's shebang resolves to. + STDLIB_ALLOWED = frozenset({"re", "json", "os", "sys", "time", "shlex", "itertools", "typing"}) + + def __init__(self, filepath: str) -> None: + super().__init__(filepath) + self._embedded = bool(self._EMBEDDED.search(filepath)) + + def _check(self, node: ast.AST, module: str | None) -> None: + if not self._embedded or module is None: + return + root = module.split(".")[0] + if root in self.STDLIB_ALLOWED: + return + self.violation( + node, + f"'{module}' is imported by a module embedded into generated sandbox shims, which run " + "where coder_eval and its dependencies are not installed. Use the standard library, or " + f"add '{root}' to CE047's STDLIB_ALLOWED if it really is stdlib.", + ) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + # A relative import (level > 0) is a package import by definition. + self._check(node, node.module if node.level == 0 else f".{node.module or ''}") + self.generic_visit(node) + + def visit_Import(self, node: ast.Import) -> None: + for alias in node.names: + self._check(node, alias.name) + self.generic_visit(node) diff --git a/tests/lint/runner.py b/tests/lint/runner.py index 092e97a6..e9f24a33 100644 --- a/tests/lint/runner.py +++ b/tests/lint/runner.py @@ -26,6 +26,7 @@ from tests.lint.rules.ce039_config_error_escalates import ConfigErrorEscalates from tests.lint.rules.ce043_no_command_output_truncation import NoCommandOutputTruncation from tests.lint.rules.ce046_env_info_spreads_super import EnvInfoSpreadsSuper +from tests.lint.rules.ce047_embedded_shim_stdlib_only import EmbeddedShimStdlibOnly from tests.lint.rules.no_agent_timing_access import NoAgentTimingAccess from tests.lint.rules.no_blocking_io_in_async import NoBlockingIoInAsync from tests.lint.rules.no_cli_imports_in_core import NoCliImportsInCore @@ -75,6 +76,7 @@ ConfigErrorEscalates, NoCommandOutputTruncation, EnvInfoSpreadsSuper, + EmbeddedShimStdlibOnly, ] # Anti-shadow invariant (mirrors AgentRegistry / register_pricing): every CE rule diff --git a/tests/test_cli_called_criterion.py b/tests/test_cli_called_criterion.py index 8c2d79bb..0dfbaadb 100644 --- a/tests/test_cli_called_criterion.py +++ b/tests/test_cli_called_criterion.py @@ -6,7 +6,7 @@ import pytest from pydantic import ValidationError -from coder_eval.criteria.cli_called import _split_flags +from coder_eval.argv_match import split_flags from coder_eval.evaluation.checker import SuccessChecker from coder_eval.models import CliCalledCriterion, SandboxConfig from coder_eval.sandbox import Sandbox @@ -324,8 +324,8 @@ def test_malformed_line_now_fails_instead_of_being_skipped(self, sandbox_with_lo class TestArgvNormalization: def test_equals_form_and_space_form_are_equivalent(self): - space = _split_flags(["get", "--model", "pro"], frozenset(), frozenset({"model"})) - equals = _split_flags(["get", "--model=pro"], frozenset(), frozenset({"model"})) + space = split_flags(["get", "--model", "pro"], frozenset(), frozenset({"model"})) + equals = split_flags(["get", "--model=pro"], frozenset(), frozenset({"model"})) assert space == equals == (["get"], {"model": ["pro"]}) def test_output_is_ignored_by_default(self, sandbox_with_log): @@ -344,13 +344,13 @@ def test_output_is_ignored_by_default(self, sandbox_with_log): assert SuccessChecker(sandbox).check(with_json).score == 1.0 def test_boolean_switch_does_not_consume_the_next_flag(self): - positional, flags = _split_flags(["delete", "proj-1", "--yes", "--force"], frozenset(), frozenset()) + positional, flags = split_flags(["delete", "proj-1", "--yes", "--force"], frozenset(), frozenset()) assert positional == ["delete", "proj-1"] assert flags == {"yes": [""], "force": [""]} def test_flag_like_value_stays_a_value(self): """A value that merely looks like a flag is still a value when quoted as one.""" - positional, flags = _split_flags( + positional, flags = split_flags( ["confirm", "--corrections", '[{"v":"--x"}]'], frozenset(), frozenset({"corrections"}) ) assert positional == ["confirm"] @@ -358,13 +358,13 @@ def test_flag_like_value_stays_a_value(self): def test_double_dash_terminates_flag_parsing(self): """`--` is consumed as a separator; what follows is positional, not a flag.""" - positional, flags = _split_flags(["run", "--", "--not-a-flag"], frozenset(), frozenset()) + positional, flags = split_flags(["run", "--", "--not-a-flag"], frozenset(), frozenset()) assert positional == ["run", "--not-a-flag"] assert flags == {} def test_lone_dash_is_positional(self): """A bare `-` is the stdin convention, not a flag.""" - positional, flags = _split_flags(["import", "-"], frozenset(), frozenset()) + positional, flags = split_flags(["import", "-"], frozenset(), frozenset()) assert positional == ["import", "-"] assert flags == {} @@ -461,13 +461,13 @@ def test_clustered_short_flags_are_split(self, sandbox_with_log): 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"})) == ( + 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"})) == ( + assert split_flags(["g", "-ff-002"], frozenset(), frozenset({"f"}), frozenset({"f"})) == ( ["g"], {"f": ["f-002"]}, ) @@ -475,35 +475,35 @@ def test_attached_value_on_a_short_flag(self): 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()) == ( + assert split_flags(["seek", "-1"], frozenset(), frozenset(), frozenset()) == ( ["seek", "-1"], {}, ) - assert _split_flags(["seek", "-1.5"], frozenset(), frozenset(), frozenset())[0] == ["seek", "-1.5"] + 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"})) == ( + 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( + positional, flags = split_flags( ["ixp", "proj", "get", "--limit", "-1", "proj-1"], frozenset(), frozenset({"limit"}) ) assert positional == ["ixp", "proj", "get", "proj-1"] assert flags == {"limit": ["-1"]} def test_undeclared_flag_leaves_its_neighbour_positional(self): - positional, flags = _split_flags(["ixp", "fields", "delete", "--yes", "proj-1"], frozenset(), frozenset()) + positional, flags = split_flags(["ixp", "fields", "delete", "--yes", "proj-1"], frozenset(), frozenset()) assert positional == ["ixp", "fields", "delete", "proj-1"] assert flags == {"yes": [""]} def test_equals_form_keeps_a_dash_leading_value_and_invents_no_flag(self): """`--offset=-1` used to drop the value AND invent a flag named `1`.""" - positional, flags = _split_flags(["get", "--offset=-1"], frozenset(), frozenset()) + positional, flags = split_flags(["get", "--offset=-1"], frozenset(), frozenset()) assert positional == ["get"] assert flags == {"offset": ["-1"]} diff --git a/tests/test_cli_match_parity.py b/tests/test_cli_match_parity.py new file mode 100644 index 00000000..62dac0d4 --- /dev/null +++ b/tests/test_cli_match_parity.py @@ -0,0 +1,102 @@ +"""Parity between the two argv-matching surfaces. + +``CliMatch`` (a ``record_cli`` response rule's ``when:``) and +``CliCalledCriterion`` declare the same argv facets separately, so each can carry +guidance written for its own job. Separate declarations can drift, and the drift +is silent in the worst direction: a task author stubs a response with a facet the +criterion cannot express, or grades on one no rule can dispatch on, and finds out +only when a suite scores wrong. + +The matching *semantics* cannot drift — both lower to one spec dict that +``coder_eval.argv_match`` evaluates — which is what these tests pin. +""" + +import pytest +from pydantic import ValidationError + +from coder_eval.argv_match import argv_matches +from coder_eval.models import CliCalledCriterion, CliMatch, CliResponse +from coder_eval.models.cli_match import MATCH_FACET_FIELDS + + +class TestFacetParity: + def test_both_surfaces_declare_every_match_facet(self): + for field in MATCH_FACET_FIELDS: + assert field in CliMatch.model_fields, f"CliMatch is missing match facet {field!r}" + assert field in CliCalledCriterion.model_fields, f"cli_called is missing match facet {field!r}" + + def test_criterion_adds_only_non_argv_fields(self): + """A facet on the criterion that CliMatch lacks is a rule authors cannot write.""" + # Everything the criterion adds is about the LOG (where to read, which + # record, how many), not about the arguments of one invocation. + non_argv = {"log", "tool", "min_count", "max_count"} + base = set(CliCalledCriterion.model_fields) - set(MATCH_FACET_FIELDS) + # Fields inherited from BaseSuccessCriterion are not match surface either. + from coder_eval.models import BaseSuccessCriterion + + added = base - set(BaseSuccessCriterion.model_fields) + assert added == non_argv, f"cli_called gained non-facet field(s) {sorted(added - non_argv)}; add to CliMatch" + + +# One pattern, both surfaces: (pattern, argv, expected verdict). +SHARED_CASES = [ + ({"verb": "ixp dummy1"}, ["ixp", "dummy1"], True), + ({"verb": "ixp dummy1"}, ["ixp", "dummy2"], False), + # Prefix semantics: tokens after the verb are unconstrained. + ({"verb": "ixp dummy1"}, ["ixp", "dummy1", "extra"], True), + # Token-wise, so a longer word never satisfies a shorter one. + ({"verb": "projects list"}, ["projects", "lists"], False), + ({"verb_any_of": ["projects list", "projects get"]}, ["projects", "get", "p1"], True), + ({"positional": ["proj-1"]}, ["proj-1", "tail"], True), + ({"verb": "projects get", "positional": ["proj-1"]}, ["projects", "get", "proj-2"], False), + # Not `output`: the criterion ignores that one by default (see the + # deliberate-divergence test below), so a shared case cannot use it. + ({"verb": "projects get", "flags": {"model": "pro"}}, ["projects", "get", "--model", "pro"], True), + ({"verb": "projects get", "flags": {"model": "pro"}}, ["projects", "get", "--model", "lite"], False), + ({"flags": {"force": {"present": True}}}, ["delete", "--force", "proj-1"], True), + ({"flags": {"force": {"absent": True}}}, ["delete", "proj-1"], True), +] + + +class TestSemanticParity: + """The same pattern, written on either surface, matches the same argv.""" + + @pytest.mark.parametrize(("pattern", "argv", "expected"), SHARED_CASES) + def test_rule_and_criterion_agree(self, pattern, argv, expected): + rule_spec = CliMatch.model_validate(pattern).match_spec + criterion = CliCalledCriterion(description="d", **pattern) + assert argv_matches(rule_spec, argv) is expected + assert argv_matches(criterion.match_spec, argv) is expected + + def test_ignore_flags_default_differs_and_that_is_deliberate(self): + """The criterion drops --output by default; a response rule does not. + + Grading must not depend on a flag that changes nothing about the outcome; + dispatch may legitimately answer differently for `--output json`. + """ + assert CliCalledCriterion(description="d", verb="get").ignore_flags == ["output"] + assert CliMatch(verb="get").ignore_flags == [] + + +class TestSharedValidation: + """One validator, so a pattern rejected on one surface is rejected on both.""" + + @pytest.mark.parametrize("verb", ["ixp projects get --output json", "ixp projects get -o", "delete --yes"]) + def test_a_flag_inside_a_verb_is_rejected_everywhere(self, verb): + """It validated, then matched nothing: the verb is compared to the NON-flag + arguments, so the criterion scored 0 against a log holding that very call and + a response rule fell through to the tool's default.""" + for build in ( + lambda v: CliMatch(verb=v), + lambda v: CliMatch(verb_any_of=[v]), + lambda v: CliCalledCriterion(description="d", verb=v), + lambda v: CliResponse(when={"verb": v}), + ): + with pytest.raises(ValidationError, match="looks like a flag"): + build(verb) + + @pytest.mark.parametrize("verb", ["ixp projects get", "head -1", "seek -1.5"]) + def test_tokens_the_matcher_would_really_see_stay_legal(self, verb): + """`-1` is a value to the splitter, not a flag, so the check must not forbid it.""" + assert CliMatch(verb=verb).verb_spellings == [verb.split()] + assert CliCalledCriterion(description="d", verb=verb).verb_spellings == [verb.split()] diff --git a/tests/test_custom_lint.py b/tests/test_custom_lint.py index 64508fe7..efedc2b1 100644 --- a/tests/test_custom_lint.py +++ b/tests/test_custom_lint.py @@ -146,6 +146,38 @@ def test_ignores_classes_without_the_method(self): assert not self._run("class FooAgent:\n def other(self):\n return {}") +@pytest.mark.lint +class TestCE047EmbeddedShimStdlibOnly: + """CE047 flags a non-stdlib import in a module embedded into generated shims.""" + + @staticmethod + def _run(src: str, *, embedded: bool = True): + import ast + + from tests.lint.rules.ce047_embedded_shim_stdlib_only import EmbeddedShimStdlibOnly + + path = "src/coder_eval/argv_match.py" if embedded else "src/coder_eval/invocation_log.py" + return EmbeddedShimStdlibOnly(path).check(ast.parse(src)) + + def test_flags_package_import(self): + assert self._run("from coder_eval.models import FlagMatch") + assert self._run("import coder_eval.models") + + def test_flags_third_party_import(self): + assert self._run("import pydantic") + assert self._run("from pydantic import BaseModel") + + def test_flags_relative_import(self): + assert self._run("from .models import FlagMatch") + + def test_allows_stdlib(self): + assert not self._run("import re\nimport json") + + def test_ignores_files_that_are_not_embedded(self): + # invocation_log.py renders the shim; it is not itself copied into one. + assert not self._run("from coder_eval.models import RecordedCli", embedded=False) + + @pytest.mark.lint class TestCE017ModelsLazyAgentImports: """CE017 flags only module-level agents/plugins imports inside models/.""" diff --git a/tests/test_merge_strategy_annotations.py b/tests/test_merge_strategy_annotations.py index 4d66ec43..dd706004 100644 --- a/tests/test_merge_strategy_annotations.py +++ b/tests/test_merge_strategy_annotations.py @@ -16,6 +16,7 @@ DockerDriverConfig, NodeEnvConfig, PythonEnvConfig, + RecordedCli, SandboxConfig, merge_strategy_of, parse_agent_config, @@ -33,6 +34,7 @@ class TestSandboxStrategies: (SandboxConfig, "template_sources", "append"), (SandboxConfig, "mock_path_dirs", "replace"), (SandboxConfig, "record_cli", "replace"), + (RecordedCli, "responses", "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 ef8a9bac..ffe251b5 100644 --- a/tests/test_sandbox_record_cli.py +++ b/tests/test_sandbox_record_cli.py @@ -10,6 +10,7 @@ import os import subprocess import sys +from pathlib import Path import pytest from pydantic import ValidationError @@ -20,6 +21,7 @@ RECORD_CLI_DIR, RECORD_CLI_LOG, CliCalledCriterion, + CliResponse, RecordedCli, SandboxConfig, StarterFile, @@ -504,3 +506,165 @@ def test_parse_log_separates_usable_from_unusable(self): assert [argv for argv, _ in usable] == [["a"]] # An argv that is not list[str] is unusable, not a non-match. assert unusable == 2 + + +class TestPerInvocationResponses: + """`responses:` — one shadowed tool answering each subcommand differently. + + The reason the shim is more than a recorder: an agent that reads + `ixp projects list` and acts on what came back cannot be evaluated by a stub + that returns the same line for everything it types. + """ + + @staticmethod + def _spec() -> RecordedCli: + return RecordedCli( + tool="uip", + exit_code=1, + stderr="uip: unknown command\n", + responses=[ + CliResponse(when={"verb": "ixp dummy1"}, stdout="response1\n"), + CliResponse(when={"verb": "ixp dummy2"}, stdout="response2\n"), + ], + ) + + def test_each_verb_gets_its_own_response(self): + sandbox = _sandbox("record_responses", record_cli=[self._spec()]) + try: + sandbox_dir = sandbox.setup() + first = _run_shim(sandbox_dir, "uip", ["ixp", "dummy1"]) + second = _run_shim(sandbox_dir, "uip", ["ixp", "dummy2"]) + assert (first.returncode, first.stdout) == (0, "response1\n") + assert (second.returncode, second.stdout) == (0, "response2\n") + finally: + sandbox.cleanup(preserve=False) + + def test_unmatched_invocation_falls_back_to_the_entry_defaults(self): + sandbox = _sandbox("record_responses_fallback", record_cli=[self._spec()]) + try: + sandbox_dir = sandbox.setup() + proc = _run_shim(sandbox_dir, "uip", ["ixp", "dummy3"]) + assert proc.returncode == 1 + assert proc.stdout == "" + assert "unknown command" in proc.stderr + finally: + sandbox.cleanup(preserve=False) + + def test_log_names_the_rule_that_answered(self): + """ "Returned the default" and "rule 1 answered" are otherwise the same line.""" + sandbox = _sandbox("record_responses_log", record_cli=[self._spec()]) + try: + sandbox_dir = sandbox.setup() + _run_shim(sandbox_dir, "uip", ["ixp", "dummy2"]) + _run_shim(sandbox_dir, "uip", ["ixp", "dummy3"]) + records = _records((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8")) + assert records[0]["rule"] == 1 + assert records[0]["exit"] == 0 + assert "rule" not in records[1], "no rule matched, so none may be claimed" + assert records[1]["exit"] == 1 + finally: + sandbox.cleanup(preserve=False) + + def test_first_matching_rule_wins(self): + """Order is the author's disambiguation tool, so the general rule last.""" + spec = RecordedCli( + tool="uip", + responses=[ + CliResponse(when={"verb": "ixp projects get proj-1"}, stdout="specific\n"), + CliResponse(when={"verb": "ixp projects get"}, stdout="generic\n"), + ], + ) + sandbox = _sandbox("record_responses_order", record_cli=[spec]) + try: + sandbox_dir = sandbox.setup() + assert _run_shim(sandbox_dir, "uip", ["ixp", "projects", "get", "proj-1"]).stdout == "specific\n" + assert _run_shim(sandbox_dir, "uip", ["ixp", "projects", "get", "proj-9"]).stdout == "generic\n" + finally: + sandbox.cleanup(preserve=False) + + def test_rule_can_match_on_flags_and_positional(self): + spec = RecordedCli( + tool="uip", + responses=[ + CliResponse( + when={"verb": "ixp projects get", "positional": ["proj-1"], "flags": {"output": "json"}}, + stdout='{"id": "proj-1"}', + ), + CliResponse(when={"verb": "ixp projects get"}, stdout="proj-1 (table)\n"), + ], + ) + sandbox = _sandbox("record_responses_flags", record_cli=[spec]) + try: + sandbox_dir = sandbox.setup() + asked_json = _run_shim(sandbox_dir, "uip", ["ixp", "projects", "get", "proj-1", "--output", "json"]) + asked_table = _run_shim(sandbox_dir, "uip", ["ixp", "projects", "get", "proj-1"]) + other_project = _run_shim(sandbox_dir, "uip", ["ixp", "projects", "get", "proj-2", "--output", "json"]) + assert asked_json.stdout == '{"id": "proj-1"}' + assert asked_table.stdout == "proj-1 (table)\n" + assert other_project.stdout == "proj-1 (table)\n" + finally: + sandbox.cleanup(preserve=False) + + def test_stderr_and_exit_code_are_per_rule(self): + spec = RecordedCli( + tool="uip", + exit_code=0, + responses=[CliResponse(when={"verb": "ixp projects get missing"}, exit_code=4, stderr="not found\n")], + ) + sandbox = _sandbox("record_responses_failure", record_cli=[spec]) + try: + sandbox_dir = sandbox.setup() + proc = _run_shim(sandbox_dir, "uip", ["ixp", "projects", "get", "missing"]) + assert (proc.returncode, proc.stderr) == (4, "not found\n") + # The entry default still applies to everything else, including its 0. + assert _run_shim(sandbox_dir, "uip", ["ixp", "projects", "list"]).returncode == 0 + finally: + sandbox.cleanup(preserve=False) + + def test_the_pattern_that_served_the_response_also_grades_it(self): + """One semantic across both surfaces: same facets, same verdict. + + A rule and a criterion written from the same pattern must agree, or a task + stubs one invocation and grades another. + """ + pattern = {"verb": "ixp projects configure-model", "positional": ["proj-1"], "flags": {"model": "pro"}} + spec = RecordedCli(tool="uip", responses=[CliResponse(when=dict(pattern), stdout="ok\n")]) + sandbox = _sandbox("record_responses_parity", record_cli=[spec]) + try: + sandbox_dir = sandbox.setup() + served = _run_shim(sandbox_dir, "uip", ["ixp", "projects", "configure-model", "proj-1", "--model", "pro"]) + assert served.stdout == "ok\n", "the rule did not match, so the grading half proves nothing" + criterion = CliCalledCriterion(description="configured the model", **pattern) + assert SuccessChecker(sandbox).check(criterion).score == 1.0 + finally: + sandbox.cleanup(preserve=False) + + def test_matcher_is_embedded_only_when_rules_exist(self): + """A shim with no rules never consults the matcher, so it does not carry it.""" + plain = render_recorder(RecordedCli(tool="uip")) + with_rules = render_recorder(RecordedCli(tool="uip", responses=[CliResponse(when={"verb": "ixp dummy1"})])) + assert "argv_match.py" not in plain + assert "def argv_matches" not in plain + assert "def argv_matches" in with_rules + # Both must be valid Python: the embedded half lands mid-file. + compile(plain, "shim", "exec") + compile(with_rules, "shim", "exec") + + def test_embedded_matcher_is_the_shipped_source_verbatim(self): + """Not a paraphrase: the shim's matcher IS coder_eval/argv_match.py.""" + from coder_eval import argv_match + + shipped = Path(argv_match.__file__).read_text(encoding="utf-8") + rendered = render_recorder(RecordedCli(tool="uip", responses=[CliResponse(when={"verb": "ixp dummy1"})])) + assert shipped.strip() in rendered + + def test_response_rule_needs_a_facet(self): + """A catch-all rule is the entry's own default; two ways to say it is one too many.""" + with pytest.raises(ValidationError, match="at least one of verb"): + CliResponse(when={}) + + def test_a_bare_string_when_is_rejected_with_the_fix(self): + """One shape for a pattern. A lone string leaves which of six facets it sets + to inference, and reads enough like a command line to invite flags.""" + with pytest.raises(ValidationError, match=r'use \{verb: "ixp dummy1"\}'): + CliResponse(when="ixp dummy1") From dbc5afb5805ae799ff3ed20e9ac5d1824ba46cc9 Mon Sep 17 00:00:00 2001 From: Alexandru Jircan Date: Wed, 2 Sep 2026 19:29:40 +0300 Subject: [PATCH 2/3] fix(record_cli): reject an unusable response rule at load, and trace a shim fault Addresses review on #150. Blocker: `FlagMatch.matches_regex` was never compiled at validation. The criterion's checker pre-flighted it, but the response-rule surface that now shares the model evaluates the pattern INSIDE the sandbox, where the shim swallowed the PatternError and served its fallback -- a log line byte-identical to a legitimate no-match. The task scored differently for identical agent behaviour, with nothing on any report surface. The compile moved into `FlagMatch`, so both surfaces refuse the pattern at load, and the now-unreachable checker pre-flight is gone. Second half of the same chain: when the shim's rule evaluation does raise, it returns the error and `record()` books it as `rule_error`, so an eval-config fault can no longer read as a clean no-match; `cli_called` fails the whole log on it, the way it already fails on the write-failure sentinel. Tested by corrupting a rendered shim -- the only route left now that the pattern cannot load. Also from the review: - The lowered spec crossed the model/matcher seam as `dict[str, Any]` read with permissive `.get(...) or `, whose failure direction is always "unconstrained" -- a rule that matches everything, or a criterion that scores 1.0 on any log. It is now `MatchSpec` / `FlagPredicate` / `ResponseRule` TypedDicts with required keys indexed directly, so a key renamed on either side is a pyright error on both. Tests pin that the TypedDict key sets equal the model field sets, which is what makes the one cast honest. - `FlagMatch.needs_value` was dead after the matcher extraction while `argv_match.predicate_needs_value` documented a mirror contract nothing enforced. Deleted; the survivor now says it is the only implementation. - A `responses` rule an earlier rule already claims was accepted silently, unlike every other unusable declaration on this surface. Now a load error for the two decidable cases: an exact duplicate, and a verb-only rule whose verb prefixes a later one under the same flag parsing. - CE047 grew a namespace half: an embedded module may not bind a top-level name the shim binds itself, since the shim's definition wins and the resulting TypeError is swallowed into "every invocation gets the fallback". Its target set now derives from `invocation_log.EMBEDDED_MODULES` instead of a second hardcoded path, and a test asserts it matches a file that exists -- a rule guarding zero files must fail, not pass. - The three rendered-shim invariants only ever rendered the rules-less shape. Parametrized over both; the spliced shape did violate the ASCII one, so `argv_match.py` is ASCII-only now, by rule rather than by luck. - Parity test closed the other direction (a facet added to `CliMatch` alone passed before), `MergeField` dropped from `RecordedCli.responses` (never a merge root, so the strategy was inert metadata that read as a knob), doubled paren in CLAUDE.md, and the guide's example no longer uses the one flag a rule may key on but the criterion rejects. BREAKING CHANGE: a flag written inside a `verb:` (e.g. `verb: "ixp projects get --output json"`) is now rejected when a task loads, on `cli_called` and on a `record_cli` response rule. It previously validated and then matched nothing, so the criterion scored 0 against a log holding that exact call. Move the flag to `flags:`. An invalid `matches_regex` is likewise a load error rather than a check-time one. Co-Authored-By: Claude Opus 5 (1M context) --- CLAUDE.md | 2 +- docs/TASK_DEFINITION_GUIDE.md | 4 +- src/coder_eval/argv_match.py | 120 ++++++++++++------ src/coder_eval/criteria/cli_called.py | 40 +++--- src/coder_eval/invocation_log.py | 57 +++++++-- src/coder_eval/models/cli_match.py | 50 +++++--- src/coder_eval/models/criteria.py | 5 +- src/coder_eval/models/sandbox.py | 54 +++++++- .../rules/ce047_embedded_shim_stdlib_only.py | 79 ++++++++---- tests/test_cli_called_criterion.py | 44 +++---- tests/test_cli_match_parity.py | 23 +++- tests/test_custom_lint.py | 23 ++++ tests/test_merge_strategy_annotations.py | 2 - tests/test_sandbox_record_cli.py | 99 ++++++++++++++- 14 files changed, 448 insertions(+), 154 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 0fdb1b31..aa2f047a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -220,7 +220,7 @@ make plugin-reference # the plugin's bundled criteria reference from the models Editing `src/coder_eval/pricing.py` means editing `evalboard/lib/pricing.ts` too — it is a hand-copied mirror, and `evalboard/lib/__tests__/pricing-parity.test.ts` fails the build on drift in either direction. -Recent additions, each traceable to a shipped defect: **CE037** (no unreferenced module-level private helper in `src/` — a helper whose docstring documents a bug the live code still has is worse than none), **CE038** (in an `@asynccontextmanager`, the acquire must sit INSIDE the `try` whose `finally` releases it — `asyncio.shield` protects the inner task, NOT the await, so a cancel on `__aenter__` skips the unwind while the work completes), **CE039** (a criterion checker must not return a gating `score=0.0` from an `except OSError` over a path the *task author* named — that books an eval-config error as an agent failure; raise `CheckerMisuseError` instead, and `# noqa: CE039` the cases that really are the agent's)), **CE047** (a module whose SOURCE is embedded into a generated sandbox shim — `argv_match.py` — may import stdlib only: the shim runs where `coder_eval` and its dependencies are not installed, so one package import makes every shadowed CLI die with an ImportError the agent reads as "the tool is broken"). +Recent additions, each traceable to a shipped defect: **CE037** (no unreferenced module-level private helper in `src/` — a helper whose docstring documents a bug the live code still has is worse than none), **CE038** (in an `@asynccontextmanager`, the acquire must sit INSIDE the `try` whose `finally` releases it — `asyncio.shield` protects the inner task, NOT the await, so a cancel on `__aenter__` skips the unwind while the work completes), **CE039** (a criterion checker must not return a gating `score=0.0` from an `except OSError` over a path the *task author* named — that books an eval-config error as an agent failure; raise `CheckerMisuseError` instead, and `# noqa: CE039` the cases that really are the agent's), **CE047** (a module whose SOURCE is spliced into a generated sandbox shim — `invocation_log.EMBEDDED_MODULES`, currently `argv_match.py` — may import stdlib only AND may not bind a top-level name the shim binds itself (`invocation_log.SHIM_GLOBALS`). Both failures are silent: the shim runs where `coder_eval` and its dependencies are not installed, so one package import makes every shadowed CLI die with an ImportError the agent reads as "the tool is broken"; and a name the shim rebinds turns into a TypeError its `respond()` swallows, so every invocation quietly gets the fallback response. The rule derives its target set from that exported tuple and a test asserts it matches a file that exists — a lint rule guarding zero files must fail, not pass). When fixing a bug, ask: *could a custom lint rule have prevented this?* If the root cause is a mechanically detectable pattern (e.g., "always import from `coder_eval.models`", "never call blocking IO in async"), add a rule to `tests/lint/rules/` following the CE001+ pattern and wire it up in `tests/lint/runner.py`. This turns a one-time fix into permanent enforcement. See `tests/test_custom_lint.py` for how rules are tested. (Doc-surface / whole-tree rules that reason over Markdown/YAML or the entire `src/` tree rather than one `.py` AST at a time — CE026–CE031, CE033–CE036 — are not `BaseRule`s in the runner; they are wired as dedicated `@pytest.mark.lint` test classes. CE036 enforces the `live_verdict` determinism + monotonicity contract (`criteria/base.py`) that `EarlyStopWatcher`'s latching, deferred fail-stop, and flip-attribution silently depend on: monotonicity over arbitrary Python is undecidable, so instead of a static check it REPLAYS each live criterion against every prefix of recorded trajectories (`tests/lint/live_verdict_contract.py::CASES`) — on the authored ordering AND under seeded shuffles (`permuted_violations`, which catch order-sensitive bugs the authored walk misses) — and asserts the property directly, plus registry-derived coverage — every `LiveSuccessCriterion` in the union must have cases, and every polarity its instances claim via `live_decidable_polarities()` must actually be reached by one (otherwise a single always-`undecided` fixture would "cover" a type while proving nothing). Adding a live criterion therefore means adding `ContractCase`s in the same change. CE035 resolves every `steps..outputs.` / `needs..outputs.` reference in `.github/workflows/**` to a writer that actually produces that key — GitHub expands an unwritten output to the empty string, so a typo degrades a gate silently and actionlint models `steps.*.outputs` as an open string map. CE034 scans `tasks/` and forces an armed, live-*passable* `command_executed` to set `require_success` — a crashed invocation would otherwise latch a live PASS, fire `on_pass: stop`, and let FIRED-ONLY armed gating report SUCCESS without ever consulting the unarmed criteria (negative assertions are fail-only and are exempt). CE033 keeps the plugin's bundled `reference/criteria.md` in parity with the `SuccessCriterion` union that generates it (`make plugin-reference` writes it; the rule re-renders and diffs — never hand-edit the file). CE031 guards against dead config: a behavior-driving field on `SimulationConfig`/`RunLimits`/`Dataset` that no code reads by name. CE026 keeps the GitHub Action's onboarding surfaces honest — `README.md`, `docs/CI_GATE.md`, `docs/tutorials/02-ci-pipeline.md`, and the plugin's `ci` skill, whose emitted workflow users copy into their own repos: a page's *first* Action snippet must show the agent-runtime prerequisite steps (pinned to the `action-dogfood` job that proves them in CI), a zero-install absolute next to such a snippet must name the channel it means, every `github.com/marketplace/actions/` link plus the shields badge label must match `action.yml`'s `name:`, and every `with:` key on a snippet's action step must be a real `action.yml` input (GitHub ignores unknown inputs, so a rename would silently degrade every copied workflow). Renaming an action input or changing its runtime prerequisites therefore means updating the skill too.) diff --git a/docs/TASK_DEFINITION_GUIDE.md b/docs/TASK_DEFINITION_GUIDE.md index c1b2fbe0..0797e9ba 100644 --- a/docs/TASK_DEFINITION_GUIDE.md +++ b/docs/TASK_DEFINITION_GUIDE.md @@ -587,7 +587,7 @@ sandbox: - when: # any cli_called facet, ANDed verb: "ixp projects get" positional: ["proj-1"] - flags: {output: json} + flags: {model: gemini_2_5_pro} stdout: '{"id": "proj-1", "name": "Invoices"}' - when: {verb: "ixp projects get missing"} exit_code: 4 @@ -597,7 +597,7 @@ sandbox: - **`when` takes the same facets as [`cli_called`](#cli_called)** — `verb`, `verb_any_of`, `positional`, `flags`, `value_flags`, `ignore_flags` — evaluated by the same matcher, so the pattern that *serves* a response is the pattern that *grades* it. Always a mapping: a bare `when: "ixp dummy1"` is rejected (with the `{verb: ...}` spelling in the message), since a pattern has six facets and a lone string leaves which one you meant to inference. - **First match wins**, in declaration order: put the specific rule above the general one. An invocation no rule claims gets the entry's own `exit_code` / `stdout` / `stderr`. - **`exit_code` defaults to 0 on a rule** — the opposite of the entry default of 1. A rule exists because you described that invocation, so the natural reading is "and this is what it answers"; an undescribed one should still look like a tool that failed. -- **`ignore_flags` is empty on a rule**, unlike the criterion's `[output]`: grading must not depend on a flag that changes nothing about the outcome, but a rule may legitimately answer differently for `--output json`. +- **`ignore_flags` is empty on a rule**, unlike the criterion's `[output]`: grading must not depend on a flag that changes nothing about the outcome, but a rule may legitimately answer differently for `--output json`. That is the one place a rule is *not* copy-pastable into a criterion — `flags: {output: ...}` is valid on a rule and rejected on the criterion, which ignores that flag by default. - **The log names the rule that answered** (`"rule": 1`), and omits the key when none did — the first thing you want to know when an expected canned response does not arrive. - **Still stateless.** A rule answers the same way however many times it matches; a counter would have to survive concurrent agent commands. For a tool whose reply must change over a run, hand-write a mock under `mock_path_dirs`. diff --git a/src/coder_eval/argv_match.py b/src/coder_eval/argv_match.py index 136333b2..d6fa9926 100644 --- a/src/coder_eval/argv_match.py +++ b/src/coder_eval/argv_match.py @@ -1,4 +1,8 @@ -"""Structured argv matching — the one engine both CLI surfaces share. +"""Structured argv matching: the one engine both CLI surfaces share. + +ASCII-only by rule, not by accident: this module's source is spliced into +generated shims, and `test_rendered_shim_is_pure_ascii` covers the spliced +shape, so an em-dash here fails that test rather than a sandbox somewhere. Two places ask the same question about one invocation. The ``cli_called`` criterion reads a recorded ``argv`` back afterwards and asks *did this happen*; @@ -13,20 +17,53 @@ the sandbox, where ``coder_eval`` is not installed. Lint rule CE047 keeps the imports stdlib-only. -The spec dict is what ``CliMatch.match_spec`` emits:: - - {"verb_spellings": [["ixp", "projects", "get"]], - "positional": ["proj-1"], - "flags": {"model": {"equals": "pro", "aliases": [], "flags": 0, ...}}, - "value_flags": ["output"], - "ignore_flags": []} - -``verb_spellings`` is empty when there is no verb constraint; ``positional`` and -``flags`` are absent or None when unconstrained. +:class:`MatchSpec` is what ``CliMatch.match_spec`` emits. It is a ``TypedDict`` +rather than a bare dict on purpose: it is the seam where every guarantee the +pydantic models establish would otherwise be erased, and the fallback for a key +the reader failed to find is always "unconstrained" -- the direction that makes a +rule match everything, or a criterion score 1.0 against any log. TypedDict is +closed, so a key renamed on either side is a pyright error on both. """ import re -from typing import Any +from typing import TypedDict + + +class FlagPredicate(TypedDict): + """One ``FlagMatch``, lowered. Every key present: the producer dumps the model.""" + + equals: str | None + contains: str | None + matches_regex: str | None + any_of: list[str] | None + absent: bool + present: bool + aliases: list[str] + flags: int + + +class MatchSpec(TypedDict): + """One authored pattern, lowered. ``verb_spellings`` is empty for no constraint. + + ``positional`` / ``flags`` are None when unconstrained -- None rather than + absent, so a reader indexes required keys directly and a missing one is a + KeyError rather than a silently wider match. + """ + + verb_spellings: list[list[str]] + positional: list[str] | None + flags: dict[str, FlagPredicate] | None + value_flags: list[str] + ignore_flags: list[str] + + +class ResponseRule(TypedDict): + """One ``CliResponse``, lowered -- what the shim carries and dispatches on.""" + + when: MatchSpec + exit: int + stdout: str + stderr: str def split_flags( @@ -118,63 +155,70 @@ def is_number(text: str) -> bool: return True -def predicate_needs_value(predicate: dict[str, Any]) -> bool: +def predicate_needs_value(predicate: FlagPredicate) -> bool: """Whether evaluating this flag predicate requires the flag's VALUE. Presence predicates (``present`` / ``absent``) do not, so they must not make a flag value-bearing: asserting a boolean switch would otherwise make it - consume the following token, dropping that token from the positionals. - Mirrors ``FlagMatch.needs_value``; both sides of the spec boundary must agree - or a rule and the criterion grading it would parse the same argv differently. + consume the following token, dropping that token from the positionals. That + is how `flags: {yes: {present: true}}` on a guard over `delete --yes proj-1` + once bound ``yes=proj-1``, dropped the project name, and handed the guard a + false PASS. + + The ONLY implementation of the rule. ``FlagMatch`` deliberately does not + carry a pydantic-side twin: two spellings of one predicate rule is how a rule + and the criterion grading it come to parse the same argv differently. """ - return not (predicate.get("present") or predicate.get("absent")) + return not (predicate["present"] or predicate["absent"]) -def flag_matches(predicate: dict[str, Any], values: list[str] | None) -> bool: +def flag_matches(predicate: FlagPredicate, values: list[str] | None) -> bool: """Whether a recorded flag satisfies one flag predicate. ``values`` is None when the flag was not passed at all. Every non-``absent`` predicate is satisfied by ANY of a repeated flag's values. """ - if predicate.get("absent"): + if predicate["absent"]: return values is None - if predicate.get("present"): + if predicate["present"]: return values is not None if values is None: return False - if predicate.get("equals") is not None: - return any(value == predicate["equals"] for value in values) - if predicate.get("contains") is not None: - return any(predicate["contains"] in value for value in values) - if predicate.get("any_of") is not None: - allowed = set(predicate["any_of"]) + if (equals := predicate["equals"]) is not None: + return any(value == equals for value in values) + if (contains := predicate["contains"]) is not None: + return any(contains in value for value in values) + if (any_of := predicate["any_of"]) is not None: + allowed = set(any_of) return any(value in allowed for value in values) - if predicate.get("matches_regex") is not None: - regex = re.compile(predicate["matches_regex"], predicate.get("flags", 0)) + if (pattern := predicate["matches_regex"]) is not None: + # Compiled at load by FlagMatch, so this cannot raise on a spec the models + # produced -- and re caches, so recompiling per invocation is not a cost. + regex = re.compile(pattern, predicate["flags"]) return any(regex.search(value) is not None for value in values) # Unreachable: the model guarantees exactly one predicate. Raise rather than # return False so a predicate added without a matcher arm here fails loudly. raise AssertionError(f"flag predicate has no matcher arm: {predicate!r}") -def argv_matches(spec: dict[str, Any], argv: list[str]) -> bool: +def argv_matches(spec: MatchSpec, argv: list[str]) -> bool: """Whether ``argv`` satisfies every configured facet of one match spec.""" - flag_specs: dict[str, Any] = spec.get("flags") or {} + flag_specs = spec["flags"] or {} - def names_of(flag: str, predicate: dict[str, Any]) -> tuple[str, ...]: - return (flag, *(predicate.get("aliases") or ())) + def names_of(flag: str, predicate: FlagPredicate) -> tuple[str, ...]: + return (flag, *predicate["aliases"]) # Declarations only. Folding `ignore_flags` into value_flags made ignored # SWITCHES value-bearing, which swallowed the next positional and reopened a # guard false-PASS; an ignored flag that takes a value declares it in # value_flags. - ignore = frozenset(spec.get("ignore_flags") or ()) + ignore = frozenset(spec["ignore_flags"]) value_flags = frozenset( name for flag, predicate in flag_specs.items() if predicate_needs_value(predicate) for name in names_of(flag, predicate) - ) | frozenset(spec.get("value_flags") or ()) + ) | frozenset(spec["value_flags"]) known_names = ( frozenset(name for flag, predicate in flag_specs.items() for name in names_of(flag, predicate)) | ignore ) @@ -182,7 +226,7 @@ def names_of(flag: str, predicate: dict[str, Any]) -> tuple[str, ...]: positional, flags = split_flags(argv, ignore, value_flags, known_names) offset = 0 - spellings = spec.get("verb_spellings") or [] + spellings = spec["verb_spellings"] if spellings: # Token-wise, not a subset and not a string startswith: `labellings confirm` # must never be satisfied by `labellings unconfirm`. Taking the first match is @@ -194,7 +238,7 @@ def names_of(flag: str, predicate: dict[str, Any]) -> tuple[str, ...]: # Measured from the spelling that matched, since spellings can differ in length. offset = len(matched) - expected = spec.get("positional") + expected = spec["positional"] if expected is not None and positional[offset : offset + len(expected)] != list(expected): return False @@ -208,7 +252,7 @@ def names_of(flag: str, predicate: dict[str, Any]) -> tuple[str, ...]: return True -def select_rule(rules: list[dict[str, Any]], argv: list[str]) -> tuple[int, dict[str, Any]] | None: +def select_rule(rules: list[ResponseRule], argv: list[str]) -> tuple[int, ResponseRule] | None: """``(index, rule)`` of the first rule whose ``when`` spec matches ``argv``, or None. First match wins, so ordering is the author's disambiguation tool: the @@ -221,6 +265,6 @@ def select_rule(rules: list[dict[str, Any]], argv: list[str]) -> tuple[int, dict same line in the log. """ for index, rule in enumerate(rules): - if argv_matches(rule.get("when") or {}, argv): + if argv_matches(rule["when"], argv): return index, rule return None diff --git a/src/coder_eval/criteria/cli_called.py b/src/coder_eval/criteria/cli_called.py index ea970519..81ab265c 100644 --- a/src/coder_eval/criteria/cli_called.py +++ b/src/coder_eval/criteria/cli_called.py @@ -1,9 +1,8 @@ """CLI-called criterion checker — structured matching over an invocation log.""" import logging -import re import shlex -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING from coder_eval.argv_match import argv_matches from coder_eval.criteria.base import BaseCriterion, CheckContext, register_criterion @@ -18,7 +17,7 @@ logger = logging.getLogger(__name__) -def _record_matches(criterion: CliCalledCriterion, argv: list[str], record: dict[str, Any]) -> bool: +def _record_matches(criterion: CliCalledCriterion, argv: list[str], record: dict[str, object]) -> bool: """Whether one log record satisfies every configured facet of the criterion. ``tool`` is checked here rather than in :func:`argv_matches` because it is a @@ -54,21 +53,10 @@ def _check_impl( Result with binary score (1.0 when the match count is within [min_count, max_count], 0.0 otherwise) """ - # Up front so a bad pattern names its flag, rather than surfacing as a - # generic caught exception when some record first reaches that predicate. - for name, predicate in (criterion.flags or {}).items(): - if predicate.matches_regex is None: - continue - try: - re.compile(predicate.matches_regex, predicate.flags) - except (re.error, ValueError) as exc: - return CriterionResult( - criterion_type=criterion.type, - description=criterion.description, - score=0.0, - error=f"Invalid matches_regex for flag '{name}': {exc}", - ) - + # No pre-flight re.compile here: `FlagMatch` compiles the pattern at + # validation, so an uncompilable one never reaches a checker -- and it has + # to be caught there, because the response-rule surface that shares this + # model cannot report an error at all. if not sandbox.file_exists(criterion.log): # Harness fault, not agent behaviour. Failing stops a max_count: 0 # guard passing vacuously against a log that never existed. @@ -98,6 +86,22 @@ def _check_impl( usable, unusable = parse_log(content) + # The shim books this when its own rule evaluation raised. Defense in + # depth (FlagMatch compiles at load), but if it ever fires, the responses + # the agent saw were not the ones the task described, so no verdict over + # this log means anything -- same treatment as the write-failure sentinel. + faults = [record for _, record in usable if record.get("rule_error") is not None] + if faults: + return CriterionResult( + criterion_type=criterion.type, + description=criterion.description, + score=0.0, + error=( + f"Recorder could not evaluate its response rules on {len(faults)} invocation(s), so the " + f"agent saw fallback output the task did not describe. First: {faults[0].get('rule_error')!r}" + ), + ) + if unusable: # A record we cannot read might BE the call a max_count: 0 guard # forbids, so scoring it "did not match" would let the guard pass. diff --git a/src/coder_eval/invocation_log.py b/src/coder_eval/invocation_log.py index f1efa0d7..6ac594ab 100644 --- a/src/coder_eval/invocation_log.py +++ b/src/coder_eval/invocation_log.py @@ -26,6 +26,32 @@ # travels with them if the sandbox root moves. LOG_FILENAME = "calls.jsonl" +# Modules whose SOURCE is spliced into a generated shim. Exported because lint +# rule CE047 keeps their imports stdlib-only and their module-level names clear +# of SHIM_GLOBALS: a rule that hardcodes its own copy of this list guards nothing +# the day the module moves, and would pass vacuously rather than fail. +EMBEDDED_MODULES = ("argv_match.py",) + +# Top-level names the generated shim binds itself. An embedded module that binds +# any of them is rebound by the shim's own definition further down the file -- +# and the resulting TypeError is swallowed by respond(), so EVERY invocation +# would quietly fall back to the entry defaults. +SHIM_GLOBALS = frozenset( + { + "TOOL", + "EXIT_CODE", + "STDOUT_TEXT", + "STDERR_TEXT", + "RULES", + "SHIM_DIR", + "LOG_PATH", + "LOG_ERROR_PATH", + "record", + "respond", + "main", + } +) + _TEMPLATE = '''\ #!{interpreter} """Recording shim for `{tool}` - generated by coder_eval SandboxConfig.record_cli. @@ -53,7 +79,7 @@ LOG_ERROR_PATH = LOG_PATH + ".error" -def record(argv, exit_code, rule): +def record(argv, exit_code, rule, rule_error): """Append this invocation to the log. Best-effort: a logging failure must never break the command the agent ran, @@ -70,6 +96,11 @@ def record(argv, exit_code, rule): answered, and happens to look like the default" are indistinguishable in the log -- the first question asked when an expected canned response does not arrive. + + `rule_error` is booked when rule evaluation RAISED, and it is what stops an + eval-config fault from reading as a clean no-match: the agent got fallback + output the task never described, so `cli_called` fails the whole log on it + rather than scoring a run whose responses were wrong. """ entry = {{ "ts": round(time.time(), 3), @@ -79,6 +110,8 @@ def record(argv, exit_code, rule): }} if rule is not None: entry["rule"] = rule + if rule_error is not None: + entry["rule_error"] = rule_error try: # ensure_ascii escapes non-ASCII and any stray surrogate from # undecodable argv bytes, so an exotic argument cannot make this write @@ -97,26 +130,29 @@ def record(argv, exit_code, rule): def respond(argv): - """Pick this invocation's (exit code, stdout, stderr, matched rule index). + """Pick this invocation's (exit code, stdout, stderr, rule index, rule error). First matching rule wins; whatever no rule claims gets the defaults. The matcher above is embedded only when RULES is non-empty, so this guard is what keeps `select_rule` from being named when it was not embedded. """ if not RULES: - return EXIT_CODE, STDOUT_TEXT, STDERR_TEXT, None + return EXIT_CODE, STDOUT_TEXT, STDERR_TEXT, None, None try: selected = select_rule(RULES, list(argv)) except Exception as exc: # Best-effort, like the log write: a matcher fault must not turn the stub # into a crashing executable, which the agent would read as the tool - # itself breaking in a way the task never described. + # itself breaking in a way the task never described. Unlike the log + # write it is also RETURNED, so the record says what happened -- an + # untraceable fallback here scores the task as if the agent had never + # made the call at all. sys.stderr.write("coder_eval recorder: response matching failed: %r\\n" % (exc,)) - return EXIT_CODE, STDOUT_TEXT, STDERR_TEXT, None + return EXIT_CODE, STDOUT_TEXT, STDERR_TEXT, None, repr(exc) if selected is None: - return EXIT_CODE, STDOUT_TEXT, STDERR_TEXT, None + return EXIT_CODE, STDOUT_TEXT, STDERR_TEXT, None, None index, rule = selected - return rule["exit"], rule["stdout"], rule["stderr"], index + return rule["exit"], rule["stdout"], rule["stderr"], index, None def main(argv): @@ -129,8 +165,8 @@ def main(argv): one, and this shim deliberately does only the second. """ args = argv[1:] - exit_code, stdout_text, stderr_text, rule = respond(args) - record(args, exit_code, rule) + exit_code, stdout_text, stderr_text, rule, rule_error = respond(args) + record(args, exit_code, rule, rule_error) if stdout_text: sys.stdout.write(stdout_text) if stderr_text: @@ -160,7 +196,8 @@ def _matcher_source() -> str: shim must dispatch on the SAME matcher the ``cli_called`` criterion grades with, and every transformation in between is a place the two could diverge. """ - return resources.files("coder_eval").joinpath("argv_match.py").read_text(encoding="utf-8") + (module,) = EMBEDDED_MODULES + return resources.files("coder_eval").joinpath(module).read_text(encoding="utf-8") def render_recorder(spec: RecordedCli, interpreter: str | None = None) -> str: diff --git a/src/coder_eval/models/cli_match.py b/src/coder_eval/models/cli_match.py index e60fb88d..c38a9a45 100644 --- a/src/coder_eval/models/cli_match.py +++ b/src/coder_eval/models/cli_match.py @@ -15,11 +15,12 @@ from __future__ import annotations import itertools -from typing import Any +import re +from typing import Any, cast from pydantic import BaseModel, ConfigDict, Field, model_validator -from coder_eval.argv_match import is_number +from coder_eval.argv_match import FlagPredicate, MatchSpec, is_number class FlagMatch(BaseModel): @@ -83,20 +84,6 @@ class FlagMatch(BaseModel): ), ) - @property - def needs_value(self) -> bool: - """Whether evaluating this predicate requires the flag's VALUE. - - Presence predicates (``present`` / ``absent``) do not, so they must not - make a flag value-bearing. Otherwise asserting a boolean switch would - make it consume the following token: adding ``flags: {yes: {present: - true}}`` to a guard on ``delete --yes proj-1`` would bind - ``yes=proj-1``, drop ``proj-1`` from the positionals, and hand the guard - a false PASS -- reintroducing the very defect declared value-binding - exists to prevent. - """ - return not (self.present or self.absent) - @model_validator(mode="before") @classmethod def _coerce_scalar_shorthand(cls, value: Any) -> Any: @@ -125,6 +112,19 @@ def _exactly_one_predicate(self) -> FlagMatch: if self.flags and self.matches_regex is None: msg = f"FlagMatch.flags applies only to matches_regex, but the predicate is {set_predicates[0]!r}" raise ValueError(msg) + # Compile HERE, not in a checker: this model now feeds two consumers, and + # only one of them can report. A `record_cli` response rule evaluates the + # pattern inside the sandbox, where a PatternError is swallowed and the + # tool serves its fallback -- a log line indistinguishable from a + # legitimate no-match, so the task scores differently for identical agent + # behaviour with nothing on any report surface. At load, both surfaces + # refuse the pattern instead. + if self.matches_regex is not None: + try: + re.compile(self.matches_regex, self.flags) + except (re.error, ValueError) as exc: + msg = f"FlagMatch.matches_regex is not a valid regex with flags={self.flags}: {exc}" + raise ValueError(msg) from exc return self @@ -248,17 +248,25 @@ def build_match_spec( flags: dict[str, FlagMatch] | None, value_flags: list[str], ignore_flags: list[str], -) -> dict[str, Any]: - """Lower an authored match surface to the plain dict :mod:`coder_eval.argv_match` reads. +) -> MatchSpec: + """Lower an authored match surface to what :mod:`coder_eval.argv_match` reads. JSON-serializable on purpose: the same dict is embedded verbatim into a generated shim, so a spec the criterion evaluates in-process and a spec the shim evaluates in the sandbox are the same bytes. + + The cast is honest because ``FlagMatch``'s field set IS ``FlagPredicate``'s key + set -- asserted in tests/test_cli_match_parity.py, so adding a field to one and + not the other fails rather than silently dropping out of the lowered spec. """ return { "verb_spellings": verb_spellings, "positional": positional, - "flags": {name: predicate.model_dump() for name, predicate in flags.items()} if flags else None, + "flags": ( + {name: cast("FlagPredicate", predicate.model_dump()) for name, predicate in flags.items()} + if flags + else None + ), "value_flags": list(value_flags), "ignore_flags": list(ignore_flags), } @@ -343,8 +351,8 @@ def verb_spellings(self) -> list[list[str]]: return verb_spellings_of(self.verb, self.verb_any_of) @property - def match_spec(self) -> dict[str, Any]: - """This pattern as the plain dict :func:`coder_eval.argv_match.argv_matches` reads.""" + def match_spec(self) -> MatchSpec: + """This pattern as what :func:`coder_eval.argv_match.argv_matches` reads.""" return build_match_spec( verb_spellings=self.verb_spellings, positional=self.positional, diff --git a/src/coder_eval/models/criteria.py b/src/coder_eval/models/criteria.py index ea0daaae..bfe8d5a5 100644 --- a/src/coder_eval/models/criteria.py +++ b/src/coder_eval/models/criteria.py @@ -14,6 +14,7 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator +from coder_eval.argv_match import MatchSpec from coder_eval.models.agent_config import AgentConfig, ClaudeCodeAgentConfig, parse_agent_config from coder_eval.models.cli_match import ( FlagMatch, @@ -568,8 +569,8 @@ def verb_spellings(self) -> list[list[str]]: return verb_spellings_of(self.verb, self.verb_any_of) @property - def match_spec(self) -> dict[str, Any]: - """This criterion's argv facets as the dict :mod:`coder_eval.argv_match` reads. + def match_spec(self) -> MatchSpec: + """This criterion's argv facets as what :mod:`coder_eval.argv_match` reads. The same lowering a ``record_cli`` response rule uses, so a rule that serves a response and the criterion that grades it cannot read one argv diff --git a/src/coder_eval/models/sandbox.py b/src/coder_eval/models/sandbox.py index afe9eb9b..151a59d6 100644 --- a/src/coder_eval/models/sandbox.py +++ b/src/coder_eval/models/sandbox.py @@ -409,18 +409,64 @@ class RecordedCli(BaseModel): "would, so an agent reads a plausible error rather than silence" ), ) - responses: list[CliResponse] = MergeField( - strategy="replace", + # Plain Field, not MergeField: `RecordedCli` is never a merge root. The + # enclosing `SandboxConfig.record_cli` is a `replace` list, so a later layer + # substitutes the whole list of entries and no per-entry strategy is ever + # consulted. A strategy annotation here would read as a knob and be inert. + responses: list[CliResponse] = Field( default_factory=list, description=( "Per-invocation responses, tried in order until one matches; the fields above are the " "fallback for an invocation none of them claim. Use it when the agent's next step " "depends on what the tool answered -- `ixp projects list` returning a project the agent " - "then acts on, say -- instead of one fixed reply to everything. Replaced (not merged) " - "across config layers, like the enclosing record_cli list" + "then acts on, say -- instead of one fixed reply to everything. A config layer that sets " + "record_cli replaces the whole list of entries, this one included" ), ) + @model_validator(mode="after") + def _validate_responses_are_reachable(self) -> RecordedCli: + """Reject a rule an earlier rule already claims. + + First-match-wins means a rule below a more general one can never answer. + Silence there would be out of step with the rest of this authoring + surface, which hard-errors on every declaration that cannot take effect: + a `verb_any_of` entry prefixed by another, a predicate on an ignored + flag, an empty `positional`, two entries writing the same shim filename. + + Deliberately narrow, because "A matches everything B matches" is not + decidable in general. Two sound cases only: an exact duplicate, and a + verb-only A whose verb prefixes B's under the same flag parsing. + """ + specs = [response.when.match_spec for response in self.responses] + for later, spec in enumerate(specs): + for earlier, prior in enumerate(specs[:later]): + if prior == spec: + reason = "is an exact duplicate of" + elif ( + prior["positional"] is None + and prior["flags"] is None + # Same parsing, or the two disagree on which tokens are even + # positional and neither claim covers the other. + and prior["value_flags"] == spec["value_flags"] + and prior["ignore_flags"] == spec["ignore_flags"] + and spec["verb_spellings"] + and all( + any(tokens[: len(prefix)] == prefix for prefix in prior["verb_spellings"]) + for tokens in spec["verb_spellings"] + ) + ): + reason = "is already claimed by the more general" + else: + continue + msg = ( + f"record_cli tool {self.tool!r}: responses[{later}] {reason} responses[{earlier}], " + "so it can never answer -- the first matching rule wins. Put the specific rule " + "above the general one, or drop the duplicate." + ) + raise ValueError(msg) + return self + @field_validator("tool") @classmethod def validate_tool_name(cls, v: str) -> str: diff --git a/tests/lint/rules/ce047_embedded_shim_stdlib_only.py b/tests/lint/rules/ce047_embedded_shim_stdlib_only.py index 5e1e9c9f..d3193529 100644 --- a/tests/lint/rules/ce047_embedded_shim_stdlib_only.py +++ b/tests/lint/rules/ce047_embedded_shim_stdlib_only.py @@ -1,19 +1,27 @@ -"""CE047: modules embedded into a generated sandbox shim import stdlib only. - -``invocation_log.render_recorder`` embeds the SOURCE of ``coder_eval/argv_match.py`` -into every ``record_cli`` shim that declares response rules. That shim runs inside -the sandbox, where ``coder_eval`` is not installed and no project dependency is -guaranteed — so a single ``from coder_eval.models import ...`` or ``import -pydantic`` added to the embedded module makes every shadowed CLI die with an -ImportError the moment the agent runs it. The failure surfaces as "the tool is -broken", never as "the harness embedded an unimportable module", and it costs a -whole run to diagnose. - -Import-time enforcement (a test that renders and executes a shim) only catches it -when a test happens to declare a response rule; this rule catches the import the -moment it is written. - -A stdlib module that is genuinely needed is added to ``STDLIB_ALLOWED`` below — +"""CE047: a module embedded into a generated sandbox shim stays stdlib-only and namespace-clean. + +``invocation_log.render_recorder`` splices the SOURCE of every module in +``invocation_log.EMBEDDED_MODULES`` into each ``record_cli`` shim that declares +response rules. That shim runs inside the sandbox, where ``coder_eval`` is not +installed and no project dependency is guaranteed, and the spliced source shares +one module namespace with the shim's own definitions. Two ways to break it, both +silent: + +* **An import.** One ``from coder_eval.models import ...`` or ``import pydantic`` + makes every shadowed CLI die with an ImportError the moment the agent runs it. + It surfaces as "the tool is broken", never as "the harness embedded an + unimportable module", and it costs a whole run to diagnose. +* **A name collision.** An embedded module that binds a top-level name the shim + also binds (``record``, ``main``, ``RULES``, ...) is rebound by the shim's own + definition further down the file. The resulting TypeError is caught by the + shim's ``respond()``, so every invocation quietly falls back to the entry + defaults instead of its canned response. + +Import-time enforcement (a test that renders and executes a shim) only catches +either when a test happens to declare a response rule; this rule catches both the +moment they are written. + +A stdlib module that is genuinely needed is added to ``STDLIB_ALLOWED`` below -- deliberately an allowlist rather than a check against ``sys.stdlib_module_names``, so growing the shim's surface is a decision someone makes on purpose. """ @@ -21,15 +29,18 @@ import ast import re +from coder_eval.invocation_log import EMBEDDED_MODULES, SHIM_GLOBALS from tests.lint.rules.base import BaseRule class EmbeddedShimStdlibOnly(BaseRule): id = "CE047" - # Modules whose source is embedded into a generated shim. Keyed by path - # fragment so the rule fires on the file itself, wherever the tree is rooted. - _EMBEDDED = re.compile(r"[/\\]coder_eval[/\\]argv_match\.py$") + # Derived from the writer's own list, so moving the module moves the rule + # with it. A hardcoded second copy would match nothing after such a move and + # pass vacuously -- guarding zero files while reading as a guarantee. + # tests/test_custom_lint.py asserts the pattern matches a file that exists. + _EMBEDDED = re.compile(r"[/\\]coder_eval[/\\](?:" + "|".join(re.escape(m) for m in EMBEDDED_MODULES) + ")$") # Small on purpose: everything here has to exist in whatever interpreter the # sandbox's shebang resolves to. @@ -39,7 +50,7 @@ def __init__(self, filepath: str) -> None: super().__init__(filepath) self._embedded = bool(self._EMBEDDED.search(filepath)) - def _check(self, node: ast.AST, module: str | None) -> None: + def _check_import(self, node: ast.AST, module: str | None) -> None: if not self._embedded or module is None: return root = module.split(".")[0] @@ -52,12 +63,36 @@ def _check(self, node: ast.AST, module: str | None) -> None: f"add '{root}' to CE047's STDLIB_ALLOWED if it really is stdlib.", ) + def _check_name(self, node: ast.AST, name: str) -> None: + if not self._embedded or name not in SHIM_GLOBALS: + return + self.violation( + node, + f"module-level '{name}' collides with a name the generated shim binds itself " + "(invocation_log.SHIM_GLOBALS). The shim's own definition wins, and the resulting failure " + "is swallowed into 'every invocation gets the fallback response'. Rename it.", + ) + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: # A relative import (level > 0) is a package import by definition. - self._check(node, node.module if node.level == 0 else f".{node.module or ''}") + self._check_import(node, node.module if node.level == 0 else f".{node.module or ''}") self.generic_visit(node) def visit_Import(self, node: ast.Import) -> None: for alias in node.names: - self._check(node, alias.name) + self._check_import(node, alias.name) + self.generic_visit(node) + + def visit_Module(self, node: ast.Module) -> None: + # Top level only: a name bound inside a function is not in the namespace + # the splice shares with the shim. + for statement in node.body: + if isinstance(statement, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef): + self._check_name(statement, statement.name) + elif isinstance(statement, ast.Assign): + for target in statement.targets: + if isinstance(target, ast.Name): + self._check_name(statement, target.id) + elif isinstance(statement, ast.AnnAssign) and isinstance(statement.target, ast.Name): + self._check_name(statement, statement.target.id) self.generic_visit(node) diff --git a/tests/test_cli_called_criterion.py b/tests/test_cli_called_criterion.py index 0dfbaadb..6b2919d7 100644 --- a/tests/test_cli_called_criterion.py +++ b/tests/test_cli_called_criterion.py @@ -174,18 +174,16 @@ def test_dotall_flag_lets_a_pattern_cross_newlines(self, sandbox_with_log): assert checker.check(without_dotall).score == 0.0 assert checker.check(with_dotall).score == 1.0 - def test_invalid_regex_reports_the_offending_flag(self, sandbox_with_log): - sandbox, sandbox_dir = sandbox_with_log - _write_log(sandbox_dir, [_call(["ixp", "projects", "get", "--val", "x"])]) - criterion = CliCalledCriterion( - description="bad pattern", - log=LOG, - verb="ixp projects get", - flags={"val": {"matches_regex": "([unclosed"}}, - ) - result = SuccessChecker(sandbox).check(criterion) - assert result.score == 0.0 - assert "Invalid matches_regex for flag 'val'" in (result.error or "") + def test_invalid_regex_is_refused_at_load_naming_the_flag(self): + """Load-time, not check-time: the same FlagMatch feeds a record_cli response + rule, which evaluates the pattern inside the sandbox and cannot report.""" + with pytest.raises(ValidationError, match="matches_regex is not a valid regex"): + CliCalledCriterion( + description="bad pattern", + log=LOG, + verb="ixp projects get", + flags={"val": {"matches_regex": "([unclosed"}}, + ) def test_absent_distinguishes_missing_from_different_value(self, sandbox_with_log): """`absent` is why flags is a predicate map, not dict[str, str].""" @@ -634,19 +632,15 @@ def test_present_requires_the_flag(self, sandbox_with_log): ) assert SuccessChecker(sandbox).check(criterion).score == 0.0 - def test_bad_regex_flags_value_names_the_flag(self, sandbox_with_log): - """re.error is not a ValueError, so the pre-flight guard missed this.""" - sandbox, sandbox_dir = sandbox_with_log - _write_log(sandbox_dir, [_call(["ixp", "projects", "get", "--val", "x"])]) - criterion = CliCalledCriterion( - description="bad flags int", - log=LOG, - verb="ixp projects get", - flags={"val": {"matches_regex": "a", "flags": 99999999}}, - ) - result = SuccessChecker(sandbox).check(criterion) - assert result.score == 0.0 - assert "flag 'val'" in (result.error or "") + def test_bad_regex_flags_value_is_refused_at_load(self): + """re.error is not a ValueError, so the old pre-flight guard missed this.""" + with pytest.raises(ValidationError, match="not a valid regex with flags=99999999"): + CliCalledCriterion( + description="bad flags int", + log=LOG, + verb="ixp projects get", + flags={"val": {"matches_regex": "a", "flags": 99999999}}, + ) class TestModelValidation: diff --git a/tests/test_cli_match_parity.py b/tests/test_cli_match_parity.py index 62dac0d4..90fb3352 100644 --- a/tests/test_cli_match_parity.py +++ b/tests/test_cli_match_parity.py @@ -14,8 +14,8 @@ import pytest from pydantic import ValidationError -from coder_eval.argv_match import argv_matches -from coder_eval.models import CliCalledCriterion, CliMatch, CliResponse +from coder_eval.argv_match import FlagPredicate, MatchSpec, argv_matches +from coder_eval.models import CliCalledCriterion, CliMatch, CliResponse, FlagMatch from coder_eval.models.cli_match import MATCH_FACET_FIELDS @@ -25,6 +25,11 @@ def test_both_surfaces_declare_every_match_facet(self): assert field in CliMatch.model_fields, f"CliMatch is missing match facet {field!r}" assert field in CliCalledCriterion.model_fields, f"cli_called is missing match facet {field!r}" + def test_the_rule_surface_declares_no_facet_outside_the_shared_tuple(self): + """Closes the loop the other direction: without this, a facet added to + CliMatch alone passes, since the loop above only walks MATCH_FACET_FIELDS.""" + assert set(CliMatch.model_fields) == set(MATCH_FACET_FIELDS) + def test_criterion_adds_only_non_argv_fields(self): """A facet on the criterion that CliMatch lacks is a rule authors cannot write.""" # Everything the criterion adds is about the LOG (where to read, which @@ -100,3 +105,17 @@ def test_tokens_the_matcher_would_really_see_stay_legal(self, verb): """`-1` is a value to the splitter, not a flag, so the check must not forbid it.""" assert CliMatch(verb=verb).verb_spellings == [verb.split()] assert CliCalledCriterion(description="d", verb=verb).verb_spellings == [verb.split()] + + +class TestLoweredSpecKeys: + """The lowered spec is a TypedDict, so pyright catches a renamed key. These + pin what pyright cannot: that the models and the TypedDicts hold the same + field set, which is what makes `build_match_spec`'s cast honest.""" + + def test_flag_predicate_keys_are_exactly_the_model_fields(self): + assert set(FlagPredicate.__annotations__) == set(FlagMatch.model_fields) + + def test_match_spec_keys_are_exactly_what_lowering_emits(self): + emitted = set(CliMatch(verb="ixp dummy1").match_spec) + assert emitted == set(MatchSpec.__annotations__) + assert emitted == set(CliCalledCriterion(description="d", verb="ixp dummy1").match_spec) diff --git a/tests/test_custom_lint.py b/tests/test_custom_lint.py index efedc2b1..4a388ae3 100644 --- a/tests/test_custom_lint.py +++ b/tests/test_custom_lint.py @@ -177,6 +177,29 @@ def test_ignores_files_that_are_not_embedded(self): # invocation_log.py renders the shim; it is not itself copied into one. assert not self._run("from coder_eval.models import RecordedCli", embedded=False) + def test_flags_a_name_the_shim_binds_itself(self): + """The shim's own `def record` wins, and respond() swallows the TypeError, + so every invocation would silently get the fallback response.""" + assert self._run("def record(argv):\n return argv") + assert self._run("RULES = []") + + def test_allows_a_colliding_name_that_is_not_module_level(self): + assert not self._run("def matcher():\n record = 1\n return record") + + def test_the_rule_guards_a_file_that_actually_exists(self): + """A rule matching nothing passes vacuously while reading as a guarantee -- + which is what a move of the embedded module would otherwise cause.""" + from pathlib import Path + + from coder_eval.invocation_log import EMBEDDED_MODULES + from tests.lint.rules.ce047_embedded_shim_stdlib_only import EmbeddedShimStdlibOnly + + package = Path(__file__).resolve().parents[1] / "src" / "coder_eval" + for module in EMBEDDED_MODULES: + target = package / module + assert target.is_file(), f"EMBEDDED_MODULES names {module}, which does not exist" + assert EmbeddedShimStdlibOnly(str(target))._embedded, f"CE047 does not match {target}" + @pytest.mark.lint class TestCE017ModelsLazyAgentImports: diff --git a/tests/test_merge_strategy_annotations.py b/tests/test_merge_strategy_annotations.py index dd706004..4d66ec43 100644 --- a/tests/test_merge_strategy_annotations.py +++ b/tests/test_merge_strategy_annotations.py @@ -16,7 +16,6 @@ DockerDriverConfig, NodeEnvConfig, PythonEnvConfig, - RecordedCli, SandboxConfig, merge_strategy_of, parse_agent_config, @@ -34,7 +33,6 @@ class TestSandboxStrategies: (SandboxConfig, "template_sources", "append"), (SandboxConfig, "mock_path_dirs", "replace"), (SandboxConfig, "record_cli", "replace"), - (RecordedCli, "responses", "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 ffe251b5..21df87c5 100644 --- a/tests/test_sandbox_record_cli.py +++ b/tests/test_sandbox_record_cli.py @@ -47,6 +47,14 @@ def _run_shim(sandbox_dir, tool: str, args: list[str]) -> subprocess.CompletedPr ) +# The two rendered shim shapes. Only the second splices in argv_match.py, so an +# invariant asserted on the first alone proves nothing about the interesting half. +SHIM_SHAPES = ( + RecordedCli(tool="uip"), + RecordedCli(tool="uip", responses=[CliResponse(when={"verb": "ixp dummy1"}, stdout="ok")]), +) + + def _records(text: str) -> list[dict]: """Just the records; parse_log also returns the unusable count.""" usable, _ = parse_log(text) @@ -474,15 +482,17 @@ def test_rendered_shim_is_valid_python_and_embeds_config(self): assert namespace["EXIT_CODE"] == 3 assert namespace["STDERR_TEXT"] == "boom\n" - def test_rendered_shim_does_not_execute_anything(self): + @pytest.mark.parametrize("spec", SHIM_SHAPES, ids=("no_rules", "with_rules")) + def test_rendered_shim_does_not_execute_anything(self, spec): """It stubs a tool rather than proxying one: no subprocess, no exec.""" - source = render_recorder(RecordedCli(tool="uip")) + source = render_recorder(spec) for forbidden in ("subprocess", "execv", "execvp", "popen", "system("): assert forbidden not in source - def test_rendered_shim_imports_nothing_from_coder_eval(self): + @pytest.mark.parametrize("spec", SHIM_SHAPES, ids=("no_rules", "with_rules")) + def test_rendered_shim_imports_nothing_from_coder_eval(self, spec): """It runs inside the sandbox, where this package is not installed.""" - source = render_recorder(RecordedCli(tool="uip")) + source = render_recorder(spec) imports = [ line.strip() for line in source.splitlines() @@ -490,9 +500,15 @@ def test_rendered_shim_imports_nothing_from_coder_eval(self): ] 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")) + @pytest.mark.parametrize("spec", SHIM_SHAPES, ids=("no_rules", "with_rules")) + def test_rendered_shim_is_pure_ascii(self, spec): + """Written into arbitrary sandboxes and read by whatever python3 is there. + + Parametrized over both shapes because only the rules-bearing one splices in + another module's source -- the half that can actually break any of these + three invariants, and the half the unparametrized versions never rendered. + """ + source = render_recorder(spec) source.encode("ascii") def test_parse_log_separates_usable_from_unusable(self): @@ -639,6 +655,40 @@ def test_the_pattern_that_served_the_response_also_grades_it(self): finally: sandbox.cleanup(preserve=False) + def test_a_rule_evaluation_fault_is_recorded_and_fails_the_grading(self): + """The shim swallows a matcher fault so the stub does not crash, but the + record must say so: without it, an eval-config fault is byte-identical to a + legitimate no-match and the task scores as if the agent never made the call. + + FlagMatch compiles at load, so the only way to reach this is to corrupt a + rendered shim -- which is the point: the branch is defense in depth, and + nothing else exercises it. + """ + sandbox = _sandbox("record_rule_fault", record_cli=[self._spec()]) + try: + sandbox_dir = sandbox.setup() + shim = sandbox_dir / RECORD_CLI_DIR / "uip" + source = shim.read_text(encoding="utf-8") + # A spec no matcher can evaluate, standing in for any future shim fault. + broken = source.replace("'verb_spellings': [['ixp', 'dummy1']]", "'verb_spellings': 5", 1) + assert broken != source, "the rule literal moved; update this test" + shim.write_text(broken, encoding="utf-8") + + proc = _run_shim(sandbox_dir, "uip", ["ixp", "dummy1"]) + assert proc.returncode == 1, "the stub must still answer, not crash" + assert "response matching failed" in proc.stderr + + record = _records((sandbox_dir / RECORD_CLI_LOG).read_text(encoding="utf-8"))[0] + assert "rule" not in record + assert "TypeError" in record["rule_error"] + + criterion = CliCalledCriterion(description="called dummy1", verb="ixp dummy1") + result = SuccessChecker(sandbox).check(criterion) + assert result.score == 0.0 + assert "could not evaluate its response rules" in (result.error or "") + finally: + sandbox.cleanup(preserve=False) + def test_matcher_is_embedded_only_when_rules_exist(self): """A shim with no rules never consults the matcher, so it does not carry it.""" plain = render_recorder(RecordedCli(tool="uip")) @@ -663,6 +713,41 @@ def test_response_rule_needs_a_facet(self): with pytest.raises(ValidationError, match="at least one of verb"): CliResponse(when={}) + @pytest.mark.parametrize( + ("responses", "expected"), + [ + ([{"when": {"verb": "ixp x"}, "stdout": "a"}, {"when": {"verb": "ixp x"}, "stdout": "b"}], "duplicate"), + ([{"when": {"verb": "ixp projects"}}, {"when": {"verb": "ixp projects get"}}], "already claimed"), + ( + [{"when": {"verb": "ixp projects"}}, {"when": {"verb_any_of": ["ixp projects get", "ixp projects x"]}}], + "already claimed", + ), + ], + ids=("exact_duplicate", "general_above_specific", "every_alternative_covered"), + ) + def test_a_rule_an_earlier_rule_already_claims_is_rejected(self, responses, expected): + """First-match-wins makes such a rule dead, and the rest of this surface + hard-errors on every declaration that cannot take effect.""" + with pytest.raises(ValidationError, match=expected): + RecordedCli(tool="uip", responses=responses) + + @pytest.mark.parametrize( + "responses", + [ + [{"when": {"verb": "ixp projects get"}}, {"when": {"verb": "ixp projects"}}], + [{"when": {"verb": "ixp projects", "flags": {"o": "j"}}}, {"when": {"verb": "ixp projects get"}}], + [{"when": {"verb": "ixp projects", "positional": ["p1"]}}, {"when": {"verb": "ixp projects get"}}], + [{"when": {"verb": "ixp projects", "value_flags": []}}, {"when": {"verb": "ixp projects get"}}], + [{"when": {"verb": "ixp a"}}, {"when": {"verb": "ixp b"}}], + ], + ids=("specific_first", "general_has_flag", "general_has_positional", "parsing_differs", "unrelated"), + ) + def test_a_reachable_rule_is_not_rejected(self, responses): + """The check must stay narrow: an earlier rule that constrains anything + beyond its verb does NOT claim everything a later rule would, and two + rules parsing argv differently cannot be compared by verb prefix at all.""" + assert len(RecordedCli(tool="uip", responses=responses).responses) == 2 + def test_a_bare_string_when_is_rejected_with_the_fix(self): """One shape for a pattern. A lone string leaves which of six facets it sets to inference, and reads enough like a command line to invite flags.""" From b3d6b567378dc0114d44e2c3259465df6dc84f55 Mon Sep 17 00:00:00 2001 From: Alexandru Jircan Date: Thu, 3 Sep 2026 10:55:34 +0300 Subject: [PATCH 3/3] fix(models): reference FlagPredicate at runtime so the cast is a real use CodeQL flagged the import as unused: the only reference was inside a QUOTED `cast("FlagPredicate", ...)`, which pyright resolves but a static importer scan cannot see. Unquoting makes it a genuine runtime reference, which is what the alert was asking for and costs one name lookup per flag predicate at config-load time. Co-Authored-By: Claude Opus 5 (1M context) --- src/coder_eval/models/cli_match.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/coder_eval/models/cli_match.py b/src/coder_eval/models/cli_match.py index c38a9a45..badcbaf4 100644 --- a/src/coder_eval/models/cli_match.py +++ b/src/coder_eval/models/cli_match.py @@ -263,9 +263,7 @@ def build_match_spec( "verb_spellings": verb_spellings, "positional": positional, "flags": ( - {name: cast("FlagPredicate", predicate.model_dump()) for name, predicate in flags.items()} - if flags - else None + {name: cast(FlagPredicate, predicate.model_dump()) for name, predicate in flags.items()} if flags else None ), "value_flags": list(value_flags), "ignore_flags": list(ignore_flags),