diff --git a/CLAUDE.md b/CLAUDE.md index 392c29e2..8875b1a6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -34,7 +34,7 @@ coder_eval/ ├── models/ # Pydantic data models (subpackage) │ ├── __init__.py # Unified exports for all models │ ├── enums.py # AgentKind, AgentState, FinalStatus, ApiBackend -│ ├── criteria.py # 14 success criterion types + base + union +│ ├── 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) │ ├── mutations.py # PromptMutation variants (prefix/suffix/replace/template) @@ -144,7 +144,7 @@ action.yml # Published composite GitHub Action (coder-ev - **Run-time caps (non-criterion enforcement)**: `TaskDefinition.run_limits` (`RunLimits` model) is the single namespace for all *task-level* run-time caps — `max_turns` / `task_timeout` / `turn_timeout` (structural) and `max_input_tokens` / `max_output_tokens` / `max_total_tokens` / `max_usd` (cumulative budget). Token/USD breaches abort with `FinalStatus.TOKEN_BUDGET_EXCEEDED` or `COST_BUDGET_EXCEEDED` (both `category == "failed"`). Structural caps are set from the CLI via `-D run_limits.max_turns=…` / `-D run_limits.task_timeout=…` / `-D run_limits.turn_timeout=…` (field-merged into `run_limits`); budget caps via `-D run_limits.max_usd=…` etc. or YAML. Layered config uses field-merge — a variant block overrides individual keys without replacing the task's block. The one *per-criterion* cap, `stop_early.decide_within`, deliberately lives on `LiveSuccessCriterion` instead (see below) — the watcher must attribute a decision-step timeout to a specific criterion, which `RunLimits` (task-scoped, criterion-agnostic) cannot express. - **Early stop on criterion (opt-in, per-criterion arming)**: a `stop_early:` block (`StopEarlyPolicy`) on a criterion ends a single-shot run early once the run's **armed** criteria decide the outcome, so a raised `max_turns` isn't wasted on the smoke flavor. The block's PRESENCE is the arming and alone activates the watcher — there is **no run-level master switch**: `run_limits.stop_early: false` is the run-level KILL SWITCH that force-disarms every block (the one-line experiment-variant/`-D` override for an authoritative full run), and `run_limits.stop_early: true` (the removed master arm) is a hard `EarlyStopConfigError` at resolution. The block exists on `LiveSuccessCriterion` only (currently `skill_triggered`, `command_executed` — so arming an unobservable criterion is unrepresentable, a pydantic extra-forbid error). Arming carries one implicit trigger (a native live-fail may fail-stop the run); its keys refine it: `on_pass: stop` (pass-stop the moment the criterion live-passes; default `continue` just latches) and `decide_within: N` (still undecided after N tool-call steps latches an **effective fail**, fed through the same fail-stop rule, reported as `decision_budget_exceeded` — an ordinary weighted fail, NOT a gate-bypassing force-fail; cumulative across retry attempts of the same turn). A trigger whose polarity the instance can't decide (per the abstract, checker-independent `live_decidable_polarities()`, a pure function of the criterion's own fields, paired with the checker's `live_verdict` override by lint rule CE025, a registry-based whole-tree check) is **inert by design** — one dataset-fanned YAML line serves both positive rows (pass/timeout live) and distractor rows (fail live). Verdicts **latch**: once a criterion decides, its `live_verdict` is never polled again. Stop rule is weighted, not strict-boolean: `run_limits.stop_early_gate_threshold` (default `1.0`, reproducing strict-AND behavior exactly) is the minimum weighted score (`Σ weight·score / Σ weight` over the armed subset) required to pass; a fail-stop fires once the armed set's **ceiling** (best case for everything still undecided) can no longer reach the threshold — so a low-weight fail or timeout that can't doom the gate is absorbed and the run continues — and is **deferred while any pass-capable armed criterion is undecided** (a distractor misfire never truncates a positive row's recall signal); a pass-stop fires once the `on_pass: stop` subset's **floor** (worst case) already meets the threshold, and is symmetrically **deferred while any pass-capable armed criterion outside the `on_pass: stop` subset is undecided** (so an early pass never freezes a sibling `on_pass: continue` criterion's signal out of the trajectory). A fail-stop is therefore verdict-preserving; a pass-stop can miss a *later* distractor misfire, so authoritative P/R/F1 comes from a kill-switched (`stop_early: false`) run. Driven by `orchestration/early_stop.py::EarlyStopWatcher` (built when `early_stop_active(task)`: ≥1 armed criterion, kill switch not thrown) through the agent's cooperative `should_stop` seam (tool-call granularity, no SIGKILL); live verdicts only *trigger* the stop — the standard `check_all_async` on the frozen trajectory is authoritative. Gating is **FIRED-ONLY**: a run the watcher actually cut gates on the **armed subset** via the weighted `EvaluationResult.armed_criteria_passed`; a run that completes naturally — armed or not — gates strict-AND via `all_criteria_passed`, so adding a block never changes the verdict of a run it didn't cut. Note the gate keys on the watcher having FIRED (`result.early_stop is not None`), not on confirmed truncation — an agent that ignores `should_stop`, or a stop firing on the final message, still gates armed-only. Every resolution-time guardrail violation is a hard error at resolution (plan *and* run); the one load-time case — a `stop_early:` block on a non-live criterion — is a pydantic schema error at task load, which the run surface reports as a skipped task like any other malformed task. A runtime verdict bug **fails open** to a full run. Surfaces: `EarlyStopInfo` (incl. `gate_threshold` at stop time), report notes/badges, `stopped_early` run.json rows, `EarlyStopped`/`EarlyStopReason` telemetry dims. Worked rationale: docs/TASK_DEFINITION_GUIDE.md § `stop_early`. No blocks anywhere ⇒ behavior byte-for-byte unchanged. -## Success Criteria (14 types) +## Success Criteria (15 types) | Type | Scoring | Description | |------|---------|-------------| @@ -156,6 +156,7 @@ action.yml # Published composite GitHub Action (coder-ev | `file_matches_regex` | Binary | Regex match on file | | `reference_comparison` | Continuous | AST/token/complexity similarity | | `command_executed` | Fractional | Agent tool usage verification | +| `cli_called` | Binary | Structured match over a JSON Lines invocation log: verb / positional / per-flag predicates, with min_count/max_count bounds | | `commands_efficiency` | Continuous | Agent tool-call efficiency relative to expected budget | | `uipath_eval` | Fractional | UiPath agent evaluation results | | `classification_match` | Binary | File-based label match (observed vs expected) with `(none)`/`(other)` sentinels; emits `ClassificationCriterionResult` for suite-level P/R/F1 | diff --git a/docs/TASK_DEFINITION_GUIDE.md b/docs/TASK_DEFINITION_GUIDE.md index 3da43187..1e988237 100644 --- a/docs/TASK_DEFINITION_GUIDE.md +++ b/docs/TASK_DEFINITION_GUIDE.md @@ -28,6 +28,7 @@ Complete reference for defining evaluation tasks in Coder Eval. - [file_matches_regex](#file_matches_regex) - [reference_comparison](#reference_comparison) - [command_executed](#command_executed) + - [cli_called](#cli_called) - [uipath_eval](#uipath_eval) - [llm_judge](#llm_judge) - [agent_judge](#agent_judge) @@ -620,7 +621,7 @@ All criteria share these fields: | `stop_early` | `null` | **Only on live-observable criteria** (`skill_triggered`, `command_executed`). Presence arms the criterion for early stop (no run-level switch needed): an effective fail may end the run (weighted ceiling rule, recall deferral). Keys: `on_pass: stop\|continue` (default `continue`), `decide_within: N` (timeout → effective fail, reported as `decision_budget_exceeded`). Inert triggers by design on instances that can't decide their polarity (dataset fan-out support). See [`stop_early`](#stop_early-opt-in-early-stop). | **Scoring types:** -- **Binary** (1.0 or 0.0): `file_exists`, `run_command`, `file_matches_regex`, `classification_match`, `skill_triggered` +- **Binary** (1.0 or 0.0): `file_exists`, `run_command`, `file_matches_regex`, `cli_called`, `classification_match`, `skill_triggered` - **Fractional** (0.0–1.0): `file_contains`, `file_check`, `json_check`, `command_executed`, `uipath_eval` - **Continuous** (0.0–1.0): `reference_comparison`, `commands_efficiency`, `llm_judge`, `agent_judge` @@ -842,6 +843,113 @@ Checks whether the agent executed specific tools/commands during evaluation. Ins **Codex limitation.** Codex agents map `Read`, `Grep`, and `Glob` tools to `shell` commands (they execute via bash), so `tool_name: "Read"` on Codex returns no matches. Use `tool_name: "Bash"` or `tool_name: null` (any tool) for Codex-compatible checks. This criterion works correctly on Claude Code agents, which emit separate `Read`/`Grep`/`Glob` telemetry. +### `cli_called` + +Checks whether a CLI invocation matching a **structured** pattern was recorded, by reading a JSON Lines invocation log the sandbox produced. **Binary scoring.** + +Use this instead of `command_executed` or `file_matches_regex` when a test shadows a CLI with a recording mock and needs to assert on *what was actually executed*, field by field. + +```yaml +- type: "cli_called" + description: "Switched the project to the capable model" + log: "mocks/calls.jsonl" # Path to the JSON Lines invocation log (required) + verb: "ixp projects configure-model" # Ordered prefix of the non-flag arguments + positional: ["my_invoices-ixp"] # Non-flag arguments following the verb, in order + flags: + model: "gemini_2_5_pro" # Bare scalar == {equals: ...} + tool: "uip" # Optional: match only records with this tool + min_count: 1 # Minimum matching invocations (default: 1) + max_count: null # Maximum; null = unbounded, 0 = forbidden + ignore_flags: ["output"] # Flags dropped before matching (default: ["output"]) +``` + +**Log format.** One JSON object per line. Only `argv` is required; `tool` lets one log serve several shadowed executables, and `exit`/`ts` are recorded for reporting rather than matched. Unknown keys are ignored, so a mock may record more. + +```json +{"ts": 1785416844.987, "tool": "uip", "argv": ["ixp", "projects", "get", "proj-1"], "exit": 1} +``` + +**Flag predicates.** Each entry under `flags:` takes **exactly one** of: + +| Predicate | Matches when the flag value… | +|-----------|------------------------------| +| `equals` | equals the string exactly (the bare-scalar shorthand) | +| `contains` | contains the substring | +| `matches_regex` | matches the regex — scoped to one value, not the whole line | +| `any_of` | equals one of the listed strings | +| `present: true` | the flag was passed, whatever its value — the right predicate for a boolean switch | +| `absent: true` | the flag was **not** passed at all | + +`matches_regex` also accepts `flags:` (the `re` module's integers, e.g. `2` = `IGNORECASE`, `8` = `MULTILINE`, `16` = `DOTALL`), mirroring [`file_matches_regex`](#file_matches_regex). Setting `flags` next to any other predicate is rejected rather than silently ignored. + +Flags the criterion does not mention are ignored, so an extra `--output json` never breaks a match. Repeated flags (`--fields a --fields b`) are satisfied by any one value. A predicate on a flag also listed in `ignore_flags` is rejected at load time — the flag is dropped before predicates run, so it could never be evaluated. + +**Which flags carry a value is declared, not guessed.** A flag consumes the following token only if it appears in `flags:`, in `value_flags:`, or in `ignore_flags:`. Everything else is a switch, and the token after it stays positional: + +```yaml +# `uip ixp fields delete --yes proj-1` +- type: "cli_called" + description: "Did not delete proj-1" + verb: "ixp fields delete" + positional: ["proj-1"] # --yes is a switch, so proj-1 stays positional + min_count: 0 + max_count: 0 # correctly FAILS -- the log proves the delete happened + +# `uip ixp projects list --folder Finance proj-1` +- type: "cli_called" + description: "Listed proj-1" + verb: "ixp projects list" + positional: ["proj-1"] + value_flags: ["folder"] # without this, "Finance" would count as a positional +``` + +Defaulting to "switch" is deliberate: `--yes` / `--force` / `-y` before the target is how destructive CLIs are invoked, so guessing that the flag swallows its neighbour is precisely how a `max_count: 0` guard ends up passing on the call it exists to forbid. The equals form (`--offset=-1`) is unambiguous and always binds directly, and a declared value flag binds even a dash-leading value (`--limit -1`). + +`ignore_flags` drops a flag from matching but does **not** make it value-bearing — an ignored flag that takes a value must also appear in `value_flags` (as `output` does by default). Otherwise `ignore_flags: ["verbose"]` on `delete --verbose proj-1` would let `--verbose` eat `proj-1`. + +**Limitation: bundled short flags are not split.** `-rf` parses as one flag named `rf`, so a predicate on `f` will not see it — including `absent: true`, which passes despite `-rf` being present. Assert on the long spelling, or add the bundled form via `aliases`. Likewise a bare negative number in flag position (`seek -1`) is read as a flag named `1`. + +**Negative guards want the FEWEST facets that capture the forbidden act.** This is the opposite of a positive assertion, and it is easy to get backwards. `max_count: 0` passes when *nothing matches*, so every facet you add is another way for the real invocation to slip past the pattern and report a false PASS. + +In the delete example above, it is tempting to also assert `--yes`. Don't: + +- `--yes` is not the forbidden thing — the deletion is. The CLI *requires* a confirmation flag, so asserting it adds no discriminating power. +- It adds escape routes: `-y` instead of `--yes` (a different flag name) no longer matches, and the guard passes on a delete that did happen. + +Use `present: true` — not `equals: ""` — when you do need to assert a switch. `present` needs no value, so it never makes the flag value-bearing; `equals: ""` depends on how the mock happens to record a switch and breaks if the CLI spells it `--force true`. + +**Short and long spellings are one flag via `aliases`.** A predicate matches a flag *name*, so `--yes` and `-y` are otherwise unrelated flags: + +```yaml +flags: + yes: + present: true + aliases: ["y"] # values gathered across --yes AND -y +``` + +`present` holds if any listed name appeared, `absent` only if none did, and a value predicate matches if any value under any name satisfies it — so `-f f-002` binds like `--fields f-002`. Splitting the spellings into one criterion each works for a *guard* (both forbidden, and criteria are ANDed) but cannot express "either spelling" positively, and makes `absent` flag **every** invocation, because whichever spelling was not used is always absent. A flag may belong to only one predicate: an alias that is also another key, or that appears in `ignore_flags`, is rejected at load time. + +The mirror rule for positive assertions: add every facet that distinguishes the right call from a near-miss, because there a missing facet makes the assertion *too easy* to satisfy. + +**Unusable records fail the criterion.** A line that is not JSON, not an object, or whose `argv` is not a list of strings scores 0.0 with an error, on the same footing as a missing log — a record that cannot be read might *be* the invocation a negative guard forbids. + +**One predicate per flag** — so a conjunction on a single flag ("contains *both* A and B") is not expressible directly. Two ways to write it: + +```yaml +# 1. One matches_regex spanning both. DOTALL (16) is usually needed: a payload +# built with a heredoc contains newlines, and without it `.` stops at the first. +flags: + updates: + matches_regex: '"name": "Invoice Number".*Do NOT use the Purchase Order' + flags: 16 + +# 2. Or two criteria over the same log, which scores and reports each part separately. +``` + +**Negative guards.** Set `min_count: 0` and `max_count: 0` to assert a call did **not** happen. A missing log file *fails* rather than counting as zero matches — otherwise a mock writing to the wrong path would make every negative guard pass vacuously. + +**Why not a regex over a flattened log line.** A flat `cmd arg arg` string cannot express "verb X was called AND flag Y had value Z" without stacked lookaheads; cannot distinguish a quoted argument containing spaces from two arguments; and cannot stop a match from running across shell operators. Matching `argv` element-wise removes all three problems. `verb` is an **ordered prefix**, so `ixp labellings confirm` is never satisfied by `ixp labellings unconfirm`. + ### `commands_efficiency` Scores how economically the agent worked, relative to a budget of expected tool calls. **Continuous scoring:** `score = expected_commands / max(actual_commands, expected_commands)` — so a run at or under budget scores `1.0`, and the score decays as the agent takes more calls than expected (e.g. twice the budget → `0.5`). diff --git a/src/coder_eval/criteria/cli_called.py b/src/coder_eval/criteria/cli_called.py new file mode 100644 index 00000000..6e196b4c --- /dev/null +++ b/src/coder_eval/criteria/cli_called.py @@ -0,0 +1,291 @@ +"""CLI-called criterion checker — structured matching over an invocation log.""" + +import json +import logging +import re +import shlex +from typing import TYPE_CHECKING, Any + +from coder_eval.criteria.base import BaseCriterion, CheckContext, register_criterion +from coder_eval.models import CliCalledCriterion, CriterionResult, FlagMatch + + +if TYPE_CHECKING: + from coder_eval.models.results import TurnRecord + from coder_eval.sandbox import Sandbox + +logger = logging.getLogger(__name__) + + +def _split_flags( + argv: list[str], + ignore: frozenset[str], + value_flags: frozenset[str], +) -> 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 limitation: bundled short flags are not split, so ``-rf`` is one flag + named ``rf`` and a predicate on ``f`` will not see it. + """ + 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("-") + 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 _flag_matches(predicate: FlagMatch, values: list[str] | None) -> bool: + """Whether a recorded flag satisfies one :class:`FlagMatch` 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.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 _usable_argv(record: dict[str, Any]) -> list[str] | None: + """The record's ``argv`` when it is a list of strings, else None. + + None means the record cannot be evaluated at all — a different thing from + "evaluated and did not match", which is why the caller reports it rather than + quietly treating it as a non-match. + """ + argv = record.get("argv") + if isinstance(argv, list) and all(isinstance(item, str) for item in argv): + return argv + return None + + +def _record_matches(criterion: CliCalledCriterion, argv: list[str], record: dict[str, Any]) -> bool: + """Whether one log record satisfies every configured facet of the criterion.""" + if criterion.tool is not None and record.get("tool") != criterion.tool: + 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), + ) + + offset = 0 + if criterion.verb is not None: + verb_tokens = criterion.verb.split() + # ORDERED prefix, not a token subset: `labellings confirm` must never be + # satisfied by `labellings unconfirm`, and a project name that happens to + # equal a subcommand must not stand in for the subcommand. + if positional[: len(verb_tokens)] != verb_tokens: + return False + offset = len(verb_tokens) + + 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 + + +@register_criterion +class CliCalledChecker(BaseCriterion[CliCalledCriterion]): + """Checker for CliCalledCriterion.""" + + criterion_type = "cli_called" + + def _check_impl( + self, + criterion: CliCalledCriterion, + sandbox: "Sandbox", + reference_code: str | None = None, + *, + turn_records: list["TurnRecord"] | None = None, + context: CheckContext | None = None, + ) -> CriterionResult: + """Count invocations in the structured log that match the criterion. + + Args: + criterion: CLI-called criterion + sandbox: Sandbox instance for file access + reference_code: Not used for this criterion + + Returns: + 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}", + ) + + 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. + return CriterionResult( + criterion_type=criterion.type, + description=criterion.description, + score=0.0, + error=f"Invocation log '{criterion.log}' does not exist", + ) + + content = sandbox.get_file_content(criterion.log) + + usable: list[tuple[list[str], dict[str, Any]]] = [] + unusable = 0 + for line in content.splitlines(): + stripped = line.strip() + if not stripped: + continue + try: + parsed = json.loads(stripped) + except ValueError: + unusable += 1 + continue + if not isinstance(parsed, dict): + unusable += 1 + continue + argv = _usable_argv(parsed) + if argv is None: + unusable += 1 + continue + usable.append((argv, parsed)) + + 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. + logger.warning( + f"cli_called: {unusable} unusable record(s) in '{criterion.log}'" + + " (unparseable line, non-object line, or argv that is not a list of strings)" + ) + return CriterionResult( + criterion_type=criterion.type, + description=criterion.description, + score=0.0, + error=( + f"Invocation log '{criterion.log}' has {unusable} unusable record(s): a line that is " + "not JSON, not an object, or whose 'argv' is not a list of strings. The verdict " + "cannot be trusted, so the criterion fails rather than scoring an incomplete log." + ), + ) + + matches = [record for argv, record in usable if _record_matches(criterion, argv, record)] + count = len(matches) + records = usable + + within_lower = count >= criterion.min_count + within_upper = criterion.max_count is None or count <= criterion.max_count + score = 1.0 if within_lower and within_upper else 0.0 + + bound = f"min_count={criterion.min_count}" + if criterion.max_count is not None: + bound += f", max_count={criterion.max_count}" + + facets = [] + if criterion.tool is not None: + facets.append(f"tool={criterion.tool!r}") + if criterion.verb is not None: + facets.append(f"verb={criterion.verb!r}") + if criterion.positional is not None: + facets.append(f"positional={criterion.positional!r}") + if criterion.flags: + facets.append(f"flags={sorted(criterion.flags)}") + wanted = ", ".join(facets) + + if score == 1.0: + details = f"{count} invocation(s) matched ({wanted}); satisfies {bound}" + elif not within_lower: + # A bare count sends the reader to the sandbox; this criterion exists + # to answer "what did it actually run". + sample = "; ".join(shlex.join(argv)[:120] for argv, _ in usable[:3]) + more = f" (+{len(usable) - 3} more)" if len(usable) > 3 else "" + recorded = f" Recorded: {sample}{more}" if sample else "" + details = ( + f"{count} invocation(s) matched ({wanted}); needs {bound}. " + f"{len(records)} invocation(s) recorded in '{criterion.log}'.{recorded}" + ) + else: + details = f"{count} invocation(s) matched ({wanted}) but {bound} forbids it" + + return CriterionResult( + criterion_type=criterion.type, + description=criterion.description, + score=score, + details=details, + ) diff --git a/src/coder_eval/models/__init__.py b/src/coder_eval/models/__init__.py index cf081ee1..ad33fdfd 100644 --- a/src/coder_eval/models/__init__.py +++ b/src/coder_eval/models/__init__.py @@ -32,12 +32,14 @@ AgentJudgeCriterion, BaseSuccessCriterion, ClassificationMatchCriterion, + CliCalledCriterion, CommandExecutedCriterion, CommandsEfficiencyCriterion, FileCheckCriterion, FileContainsCriterion, FileExistsCriterion, FileMatchesRegexCriterion, + FlagMatch, JMESPathAssertion, JsonCheckCriterion, LivePolarity, @@ -230,6 +232,8 @@ "FileMatchesRegexCriterion", "ReferenceComparisonCriterion", "CommandExecutedCriterion", + "CliCalledCriterion", + "FlagMatch", "CommandsEfficiencyCriterion", "UiPathEvalCriterion", "LLMJudgeCriterion", diff --git a/src/coder_eval/models/criteria.py b/src/coder_eval/models/criteria.py index 77be9c9e..c368ca3b 100644 --- a/src/coder_eval/models/criteria.py +++ b/src/coder_eval/models/criteria.py @@ -409,6 +409,273 @@ 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. + + Reads a **structured invocation log** the sandbox produced: JSON Lines, one + object per invocation, each with at minimum an ``argv`` list. A test harness + that shadows a CLI with a recording mock writes this log; this criterion + matches against it field-by-field instead of regexing a flattened command + string. + + Record schema (extra keys ignored):: + + {"argv": ["ixp", "projects", "get", "proj-1", "--output", "json"], + "tool": "uip", "exit": 1, "ts": 1785416844.987} + + Only ``argv`` is required. ``tool`` enables one log to serve several shadowed + executables; ``exit`` and ``ts`` are recorded for reporting, not matched. + + Why not ``file_matches_regex`` over a flattened log line: a flat line cannot + express "verb X was called AND flag Y had value Z" without stacked + lookaheads, cannot tell a quoted argument containing spaces from two + arguments, and cannot stop a match from running across shell operators. + + Pure data model - checking logic in CliCalledChecker._check_impl() + + Example YAML (positive — flag value must match):: + + success_criteria: + - type: "cli_called" + description: "Switched the project to the capable model" + log: "mocks/calls.jsonl" + verb: "ixp projects configure-model" + positional: ["my_invoices-f1afa9ef-ixp"] + flags: + model: "gemini_2_5_pro" + min_count: 1 + + Example YAML (negative — must NOT have been called; ``min_count: 0`` + ``max_count: 0``):: + + success_criteria: + - type: "cli_called" + description: "Did not use --corrections to flip a boolean field" + log: "mocks/calls.jsonl" + verb: "ixp labellings confirm" + flags: + corrections: {contains: "f-100"} + min_count: 0 + max_count: 0 + """ + + type: Literal["cli_called"] = "cli_called" + log: str = Field(description="Path to the JSON Lines invocation log, relative to the sandbox working directory") + verb: str | None = Field( + default=None, + min_length=1, + description=( + "Whitespace-separated subcommand chain that must be an ORDERED PREFIX of the invocation's " + "non-flag arguments. Order matters, so 'labellings confirm' never matches " + "'labellings unconfirm'" + ), + ) + tool: str | None = Field( + default=None, + description="Match only records whose 'tool' equals this (e.g. 'uip'). None matches any tool", + ) + positional: list[str] | None = Field( + default=None, + description="Non-flag arguments that must follow the verb, in order", + ) + flags: dict[str, FlagMatch] | None = Field( + default=None, + description=( + "Flag name (without leading dashes) to predicate. A bare scalar means 'equals'. " + "Flags not listed here are ignored, so an extra --output json never breaks a match" + ), + ) + 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]" + ), + ) + min_count: int = Field( + default=1, + ge=0, + description=( + "Minimum matching invocations. Combine min_count: 0 with max_count: 0 for must-NOT-match. " + "Scoring is BINARY (in vs out of bounds), unlike command_executed's fractional field of " + "the same name" + ), + ) + max_count: int | None = Field( + default=None, + ge=0, + description="Maximum number of matching invocations. None means no upper bound; 0 forbids the call", + ) + ignore_flags: list[str] = Field( + default_factory=lambda: ["output"], + description=( + "Flag names dropped before matching. Defaults to ['output'] so grading never depends on " + "--output json, which is outcome-invisible. An ignored flag that takes a value must also " + "appear in value_flags. Pass [] to disable" + ), + ) + + @model_validator(mode="after") + def _validate_bounds(self) -> CliCalledCriterion: + # min_count 0 with no upper bound is satisfied by every possible log, so + # the criterion can never fail -- the same vacuity class as a blank verb. + if self.min_count == 0 and self.max_count is None: + msg = ( + "cli_called with min_count: 0 and no max_count can never fail. Set max_count: 0 for a " + "negative guard, or raise min_count for a positive assertion." + ) + raise ValueError(msg) + if self.max_count is not None and self.max_count < self.min_count: + msg = f"max_count ({self.max_count}) must be >= min_count ({self.min_count})" + raise ValueError(msg) + # min_length=1 counts characters, so " " passes it — and `" ".split()` + # is `[]`, an empty prefix that matches every record. + if self.verb is not None and not self.verb.strip(): + msg = "cli_called verb must not be blank: a blank verb is an empty prefix and matches every record" + raise ValueError(msg) + # Falsiness-symmetric on purpose: `verb: ""` used to slip past an `is None` + # check here and then match EVERY record (empty prefix), silently scoring 1.0. + if not self.verb and not self.positional and not self.flags and not self.tool: + msg = "cli_called requires at least one of verb / 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) + return self + + class RegexPattern(BaseModel): """A single regex pattern check within FileCheckCriterion.""" @@ -1161,6 +1428,7 @@ def _reject_verdict_channel(cls, data: Any) -> Any: | JsonCheckCriterion | ReferenceComparisonCriterion | CommandExecutedCriterion + | CliCalledCriterion | CommandsEfficiencyCriterion | UiPathEvalCriterion | ClassificationMatchCriterion diff --git a/tests/test_cli_called_criterion.py b/tests/test_cli_called_criterion.py new file mode 100644 index 00000000..1ba57207 --- /dev/null +++ b/tests/test_cli_called_criterion.py @@ -0,0 +1,698 @@ +"""Tests for CliCalledCriterion — structured matching over an invocation log.""" + +import json +import re + +import pytest +from pydantic import ValidationError + +from coder_eval.criteria.cli_called import _split_flags +from coder_eval.evaluation.checker import SuccessChecker +from coder_eval.models import CliCalledCriterion, SandboxConfig +from coder_eval.sandbox import Sandbox + + +LOG = "mocks/calls.jsonl" + + +def _write_log(sandbox_dir, records: list[dict]) -> None: + """Write an invocation log in the shape a recording CLI mock produces.""" + log_path = sandbox_dir / LOG + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "".join(json.dumps(record) + "\n" for record in records), + encoding="utf-8", + ) + + +def _call(argv: list[str], tool: str = "uip", exit_code: int = 1) -> dict: + return {"ts": 1785416844.987, "tool": tool, "argv": argv, "exit": exit_code} + + +@pytest.fixture +def sandbox_with_log(request): + """A tempdir sandbox whose log is populated per-test, then cleaned up.""" + config = SandboxConfig(driver="tempdir", python=None) + # Sanitize: parametrized test names carry [] {} " which are illegal in a + # Windows path, and the tempdir driver builds the sandbox dir from task_id. + safe_task_id = re.sub(r"[^A-Za-z0-9_]", "_", request.node.name)[:50] + sandbox = Sandbox(config, task_id=safe_task_id) + sandbox_dir = sandbox.setup() + yield sandbox, sandbox_dir + sandbox.cleanup(preserve=False) + + +class TestVerbAndFlagMatching: + def test_matches_verb_and_flag_value(self, sandbox_with_log): + """The canonical positive: verb chain plus one flag value.""" + sandbox, sandbox_dir = sandbox_with_log + _write_log( + sandbox_dir, + [_call(["ixp", "projects", "configure-model", "proj-1", "--model", "gemini_2_5_pro"])], + ) + criterion = CliCalledCriterion( + description="switched model", + log=LOG, + verb="ixp projects configure-model", + positional=["proj-1"], + flags={"model": "gemini_2_5_pro"}, + ) + result = SuccessChecker(sandbox).check(criterion) + assert result.score == 1.0 + assert result.error is None + + def test_wrong_flag_value_does_not_match(self, sandbox_with_log): + sandbox, sandbox_dir = sandbox_with_log + _write_log( + sandbox_dir, + [_call(["ixp", "projects", "configure-model", "proj-1", "--model", "gemini_2_5_flash"])], + ) + criterion = CliCalledCriterion( + description="switched model", + log=LOG, + verb="ixp projects configure-model", + flags={"model": "gemini_2_5_pro"}, + ) + assert SuccessChecker(sandbox).check(criterion).score == 0.0 + + def test_verb_prefix_is_ordered_not_a_token_subset(self, sandbox_with_log): + """`labellings confirm` must NOT be satisfied by `labellings unconfirm`. + + This is the property that distinguishes an assertion matcher from a + permissive dispatch matcher. + """ + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "labellings", "unconfirm", "proj-1", "doc-1"])]) + criterion = CliCalledCriterion( + description="confirmed", + log=LOG, + verb="ixp labellings confirm", + ) + assert SuccessChecker(sandbox).check(criterion).score == 0.0 + + def test_positional_must_follow_the_verb_in_order(self, sandbox_with_log): + """A value in the wrong position must not satisfy a positional match.""" + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "labellings", "confirm", "doc-1", "proj-1"])]) + criterion = CliCalledCriterion( + description="confirmed the right project", + log=LOG, + verb="ixp labellings confirm", + positional=["proj-1", "doc-1"], + ) + assert SuccessChecker(sandbox).check(criterion).score == 0.0 + + def test_unlisted_flags_are_ignored(self, sandbox_with_log): + """Extra flags the criterion does not mention never break a match.""" + sandbox, sandbox_dir = sandbox_with_log + _write_log( + sandbox_dir, + [_call(["ixp", "projects", "get", "proj-1", "--verbose", "--folder", "F"])], + ) + criterion = CliCalledCriterion(description="got it", log=LOG, verb="ixp projects get") + assert SuccessChecker(sandbox).check(criterion).score == 1.0 + + def test_tool_filter_separates_shadowed_executables(self, sandbox_with_log): + """One log serves several mocks; `tool` selects among them.""" + sandbox, sandbox_dir = sandbox_with_log + _write_log( + sandbox_dir, + [_call(["-s", "https://example.invalid/projects"], tool="curl")], + ) + wants_curl = CliCalledCriterion(description="used curl", log=LOG, tool="curl") + wants_uip = CliCalledCriterion(description="used uip", log=LOG, tool="uip") + checker = SuccessChecker(sandbox) + assert checker.check(wants_curl).score == 1.0 + assert checker.check(wants_uip).score == 0.0 + + +class TestFlagPredicates: + @pytest.mark.parametrize( + ("predicate", "recorded", "expected"), + [ + ({"equals": "gemini_2_5_pro"}, "gemini_2_5_pro", 1.0), + ({"equals": "gemini_2_5_pro"}, "gpt_4o_2024_05_13", 0.0), + ({"contains": "f-100"}, '[{"field_id":"f-100"}]', 1.0), + ({"contains": "f-100"}, '[{"field_id":"f-200"}]', 0.0), + ({"matches_regex": r"^gemini_\d"}, "gemini_2_5_pro", 1.0), + ({"matches_regex": r"^gemini_\d"}, "gpt_4o", 0.0), + ({"any_of": ["a", "b"]}, "b", 1.0), + ({"any_of": ["a", "b"]}, "c", 0.0), + ], + ) + def test_predicate_forms(self, sandbox_with_log, predicate, recorded, expected): + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "projects", "get", "--val", recorded])]) + criterion = CliCalledCriterion( + description="predicate", + log=LOG, + verb="ixp projects get", + flags={"val": predicate}, + ) + assert SuccessChecker(sandbox).check(criterion).score == expected + + def test_dotall_flag_lets_a_pattern_cross_newlines(self, sandbox_with_log): + """A heredoc-built payload spans lines; without DOTALL `.` stops at the first.""" + payload = '[\n {"name": "Invoice Number",\n "instructions": "Extract it."}\n]' + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "fields", "update-prompts", "--updates", payload])]) + pattern = r'"name": "Invoice Number".*"instructions"' + + without_dotall = CliCalledCriterion( + description="no flags", + log=LOG, + verb="ixp fields update-prompts", + flags={"updates": {"matches_regex": pattern}}, + ) + with_dotall = CliCalledCriterion( + description="DOTALL", + log=LOG, + verb="ixp fields update-prompts", + flags={"updates": {"matches_regex": pattern, "flags": re.DOTALL}}, + ) + checker = SuccessChecker(sandbox) + 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_absent_distinguishes_missing_from_different_value(self, sandbox_with_log): + """`absent` is why flags is a predicate map, not dict[str, str].""" + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "labellings", "confirm", "proj-1"])]) + criterion = CliCalledCriterion( + description="confirmed without corrections", + log=LOG, + verb="ixp labellings confirm", + flags={"corrections": {"absent": True}}, + ) + assert SuccessChecker(sandbox).check(criterion).score == 1.0 + + def test_absent_fails_when_flag_present(self, sandbox_with_log): + sandbox, sandbox_dir = sandbox_with_log + _write_log( + sandbox_dir, + [_call(["ixp", "labellings", "confirm", "proj-1", "--corrections", "[]"])], + ) + criterion = CliCalledCriterion( + description="confirmed without corrections", + log=LOG, + verb="ixp labellings confirm", + flags={"corrections": {"absent": True}}, + ) + assert SuccessChecker(sandbox).check(criterion).score == 0.0 + + def test_repeated_flag_satisfied_by_any_value(self, sandbox_with_log): + sandbox, sandbox_dir = sandbox_with_log + _write_log( + sandbox_dir, + [_call(["ixp", "labellings", "confirm", "--fields", "f-1", "--fields", "f-2"])], + ) + criterion = CliCalledCriterion( + description="confirmed f-2", + log=LOG, + verb="ixp labellings confirm", + flags={"fields": "f-2"}, + ) + assert SuccessChecker(sandbox).check(criterion).score == 1.0 + + +class TestCounts: + def test_max_count_zero_is_the_negative_guard(self, sandbox_with_log): + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "fields", "rename", "proj-1"])]) + forbidden = CliCalledCriterion( + description="did not delete the field", + log=LOG, + verb="ixp fields delete", + min_count=0, + max_count=0, + ) + assert SuccessChecker(sandbox).check(forbidden).score == 1.0 + + def test_max_count_zero_fails_when_called(self, sandbox_with_log): + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "fields", "delete", "proj-1"])]) + forbidden = CliCalledCriterion( + description="did not delete the field", + log=LOG, + verb="ixp fields delete", + min_count=0, + max_count=0, + ) + result = SuccessChecker(sandbox).check(forbidden) + assert result.score == 0.0 + assert "forbids" in result.details + + def test_min_count_requires_repetition(self, sandbox_with_log): + sandbox, sandbox_dir = sandbox_with_log + _write_log( + sandbox_dir, + [_call(["ixp", "documents", "upload", "proj-1", f"doc{n}.pdf"]) for n in range(3)], + ) + criterion = CliCalledCriterion( + description="uploaded three documents", + log=LOG, + verb="ixp documents upload", + min_count=3, + ) + assert SuccessChecker(sandbox).check(criterion).score == 1.0 + stricter = criterion.model_copy(update={"min_count": 4}) + assert SuccessChecker(sandbox).check(stricter).score == 0.0 + + +class TestLogHandling: + def test_missing_log_fails_even_a_negative_guard(self, sandbox_with_log): + """A missing log is a harness fault, so `max_count: 0` must NOT pass on it. + + Otherwise re-pointing the mock's sink would make every negative guard + pass vacuously. + """ + sandbox, _ = sandbox_with_log + forbidden = CliCalledCriterion( + description="did not delete", + log=LOG, + verb="ixp fields delete", + min_count=0, + max_count=0, + ) + result = SuccessChecker(sandbox).check(forbidden) + assert result.score == 0.0 + assert "does not exist" in (result.error or "") + + def test_empty_log_is_zero_calls_not_an_error(self, sandbox_with_log): + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, []) + forbidden = CliCalledCriterion( + description="did not delete", + log=LOG, + verb="ixp fields delete", + min_count=0, + max_count=0, + ) + result = SuccessChecker(sandbox).check(forbidden) + assert result.score == 1.0 + assert result.error is None + + def test_malformed_line_now_fails_instead_of_being_skipped(self, sandbox_with_log): + """Superseded behaviour: an unparseable line used to be skipped with the + score untouched, which let a max_count: 0 guard pass on a truncated record + of the forbidden call. It is now a harness fault, like a missing log.""" + sandbox, sandbox_dir = sandbox_with_log + log_path = sandbox_dir / LOG + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + json.dumps(_call(["ixp", "projects", "get", "proj-1"])) + "\nnot json at all\n", + encoding="utf-8", + ) + criterion = CliCalledCriterion(description="got it", log=LOG, verb="ixp projects get") + result = SuccessChecker(sandbox).check(criterion) + assert result.score == 0.0 + assert "1 unusable record" in (result.error or "") + + +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"})) + assert space == equals == (["get"], {"model": ["pro"]}) + + def test_output_is_ignored_by_default(self, sandbox_with_log): + """--output is outcome-invisible, so grading must not depend on it.""" + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "projects", "get", "proj-1"])]) + with_json = CliCalledCriterion( + description="got it", + log=LOG, + verb="ixp projects get", + positional=["proj-1"], + ) + assert SuccessChecker(sandbox).check(with_json).score == 1.0 + + _write_log(sandbox_dir, [_call(["ixp", "projects", "get", "--output", "json", "proj-1"])]) + 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()) + 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( + ["confirm", "--corrections", '[{"v":"--x"}]'], frozenset(), frozenset({"corrections"}) + ) + assert positional == ["confirm"] + assert flags == {"corrections": ['[{"v":"--x"}]']} + + 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()) + 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()) + assert positional == ["import", "-"] + assert flags == {} + + +class TestRegressionsFromReview: + """One test per defect found reviewing PR #72, each written in the failing + direction — the guard that reported a PASS while the log proved otherwise.""" + + def test_boolean_switch_before_a_positional_does_not_swallow_it(self, sandbox_with_log): + """`delete --yes proj-1`: the guard must CATCH the delete, not pass. + + The old heuristic bound `yes=proj-1`, emptied the positionals, and scored + a `max_count: 0` guard 1.0 on the very invocation it forbids. + """ + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "fields", "delete", "--yes", "proj-1"])]) + forbidden = CliCalledCriterion( + description="did NOT delete proj-1", + log=LOG, + verb="ixp fields delete", + positional=["proj-1"], + min_count=0, + max_count=0, + ) + assert SuccessChecker(sandbox).check(forbidden).score == 0.0 + # ...and the positive form of the same assertion must hold. + positive = forbidden.model_copy(update={"min_count": 1, "max_count": None}) + assert SuccessChecker(sandbox).check(positive).score == 1.0 + + def test_ignored_switch_does_not_swallow_the_next_positional(self, sandbox_with_log): + """Mirror of the --yes case, through ignore_flags. + + Folding ignore_flags into the value-bearing set was right for + `--output json` but made every ignored SWITCH consume its neighbour, + reopening the same false PASS on a destructive call. + """ + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "fields", "delete", "--verbose", "proj-1"])]) + guard = CliCalledCriterion( + description="did NOT delete proj-1", + log=LOG, + verb="ixp fields delete", + positional=["proj-1"], + ignore_flags=["verbose"], + min_count=0, + max_count=0, + ) + assert SuccessChecker(sandbox).check(guard).score == 0.0 + + def test_ignored_value_flag_still_consumes_its_value(self, sandbox_with_log): + """The control: `output` is ignored AND declared value-bearing by default, + so `json` must not leak into the positionals.""" + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "projects", "get", "--output", "json", "proj-1"])]) + criterion = CliCalledCriterion( + description="got proj-1", log=LOG, verb="ixp projects get", positional=["proj-1"] + ) + assert SuccessChecker(sandbox).check(criterion).score == 1.0 + + def test_failure_details_show_what_was_actually_recorded(self, sandbox_with_log): + """A bare count sends the reader to the sandbox; this criterion exists to + answer 'what did it actually run'.""" + sandbox, sandbox_dir = sandbox_with_log + _write_log( + sandbox_dir, + [ + _call(["ixp", "projects", "get", "proj-1"]), + _call(["ixp", "projects", "list"]), + _call(["ixp", "fields", "rename", "proj-1"]), + _call(["ixp", "track"]), + ], + ) + criterion = CliCalledCriterion(description="configured the model", log=LOG, verb="ixp projects configure-model") + details = SuccessChecker(sandbox).check(criterion).details or "" + assert "Recorded:" in details + assert "ixp projects get proj-1" in details + assert "(+1 more)" in details + + 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( + ["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()) + 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()) + assert positional == ["get"] + assert flags == {"offset": ["-1"]} + + def test_unparseable_line_fails_a_negative_guard(self, sandbox_with_log): + """An unreadable record might BE the forbidden call, so the guard must fail.""" + sandbox, sandbox_dir = sandbox_with_log + log_path = sandbox_dir / LOG + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + json.dumps(_call(["ixp", "projects", "get", "p1"])) + + '\n{"tool": "uip", "argv": ["ixp", "fields", "delete"\n', + encoding="utf-8", + ) + forbidden = CliCalledCriterion( + description="did NOT delete", + log=LOG, + verb="ixp fields delete", + min_count=0, + max_count=0, + ) + result = SuccessChecker(sandbox).check(forbidden) + assert result.score == 0.0 + assert "unusable record" in (result.error or "") + + def test_argv_not_a_list_of_strings_fails_loudly(self, sandbox_with_log): + """A mock recording argv as a string is a harness fault, not a non-match.""" + sandbox, sandbox_dir = sandbox_with_log + log_path = sandbox_dir / LOG + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text(json.dumps({"tool": "uip", "argv": "ixp fields delete proj-1"}) + "\n", encoding="utf-8") + forbidden = CliCalledCriterion( + description="did NOT delete", log=LOG, verb="ixp fields delete", min_count=0, max_count=0 + ) + result = SuccessChecker(sandbox).check(forbidden) + assert result.score == 0.0 + assert "unusable record" in (result.error or "") + + def test_required_flag_missing_entirely_scores_zero(self, sandbox_with_log): + """The branch separating `equals` from `absent`, previously uncovered.""" + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "projects", "configure-model", "proj-1"])]) + criterion = CliCalledCriterion( + description="passed --model at all", + log=LOG, + verb="ixp projects configure-model", + flags={"model": "gemini_2_5_pro"}, + ) + assert SuccessChecker(sandbox).check(criterion).score == 0.0 + + def test_asserting_a_switch_does_not_swallow_the_next_positional(self, sandbox_with_log): + """Adding a switch predicate must not weaken the guard it is added to. + + A presence predicate needs no value, so it must not make the flag + value-bearing — otherwise `flags: {yes: {present: true}}` on a guard over + `delete --yes proj-1` would rebind `yes=proj-1`, empty the positionals, + and hand the guard a false PASS: the exact defect declared value-binding + exists to prevent, reintroduced by trying to assert more. + """ + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "fields", "delete", "--yes", "proj-1"])]) + guard = CliCalledCriterion( + description="did NOT delete proj-1 with --yes", + log=LOG, + verb="ixp fields delete", + positional=["proj-1"], + flags={"yes": {"present": True}}, + min_count=0, + max_count=0, + ) + assert SuccessChecker(sandbox).check(guard).score == 0.0 + positive = guard.model_copy(update={"min_count": 1, "max_count": None}) + assert SuccessChecker(sandbox).check(positive).score == 1.0 + + @pytest.mark.parametrize( + ("argv_tail", "expected"), + [(["--yes", "proj-1"], 1.0), (["-y", "proj-1"], 1.0), (["proj-1"], 0.0)], + ) + def test_aliases_make_short_and_long_one_flag(self, sandbox_with_log, argv_tail, expected): + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "fields", "delete", *argv_tail])]) + criterion = CliCalledCriterion( + description="confirmed, either spelling", + log=LOG, + verb="ixp fields delete", + flags={"yes": {"present": True, "aliases": ["y"]}}, + ) + assert SuccessChecker(sandbox).check(criterion).score == expected + + @pytest.mark.parametrize( + ("argv_tail", "expected"), + [(["--yes", "proj-1"], 1.0), (["-y", "proj-1"], 1.0), (["proj-1"], 0.0)], + ) + def test_absent_across_aliases_needs_all_spellings_missing(self, sandbox_with_log, argv_tail, expected): + """Without aliases this was silently wrong: an ANDed pair of absent-guards + (one per spelling) flagged EVERY invocation, whatever it did, because the + spelling not used was always absent.""" + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "fields", "delete", *argv_tail])]) + guard = CliCalledCriterion( + description="never deleted without confirming", + log=LOG, + verb="ixp fields delete", + flags={"yes": {"absent": True, "aliases": ["y"]}}, + min_count=0, + max_count=0, + ) + assert SuccessChecker(sandbox).check(guard).score == expected + + def test_alias_of_a_value_flag_binds_its_value(self, sandbox_with_log): + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "labellings", "confirm", "-f", "f-002"])]) + criterion = CliCalledCriterion( + description="confirmed f-002 via the short flag", + log=LOG, + verb="ixp labellings confirm", + flags={"fields": {"equals": "f-002", "aliases": ["f"]}}, + ) + assert SuccessChecker(sandbox).check(criterion).score == 1.0 + + def test_present_requires_the_flag(self, sandbox_with_log): + sandbox, sandbox_dir = sandbox_with_log + _write_log(sandbox_dir, [_call(["ixp", "fields", "delete", "proj-1"])]) + criterion = CliCalledCriterion( + description="passed --yes", + log=LOG, + verb="ixp fields delete", + flags={"yes": {"present": True}}, + ) + 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 "") + + +class TestModelValidation: + @pytest.mark.parametrize("verb", ["", " ", "\t"]) + def test_blank_verb_rejected(self, verb): + """A blank verb is an empty prefix: it matched every record and scored 1.0.""" + with pytest.raises(ValidationError): + CliCalledCriterion(description="d", log=LOG, verb=verb) + + def test_empty_any_of_rejected(self): + with pytest.raises(ValidationError): + CliCalledCriterion(description="d", log=LOG, verb="v", flags={"m": {"any_of": []}}) + + def test_min_count_zero_without_max_count_rejected(self): + """Satisfied by every possible log, so the criterion could never fail.""" + with pytest.raises(ValidationError, match="can never fail"): + CliCalledCriterion(description="d", log=LOG, verb="v", min_count=0) + + def test_alias_claimed_by_two_predicates_rejected(self): + """Ambiguous ownership would make the verdict depend on dict order.""" + with pytest.raises(ValidationError, match="claimed by both"): + CliCalledCriterion( + description="d", + log=LOG, + verb="v", + flags={"yes": {"present": True, "aliases": ["y"]}, "y": {"present": True}}, + ) + + def test_self_alias_rejected(self): + with pytest.raises(ValidationError, match="itself in aliases"): + CliCalledCriterion(description="d", log=LOG, verb="v", flags={"yes": {"present": True, "aliases": ["yes"]}}) + + def test_alias_in_ignore_flags_rejected(self): + with pytest.raises(ValidationError, match="ignore_flags"): + CliCalledCriterion( + description="d", + log=LOG, + verb="v", + flags={"out": {"present": True, "aliases": ["output"]}}, + ) + + def test_predicate_on_an_ignored_flag_rejected(self): + """ignore_flags drops the flag before predicates run, so this can never work.""" + with pytest.raises(ValidationError, match="ignore_flags"): + CliCalledCriterion(description="d", log=LOG, verb="v", flags={"output": "json"}) + + def test_docstring_negative_example_is_constructible(self): + """The model docstring's negative example must actually validate.""" + CliCalledCriterion( + description="Did not use --corrections to flip a boolean field", + log="mocks/calls.jsonl", + verb="ixp labellings confirm", + flags={"corrections": {"contains": "f-100"}}, + min_count=0, + max_count=0, + ) + + def test_scalar_shorthand_equals_predicate(self): + shorthand = CliCalledCriterion(description="d", log=LOG, verb="ixp projects get", flags={"model": "pro"}) + explicit = CliCalledCriterion( + description="d", log=LOG, verb="ixp projects get", flags={"model": {"equals": "pro"}} + ) + assert shorthand.flags == explicit.flags + + def test_two_predicates_on_one_flag_rejected(self): + with pytest.raises(ValidationError, match="exactly one"): + CliCalledCriterion( + description="d", + log=LOG, + verb="ixp projects get", + flags={"model": {"equals": "a", "contains": "b"}}, + ) + + def test_no_predicate_on_one_flag_rejected(self): + with pytest.raises(ValidationError, match="exactly one"): + CliCalledCriterion(description="d", log=LOG, verb="v", flags={"model": {}}) + + def test_flags_without_matches_regex_rejected(self): + """Setting flags beside another predicate would be a silent no-op.""" + with pytest.raises(ValidationError, match="applies only to matches_regex"): + CliCalledCriterion(description="d", log=LOG, verb="v", flags={"model": {"equals": "x", "flags": 16}}) + + def test_max_count_below_min_count_rejected(self): + with pytest.raises(ValidationError, match="must be >="): + CliCalledCriterion(description="d", log=LOG, verb="v", min_count=2, max_count=1) + + def test_criterion_with_nothing_to_match_rejected(self): + """A criterion matching on nothing would count every invocation.""" + with pytest.raises(ValidationError, match="at least one of"): + CliCalledCriterion(description="d", log=LOG) + + def test_unknown_field_rejected(self): + with pytest.raises(ValidationError, match="Extra inputs are not permitted"): + CliCalledCriterion(description="d", log=LOG, verb="v", pattern="oops") diff --git a/tests/test_success_criterion_union.py b/tests/test_success_criterion_union.py index 34f5aad7..3163cb7f 100644 --- a/tests/test_success_criterion_union.py +++ b/tests/test_success_criterion_union.py @@ -37,6 +37,7 @@ def _make_task(criteria: list[dict]) -> TaskDefinition: "json_check": {"description": "d", "path": "f.json"}, "reference_comparison": {"description": "d", "agent_file": "f.py"}, "command_executed": {"description": "d"}, + "cli_called": {"description": "d", "log": "calls.jsonl", "verb": "ixp projects get"}, "commands_efficiency": {"description": "d", "expected_commands": 3}, "uipath_eval": {"description": "d", "agent_name": "a", "eval_set": "e", "thresholds": {"accuracy": 0.8}}, "classification_match": {