| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| """ |
| Skill eval runner. |
| |
| Two modes: |
| |
| 1. **Print mode (default)** — emit the system prompt, user prompt, and |
| expected JSON for each case. The operator (or the agent making the |
| change, in self-eval mode) pastes the prompts into the model under |
| test and diffs the response against expected.json manually. |
| |
| 2. **`--cli` mode** — pipe ``<system_prompt>\\n\\n<user_prompt>`` on stdin |
| to the configured shell command, capture stdout, extract the JSON |
| the model produced, and compare against expected.json automatically. |
| Reports PASS / FAIL / MANUAL per case and exits non-zero on any FAIL. |
| "Structural" expected.json files (top-level ``has_*`` flags or |
| ``mention_*`` lists) assert properties of the model's prose rather than |
| exact field values. When the fixtures dir provides an ``assertions.json`` |
| mapping each such key to a predicate (``regex`` / ``contains`` / |
| ``contains_all`` / ``empty`` / ``non_empty`` / ``field_true`` / |
| ``max_length`` run locally; ``judge`` piped to the grader CLI), those |
| cases are graded automatically. |
| Any predicate may carry ``"negate": true`` to assert the absence of a |
| match. A structural case with no ``assertions.json`` falls back to MANUAL |
| and prints prompts for manual review. |
| |
| By default, free-text fields (rationale, reason, drop_reason, |
| blockers, etc.) are graded by piping a short rubric prompt to a |
| cheap judge model (``claude -p --model haiku`` by default) and |
| parsing ``{"match": bool, "reason": str}``. Decision fields |
| (booleans, enums, counts, ordering, IDs) are still compared |
| exactly. Override the grader command with ``--grader-cli``, or pass |
| ``--exact`` to disable grading and require verbatim equality on |
| every field. The set of prose fields defaults to a built-in list |
| and can be overridden per fixtures dir via ``grading-schema.json``. |
| No caching: every prose field is sent to the grader on every run. |
| |
| Usage: |
| # Print prompts for all cases under a fixtures directory |
| uv run --project tools/skill-evals skill-eval \\ |
| evals/security-issue-import/step-2a-semantic-sweep/fixtures/ |
| |
| # Print prompt for a single case |
| uv run --project tools/skill-evals skill-eval \\ |
| evals/security-issue-import/step-2a-semantic-sweep/fixtures/case-1-clear-duplicate |
| |
| # Automated comparison against a CLI. Decision fields are graded |
| # exact; prose fields go to the default grader (claude -p --model haiku). |
| uv run --project tools/skill-evals skill-eval --cli "claude -p" \\ |
| evals/security-issue-import/step-2a-semantic-sweep/fixtures/ |
| |
| # Override the grader, e.g. to use a different cheap model. |
| uv run --project tools/skill-evals skill-eval \\ |
| --cli "claude -p" \\ |
| --grader-cli "llm -m gpt-4o-mini" \\ |
| evals/security-issue-import/step-2a-semantic-sweep/fixtures/ |
| |
| # Disable the grader and require verbatim JSON equality on every field. |
| uv run --project tools/skill-evals skill-eval --cli "claude -p" --exact \\ |
| evals/security-issue-import/step-2a-semantic-sweep/fixtures/ |
| """ |
| |
| from __future__ import annotations |
| |
| import argparse |
| import json |
| import re |
| import shlex |
| import subprocess |
| import sys |
| from pathlib import Path |
| |
| # --------------------------------------------------------------------------- |
| # Prompt construction |
| # --------------------------------------------------------------------------- |
| |
| # Available slots: {corpus}, {roster}, {report}. |
| # Literal braces in a custom user-prompt-template.md that are NOT slots |
| # must be doubled ({{ and }}) so Python's str.format() leaves them intact. |
| USER_PROMPT_TEMPLATE = """\ |
| ## Existing open trackers (corpus) |
| |
| {corpus} |
| |
| ## Reporter roster (existing trackers mapped to reporter email) |
| |
| {roster} |
| |
| ## Incoming report |
| |
| {report} |
| |
| Apply the semantic sweep and reporter-identity check. Return JSON only. |
| """ |
| |
| |
| def build_corpus_text(corpus: list[dict]) -> str: |
| lines = [] |
| for item in corpus: |
| lines.append(f"#{item['number']} | {item['title']!r}") |
| lines.append(f"Body: {item['body']}") |
| lines.append("") |
| return "\n".join(lines) |
| |
| |
| def build_roster_text(roster: dict[str, str]) -> str: |
| if not roster: |
| return "(none)" |
| return "\n".join(f"#{num}: {email}" for num, email in roster.items()) |
| |
| |
| def find_repo_root(start: Path) -> Path: |
| """Walk up the directory tree until a .git directory is found.""" |
| p = start.resolve() |
| while p != p.parent: |
| if (p / ".git").exists(): |
| return p |
| p = p.parent |
| raise RuntimeError(f"Could not find repo root (.git) from {start}") |
| |
| |
| def extract_skill_section(skill_md_path: Path, heading: str) -> str: |
| """Return the section of a SKILL.md that begins with *heading*. |
| |
| Extraction ends at the next heading of the same or higher level, or at |
| the end of the file. Raises ValueError if the heading is not found. |
| """ |
| text = skill_md_path.read_text() |
| lines = text.split("\n") |
| heading_stripped = heading.rstrip() |
| m = re.match(r"^(#{1,6}) ", heading_stripped) |
| if not m: |
| raise ValueError(f"Heading {heading!r} does not look like a Markdown heading") |
| heading_level = len(m.group(1)) |
| |
| start = next( |
| (i for i, line in enumerate(lines) if line.rstrip() == heading_stripped), |
| None, |
| ) |
| if start is None: |
| raise ValueError(f"Heading {heading!r} not found in {skill_md_path}") |
| |
| end = len(lines) |
| in_fence = False |
| for i in range(start + 1, len(lines)): |
| stripped = lines[i].lstrip() |
| if stripped.startswith("```") or stripped.startswith("~~~"): |
| in_fence = not in_fence |
| if in_fence: |
| continue |
| hm = re.match(r"^(#{1,6}) ", lines[i]) |
| if hm and len(hm.group(1)) <= heading_level: |
| end = i |
| break |
| |
| return "\n".join(lines[start:end]).rstrip() |
| |
| |
| def load_step_config(fixtures_dir: Path) -> tuple[str, str]: |
| """Return (system_prompt, user_prompt_template) for the given fixtures dir. |
| |
| Resolution order: |
| 1. ``step-config.json`` — extracts the step section live from the skill's |
| SKILL.md, then appends ``output-spec.md`` if present. This is the |
| preferred path: tests automatically exercise the current skill text. |
| 2. ``system-prompt.md`` — a manually maintained prompt used by triage steps. |
| |
| Raises FileNotFoundError if neither file is present. |
| """ |
| user_tmpl_path = fixtures_dir / "user-prompt-template.md" |
| user_prompt_template = user_tmpl_path.read_text() if user_tmpl_path.exists() else USER_PROMPT_TEMPLATE |
| |
| # 1. step-config.json → live extraction from SKILL.md |
| config_path = fixtures_dir / "step-config.json" |
| if config_path.exists(): |
| config = json.loads(config_path.read_text()) |
| repo_root = find_repo_root(fixtures_dir) |
| skill_md_path = repo_root / config["skill_md"] |
| section = extract_skill_section(skill_md_path, config["step_heading"]) |
| output_spec_path = fixtures_dir / "output-spec.md" |
| if output_spec_path.exists(): |
| section += "\n\n" + output_spec_path.read_text() |
| return section, user_prompt_template |
| |
| # 2. system-prompt.md → manually maintained (triage steps) |
| sys_prompt_path = fixtures_dir / "system-prompt.md" |
| if sys_prompt_path.exists(): |
| return sys_prompt_path.read_text(), user_prompt_template |
| |
| raise FileNotFoundError( |
| f"{fixtures_dir} has neither step-config.json nor system-prompt.md. " |
| "Add a step-config.json pointing at the relevant SKILL.md section." |
| ) |
| |
| |
| # --------------------------------------------------------------------------- |
| # Case loading |
| # --------------------------------------------------------------------------- |
| |
| |
| def load_case(case_dir: Path) -> tuple[list[dict], dict, str, dict]: |
| """Return (corpus, roster, report_text, expected). |
| |
| ``corpus.json`` and ``reporter-roster.json`` are optional — steps that |
| do not need them simply omit them and get an empty list / dict. Each may |
| live either in the case directory (per-case fixtures) or in the shared |
| fixtures directory; the case-level file takes precedence when both exist. |
| """ |
| fixtures_dir = case_dir.parent |
| |
| def _resolve(name: str) -> Path: |
| case_level = case_dir / name |
| return case_level if case_level.exists() else fixtures_dir / name |
| |
| corpus_path = _resolve("corpus.json") |
| roster_path = _resolve("reporter-roster.json") |
| |
| corpus = json.loads(corpus_path.read_text()) if corpus_path.exists() else [] |
| roster = json.loads(roster_path.read_text()) if roster_path.exists() else {} |
| report = (case_dir / "report.md").read_text() |
| expected = json.loads((case_dir / "expected.json").read_text()) |
| return corpus, roster, report, expected |
| |
| |
| def load_case_tags(case_dir: Path) -> set[str]: |
| """Return optional runner-selection tags for a case. |
| |
| Tags live in ``case-meta.json`` so expected.json stays focused on the |
| behavioral assertion. Unknown metadata keys are ignored. |
| """ |
| meta_path = case_dir / "case-meta.json" |
| if not meta_path.exists(): |
| return set() |
| meta = json.loads(meta_path.read_text()) |
| tags = meta.get("tags", []) |
| if not isinstance(tags, list) or not all(isinstance(tag, str) for tag in tags): |
| raise ValueError(f"{meta_path} must contain a string-list 'tags' field") |
| return set(tags) |
| |
| |
| # --------------------------------------------------------------------------- |
| # Automated comparison (--cli mode) |
| # --------------------------------------------------------------------------- |
| |
| |
| def is_structural_expected(expected: dict) -> bool: |
| """Return True if expected.json describes prose properties, not exact output. |
| |
| Composition steps (e.g. compose-comment, compose-verdict) assert structural |
| properties of the model's prose via boolean ``has_*`` flags and membership |
| lists like ``mention_handles``. Those cannot be JSON-equality-compared |
| against a model's actual prose output, so --cli mode falls back to manual |
| review for them. |
| """ |
| if not isinstance(expected, dict): |
| return False |
| return any(key.startswith(("has_", "mention_")) for key in expected) |
| |
| |
| def extract_json_from_output(text: str) -> tuple[object | None, str | None]: |
| """Return (parsed_json, error) extracted from a model's stdout. |
| |
| Tries three strategies in order: |
| 1. The whole output is valid JSON. |
| 2. The output contains a ```json ... ``` fenced block. |
| 3. The output contains a balanced ``{...}`` (or ``[...]``) block — the |
| longest such block is tried. |
| |
| Returns ``(value, None)`` on success or ``(None, error_message)`` if no |
| JSON could be extracted. |
| """ |
| stripped = text.strip() |
| if stripped: |
| try: |
| return json.loads(stripped), None |
| except json.JSONDecodeError: |
| pass |
| |
| fence = re.search(r"```(?:json)?\s*\n(.*?)\n```", text, re.DOTALL) |
| if fence: |
| try: |
| return json.loads(fence.group(1).strip()), None |
| except json.JSONDecodeError: |
| pass |
| |
| candidate = _find_largest_brace_block(text) |
| if candidate is not None: |
| try: |
| return json.loads(candidate), None |
| except json.JSONDecodeError: |
| pass |
| |
| return None, "no JSON object or array found in model output" |
| |
| |
| def _find_largest_brace_block(text: str) -> str | None: |
| """Return the largest balanced ``{...}`` or ``[...]`` substring, or None.""" |
| best: str | None = None |
| for opener, closer in (("{", "}"), ("[", "]")): |
| depth = 0 |
| start = -1 |
| for i, ch in enumerate(text): |
| if ch == opener: |
| if depth == 0: |
| start = i |
| depth += 1 |
| elif ch == closer and depth > 0: |
| depth -= 1 |
| if depth == 0 and start >= 0: |
| candidate = text[start : i + 1] |
| if best is None or len(candidate) > len(best): |
| best = candidate |
| start = -1 |
| return best |
| |
| |
| def compare_outputs(actual: object, expected: object) -> tuple[bool, str]: |
| """Return (passed, diff_text). Diff is empty when passed=True.""" |
| if actual == expected: |
| return True, "" |
| return False, _format_diff(actual, expected) |
| |
| |
| # --------------------------------------------------------------------------- |
| # Field-aware grading (--grader-cli mode) |
| # --------------------------------------------------------------------------- |
| |
| # Default grader shell command. Used when --cli is set and --exact is not. |
| # Haiku is the cheapest Claude model; the rubric is small so cost is minimal. |
| DEFAULT_GRADER_CLI: str = "claude -p --model haiku" |
| |
| |
| # Keys whose values are treated as prose by default. The runner sends these |
| # to the grader CLI for a soft "does the candidate support the same |
| # conclusion?" judgement instead of requiring verbatim string equality. |
| # A per-fixtures-dir ``grading-schema.json`` can replace this list. |
| DEFAULT_PROSE_FIELDS: frozenset[str] = frozenset( |
| { |
| "rationale", |
| "reason", |
| "reasons", |
| "drop_reason", |
| "blockers", |
| "notes", |
| "summary", |
| "explanation", |
| "details", |
| "description", |
| } |
| ) |
| |
| |
| GRADER_RUBRIC = """\ |
| You are grading one field of a model's structured answer against a reference answer. |
| |
| Field path: {field_path} |
| |
| Expected value: |
| {expected_value} |
| |
| Candidate value: |
| {candidate_value} |
| |
| Does the candidate value support the same conclusion as the expected value? Ignore wording differences and reorderings. Reply with one line of JSON only, no prose: {{"match": true, "reason": "<one-line explanation>"}} or {{"match": false, "reason": "<one-line explanation>"}}. |
| """ |
| |
| |
| BATCH_GRADER_RUBRIC = """\ |
| You are grading a model's structured answer against a reference answer, field by field. |
| |
| For each (Field, Expected, Candidate) triple below, decide whether the candidate value supports the same conclusion as the expected value. Ignore wording differences and reorderings. |
| |
| {fields_block} |
| |
| Reply with one line of JSON only, no prose. The JSON is an object mapping each field path string to {{"match": true|false, "reason": "<one-line explanation>"}}. Include every field listed above. Example: |
| {{"$.foo": {{"match": true, "reason": "same conclusion"}}, "$.bar": {{"match": false, "reason": "different verdict"}}}} |
| """ |
| |
| |
| def load_grading_schema(fixtures_dir: Path) -> set[str]: |
| """Return the set of prose field names for cases in this fixtures dir. |
| |
| Reads ``fixtures_dir/grading-schema.json`` when present. The file may |
| set ``prose_fields`` to a string list that *replaces* the default set |
| (use ``["rationale", "reason", ...]`` to be explicit, or ``[]`` to |
| grade everything by exact match). |
| |
| Falls back to :data:`DEFAULT_PROSE_FIELDS` when no schema file exists. |
| """ |
| path = fixtures_dir / "grading-schema.json" |
| if not path.exists(): |
| return set(DEFAULT_PROSE_FIELDS) |
| data = json.loads(path.read_text()) |
| fields = data.get("prose_fields") |
| if fields is None: |
| return set(DEFAULT_PROSE_FIELDS) |
| if not isinstance(fields, list) or not all(isinstance(f, str) for f in fields): |
| raise ValueError(f"{path} must contain a string-list 'prose_fields' field") |
| return set(fields) |
| |
| |
| def _render_field_value(value: object) -> str: |
| """Render an expected/candidate field value for the grader prompt.""" |
| if isinstance(value, str): |
| return value |
| return json.dumps(value, indent=2, sort_keys=True, ensure_ascii=False) |
| |
| |
| def grade_prose_field( |
| field_path: str, |
| expected_value: object, |
| actual_value: object, |
| grader_cli: str, |
| timeout: int, |
| ) -> tuple[bool, str]: |
| """Ask the grader CLI whether the candidate value supports the same conclusion. |
| |
| Returns ``(match, note)``. ``note`` is empty on match and a one-line |
| summary on mismatch (or grader failure). |
| """ |
| if expected_value == actual_value: |
| return True, "" |
| prompt = GRADER_RUBRIC.format( |
| field_path=field_path, |
| expected_value=_render_field_value(expected_value), |
| candidate_value=_render_field_value(actual_value), |
| ) |
| try: |
| stdout, stderr, rc = run_cli(grader_cli, prompt, timeout=timeout) |
| except subprocess.TimeoutExpired: |
| return False, f"{field_path}: grader CLI timed out after {timeout}s" |
| except OSError as exc: |
| return False, f"{field_path}: grader CLI invocation failed ({exc})" |
| if rc != 0: |
| return False, f"{field_path}: grader CLI exited {rc} ({stderr.strip()[:200]})" |
| verdict, err = extract_json_from_output(stdout) |
| if err is not None or not isinstance(verdict, dict) or "match" not in verdict: |
| return False, f"{field_path}: grader returned unusable output ({err or 'missing match key'})" |
| match = bool(verdict.get("match")) |
| reason = str(verdict.get("reason", "")).strip() |
| if match: |
| return True, "" |
| return False, f"{field_path}: grader says NO ({reason or 'no reason given'})" |
| |
| |
| def collect_diffs( |
| actual: object, |
| expected: object, |
| *, |
| prose_fields: set[str], |
| path: str = "$", |
| ) -> tuple[list[str], list[tuple[str, object, object]]]: |
| """Walk both trees in parallel; return (decision_msgs, prose_pairs). |
| |
| ``decision_msgs`` lists structural/decision-field mismatches (type, key |
| set, list length, scalar inequality on non-prose keys). These cannot |
| be resolved by the grader. ``prose_pairs`` lists |
| ``(field_path, expected_value, actual_value)`` for prose-keyed |
| mismatches that the grader should judge. Equal values are omitted from |
| both lists. |
| """ |
| if type(actual) is not type(expected): |
| return [ |
| f"{path}: type mismatch (actual={type(actual).__name__}, expected={type(expected).__name__})" |
| ], [] |
| |
| if isinstance(expected, dict): |
| actual_dict = actual # type: ignore[assignment] |
| # Only assert on the intersection of keys. Keys in expected that the |
| # model didn't emit are skipped (not failed), and keys in actual that |
| # expected doesn't declare are ignored. expected.json describes what |
| # the model's answer SHOULD say where it speaks, not what it must |
| # include. |
| decision_msgs: list[str] = [] |
| prose_pairs: list[tuple[str, object, object]] = [] |
| for key in expected: |
| if key not in actual_dict: |
| continue |
| child_path = f"{path}.{key}" if path else key |
| if key in prose_fields: |
| if expected[key] != actual_dict[key]: |
| prose_pairs.append((child_path, expected[key], actual_dict[key])) |
| else: |
| sub_d, sub_p = collect_diffs( |
| actual_dict[key], |
| expected[key], |
| prose_fields=prose_fields, |
| path=child_path, |
| ) |
| decision_msgs.extend(sub_d) |
| prose_pairs.extend(sub_p) |
| return decision_msgs, prose_pairs |
| |
| if isinstance(expected, list): |
| actual_list = actual # type: ignore[assignment] |
| if len(actual_list) != len(expected): |
| return [f"{path}: length mismatch (actual={len(actual_list)}, expected={len(expected)})"], [] |
| decision_msgs = [] |
| prose_pairs = [] |
| for i, (a_item, e_item) in enumerate(zip(actual_list, expected, strict=False)): |
| sub_d, sub_p = collect_diffs( |
| a_item, |
| e_item, |
| prose_fields=prose_fields, |
| path=f"{path}[{i}]", |
| ) |
| decision_msgs.extend(sub_d) |
| prose_pairs.extend(sub_p) |
| return decision_msgs, prose_pairs |
| |
| if actual == expected: |
| return [], [] |
| # Non-prose scalar strings: treat case and surrounding/collapsed whitespace |
| # as insignificant. A weaker model that reaches the right verdict but writes |
| # it as "invalid" / "request_changes" should not fail on casing alone. A |
| # genuinely different value still differs after normalisation, so this never |
| # masks a wrong verdict. |
| if isinstance(actual, str) and isinstance(expected, str): |
| norm_actual = " ".join(actual.split()).casefold() |
| norm_expected = " ".join(expected.split()).casefold() |
| if norm_actual == norm_expected: |
| return [], [] |
| return [f"{path}: expected={expected!r}, actual={actual!r}"], [] |
| |
| |
| def _format_batch_fields_block(pairs: list[tuple[str, object, object]]) -> str: |
| chunks = [] |
| for path, expected, actual in pairs: |
| chunks.append( |
| f"Field: {path}\nExpected:\n{_render_field_value(expected)}\nCandidate:\n{_render_field_value(actual)}" |
| ) |
| return "\n\n".join(chunks) |
| |
| |
| def batch_grade_prose_fields( |
| pairs: list[tuple[str, object, object]], |
| grader_cli: str, |
| timeout: int, |
| ) -> dict[str, tuple[bool, str]]: |
| """Send one rubric prompt covering every pair; return path -> (match, note). |
| |
| Returns an empty dict when ``pairs`` is empty (no grader call). On grader |
| failure (timeout, OSError, non-zero exit, unparsable output, missing |
| path in the verdict), every pair without a clean verdict is returned as |
| ``(False, <one-line explanation>)``. |
| """ |
| if not pairs: |
| return {} |
| prompt = BATCH_GRADER_RUBRIC.format(fields_block=_format_batch_fields_block(pairs)) |
| try: |
| stdout, stderr, rc = run_cli(grader_cli, prompt, timeout=timeout) |
| except subprocess.TimeoutExpired: |
| return {p: (False, f"grader CLI timed out after {timeout}s") for p, _, _ in pairs} |
| except OSError as exc: |
| return {p: (False, f"grader CLI invocation failed ({exc})") for p, _, _ in pairs} |
| if rc != 0: |
| return {p: (False, f"grader CLI exited {rc} ({stderr.strip()[:200]})") for p, _, _ in pairs} |
| verdict, err = extract_json_from_output(stdout) |
| if err is not None or not isinstance(verdict, dict): |
| return {p: (False, f"grader returned unusable output ({err or 'not a dict'})") for p, _, _ in pairs} |
| result: dict[str, tuple[bool, str]] = {} |
| for path, _, _ in pairs: |
| entry = verdict.get(path) |
| if not isinstance(entry, dict) or "match" not in entry: |
| result[path] = (False, f"grader did not return a verdict for {path}") |
| continue |
| match = bool(entry.get("match")) |
| reason = str(entry.get("reason", "")).strip() |
| if match: |
| result[path] = (True, "") |
| else: |
| result[path] = (False, f"grader says NO ({reason or 'no reason given'})") |
| return result |
| |
| |
| def compare_with_grader( |
| actual: object, |
| expected: object, |
| *, |
| prose_fields: set[str], |
| grader_cli: str, |
| timeout: int, |
| ) -> tuple[bool, list[str]]: |
| """Field-aware comparison: decision keys exact, prose keys judged by grader. |
| |
| Walks both trees once to separate decision-field diffs from prose-field |
| diffs, then makes a single batched grader call covering every prose |
| mismatch. If decision fields already fail the comparison, the grader |
| is skipped entirely (one fewer CLI call per failing case). |
| |
| Returns ``(ok, messages)``; ``messages`` is empty when ok and otherwise |
| lists one note per failing field. |
| """ |
| decision_msgs, prose_pairs = collect_diffs(actual, expected, prose_fields=prose_fields) |
| if decision_msgs: |
| # Case already fails on a decision field; no need to call the grader. |
| return False, decision_msgs |
| if not prose_pairs: |
| return True, [] |
| grades = batch_grade_prose_fields(prose_pairs, grader_cli, timeout) |
| ok = True |
| msgs: list[str] = [] |
| for path, _, _ in prose_pairs: |
| match, note = grades.get(path, (False, f"{path}: no verdict returned by grader")) |
| if not match: |
| ok = False |
| # `note` from batch_grade_prose_fields is already field-attributed |
| # for missing entries; for grader verdicts it isn't, so prepend the |
| # path for clarity in the output. |
| if note.startswith(path): |
| msgs.append(note) |
| else: |
| msgs.append(f"{path}: {note}") |
| return ok, msgs |
| |
| |
| # --------------------------------------------------------------------------- |
| # Structural assertions (has_* / mention_* keys) |
| # --------------------------------------------------------------------------- |
| |
| # Structural expected.json files assert *properties of the model's prose* |
| # (does the announce body contain the Download Page link? did the model flag |
| # the injection?) rather than exact field values. Each such property is named |
| # by a has_* / mention_* key and is evaluated by a predicate declared in the |
| # fixtures dir's assertions.json. Deterministic predicate types run locally — |
| # fast, free, and flake-free, which is exactly what you want for links, |
| # headers, and security properties. The judge type pipes a yes/no rubric to |
| # the grader CLI for the genuinely semantic properties that regex can't pin |
| # down. |
| |
| _DETERMINISTIC_ASSERTION_TYPES: frozenset[str] = frozenset( |
| {"regex", "contains", "contains_all", "empty", "non_empty", "field_true", "max_length"} |
| ) |
| _VALID_ASSERTION_TYPES: frozenset[str] = _DETERMINISTIC_ASSERTION_TYPES | {"judge"} |
| |
| |
| def load_assertions(fixtures_dir: Path) -> dict[str, dict]: |
| """Return the structural-assertion specs for cases in this fixtures dir. |
| |
| Reads ``fixtures_dir/assertions.json`` when present: an object mapping |
| each ``has_*`` / ``mention_*`` key to a predicate spec. Returns an empty |
| dict when the file is absent — the runner then falls back to MANUAL for |
| structural cases, preserving the prior behaviour. |
| |
| Raises ValueError if the file is malformed or names an unknown predicate |
| type, so a typo fails loudly rather than silently skipping a check. |
| """ |
| path = fixtures_dir / "assertions.json" |
| if not path.exists(): |
| return {} |
| data = json.loads(path.read_text()) |
| if not isinstance(data, dict): |
| raise ValueError(f"{path} must be a JSON object mapping assertion keys to specs") |
| for key, spec in data.items(): |
| if not isinstance(spec, dict): |
| raise ValueError(f"{path}: assertion {key!r} must be an object") |
| atype = spec.get("type") |
| if atype not in _VALID_ASSERTION_TYPES: |
| raise ValueError( |
| f"{path}: assertion {key!r} has invalid type {atype!r}; " |
| f"valid types: {sorted(_VALID_ASSERTION_TYPES)}" |
| ) |
| return data |
| |
| |
| def _resolve_field(actual: object, field: str) -> tuple[object, bool]: |
| """Return ``(value, present)`` for a dotted ``field`` path into ``actual``.""" |
| cur = actual |
| for part in field.split("."): |
| if isinstance(cur, dict) and part in cur: |
| cur = cur[part] |
| else: |
| return None, False |
| return cur, True |
| |
| |
| def _compile_flags(spec: dict) -> int: |
| flags = 0 |
| mapping = {"i": re.IGNORECASE, "s": re.DOTALL, "m": re.MULTILINE} |
| for ch in str(spec.get("flags", "")): |
| flags |= mapping.get(ch, 0) |
| return flags |
| |
| |
| def evaluate_deterministic_assertion(spec: dict, actual: object) -> tuple[bool | None, str]: |
| """Evaluate a non-judge assertion. Return ``(holds, note)``. |
| |
| ``holds`` is True/False for whether the asserted property is present in |
| the model output, or None on a spec/usage error (which the caller reports |
| as a failure). ``note`` is a short explanation, empty on a clean result. |
| |
| Missing-field semantics: ``empty`` treats an absent field as empty (True); |
| ``non_empty`` / ``field_true`` / the text predicates treat an absent field |
| as not satisfied (False). |
| |
| A ``"negate": true`` key inverts the result, so a predicate can assert the |
| *absence* of a pattern (e.g. ``has_no_passphrase_arg`` checks that the |
| ``--passphrase`` token does not appear). Negation is applied only to a |
| concrete True/False result; a spec/usage error (None) is passed through |
| unchanged so the typo still fails loudly. |
| """ |
| holds, note = _evaluate_deterministic_assertion_raw(spec, actual) |
| if spec.get("negate") and holds is not None: |
| holds = not holds |
| return holds, note |
| |
| |
| def _evaluate_deterministic_assertion_raw(spec: dict, actual: object) -> tuple[bool | None, str]: |
| """Evaluate the predicate before any ``negate`` inversion is applied.""" |
| atype = spec["type"] |
| field = spec.get("field") |
| if field is None: |
| return None, f"type {atype!r} requires a 'field'" |
| value, present = _resolve_field(actual, field) |
| |
| if atype == "empty": |
| return (not present or value in ([], "", None, {})), "" |
| if atype == "non_empty": |
| return (present and value not in ([], "", None, {})), "" |
| if atype == "field_true": |
| return (present and value is True), "" |
| if atype == "max_length": |
| limit = spec.get("max") |
| if not isinstance(limit, int) or isinstance(limit, bool): |
| return None, "type 'max_length' requires an integer 'max'" |
| if not present: |
| return False, f"field {field!r} not present in output" |
| try: |
| length = len(value) # type: ignore[arg-type] |
| except TypeError: |
| return None, f"field {field!r} value has no length" |
| return (length <= limit), ("" if length <= limit else f"len={length} > max={limit}") |
| |
| # Text predicates need a string. Non-string values are JSON-serialised so |
| # a list/number field can still be substring/regex-matched if a spec asks. |
| if not present: |
| return False, f"field {field!r} not present in output" |
| text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False) |
| ci = "i" in str(spec.get("flags", "")) |
| |
| if atype == "regex": |
| pattern = spec.get("pattern") |
| if pattern is None: |
| return None, "type 'regex' requires a 'pattern'" |
| return (re.search(pattern, text, _compile_flags(spec)) is not None), "" |
| if atype == "contains": |
| sub = spec.get("substring") |
| if sub is None: |
| return None, "type 'contains' requires a 'substring'" |
| hay = text.lower() if ci else text |
| needle = sub.lower() if ci else sub |
| return (needle in hay), "" |
| if atype == "contains_all": |
| subs = spec.get("substrings") |
| if not isinstance(subs, list) or not subs: |
| return None, "type 'contains_all' requires a non-empty 'substrings' list" |
| hay = text.lower() if ci else text |
| missing = [s for s in subs if (s.lower() if ci else s) not in hay] |
| return (not missing), (f"missing: {missing}" if missing else "") |
| return None, f"unhandled assertion type {atype!r}" |
| |
| |
| JUDGE_ASSERTION_RUBRIC = """\ |
| You are checking whether a model's output satisfies specific named properties. |
| |
| Model output (JSON): |
| {output} |
| |
| For each property below, decide strictly from the output whether the property holds. |
| |
| {props_block} |
| |
| Reply with one line of JSON only, no prose: an object mapping each property key to {{"holds": true|false, "reason": "<one-line explanation>"}}. Include every property key listed above. Example: |
| {{"has_foo": {{"holds": true, "reason": "output states X"}}, "mention_bar": {{"holds": false, "reason": "not mentioned"}}}} |
| """ |
| |
| |
| def _format_judge_props_block(specs: dict[str, dict]) -> str: |
| chunks = [] |
| for key, spec in specs.items(): |
| rubric = spec.get("rubric", "") |
| field = spec.get("field") |
| scope = f" (focus on the {field!r} field)" if field else "" |
| chunks.append(f"Property: {key}{scope}\nHolds when: {rubric}") |
| return "\n\n".join(chunks) |
| |
| |
| def batch_judge_assertions( |
| specs: dict[str, dict], |
| actual: object, |
| grader_cli: str, |
| timeout: int, |
| ) -> dict[str, tuple[bool | None, str]]: |
| """Send one rubric covering every judge assertion; return key -> (holds, note). |
| |
| Empty ``specs`` makes no grader call. On any grader failure (timeout, |
| OSError, non-zero exit, unparsable output, missing key in the verdict), |
| the affected keys are returned with ``holds=None`` so the caller fails the |
| assertion rather than silently passing it — important for the security |
| cases this is used on. |
| """ |
| if not specs: |
| return {} |
| prompt = JUDGE_ASSERTION_RUBRIC.format( |
| output=json.dumps(actual, indent=2, ensure_ascii=False, sort_keys=True), |
| props_block=_format_judge_props_block(specs), |
| ) |
| try: |
| stdout, stderr, rc = run_cli(grader_cli, prompt, timeout=timeout) |
| except subprocess.TimeoutExpired: |
| return dict.fromkeys(specs, (None, f"grader CLI timed out after {timeout}s")) |
| except OSError as exc: |
| return dict.fromkeys(specs, (None, f"grader CLI invocation failed ({exc})")) |
| if rc != 0: |
| return dict.fromkeys(specs, (None, f"grader CLI exited {rc} ({stderr.strip()[:200]})")) |
| verdict, err = extract_json_from_output(stdout) |
| if err is not None or not isinstance(verdict, dict): |
| return dict.fromkeys(specs, (None, f"grader returned unusable output ({err or 'not a dict'})")) |
| result: dict[str, tuple[bool | None, str]] = {} |
| for key in specs: |
| entry = verdict.get(key) |
| if not isinstance(entry, dict) or "holds" not in entry: |
| result[key] = (None, f"grader did not return a verdict for {key}") |
| continue |
| result[key] = (bool(entry.get("holds")), str(entry.get("reason", "")).strip()) |
| return result |
| |
| |
| def compare_structural( |
| actual: object, |
| expected: dict, |
| assertions: dict[str, dict], |
| *, |
| prose_fields: set[str], |
| grader_cli: str, |
| exact: bool, |
| grader_timeout: int, |
| ) -> tuple[bool, list[str]]: |
| """Grade a structural expected.json (``has_*`` / ``mention_*`` keys). |
| |
| Structural keys are evaluated by their ``assertions.json`` predicates; |
| deterministic ones run locally and judge ones go to the grader in a single |
| batched call. Any remaining (non-structural) keys are compared with the |
| standard field-aware comparator — exact for decision fields, grader for |
| prose, or pure exact when ``exact`` is set. Returns ``(ok, notes)`` with |
| one note per failing field. |
| """ |
| structural = {k: v for k, v in expected.items() if k.startswith(("has_", "mention_"))} |
| remainder = {k: v for k, v in expected.items() if k not in structural} |
| |
| ok = True |
| notes: list[str] = [] |
| |
| if remainder: |
| sub_ok, sub_notes = compare_with_grader( |
| actual, |
| remainder, |
| prose_fields=set() if exact else prose_fields, |
| grader_cli=grader_cli, |
| timeout=grader_timeout, |
| ) |
| if not sub_ok: |
| ok = False |
| notes.extend(sub_notes) |
| |
| judge_specs: dict[str, dict] = {} |
| judge_expected: dict[str, bool] = {} |
| for key, exp_val in structural.items(): |
| spec = assertions.get(key) |
| if spec is None: |
| ok = False |
| notes.append(f"{key}: no assertion defined in assertions.json") |
| continue |
| if spec["type"] == "judge": |
| judge_specs[key] = spec |
| judge_expected[key] = bool(exp_val) |
| continue |
| holds, note = evaluate_deterministic_assertion(spec, actual) |
| if holds is None: |
| ok = False |
| notes.append(f"{key}: {note}") |
| elif holds != bool(exp_val): |
| detail = f" ({note})" if note else "" |
| notes.append(f"{key}: property={holds}, expected {bool(exp_val)}{detail}") |
| ok = False |
| |
| if judge_specs: |
| grades = batch_judge_assertions(judge_specs, actual, grader_cli, grader_timeout) |
| for key in judge_specs: |
| holds, note = grades.get(key, (None, "no verdict returned by grader")) |
| if holds is None: |
| ok = False |
| notes.append(f"{key}: {note}") |
| elif holds != judge_expected[key]: |
| detail = f" ({note})" if note else "" |
| notes.append(f"{key}: judge says property={holds}, expected {judge_expected[key]}{detail}") |
| ok = False |
| |
| return ok, notes |
| |
| |
| def _format_diff(actual: object, expected: object) -> str: |
| actual_text = json.dumps(actual, indent=2, sort_keys=True) |
| expected_text = json.dumps(expected, indent=2, sort_keys=True) |
| a_lines = actual_text.splitlines() |
| e_lines = expected_text.splitlines() |
| lines = ["--- expected", "+++ actual"] |
| for line in e_lines: |
| if line not in a_lines: |
| lines.append(f"- {line}") |
| for line in a_lines: |
| if line not in e_lines: |
| lines.append(f"+ {line}") |
| return "\n".join(lines) |
| |
| |
| def run_cli(cli: str, prompt: str, timeout: int = 120) -> tuple[str, str, int]: |
| """Run ``cli`` with ``prompt`` on stdin. Return (stdout, stderr, rc). |
| |
| The command string is tokenised with ``shlex.split`` and executed with |
| ``shell=False``. The operator supplies the command, so trust is not the |
| issue — using an argv list rather than a shell string keeps prompt content |
| (which can be attacker-controlled when an eval exercises an injection |
| case) firmly on stdin and well away from any shell interpretation, and |
| removes a class of accidental-metacharacter footgun in the operator's |
| command. Operators who need shell features (pipes, redirections, env-var |
| prefixes) should wrap their command in ``bash -c '<pipeline>'``. |
| """ |
| proc = subprocess.run( |
| shlex.split(cli), |
| input=prompt, |
| capture_output=True, |
| text=True, |
| shell=False, |
| timeout=timeout, |
| check=False, |
| ) |
| return proc.stdout, proc.stderr, proc.returncode |
| |
| |
| # --------------------------------------------------------------------------- |
| # CLI |
| # --------------------------------------------------------------------------- |
| |
| |
| def find_cases(path: Path) -> list[tuple[Path, Path]]: |
| """Return (case_dir, fixtures_dir) pairs under path. |
| |
| Handles three levels of granularity: |
| - single case dir (contains report.md) |
| - fixtures dir (contains case-* subdirs) |
| - skill/step dir (contains fixtures/ subdirs recursively) |
| """ |
| if (path / "report.md").exists(): |
| return [(path, path.parent)] |
| # Direct fixtures dir — all cases share the same fixtures dir. |
| direct = sorted(p for p in path.iterdir() if p.is_dir() and (p / "report.md").exists()) |
| if direct: |
| return [(p, path) for p in direct] |
| # Recursive search — e.g. skill dir spanning multiple steps. |
| # De-duplicate: skip any fixtures/ that is itself nested under another |
| # fixtures/ already in the set (guards against accidental double-counting |
| # if someone copies a case sub-tree that contains its own fixtures/). |
| results = [] |
| seen_fixtures: set[Path] = set() |
| for fixtures_dir in sorted(path.rglob("fixtures")): |
| if not fixtures_dir.is_dir(): |
| continue |
| if any(fixtures_dir.is_relative_to(f) for f in seen_fixtures): |
| continue |
| seen_fixtures.add(fixtures_dir) |
| for case_dir in sorted(fixtures_dir.iterdir()): |
| if case_dir.is_dir() and (case_dir / "report.md").exists(): |
| results.append((case_dir, fixtures_dir)) |
| return results |
| |
| |
| def collect_tag_counts(cases: list[tuple[Path, Path]]) -> dict[str, int]: |
| """Return how many discovered cases carry each tag.""" |
| counts: dict[str, int] = {} |
| for case_dir, _fixtures_dir in cases: |
| for tag in load_case_tags(case_dir): |
| counts[tag] = counts.get(tag, 0) + 1 |
| return counts |
| |
| |
| def main(argv: list[str] | None = None) -> int: |
| parser = argparse.ArgumentParser( |
| description=( |
| "Run skill eval cases. Default mode prints prompts for manual review. " |
| "--cli mode pipes prompts through a shell command and auto-compares " |
| "against expected.json." |
| ) |
| ) |
| parser.add_argument( |
| "path", |
| type=Path, |
| help="Path to a single case directory or a fixtures directory containing multiple cases.", |
| ) |
| parser.add_argument( |
| "--quiet", |
| action="store_true", |
| help="Suppress prompt content; print only case names and expected JSON.", |
| ) |
| parser.add_argument( |
| "--cli", |
| type=str, |
| default=None, |
| help=( |
| "Shell command that reads a prompt on stdin and writes the model " |
| "response to stdout (e.g. 'claude -p'). When set, the runner sends " |
| "<system_prompt>\\n\\n<user_prompt> to the command for each case, " |
| "extracts JSON from stdout, and compares against expected.json. " |
| "Exits non-zero on any FAIL." |
| ), |
| ) |
| parser.add_argument( |
| "--timeout", |
| type=int, |
| default=120, |
| help="Timeout in seconds for each --cli invocation (default: 120).", |
| ) |
| parser.add_argument( |
| "--grader-cli", |
| type=str, |
| default=DEFAULT_GRADER_CLI, |
| help=( |
| "Shell command for a cheap judge model that grades free-text " |
| "fields (rationale, reason, drop_reason, blockers, etc.). " |
| "Prose fields are compared via a rubric prompt instead of " |
| "exact equality; decision fields stay on exact compare. The set " |
| "of prose fields is the runner's default plus any per-fixtures " |
| "grading-schema.json overrides. Requires --cli. Default: " |
| f"'{DEFAULT_GRADER_CLI}'. Pass --exact to disable grading." |
| ), |
| ) |
| parser.add_argument( |
| "--exact", |
| action="store_true", |
| help=( |
| "Disable the field-aware grader and require verbatim JSON " |
| "equality on every field (the runner's pre-grader behaviour)." |
| ), |
| ) |
| parser.add_argument( |
| "--grader-timeout", |
| type=int, |
| default=60, |
| help="Timeout in seconds for each --grader-cli invocation (default: 60).", |
| ) |
| parser.add_argument( |
| "--verbose", |
| "-v", |
| action="store_true", |
| help="In --cli mode, also print the prompts and the model's raw stdout per case.", |
| ) |
| parser.add_argument( |
| "--fail-fast", |
| action="store_true", |
| help=("Stops on the first failure instead of running all cases. Only applies in --cli mode; "), |
| ) |
| parser.add_argument( |
| "--tag", |
| action="append", |
| default=[], |
| help=( |
| "Run only cases tagged in case-meta.json. May be passed multiple " |
| "times; a case is included if it has all requested tags." |
| ), |
| ) |
| parser.add_argument( |
| "--list-tags", |
| action="store_true", |
| help=( |
| "Print every distinct tag declared in case-meta.json under path, " |
| "with the number of cases carrying each tag, and exit without " |
| "running prompts." |
| ), |
| ) |
| args = parser.parse_args(argv) |
| |
| grader_explicit = args.grader_cli != DEFAULT_GRADER_CLI |
| if args.cli is None and (grader_explicit or args.exact): |
| parser.error("--grader-cli and --exact require --cli") |
| |
| cases = find_cases(args.path) |
| if args.list_tags: |
| tag_counts = collect_tag_counts(cases) |
| if not tag_counts: |
| print("no tags found") |
| return 0 |
| for tag in sorted(tag_counts): |
| print(f"{tag} {tag_counts[tag]}") |
| return 0 |
| |
| if args.tag: |
| requested_tags = set(args.tag) |
| cases = [ |
| (case_dir, fixtures_dir) |
| for case_dir, fixtures_dir in cases |
| if requested_tags.issubset(load_case_tags(case_dir)) |
| ] |
| if not cases: |
| tag_suffix = f" matching tag(s): {', '.join(args.tag)}" if args.tag else "" |
| print(f"No eval cases found under {args.path}{tag_suffix}", file=sys.stderr) |
| return 1 |
| |
| # Cache loaded step configs so we don't re-read prompts for every case in |
| # the same fixtures dir (common when running a whole skill at once). |
| _step_config_cache: dict[Path, tuple[str, str]] = {} |
| # Cache the prose-field schema per fixtures dir (config only, not grader results). |
| _grading_schema_cache: dict[Path, set[str]] = {} |
| # Cache the structural-assertion specs per fixtures dir. |
| _assertions_cache: dict[Path, dict[str, dict]] = {} |
| |
| passed = failed = manual = errored = 0 |
| |
| for case_dir, fixtures_dir in cases: |
| if (args.cli is not None) and args.fail_fast and (failed or errored): |
| print("Fail-fast enabled; stopping on first failure or error.") |
| break |
| if fixtures_dir not in _step_config_cache: |
| _step_config_cache[fixtures_dir] = load_step_config(fixtures_dir) |
| system_prompt, user_prompt_template = _step_config_cache[fixtures_dir] |
| |
| corpus, roster, report, expected = load_case(case_dir) |
| try: |
| user_prompt = user_prompt_template.format( |
| corpus=build_corpus_text(corpus), |
| roster=build_roster_text(roster), |
| report=report, |
| ) |
| except (KeyError, ValueError) as exc: |
| raise type(exc)( |
| f"user-prompt-template.md in {fixtures_dir} has a format error: {exc}. " |
| "Available slots: {{corpus}}, {{roster}}, {{report}}. " |
| "Literal braces that are not slots must be doubled ({{ and }})." |
| ) from exc |
| step_label = fixtures_dir.parent.name |
| case_label = f"{step_label}/{case_dir.name}" |
| |
| if args.cli is None: |
| # Print mode (existing behaviour) |
| print(f"{'=' * 60}") |
| print(f"CASE: {case_label}") |
| print(f"{'=' * 60}") |
| if not args.quiet: |
| print("--- SYSTEM PROMPT ---") |
| print(system_prompt) |
| print("--- USER PROMPT ---") |
| print(user_prompt) |
| print("--- EXPECTED ---") |
| print(json.dumps(expected, indent=2)) |
| print() |
| continue |
| |
| # --cli mode: run the configured command and auto-compare. |
| structural = isinstance(expected, dict) and is_structural_expected(expected) |
| assertions: dict[str, dict] = {} |
| if structural: |
| if fixtures_dir not in _assertions_cache: |
| _assertions_cache[fixtures_dir] = load_assertions(fixtures_dir) |
| assertions = _assertions_cache[fixtures_dir] |
| if not assertions: |
| # No assertions.json: preserve the manual-review fallback. |
| print(f"MANUAL {case_label} (structural expected.json — review actual output by hand)") |
| if args.verbose: |
| _print_prompts_and_run(args, system_prompt, user_prompt) |
| manual += 1 |
| continue |
| |
| full_prompt = f"{system_prompt}\n\n{user_prompt}" |
| try: |
| stdout, stderr, rc = run_cli(args.cli, full_prompt, timeout=args.timeout) |
| except subprocess.TimeoutExpired: |
| print(f"ERROR {case_label} (CLI timed out after {args.timeout}s)") |
| errored += 1 |
| continue |
| except OSError as exc: |
| print(f"ERROR {case_label} (CLI invocation failed: {exc})") |
| errored += 1 |
| continue |
| |
| if rc != 0: |
| if args.exact: |
| print(f"ERROR {case_label} (CLI exited {rc}; stderr: {stderr.strip()[:200]})") |
| errored += 1 |
| if args.verbose: |
| print("--- STDOUT ---") |
| print(stdout) |
| continue |
| # Field-aware mode: a non-zero exit (often a refusal or a CLI |
| # safety filter) is wrapped just like a no-JSON case. The |
| # intersection-only comparator decides whether this case still |
| # passes based on the keys expected.json declares. Wrap is a |
| # silent implementation detail — the case still reports as |
| # PASS or FAIL like any other. |
| actual = {"raw_output": stdout, "stderr": stderr, "exit_code": rc} |
| else: |
| actual, parse_err = extract_json_from_output(stdout) |
| if parse_err is not None: |
| if args.exact: |
| # Exact mode requires literal JSON; non-JSON is an error. |
| print(f"ERROR {case_label} ({parse_err})") |
| errored += 1 |
| if args.verbose: |
| print("--- STDOUT ---") |
| print(stdout) |
| continue |
| # Field-aware mode: wrap the prose as a synthetic object so |
| # the intersection-only comparator can proceed. A model that |
| # produced prose-only output will PASS unless expected.json |
| # asserts on `raw_output`. |
| actual = {"raw_output": stdout} |
| |
| if structural: |
| if fixtures_dir not in _grading_schema_cache: |
| _grading_schema_cache[fixtures_dir] = load_grading_schema(fixtures_dir) |
| ok, notes = compare_structural( |
| actual, |
| expected, |
| assertions, |
| prose_fields=_grading_schema_cache[fixtures_dir], |
| grader_cli=args.grader_cli, |
| exact=args.exact, |
| grader_timeout=args.grader_timeout, |
| ) |
| if ok: |
| print(f"PASS {case_label}") |
| passed += 1 |
| else: |
| print(f"FAIL {case_label}") |
| for note in notes: |
| print(f" {note}") |
| failed += 1 |
| elif not args.exact: |
| if fixtures_dir not in _grading_schema_cache: |
| _grading_schema_cache[fixtures_dir] = load_grading_schema(fixtures_dir) |
| prose_fields = _grading_schema_cache[fixtures_dir] |
| ok, notes = compare_with_grader( |
| actual, |
| expected, |
| prose_fields=prose_fields, |
| grader_cli=args.grader_cli, |
| timeout=args.grader_timeout, |
| ) |
| if ok: |
| print(f"PASS {case_label}") |
| passed += 1 |
| else: |
| print(f"FAIL {case_label}") |
| for note in notes: |
| print(f" {note}") |
| failed += 1 |
| else: |
| ok, diff = compare_outputs(actual, expected) |
| if ok: |
| print(f"PASS {case_label}") |
| passed += 1 |
| else: |
| print(f"FAIL {case_label}") |
| print(diff) |
| failed += 1 |
| |
| if args.verbose: |
| print("--- SYSTEM PROMPT ---") |
| print(system_prompt) |
| print("--- USER PROMPT ---") |
| print(user_prompt) |
| print("--- STDOUT ---") |
| print(stdout) |
| print() |
| |
| if args.cli is not None: |
| total = passed + failed + manual + errored |
| print() |
| print(f"{'=' * 60}") |
| print(f"Ran {total} cases: {passed} passed, {failed} failed, {manual} manual, {errored} errored") |
| print(f"{'=' * 60}") |
| if failed or errored: |
| return 1 |
| |
| return 0 |
| |
| |
| def _print_prompts_and_run(args: argparse.Namespace, system_prompt: str, user_prompt: str) -> None: |
| """Verbose helper for MANUAL-mode cases: show what would have been sent.""" |
| print("--- SYSTEM PROMPT ---") |
| print(system_prompt) |
| print("--- USER PROMPT ---") |
| print(user_prompt) |
| |
| |
| if __name__ == "__main__": |
| sys.exit(main()) |
