diff --git a/dokimi/changelog.d/20261028_120100_napetrov_feat-verifier-redteam.md b/dokimi/changelog.d/20261028_120100_napetrov_feat-verifier-redteam.md new file mode 100644 index 000000000..285601c39 --- /dev/null +++ b/dokimi/changelog.d/20261028_120100_napetrov_feat-verifier-redteam.md @@ -0,0 +1,3 @@ +### Added + +- **`eval redteam` checks that grading blocks reject wrong answers** — grades a planted wrong, an empty, a hedged and a keyword-stuffed answer against each question's `grading` block, and a canary right answer built from its one literal check. A wrong answer the grader accepts, or a canary it rejects, is a violation: the command prints pass rates, exits 1 and, with `-o`, writes a `redteam.v1` artifact (it records the judge model and provider). A run in which a graded question was not measured (judge checks without `--judge`, a block of only negated checks) is `inconclusive` and also exits 1. Judge checks are unscored unless `--judge` is given, so the default run makes no model call. diff --git a/dokimi/docs/answer-formats.md b/dokimi/docs/answer-formats.md index 1bae0deb3..ffd6de105 100644 --- a/dokimi/docs/answer-formats.md +++ b/dokimi/docs/answer-formats.md @@ -90,6 +90,40 @@ artifact for an `arms run` file. Each block carries `scorer: reference`, and the run's pass rates and mean scores are under `run_metadata.summary` (or `reference_scoring`). +## Red-teaming the grading + +`eval redteam QUESTIONS.json` grades synthetic answers against each question's +`grading` block and reports how often the grader accepts them: + +| Attack | Answer | The grader must | +|---|---|---| +| `canary` | The right answer, built from the one literal check (`match`, `fuzzy`, `exact`, `choice`, `json`); a question with several checks, a `regex`, a `judge` or a `negate` has none | pass it | +| `wrong` | A planted wrong answer, picked per block so that no positive literal check accepts it, asked with the check's own rules (the probe is skipped only when no candidate is clean) | fail it | +| `empty` | `""` | fail it | +| `hedge` | A fixed non-committal sentence; skipped when a positive literal check accepts it | fail it | +| `stuffed` | The values of a block with one literal check, listed with no sentence around them: two or more values of a positive check (a single value is the right answer, not stuffing), bare for a check on the whole answer, between two decoys for a `match` at `begin` or `end`, and after a JSON decoy for a `json` check with `contains` (it reads the first JSON in the answer), or one or more values of a `negate` check, whose stuffed answer is the forbidden value; not built for a positive block that needs every value (`mode: all`), for several checks, for `fuzzy`, numeric or `location: any` checks (they find a listed value wherever it is, so accepting it is right), or for a negated check that the listed values cannot fail: a negated `exact`, `json` or `match` with `location: exact` (they never equal the list) and a negated `match` with `mode: all` (the list passes it by its location) | fail it | + +An attack the grader accepts, or a canary it rejects, is a violation, and any +violation makes the run `blocked`: the command prints the pass rates and the +violations and exits 1 (`-o` writes `redteam.v1`). A run in which a graded +question was not measured is `inconclusive` and also exits 1: a question whose +`judge` checks have no verdict without `--judge` (no probe scored), or a block +of only negated checks. A `negate` check passes an answer that lacks its value, +so a block of only negated checks gets none of `wrong`, `empty` and `hedge`: it +is marked `negated_only` in the artifact (the table prints a line "N questions with only negated checks: wrong, empty and hedge not run"), its `stuffed` probe (the negated value itself) +still runs, but the grader would accept any answer without that value, so +nothing was verified about it. The table names every question that was not +measured and counts it as `unprobed`; only a run that measured every graded +question and found no violation is `pass`. An unwritable `-o` exits 2. +A block with a positive check +is graded as written, `threshold` included, so a negated check's share of the score +counts: a block it lifts over the line is blocked. +A probe the grader cannot +score is unscored and does not block. `judge` checks are unscored unless +`--judge` is given, so the default run makes no model call; with `--judge` the +artifact records `judge_model` and `judge_provider` (both `null` without it). Nothing here +changes a score; it tells you whether a grading block is worth trusting. + ## What does not carry over | Format | Lossy | diff --git a/dokimi/docs/reference/cli-interface.json b/dokimi/docs/reference/cli-interface.json index 7f68d6a09..e3c9742e5 100644 --- a/dokimi/docs/reference/cli-interface.json +++ b/dokimi/docs/reference/cli-interface.json @@ -5420,6 +5420,117 @@ "subcommand_required": null, "subcommands": {} }, + "redteam": { + "allow_abbrev": true, + "params": [ + { + "action": "_StoreAction", + "choices": null, + "const": null, + "default": null, + "group_key": null, + "group_required": null, + "hidden": false, + "is_flag": false, + "key": "questions", + "kind": "argument", + "nargs": null, + "position": 0, + "required": true, + "type": null + }, + { + "action": "_HelpAction", + "choices": null, + "const": null, + "default": "==SUPPRESS==", + "group_key": null, + "group_required": null, + "hidden": false, + "is_flag": true, + "key": "--help/-h", + "kind": "option", + "nargs": 0, + "position": null, + "required": false, + "type": null + }, + { + "action": "_StoreTrueAction", + "choices": null, + "const": true, + "default": false, + "group_key": null, + "group_required": null, + "hidden": false, + "is_flag": true, + "key": "--judge", + "kind": "option", + "nargs": 0, + "position": null, + "required": false, + "type": null + }, + { + "action": "_StoreAction", + "choices": null, + "const": null, + "default": "claude-sonnet-5-5", + "group_key": null, + "group_required": null, + "hidden": false, + "is_flag": false, + "key": "--judge-model", + "kind": "option", + "nargs": null, + "position": null, + "required": false, + "type": null + }, + { + "action": "_StoreAction", + "choices": [ + "'amazon-bedrock'", + "'anthropic'", + "'google'", + "'google-vertex'", + "'openai'", + "'openrouter'" + ], + "const": null, + "default": "anthropic", + "group_key": null, + "group_required": null, + "hidden": false, + "is_flag": false, + "key": "--judge-provider", + "kind": "option", + "nargs": null, + "position": null, + "required": false, + "type": null + }, + { + "action": "_StoreAction", + "choices": null, + "const": null, + "default": null, + "group_key": null, + "group_required": null, + "hidden": false, + "is_flag": false, + "key": "--output/-o", + "kind": "option", + "nargs": null, + "position": null, + "required": false, + "type": null + } + ], + "path": "dokimi eval redteam", + "subcommand_required": null, + "subcommands": {} + }, "reference": { "allow_abbrev": true, "params": [ diff --git a/dokimi/dokimi/artifacts.py b/dokimi/dokimi/artifacts.py index 5731c0f46..c0b9e2746 100644 --- a/dokimi/dokimi/artifacts.py +++ b/dokimi/dokimi/artifacts.py @@ -57,6 +57,7 @@ "result": "result.v1", "repo_budget": "repo_budget.v1", "task_inspect": "task_inspect.v1", + "redteam": "redteam.v1", } # kind -> superseded versions still accepted on *load*, newest first. An artifact written under an diff --git a/dokimi/dokimi/commands/answer_formats.py b/dokimi/dokimi/commands/answer_formats.py index 035b262b6..7c808355b 100644 --- a/dokimi/dokimi/commands/answer_formats.py +++ b/dokimi/dokimi/commands/answer_formats.py @@ -35,10 +35,9 @@ def cmd_questions_import(args: argparse.Namespace) -> None: def _ask(args: argparse.Namespace): if args.no_judge: return None - from dokimi.eval.judge import Judge + from dokimi.eval.judge import judge_ask - llm = Judge(model=args.judge_model, provider=args.judge_provider).llm - return lambda prompt, draw=0, cache_if=None: llm.invoke(prompt, draw=draw, cache_if=cache_if).content + return judge_ask(args.judge_model, args.judge_provider) def cmd_eval_reference(args: argparse.Namespace) -> None: diff --git a/dokimi/dokimi/commands/evaluate.py b/dokimi/dokimi/commands/evaluate.py index 9d3602f31..74a8abe08 100644 --- a/dokimi/dokimi/commands/evaluate.py +++ b/dokimi/dokimi/commands/evaluate.py @@ -565,6 +565,10 @@ def register(sub, positive_int) -> None: answer_formats.register_eval(eval_sub, positive_int) # eval reference, eval native-import + from dokimi.commands import redteam + + redteam.register_eval(eval_sub) # eval redteam + # eval panel-score panel_p = eval_sub.add_parser("panel-score", help="Score answers using a multi-judge panel") panel_p.add_argument("--answers", required=True, help="Path to answers JSON file") diff --git a/dokimi/dokimi/commands/redteam.py b/dokimi/dokimi/commands/redteam.py new file mode 100644 index 000000000..381271a87 --- /dev/null +++ b/dokimi/dokimi/commands/redteam.py @@ -0,0 +1,65 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""``eval redteam``: check that a question set's grading rejects the answers it should. + +Reads a questions set (with ``grading`` blocks), grades synthetic attacks against it and prints +the pass rates. Exits 1 when any attack is accepted, a canary is rejected, a graded question was not +measured, or nothing was graded (``inconclusive``); exits 2 on an unreadable input or an unwritable ``-o``. Judge checks are not +scored unless ``--judge`` is given, so the default run makes no model call. +""" + +from __future__ import annotations + +import argparse +import sys + + +def _ask(args: argparse.Namespace): + if not args.judge: + return None + from dokimi.eval.judge import judge_ask + + return judge_ask(args.judge_model, args.judge_provider) + + +def cmd_eval_redteam(args: argparse.Namespace) -> None: + from dokimi.artifacts import save_artifact + from dokimi.eval.redteam import format_table, run + from dokimi.questions.artifact import load_questions + + try: + questions = load_questions(args.questions).artifact.get("questions") or [] + result = run(questions, ask=_ask(args)) + except (OSError, ValueError, RecursionError) as e: # RecursionError: nested too deep + print(f"Error: {e}", file=sys.stderr) + sys.exit(2) + result["questions_path"] = str(args.questions) + result["judge_model"] = args.judge_model if args.judge else None + result["judge_provider"] = args.judge_provider if args.judge else None + print(format_table(result, output=args.output)) + if args.output: + try: + save_artifact("redteam", result, args.output) + except (OSError, ValueError) as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(2) + print(f"Wrote redteam artifact to {args.output}") + if result["status"] != "pass": + sys.exit(1) + + +def register_eval(eval_sub) -> None: + from dokimi.defaults import DEFAULT_JUDGE_PROVIDER, JUDGE_MODEL_FOR_PROVIDER, JUDGE_MODEL_HELP + from dokimi.eval.judge import JUDGE_PROVIDERS + + r = eval_sub.add_parser( + "redteam", + help="Check grading blocks reject wrong, empty, hedged and stuffed answers and accept a canary") + r.add_argument("questions", help="questions JSON with grading blocks (from `questions import`)") + r.add_argument("--judge", action="store_true", + help="Grade judge checks with the judge model (model calls); default leaves them unscored") + r.add_argument("--judge-model", default=JUDGE_MODEL_FOR_PROVIDER, help=JUDGE_MODEL_HELP) + r.add_argument("--judge-provider", default=DEFAULT_JUDGE_PROVIDER, choices=JUDGE_PROVIDERS) + r.add_argument("-o", "--output", default=None, help="redteam.v1 JSON to write") + r.set_defaults(func=cmd_eval_redteam) diff --git a/dokimi/dokimi/eval/judge.py b/dokimi/dokimi/eval/judge.py index f07c9cc4d..77d5ddc68 100644 --- a/dokimi/dokimi/eval/judge.py +++ b/dokimi/dokimi/eval/judge.py @@ -679,3 +679,9 @@ def load_evaluations(input_path: Path) -> Dict[str, Any]: def _get_timestamp() -> str: from datetime import datetime, timezone return datetime.now(timezone.utc).replace(tzinfo=None).isoformat() + "Z" + + +def judge_ask(model: str, provider: str) -> Callable[..., str]: + """The ``Ask`` of the reference checks (one prompt in, the reply text out) backed by a ``Judge``'s model.""" + llm = Judge(model=model, provider=provider).llm + return lambda prompt, draw=0, cache_if=None: llm.invoke(prompt, draw=draw, cache_if=cache_if).content diff --git a/dokimi/dokimi/eval/redteam.py b/dokimi/dokimi/eval/redteam.py new file mode 100644 index 000000000..af1a7aba3 --- /dev/null +++ b/dokimi/dokimi/eval/redteam.py @@ -0,0 +1,315 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Verifier red team: do a question's grading checks reject the answers they should? + +Each question with a ``grading`` block is graded against five synthetic answers: + +- ``canary``: the right answer, built from the one literal check; it must pass. +- ``wrong``: a planted wrong answer; it must fail. Its text is picked so that no positive literal + check accepts it; the probe is skipped only when no candidate is clean. +- ``empty``: the empty answer; it must fail. +- ``hedge``: a non-committal answer that carries no check value; it must fail. +- ``stuffed``: every literal check value listed with no sentence around it; it must fail. Built for + a positive check of two or more values (mode ``any``) that does not read the list where it stands: + bare for ``exact``, ``json`` and location-``exact`` checks, between two decoys for a ``match`` at + ``begin`` or ``end``, after a JSON decoy for a ``json`` check with ``contains``. It is not built for ``fuzzy``, numeric or location-``any`` checks, a block that + needs every value, or several checks that each accept one value: the grader is right to accept the + stuffed answer there. Also built + for a negated check of one or more values, whose stuffed answer carries the forbidden value; not + built for a negated ``exact``, ``json`` or location-``exact`` match, which the listed values never + equal, nor for a negated ``mode: all`` match, whose stuffed answer can pass the negation by its + location (``Paris Lyon`` ends with one value only). The skip covers every negated ``mode: all`` + match, which is wider than that case and never a false block. + +An answer the grader accepts when it should reject it (or rejects a canary) is a violation, and +any violation blocks the run. A run with a question whose correctness no probe measured (a judge check without a judge, a block of +only negated checks) is ``inconclusive``, and so is one with nothing graded: it is not a pass, and it exits 1 +like a blocked run. Only a run that measured every graded question and found no violation is ``pass``. +A probe the grader cannot score (a judge check with no judge model, a regex past its timeout) is +reported as unscored. The canary is only built for a question whose grading is one literal check; +other questions get the other attacks alone. + +A ``negate`` check passes a wrong, empty or hedged answer by design, so a block of only negated +checks gets none of those three attacks and is reported as ``negated_only``; its stuffed answer still +runs. A block with a positive check is graded as written, ``threshold`` included: a negated check's +share of the score counts, so a block whose negated checks lift a wrong answer over the line is blocked. + +Nothing here calls a model unless the caller passes ``ask`` (the judge of judge checks). +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any, Mapping, Optional + +from dokimi.eval.reference_checks import Ask, check_values, grade, grade_check +from dokimi.eval.reference_run import grading_by_id + +ATTACKS = ("canary", "wrong", "empty", "hedge", "stuffed") +EXPECTED = {"canary": True, "wrong": False, "empty": False, "hedge": False, "stuffed": False} +# Planted wrong answers, tried in order; the first that no positive literal check accepts is used. +WRONG_CANDIDATES = ("PLANTED-WRONG-ANSWER-7731", "QZX-BOGUS-REPLY-4096", "ZKV-MISMATCH-2508", "XJQ-NONSENSE-6349") +# JSON decoys for a json check with contains, which reads the first JSON object or array in the text. +JSON_DECOYS = ('{"planted": "wrong"}', '{"planted": "other"}') +HEDGE = ("I am not certain. It could be several things, depending on the setup, " + "and I cannot say which one applies here.") +BLOCKED, PASS, INCONCLUSIVE = "blocked", "pass", "inconclusive" +# The attacks a negated check makes unfit for a block: the answers it passes by design. +NEGATION_SKIPPED = ("wrong", "empty", "hedge") + +_LITERAL = frozenset({"match", "fuzzy", "exact", "choice", "json"}) + + +def canary_for(grading: Mapping[str, Any]) -> Optional[str]: + """The right answer for a grading block of one literal check, or None when none is built.""" + checks = grading.get("checks", []) + if len(checks) != 1 or checks[0].get("type") not in _LITERAL or checks[0].get("negate"): + return None + check = checks[0] + kind, values = check["type"], check_values(check) + mode, location = str(check.get("mode", "any")), str(check.get("location", "end")) + if not values or (len(values) > 1 and mode == "all"): + return None + text = str(values[0]) + if kind == "choice": + return "ANSWER: " + ", ".join(sorted(str(v).strip().upper() for v in values)) + if kind == "json": + return json.dumps(values[0]) + if kind == "exact" or (kind == "match" and location == "exact"): + return text + if kind == "match" and location == "begin": + return f"{text} is the answer." + if kind == "match" and location == "end" and not check.get("ignore_punctuation", True): + return f"The answer is {text}" # the full stop would be the raw end of the answer, not the value + return f"The answer is {text}." + + +def _whole_answer(check: Mapping[str, Any]) -> bool: + """Whether the check compares the whole answer, so a listed value never equals it.""" + kind = check.get("type") + return kind in ("exact", "json") or (kind == "match" and str(check.get("location", "end")) == "exact") + + +def _reads_first_json(check: Mapping[str, Any]) -> bool: + """Whether a json check with ``contains`` reads the first JSON in the answer, not the whole answer.""" + return check.get("type") == "json" and bool(check.get("contains")) + + +def stuffed_for(grading: Mapping[str, Any]) -> Optional[str]: + """The block's one literal check's values, listed with no sentence around them; None when not built. + + A positive check needs two or more values and must accept any one of them (mode ``any``): a block + that needs every value, or several checks, is satisfied by the stuffed answer by design, so + accepting it is not a violation. A single positive value listed alone is the right answer itself, + not stuffing. The list is an attack only where the check does not read it: + + - a check on the whole answer (``exact``, ``json`` without ``contains``, ``match`` at ``exact``): the bare list; + - a ``match`` at ``begin`` or ``end``: the list between two decoys the check rejects, so no accepted + value is at the place the check reads; + - a ``json`` check with ``contains`` reads the first JSON in the text, so the list goes after two JSON + decoys the check rejects: a bare list would start with a listed value and be accepted; + - ``match`` at ``any``, a numeric ``match`` and ``fuzzy`` find a listed value wherever it is, so the + grader is right to accept it and no probe is built. + + A negated check is stuffed with its one or more values: the stuffed answer carries the forbidden + value, so it must fail, unless the check compares the whole answer (see ``_whole_answer``), which the + listed values never equal, or is ``mode: all``, which the stuffed answer can pass by its location + (the skip is kept for every such check, for simplicity). A choice check is left out: its answer is + a bare letter. + """ + checks = grading.get("checks", []) + if len(checks) != 1 or checks[0].get("type") not in _LITERAL - {"choice"}: + return None + check = checks[0] + values = [str(v) for v in check_values(check)] + mode = str(check.get("mode", "any")) + if check.get("negate"): + built = bool(values) and not _whole_answer(check) and mode != "all" + return " ".join(values) if built else None + if len(values) < 2 or mode == "all": + return None + if _whole_answer(check) and not _reads_first_json(check): + return " ".join(values) + candidates: tuple[str, ...] + if _reads_first_json(check): + candidates = JSON_DECOYS + elif check.get("type") != "match" or check.get("numeric") or str(check.get("location", "end")) == "any": + return None + else: + candidates = WRONG_CANDIDATES + decoys = [w for w in candidates if not grade_check(check, "", w, None).passed] + return f"{decoys[0]} {' '.join(values)} {decoys[1]}" if len(decoys) >= 2 else None + + +def _positive_checks(grading: Mapping[str, Any]) -> list[Mapping[str, Any]]: + return [c for c in grading.get("checks", []) if not c.get("negate")] + + +def negated_only(grading: Mapping[str, Any]) -> bool: + """Whether every check of the block is negated, so no answer is wrong by its positive checks.""" + checks = grading.get("checks", []) + return bool(checks) and all(c.get("negate") for c in checks) + + +def _accepts(grading: Mapping[str, Any], text: str) -> bool: + """Whether a non-negated literal check of the block accepts the text on its own, by the grader's rules.""" + return any(grade_check(c, "", text, None).passed for c in _positive_checks(grading) if c.get("type") in _LITERAL) + + +def wrong_for(grading: Mapping[str, Any]) -> Optional[str]: + """The first planted wrong answer that no positive literal check of the block accepts, or None.""" + return next((w for w in WRONG_CANDIDATES if not _accepts(grading, w)), None) + + +def attacks(grading: Mapping[str, Any]) -> dict[str, Optional[str]]: + """The synthetic answer for each attack; None where the attack does not apply to the block. + + The wrong, empty and hedge answers apply when the block has a non-negated check, and are graded + against the whole block as written. A block of only negated checks gets none of the three. + ``stuffed`` still runs, since it carries the negated value and must fail. The planted wrong answer + is chosen per block so that no positive check accepts it, and the hedge is fixed text, so it is + skipped when a positive check accepts it (each is asked with the grader's own rules, so a numeric + match that reads a number out of the planted answer, or a short value the check does not find in the + hedge, are judged as the grader judges them). + """ + applies = bool(_positive_checks(grading)) + clashes = _accepts(grading, HEDGE) + return {"canary": canary_for(grading), + "wrong": wrong_for(grading) if applies else None, + "empty": "" if applies else None, + "hedge": None if not applies or clashes else HEDGE, + "stuffed": stuffed_for(grading)} + + +@dataclass(frozen=True) +class Probe: + attack: str + passed: Optional[bool] + detail: str = "" + + @property + def expected(self) -> bool: + return EXPECTED[self.attack] + + @property + def violation(self) -> bool: + return self.passed is not None and self.passed != self.expected + + def as_dict(self) -> dict[str, Any]: + return {"passed": self.passed, "expected": self.expected, "violation": self.violation, + "detail": self.detail} + + +def probe(grading: Mapping[str, Any], question: str, ask: Optional[Ask] = None) -> dict[str, Probe]: + """Grade every applicable attack against one grading block.""" + out: dict[str, Probe] = {} + for name, answer in attacks(grading).items(): + if answer is None: + continue + g = grade(grading, question, answer, ask) + # A check left unrun (no judge, or a regex past its timeout) makes a pass on the rest no measurement; + # with a threshold it is no result either, so the score is taken over the counted checks alone. + partial = bool(g.unjudged or g.unrun) + passed = None if partial and (g.passed or g.threshold is not None) else g.passed + out[name] = Probe(name, passed, _detail(g)) + return out + + +def _detail(g: Any) -> str: + return "; ".join(c.detail for c in g.checks if c.detail) + + +def run(questions: list[Mapping[str, Any]], *, ask: Optional[Ask] = None) -> dict[str, Any]: + """Red-team every question with a grading block. Raises ValueError for a block that cannot be graded.""" + grading_by_id(questions) # refuses a set with a block that cannot be graded; the blocks are read per question + rows: list[dict[str, Any]] = [] + for q in questions: + if "grading" not in q: + continue + qid, grading = str(q.get("id")), q["grading"] # not looked up by id: ids may repeat + text = str(q.get("question") or q.get("text") or "") + probes = probe(grading, text, ask) + rows.append({"question_id": qid, "canary": "canary" in probes, + "negated_only": negated_only(grading), + "probes": {name: p.as_dict() for name, p in probes.items()}}) + violations = [{"question_id": r["question_id"], "attack": name, "passed": p["passed"]} + for r in rows for name, p in r["probes"].items() if p["violation"]] + summary = _summary(rows) + if violations: + status = BLOCKED + else: + status = PASS if rows and not summary["unprobed"] else INCONCLUSIVE + return { + "status": status, + "attacks": _attack_rates(rows), + "violations": violations, + "questions": rows, + "summary": summary, + } + + +def _attack_rates(rows: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: + out: dict[str, dict[str, Any]] = {} + for name in ATTACKS: + scored = [r["probes"][name]["passed"] for r in rows if name in r["probes"]] + scored = [p for p in scored if p is not None] + accepted = sum(1 for p in scored if p) + out[name] = {"expected": "pass" if EXPECTED[name] else "fail", "scored": len(scored), + "accepted": accepted, "rate": round(accepted / len(scored), 4) if scored else None} + return out + + +def _unmeasured(row: Mapping[str, Any]) -> str: + """Why a question's correctness was not measured, or "" when a probe scored it. + + A block of only negated checks gets no wrong, empty or hedge probe: the real grader accepts any + answer without the forbidden value, so nothing was verified about it, whatever its stuffed probe said. + """ + if row["negated_only"]: + return "only negated checks" + if not any(p["passed"] is not None for p in row["probes"].values()): + return "no probe scored" + return "" + + +def _summary(rows: list[dict[str, Any]]) -> dict[str, int]: + probes = [p for r in rows for p in r["probes"].values()] + unprobed = sum(1 for r in rows if _unmeasured(r)) + return { + "questions": len(rows), + "probed": len(rows) - unprobed, + "unprobed": unprobed, + "no_canary": sum(1 for r in rows if not r["canary"]), + "negated_only": sum(1 for r in rows if r["negated_only"]), + "unscored": sum(1 for p in probes if p["passed"] is None), + "violations": sum(1 for p in probes if p["violation"]), + } + + +def format_table(result: dict[str, Any], output: Any = None) -> str: + """The pass-rate table; ``output`` is the ``-o`` path when an artifact is written, else None.""" + lines = [f"{'attack':<8} {'expected':<9} {'scored':>6} {'accepted':>8} {'rate':>6}"] + for name, a in result["attacks"].items(): + rate = "-" if a["rate"] is None else f"{a['rate']:.2f}" + lines.append(f"{name:<8} {a['expected']:<9} {a['scored']:>6} {a['accepted']:>8} {rate:>6}") + for v in result["violations"][:20]: + verb = "accepted" if v["attack"] != "canary" else "rejected" + lines.append(f" violation: question {v['question_id']}: {v['attack']} answer {verb}") + if len(result["violations"]) > 20: + more = len(result["violations"]) - 20 + lines.append(f" ... {more} more violations in the artifact" if output + else f" ... {more} more violations; rerun with -o FILE to list them all") + s = result["summary"] + if s["negated_only"]: + lines.append(f" {s['negated_only']} questions with only negated checks: " + f"{', '.join(NEGATION_SKIPPED[:-1])} and {NEGATION_SKIPPED[-1]} not run") + unmeasured = [f"question {r['question_id']} ({why})" for r in result["questions"] if (why := _unmeasured(r))] + if unmeasured: + lines.append(f" not measured: {', '.join(unmeasured[:20])}" + + (f", and {len(unmeasured) - 20} more" if len(unmeasured) > 20 else "")) + lines.append(f"status {result['status']}: {s['questions']} questions, {s['probed']} probed, " + f"{s['unprobed']} unprobed, {s['no_canary']} without a canary, {s['unscored']} probes unscored, " + f"{s['violations']} violations") + return "\n".join(lines) diff --git a/dokimi/dokimi/eval/reference_checks.py b/dokimi/dokimi/eval/reference_checks.py index 3ad4e3f70..45ee55262 100644 --- a/dokimi/dokimi/eval/reference_checks.py +++ b/dokimi/dokimi/eval/reference_checks.py @@ -247,7 +247,7 @@ def parse_json(answer: str, *, contains: bool = False) -> tuple[bool, Any]: # --- one check --------------------------------------------------------------------------------- -def _values(check: Mapping[str, Any]) -> list[Any]: +def check_values(check: Mapping[str, Any]) -> list[Any]: raw = check.get("values", check.get("value")) if raw is None: return [] @@ -280,7 +280,7 @@ def _regex(check: Mapping[str, Any], values: list[Any], answer: str, mode: str) def _deterministic(check: Mapping[str, Any], answer: str) -> tuple[Optional[bool], str]: kind = check["type"] - values = _values(check) + values = check_values(check) mode = str(check.get("mode", "any")) if kind == "match": location = str(check.get("location", "end")) @@ -391,7 +391,7 @@ def _check_problems(kind: str, c: Mapping[str, Any]) -> list[str]: out = ["weight is not a number"] if not _is_number(weight) else ( ["weight must not be negative"] if weight < 0 else []) if kind in DETERMINISTIC and kind != "json": - values = _values(c) + values = check_values(c) if not values: out.append(f"{kind} has no values") elif any(blank_value(kind, c, v) for v in values): @@ -402,7 +402,7 @@ def _check_problems(kind: str, c: Mapping[str, Any]) -> list[str]: if kind == "match" and str(c.get("location", "end")) not in _LOCATIONS: out.append(f"match location must be one of {', '.join(_LOCATIONS)}, not {c.get('location')!r}") if kind == "regex": - for v in _values(c): + for v in check_values(c): try: regex.compile(str(v)) except regex.error as exc: diff --git a/dokimi/dokimi/schemas/redteam.v1.json b/dokimi/dokimi/schemas/redteam.v1.json new file mode 100644 index 000000000..711fc2ce0 --- /dev/null +++ b/dokimi/dokimi/schemas/redteam.v1.json @@ -0,0 +1,84 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/intel/dokimi/schemas/redteam.v1.json", + "title": "Verifier red team v1", + "description": "Whether a question set's grading blocks reject the synthetic wrong, empty, hedged and stuffed answers they should, and accept a canary right answer. Written by `eval redteam`; status is blocked when any attack is accepted or a canary is rejected, and inconclusive when nothing was graded or a graded question was not measured (judge checks without a judge, or only negated checks).", + "type": "object", + "required": ["schema_version", "status", "questions_path", "judge_model", "judge_provider", "attacks", "violations", "questions", "summary"], + "additionalProperties": false, + "properties": { + "schema_version": {"const": "redteam.v1"}, + "status": {"enum": ["pass", "blocked", "inconclusive"]}, + "questions_path": {"type": "string"}, + "judge_model": {"type": ["string", "null"], "description": "The judge model that graded the judge checks; null without --judge, when they were left unscored."}, + "judge_provider": {"type": ["string", "null"], "description": "The provider of the judge model; null without --judge."}, + "attacks": { + "type": "object", + "additionalProperties": { + "type": "object", + "required": ["expected", "scored", "accepted", "rate"], + "additionalProperties": false, + "properties": { + "expected": {"enum": ["pass", "fail"], "description": "What the grader must do with this attack."}, + "scored": {"type": "integer", "minimum": 0}, + "accepted": {"type": "integer", "minimum": 0, "description": "Probes the grader passed."}, + "rate": {"type": ["number", "null"], "minimum": 0, "maximum": 1, "description": "accepted / scored; null when nothing was scored."} + } + } + }, + "violations": { + "type": "array", + "items": { + "type": "object", + "required": ["question_id", "attack", "passed"], + "additionalProperties": false, + "properties": { + "question_id": {"type": "string"}, + "attack": {"enum": ["canary", "wrong", "empty", "hedge", "stuffed"]}, + "passed": {"type": "boolean"} + } + } + }, + "questions": { + "type": "array", + "items": { + "type": "object", + "required": ["question_id", "canary", "negated_only", "probes"], + "additionalProperties": false, + "properties": { + "question_id": {"type": "string"}, + "canary": {"type": "boolean", "description": "True when a right answer could be built from the grading block."}, + "negated_only": {"type": "boolean", "description": "True when every check is negated: no wrong, empty or hedged probe is run."}, + "probes": { + "type": "object", + "additionalProperties": { + "type": "object", + "required": ["passed", "expected", "violation", "detail"], + "additionalProperties": false, + "properties": { + "passed": {"type": ["boolean", "null"], "description": "Null when the grader could not score the probe."}, + "expected": {"type": "boolean"}, + "violation": {"type": "boolean"}, + "detail": {"type": "string"} + } + } + } + } + } + }, + "summary": { + "type": "object", + "required": ["questions", "probed", "unprobed", "no_canary", "negated_only", "unscored", "violations"], + "additionalProperties": false, + "properties": { + "questions": {"type": "integer", "minimum": 0}, + "probed": {"type": "integer", "minimum": 0}, + "unprobed": {"type": "integer", "minimum": 0}, + "no_canary": {"type": "integer", "minimum": 0}, + "negated_only": {"type": "integer", "minimum": 0}, + "unscored": {"type": "integer", "minimum": 0}, + "violations": {"type": "integer", "minimum": 0} + } + } + } +} diff --git a/dokimi/tests/fixtures/artifacts/redteam.v1.json b/dokimi/tests/fixtures/artifacts/redteam.v1.json new file mode 100644 index 000000000..cb4d552d0 --- /dev/null +++ b/dokimi/tests/fixtures/artifacts/redteam.v1.json @@ -0,0 +1,44 @@ +{ + "schema_version": "redteam.v1", + "status": "blocked", + "questions_path": "questions.json", + "judge_model": null, + "judge_provider": null, + "attacks": { + "canary": {"expected": "pass", "scored": 2, "accepted": 1, "rate": 0.5}, + "wrong": {"expected": "fail", "scored": 1, "accepted": 0, "rate": 0.0}, + "empty": {"expected": "fail", "scored": 2, "accepted": 0, "rate": 0.0}, + "hedge": {"expected": "fail", "scored": 2, "accepted": 0, "rate": 0.0}, + "stuffed": {"expected": "fail", "scored": 1, "accepted": 1, "rate": 1.0} + }, + "violations": [ + {"question_id": "1", "attack": "stuffed", "passed": true}, + {"question_id": "2", "attack": "canary", "passed": false} + ], + "questions": [ + { + "question_id": "1", + "canary": true, + "negated_only": false, + "probes": { + "canary": {"passed": true, "expected": true, "violation": false, "detail": ""}, + "wrong": {"passed": false, "expected": false, "violation": false, "detail": ""}, + "empty": {"passed": false, "expected": false, "violation": false, "detail": ""}, + "hedge": {"passed": false, "expected": false, "violation": false, "detail": ""}, + "stuffed": {"passed": true, "expected": false, "violation": true, "detail": ""} + } + }, + { + "question_id": "2", + "canary": true, + "negated_only": false, + "probes": { + "canary": {"passed": false, "expected": true, "violation": true, "detail": "not a match"}, + "wrong": {"passed": null, "expected": false, "violation": false, "detail": "not judged: no judge model"}, + "empty": {"passed": false, "expected": false, "violation": false, "detail": ""}, + "hedge": {"passed": false, "expected": false, "violation": false, "detail": ""} + } + } + ], + "summary": {"questions": 2, "probed": 2, "unprobed": 0, "no_canary": 0, "negated_only": 0, "unscored": 1, "violations": 2} +} diff --git a/dokimi/tests/test_judge_provider_default.py b/dokimi/tests/test_judge_provider_default.py index e1d8fc4d4..31d8d8c63 100644 --- a/dokimi/tests/test_judge_provider_default.py +++ b/dokimi/tests/test_judge_provider_default.py @@ -49,7 +49,7 @@ def test_the_commands_that_take_a_judge_provider_are_all_covered(): assert set(COMMANDS) == { "evaluate", "eval score", "eval judge-validity", "eval checklist", "eval judge-wording", "eval pairwise", "benchmark run", "benchmark preflight", "benchmark batch", "arms run", - "arms matrix-run", "subjects run", "eval reference", + "arms matrix-run", "subjects run", "eval reference", "eval redteam", } diff --git a/dokimi/tests/test_redteam.py b/dokimi/tests/test_redteam.py new file mode 100644 index 000000000..ae9c33bbc --- /dev/null +++ b/dokimi/tests/test_redteam.py @@ -0,0 +1,698 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for ``dokimi.eval.redteam`` and ``eval redteam``: attacks against grading blocks. + +Gradings are built in memory; judge checks are never sent to a model (no ``ask`` is passed). +""" + +from __future__ import annotations + +import argparse +import json +from types import SimpleNamespace + +import pytest +import regex + +from dokimi.commands.redteam import cmd_eval_redteam +from dokimi.eval import redteam +from dokimi.eval.redteam import canary_for, format_table, run +from dokimi.eval.reference_checks import CheckResult, Grade + + +def _q(qid, grading, text="What is the capital of France?"): + return {"id": qid, "question": text, "grading": grading} + + +def _match(*values, mode="any"): + return {"checks": [{"type": "match", "values": list(values), "mode": mode}], "threshold": None} + + +def test_a_grader_that_rejects_every_attack_passes_the_red_team(): + result = run([_q("1", _match("Paris"))]) + assert result["status"] == "pass" and result["violations"] == [] + assert set(result["questions"][0]["probes"]) == {"canary", "wrong", "empty", "hedge"} + assert result["attacks"]["canary"] == {"expected": "pass", "scored": 1, "accepted": 1, "rate": 1.0} + + +def _accept_stuffed(monkeypatch): + """A grader that accepts the stuffed answer and rejects every other attack.""" + real = redteam.grade + + def grade(grading, question, answer, ask=None): + if answer == redteam.attacks(grading)["stuffed"]: + return Grade(checks=[CheckResult("match", True, 1.0)]) + return real(grading, question, answer, ask) + + monkeypatch.setattr(redteam, "grade", grade) + + +def test_keyword_stuffing_that_the_grader_accepts_blocks_the_run(monkeypatch): + _accept_stuffed(monkeypatch) + result = run([_q("1", _match("Paris", "Lyon"))]) + assert result["status"] == "blocked" + assert result["violations"] == [{"question_id": "1", "attack": "stuffed", "passed": True}] + + +@pytest.mark.parametrize("check", [ + {"type": "match", "values": ["4", "four"], "location": "any"}, + {"type": "match", "values": ["Paris", "Lyon"]}, # location end + {"type": "match", "values": ["Paris", "Lyon"], "location": "begin"}, + {"type": "fuzzy", "values": ["Barack Obama", "Obama"]}, +]) +def test_a_correct_grader_with_aliases_is_not_blocked_by_the_stuffed_answer(check): + # The stuffed answer is built so that a correct grader rejects it (values between decoys) or it is + # not built at all (fuzzy, numeric, location any): a correct grader is never blocked. + result = run([_q("1", {"checks": [check]})]) + assert result["status"] == "pass" and result["violations"] == [] + + +def test_the_stuffed_answer_buries_the_aliases_between_decoys_for_an_end_or_begin_check(): + stuffed = redteam.attacks(_match("Paris", "Lyon"))["stuffed"] + middle = stuffed.split() + assert middle[1:3] == ["Paris", "Lyon"] and len(middle) == 4 + assert not stuffed.endswith(("Paris", "Lyon")) and not stuffed.startswith(("Paris", "Lyon")) + assert redteam.attacks({"checks": [{"type": "match", "values": ["a", "b"], "location": "any"}]})["stuffed"] is None + + +def test_a_json_contains_check_with_list_values_is_stuffed_after_a_decoy_the_grader_rejects(): + # contains reads the first JSON in the text: the bare list "[1, 2] [3, 4]" starts with the value [1, 2], + # which a correct grader accepts, so the list goes after a JSON decoy that is not one of the values. + grading = {"checks": [{"type": "json", "values": [[1, 2], [3, 4]], "contains": True}]} + stuffed = redteam.attacks(grading)["stuffed"] + assert stuffed is not None and stuffed.startswith('{"planted"') + assert redteam.grade(grading, "", stuffed).checks[0].passed is False + result = run([_q("1", grading)]) + assert result["status"] == "pass" and result["violations"] == [] + + +def test_a_grader_that_accepts_anything_is_blocked_on_every_attack(): + grading = {"checks": [{"type": "regex", "values": [".*"]}]} + result = run([_q("1", grading)]) + attacks = {v["attack"] for v in result["violations"]} + assert result["status"] == "blocked" and attacks == {"wrong", "empty", "hedge"} + assert result["questions"][0]["canary"] is False + + +def test_a_rejected_canary_is_a_violation(monkeypatch): + def reject_all(grading, question, answer, ask=None): + return Grade(checks=[CheckResult("match", False, 0.0)]) + + monkeypatch.setattr(redteam, "grade", reject_all) + result = run([_q("1", _match("Paris"))]) + assert result["violations"] == [{"question_id": "1", "attack": "canary", "passed": False}] + assert result["status"] == "blocked" + + +def test_a_judge_check_without_a_judge_with_a_negated_check_is_unscored_not_blocked(): + # The negated check passes wrong, empty and hedge by design; the judge check was never run, so + # those passes say nothing about the block and must not count as violations. + grading = {"checks": [{"type": "judge", "criterion": "names the capital"}, + {"type": "match", "values": ["Lyon"], "negate": True}]} + result = run([_q("1", grading)]) + assert result["status"] == "inconclusive" + assert result["violations"] == [] + assert result["summary"]["probed"] == 0 + + +def test_a_judge_check_without_a_judge_under_a_threshold_is_not_a_pass(): + # With a threshold the score is taken over the counted checks alone, so a missing judge can + # turn a block that rejects nothing into one that passes the canary and rejects the attacks. + grading = {"threshold": 0.5, "checks": [{"type": "judge", "criterion": "names Paris"}, + {"type": "match", "values": ["Paris"]}]} + result = run([_q("1", grading)]) + assert result["status"] == "inconclusive" + assert result["summary"]["probed"] == 0 + + +SLOW = "slow pattern" # the regex that runs past its timeout; any other pattern runs as written + + +@pytest.fixture +def regex_past_its_timeout(monkeypatch): + """Make the regex check with the SLOW pattern run past its timeout, as a catastrophic one does.""" + real_search = regex.search + + def search(pattern, string, *args, **kwargs): + if pattern == SLOW: + raise TimeoutError("regex ran past its limit") + return real_search(pattern, string, *args, **kwargs) + + monkeypatch.setattr(regex, "search", search) + + +@pytest.mark.parametrize("grading", [ + # No threshold: the negated match passes wrong, empty and hedge by design, and the regex never ran. + {"checks": [{"type": "regex", "values": [SLOW]}, {"type": "match", "values": ["Lyon"], "negate": True}]}, + # Threshold: the score over the counted match alone would pass the canary and reject the attacks. + {"threshold": 0.5, "checks": [{"type": "regex", "values": [SLOW]}, {"type": "match", "values": ["Paris"]}]}, +]) +def test_a_regex_past_its_timeout_leaves_the_probe_unscored_not_passed(regex_past_its_timeout, grading): + result = run([_q("1", grading)]) + assert result["status"] == "inconclusive" + assert result["violations"] == [] + assert result["summary"]["probed"] == 0 + + +def test_a_judge_check_without_a_judge_is_unscored_and_does_not_block(): + grading = {"checks": [{"type": "judge", "criterion": "names the capital"}]} + result = run([_q("1", grading)]) + assert result["status"] == "inconclusive" # nothing was graded, so nothing was verified + assert result["violations"] == [] + assert result["summary"]["unscored"] == 3 # wrong, empty and hedge; no canary, no stuffing + assert result["summary"]["probed"] == 0 + + +@pytest.mark.parametrize("questions", [ + [{"id": "1", "question": "Capital of France?"}], # no grading block at all + [_q("1", {"checks": [{"type": "judge", "criterion": "names the capital"}]})], +]) +def test_a_run_that_probed_nothing_is_inconclusive_not_pass(questions): + result = run(questions) + assert result["status"] == "inconclusive" + assert result["summary"]["probed"] == 0 + + +def test_stuffing_is_not_an_attack_on_a_block_that_needs_every_value(): + # "Paris Lyon" satisfies a block that requires both values, so the grader is right to accept it. + result = run([_q("1", _match("Paris", "Lyon", mode="all"))]) + assert result["status"] == "pass" + assert "stuffed" not in result["questions"][0]["probes"] + + +def test_stuffing_is_not_an_attack_on_several_checks_that_each_accept_a_value(): + grading = {"checks": [{"type": "match", "values": ["Paris"]}, {"type": "match", "values": ["Lyon"]}]} + assert redteam.attacks(grading)["stuffed"] is None + + +@pytest.mark.parametrize("grading", [ + {"checks": [{"type": "match", "values": ["Paris", "Lyon"], "mode": "all"}]}, + {"checks": [{"type": "match", "values": ["Paris"], "negate": True}]}, + {"checks": [{"type": "regex", "values": ["Paris"]}]}, + {"checks": [{"type": "match", "values": ["Paris"]}, {"type": "match", "values": ["Lyon"]}]}, +]) +def test_no_canary_when_the_block_is_not_one_literal_check(grading): + assert canary_for(grading) is None + + +def test_a_stuffed_answer_never_ends_in_a_decoy_the_check_accepts(): + # QZX is a value, so a stuffed answer that ends in it is accepted by a correct grader; the decoy + # is dropped and the answer ends in one the check rejects. + grading = _match("Lyon", "QZX-BOGUS-REPLY-4096") + stuffed = redteam.attacks(grading)["stuffed"] + assert stuffed is not None and not stuffed.endswith("QZX-BOGUS-REPLY-4096") + result = run([_q("1", grading)]) + assert result["violations"] == [] + assert result["status"] != "blocked" + + +def test_a_stuffed_answer_needs_two_decoys_the_check_rejects(): + grading = _match("PLANTED-WRONG-ANSWER-7731", "QZX-BOGUS-REPLY-4096", "ZKV-MISMATCH-2508") + assert redteam.attacks(grading)["stuffed"] is None + + +def test_a_question_with_one_scored_probe_is_measured_when_another_probe_is_not(): + # Unmeasured means no probe scored; one unscored probe next to a scored one is not a reason + # to call the question unmeasured. + row = {"negated_only": False, "probes": {"canary": {"passed": None}, "wrong": {"passed": False}}} + assert redteam._unmeasured(row) == "" + + +def test_canary_text_follows_the_check_kind(): + assert canary_for(_match("Paris")) == "The answer is Paris." + assert canary_for({"checks": [{"type": "match", "values": ["Paris"], "location": "begin"}]}) == "Paris is the answer." + assert canary_for({"checks": [{"type": "exact", "value": "Paris"}]}) == "Paris" + assert canary_for({"checks": [{"type": "choice", "values": ["b", "a"]}]}) == "ANSWER: A, B" + assert canary_for({"checks": [{"type": "json", "value": 3}]}) == "3" + + +def test_a_block_that_cannot_be_graded_is_refused(): + with pytest.raises(ValueError, match="cannot be graded"): + run([_q("1", {"checks": [{"type": "nope"}]})]) + + +def test_table_names_each_violation_and_the_status(monkeypatch): + _accept_stuffed(monkeypatch) + result = run([_q("7", _match("Paris", "Lyon"))]) + table = format_table(result) + assert "violation: question 7: stuffed answer accepted" in table + assert "status blocked: 1 questions" in table + + +def _args(path, **overrides): + base = {"questions": str(path), "judge": False, "judge_model": "m", "judge_provider": "anthropic", + "output": None} + return argparse.Namespace(**{**base, **overrides}) + + +def _patch_questions(monkeypatch, questions): + monkeypatch.setattr("dokimi.questions.artifact.load_questions", + lambda path: SimpleNamespace(artifact={"questions": questions})) + + +def test_command_passes_and_writes_a_redteam_artifact(tmp_path, monkeypatch, capsys): + _patch_questions(monkeypatch, [_q("1", _match("Paris"))]) + out = tmp_path / "redteam.json" + cmd_eval_redteam(_args(tmp_path / "questions.json", output=out)) + saved = json.loads(out.read_text(encoding="utf-8")) + assert saved["schema_version"] == "redteam.v1" and saved["status"] == "pass" + assert saved["questions_path"].endswith("questions.json") + assert f"Wrote redteam artifact to {out}" in capsys.readouterr().out + + +def test_command_exits_1_when_the_run_is_blocked(tmp_path, monkeypatch): + _patch_questions(monkeypatch, [_q("1", {"checks": [{"type": "regex", "values": [".*"]}]})]) + with pytest.raises(SystemExit) as exc: + cmd_eval_redteam(_args(tmp_path / "questions.json")) + assert exc.value.code == 1 + + +def test_command_exits_1_when_nothing_was_graded(tmp_path, monkeypatch): + _patch_questions(monkeypatch, [{"id": "1", "question": "Capital of France?"}]) + with pytest.raises(SystemExit) as exc: + cmd_eval_redteam(_args(tmp_path / "questions.json")) + assert exc.value.code == 1 + + +def test_command_exits_2_when_the_artifact_cannot_be_written(tmp_path, monkeypatch, capsys): + _patch_questions(monkeypatch, [_q("1", _match("Paris"))]) + blocker = tmp_path / "a_file" + blocker.write_text("not a directory", encoding="utf-8") + with pytest.raises(SystemExit) as exc: + cmd_eval_redteam(_args(tmp_path / "questions.json", output=blocker / "out.json")) + assert exc.value.code == 2 # not 1: CI must not read an unwritable -o as a blocked run + assert capsys.readouterr().err.startswith("Error:") + + +def test_command_exits_2_on_a_question_set_nested_too_deep_to_read(tmp_path, capsys): + path = tmp_path / "deep.json" + path.write_text("[" * 100_000 + "]" * 100_000, encoding="utf-8") + with pytest.raises(SystemExit) as exc: + cmd_eval_redteam(_args(path)) + assert exc.value.code == 2 # not a traceback and exit 1: CI reads exit 1 as a blocked run + assert capsys.readouterr().err.startswith("Error:") + + +def test_command_exits_2_on_an_unreadable_question_set(tmp_path, capsys): + with pytest.raises(SystemExit) as exc: + cmd_eval_redteam(_args(tmp_path / "missing.json")) + assert exc.value.code == 2 + assert capsys.readouterr().err.startswith("Error:") + + +def test_a_negated_check_is_not_blocked_by_the_answers_it_passes_by_design(): + # "must not say Paris": a wrong, empty or hedged answer lacks Paris and so passes the check. + grading = {"checks": [{"type": "match", "values": ["Paris"], "negate": True}]} + probes = redteam.probe(grading, "Capital of France?") + assert "wrong" not in probes and "empty" not in probes and "hedge" not in probes + assert "canary" not in probes + + +@pytest.mark.parametrize("value", ["one", "SETUP"]) +def test_a_check_value_that_occurs_in_the_hedge_skips_the_hedge_probe(value): + # The hedge sentence says "which one" and "the setup"; a grader that accepts the word would + # be blamed for a hedge it was never asked to reject. + grading = {"checks": [{"type": "fuzzy", "values": [value]}]} + assert redteam.attacks(grading)["hedge"] is None + assert redteam.attacks({"checks": [{"type": "match", "values": ["Paris"]}]})["hedge"] == redteam.HEDGE + + +def test_a_negated_check_still_fails_the_stuffed_answer_that_carries_its_value(): + grading = {"checks": [{"type": "match", "values": ["Paris", "Lyon"], "negate": True}]} + assert redteam.attacks(grading)["stuffed"] == "Paris Lyon" + probes = redteam.probe(grading, "Q?") + assert probes["stuffed"].passed is False and not probes["stuffed"].violation + + +def test_a_negated_check_does_not_silence_the_attacks_on_the_positive_check(): + # regex ".*" accepts any answer, and "must not say Lyon" accepts the wrong one too: the block is + # gameable, so the run is blocked rather than passing with nothing probed. + grading = {"checks": [{"type": "regex", "values": [".*"]}, {"type": "match", "values": ["Lyon"], "negate": True}]} + result = run([_q("1", grading)]) + assert result["status"] == "blocked" + assert {v["attack"] for v in result["violations"]} == {"wrong", "empty", "hedge"} + assert result["summary"]["probed"] == 1 + + +def test_a_negated_check_beside_a_positive_check_leaves_the_positive_check_to_reject_the_attacks(): + grading = {"checks": [{"type": "match", "values": ["Paris"]}, {"type": "match", "values": ["Lyon"], "negate": True}]} + result = run([_q("1", grading)]) + probes = result["questions"][0]["probes"] + assert result["status"] == "pass" + assert all(probes[name]["passed"] is False for name in ("wrong", "empty", "hedge")) + + +def test_a_block_of_only_negated_checks_runs_no_wrong_empty_or_hedge_probe_and_reports_it(): + grading = {"checks": [{"type": "match", "values": ["Paris"], "negate": True}]} + result = run([_q("1", grading)]) + row = result["questions"][0] + assert row["negated_only"] is True and result["summary"]["negated_only"] == 1 + assert not {"wrong", "empty", "hedge"} & set(row["probes"]) + assert set(row["probes"]) == {"stuffed"} # the forbidden value itself, which must fail + assert result["status"] == "inconclusive" # the real grader accepts any wrong answer: nothing was verified + assert result["summary"]["probed"] == 0 and result["summary"]["unprobed"] == 1 + assert "1 questions with only negated checks: wrong, empty and hedge not run" in format_table(result) + + +def test_a_negated_single_value_is_stuffed_with_that_value_and_must_fail(): + grading = {"checks": [{"type": "match", "values": ["Paris"], "negate": True}]} + assert redteam.attacks(grading)["stuffed"] == "Paris" + probes = redteam.probe(grading, "Capital of France?") + assert set(probes) == {"stuffed"} and probes["stuffed"].passed is False and not probes["stuffed"].violation + + +@pytest.mark.parametrize("check", [ + {"type": "exact", "values": ["Paris", "Lyon"], "negate": True}, + {"type": "match", "values": ["Paris", "Lyon"], "location": "exact", "negate": True}, + {"type": "json", "values": [1, 2], "negate": True}, +]) +def test_a_negated_whole_answer_check_is_not_stuffed(check): + # "Paris Lyon" is neither "Paris" nor "Lyon", so the negated check passes it: that is not an attack. + grading = {"checks": [check]} + assert redteam.attacks(grading)["stuffed"] is None + result = run([_q("1", grading)]) + assert result["violations"] == [] and "stuffed" not in result["questions"][0]["probes"] + + +@pytest.mark.parametrize("check", [ + {"type": "match", "values": ["Paris", "Lyon"], "mode": "all", "negate": True}, # location end + {"type": "match", "values": ["Paris", "Lyon"], "mode": "all", "location": "begin", "negate": True}, +]) +def test_a_negated_all_mode_check_is_not_stuffed(check): + # "Paris Lyon" ends with only one value, so the "all" match is false and the negated check passes + # it: a correct grader would be blamed for accepting it. The positive rule skips mode all alike. + grading = {"checks": [check]} + assert redteam.attacks(grading)["stuffed"] is None + result = run([_q("1", grading)]) + assert result["violations"] == [] and "stuffed" not in result["questions"][0]["probes"] + + +@pytest.mark.parametrize("check", [ + {"type": "match", "values": ["31"]}, # the planted answer ends in 7731 + {"type": "match", "values": ["1"]}, # the planted answer ends in 1 + {"type": "fuzzy", "values": ["on"]}, # the planted answer contains "on" in "wrong" +]) +def test_a_check_value_in_the_planted_wrong_answer_does_not_block_a_grader_that_rejects_it(check): + result = run([_q("1", {"checks": [check]})]) + assert result["status"] == "pass" and result["violations"] == [] + assert result["questions"][0]["probes"]["wrong"]["passed"] is False # still probed, on a clean wrong answer + + +def test_the_planted_wrong_answer_carries_none_of_the_positive_check_values(): + grading = {"checks": [{"type": "match", "values": ["1", "on"]}]} + wrong = redteam.attacks(grading)["wrong"] + assert wrong is not None and "1" not in wrong and "on" not in wrong.lower() + + +def test_the_wrong_attack_is_skipped_when_every_candidate_carries_a_check_value(monkeypatch): + monkeypatch.setattr(redteam, "WRONG_CANDIDATES", ("PLANTED-WRONG-ANSWER-7731",)) + assert redteam.attacks({"checks": [{"type": "match", "values": ["31"]}]})["wrong"] is None + + +def test_the_summary_counts_the_questions_that_were_not_probed(): + questions = [_q("1", _match("Paris")), _q("2", {"checks": [{"type": "judge", "criterion": "names the capital"}]})] + result = run(questions) + assert result["summary"]["probed"] == 1 and result["summary"]["unprobed"] == 1 + assert result["status"] == "inconclusive" + assert "1 unprobed" in format_table(result) + + +@pytest.mark.parametrize("check", [ + {"type": "match", "values": ["several things depending on the setup"], "location": "any"}, # comma is punctuation + {"type": "fuzzy", "values": ["the setup and I"]}, # articles drop out of the grader's normalised text +]) +def test_a_value_the_grader_finds_in_the_hedge_after_normalising_skips_the_hedge_probe(check): + assert redteam.attacks({"checks": [check]})["hedge"] is None + result = run([_q("1", {"checks": [check]})]) + assert "hedge" not in {v["attack"] for v in result["violations"]} + + +def test_a_match_without_punctuation_stripping_gets_a_canary_the_grader_accepts(): + grading = {"checks": [{"type": "match", "values": ["Paris"], "ignore_punctuation": False}]} + assert canary_for(grading) == "The answer is Paris" # a full stop would end the answer, not the value + result = run([_q("1", grading)]) + assert result["status"] == "pass" and result["questions"][0]["canary"] is True + + +def test_a_numeric_check_does_not_block_on_a_planted_answer_it_reads_as_its_number(): + # The planted answer ends in 7731 and a numeric match reads that number: a clean candidate is chosen by + # asking the check itself, not by looking for the value's text in it. + check = {"type": "match", "values": ["-7731.0"], "numeric": True} + wrong = redteam.attacks({"checks": [check]})["wrong"] + assert wrong is not None and "7731" not in wrong + result = run([_q("1", {"checks": [check]})]) + assert result["status"] == "pass" and result["violations"] == [] + + +def test_a_short_value_the_check_does_not_find_in_the_hedge_keeps_the_hedge_probe(): + # "one" occurs in "which one applies", but an end match reads the last words of the hedge only. + grading = {"checks": [{"type": "match", "values": ["one"]}]} + assert redteam.attacks(grading)["hedge"] == redteam.HEDGE + result = run([_q("1", grading)]) + assert result["questions"][0]["probes"]["hedge"]["passed"] is False + + +def test_a_threshold_block_that_a_negated_check_lifts_over_the_line_is_blocked(): + # Half the score comes from "must not say London", which every wrong, empty or hedged answer earns: + # the real grader passes them at 0.5, so the block is gameable. + grading = {"threshold": 0.5, "checks": [{"type": "match", "values": ["Paris"]}, + {"type": "match", "values": ["London"], "negate": True}]} + result = run([_q("1", grading)]) + assert result["status"] == "blocked" + assert {v["attack"] for v in result["violations"]} == {"wrong", "empty", "hedge"} + + +def test_one_probed_question_does_not_make_a_set_pass_when_the_others_were_not_measured(): + judge = {"checks": [{"type": "judge", "criterion": "names the capital"}]} + questions = [_q(str(i), judge) for i in range(5)] + [_q("m", _match("Paris"))] + result = run(questions) + assert result["status"] == "inconclusive" and result["violations"] == [] + assert result["summary"]["probed"] == 1 and result["summary"]["unprobed"] == 5 + table = format_table(result) + assert "not measured: question 0 (no probe scored)" in table + assert "question m" not in table.split("not measured:")[1] + + +def test_a_question_with_only_negated_checks_is_named_in_the_table_as_not_measured(): + only_negated = {"checks": [{"type": "match", "values": ["error"], "negate": True}]} + result = run([_q("g", only_negated), _q("m", _match("Paris"))]) + assert result["status"] == "inconclusive" + assert "not measured: question g (only negated checks)" in format_table(result) + + +def test_a_violation_still_blocks_a_set_with_unmeasured_questions(): + judge = {"checks": [{"type": "judge", "criterion": "names the capital"}]} + result = run([_q("j", judge), _q("1", {"checks": [{"type": "regex", "values": [".*"]}]})]) + assert result["status"] == "blocked" + + +def test_a_repeated_question_id_grades_each_row_with_its_own_block(): + weak = {"checks": [{"type": "match", "values": ["Paris"]}], "threshold": 0.0} # accepts everything + strong = {"checks": [{"type": "exact", "value": "Madrid"}]} + result = run([_q("a", weak), _q("a", strong)]) + assert result["status"] == "blocked" + assert {v["attack"] for v in result["violations"]} == {"wrong", "empty", "hedge"} + assert [r["canary"] for r in result["questions"]] == [True, True] + assert result["summary"]["questions"] == 2 and result["summary"]["violations"] == 3 + + +def test_the_artifact_names_the_judge_that_graded_the_judge_checks(tmp_path, monkeypatch): + _patch_questions(monkeypatch, [_q("1", _match("Paris"))]) + monkeypatch.setattr("dokimi.eval.judge.Judge", lambda **kw: SimpleNamespace(llm=None)) + out = tmp_path / "judged.json" + cmd_eval_redteam(_args(tmp_path / "questions.json", output=out, judge=True, + judge_model="judge-x", judge_provider="openai")) + saved = json.loads(out.read_text(encoding="utf-8")) + assert (saved["judge_model"], saved["judge_provider"]) == ("judge-x", "openai") + + +def test_the_artifact_has_no_judge_without_the_judge_flag(tmp_path, monkeypatch): + _patch_questions(monkeypatch, [_q("1", _match("Paris"))]) + out = tmp_path / "plain.json" + cmd_eval_redteam(_args(tmp_path / "questions.json", output=out, judge_model="judge-x")) + saved = json.loads(out.read_text(encoding="utf-8")) + assert saved["judge_model"] is None and saved["judge_provider"] is None + + +def test_the_judge_ask_sends_the_prompt_to_the_named_model_and_returns_the_reply(monkeypatch): + from dokimi.eval.judge import judge_ask + + made, sent = {}, {} + + class FakeLLM: + def invoke(self, prompt, draw=0, cache_if=None): + sent.update(prompt=prompt, draw=draw) + return SimpleNamespace(content="YES") + + def fake_judge(**kw): + made.update(kw) + return SimpleNamespace(llm=FakeLLM()) + + monkeypatch.setattr("dokimi.eval.judge.Judge", fake_judge) + assert judge_ask("judge-x", "openai")("is it covered?", draw=2) == "YES" + assert made == {"model": "judge-x", "provider": "openai"} and sent == {"prompt": "is it covered?", "draw": 2} + + +def _many_violations(): + return run([_q(str(i), {"checks": [{"type": "regex", "values": [".*"]}]}) for i in range(8)]) # 24 violations + + +def test_the_table_names_dash_o_when_it_cuts_the_violations_and_no_artifact_is_written(): + table = format_table(_many_violations()) + assert "... 4 more violations; rerun with -o FILE to list them all" in table + assert "in the artifact" not in table + + +def test_the_table_points_to_the_artifact_when_dash_o_was_given(tmp_path): + table = format_table(_many_violations(), output=tmp_path / "r.json") + assert "... 4 more violations in the artifact" in table + assert "rerun with -o" not in table + + +def test_the_command_prints_the_cut_violations_with_the_output_path_it_wrote(tmp_path, monkeypatch, capsys): + _patch_questions(monkeypatch, [_q(str(i), {"checks": [{"type": "regex", "values": [".*"]}]}) for i in range(8)]) + with pytest.raises(SystemExit): + cmd_eval_redteam(_args(tmp_path / "questions.json", output=tmp_path / "r.json")) + assert "4 more violations in the artifact" in capsys.readouterr().out + + +@pytest.mark.parametrize("check", [ + {"type": "match", "value": "Paris", "location": "exact"}, + {"type": "json", "value": "Paris"}, + {"type": "json", "value": 3}, + {"type": "exact", "value": "Paris"}, + {"type": "match", "value": "Paris", "location": "begin"}, + {"type": "choice", "values": ["b", "a"]}, +]) +def test_the_canary_for_each_literal_check_is_accepted_by_the_grader(check): + from dokimi.eval.reference_checks import grade + + grading = {"checks": [check]} + assert grade(grading, "Q?", canary_for(grading)).passed is True + + +def test_the_canary_is_the_bare_value_for_a_whole_answer_match_and_a_json_string(): + assert canary_for({"checks": [{"type": "match", "value": "Paris", "location": "exact"}]}) == "Paris" + assert canary_for({"checks": [{"type": "json", "value": "Paris"}]}) == '"Paris"' + + +def _covered(covered): + """A judge that finds the point covered (or absent) in both polarities, consistently.""" + from dokimi.eval.checklist import ASK_COVERED + + def ask(prompt, draw=0, cache_if=None): + return "YES" if (ASK_COVERED in prompt) == covered else "NO" + + return ask + + +_JUDGE_ONLY = {"checks": [{"type": "judge", "criterion": "names the capital"}]} + + +def test_a_judge_check_is_scored_when_an_ask_is_given_and_a_judge_that_rejects_every_attack_passes(): + result = run([_q("1", _JUDGE_ONLY)], ask=_covered(False)) + assert result["status"] == "pass" and result["summary"]["unscored"] == 0 + assert all(p["passed"] is False for p in result["questions"][0]["probes"].values()) + + +def test_a_judge_that_accepts_every_answer_blocks_the_run(): + result = run([_q("1", _JUDGE_ONLY)], ask=_covered(True)) + assert result["status"] == "blocked" + assert {v["attack"] for v in result["violations"]} == {"wrong", "empty", "hedge"} + + +def test_the_judge_flag_grades_judge_checks_with_the_named_model(tmp_path, monkeypatch): + made = {} + + def fake_judge(**kw): + made.update(kw) + return SimpleNamespace(llm=SimpleNamespace(invoke=lambda prompt, draw=0, cache_if=None: + SimpleNamespace(content=_covered(True)(prompt)))) + + _patch_questions(monkeypatch, [_q("1", _JUDGE_ONLY)]) + monkeypatch.setattr("dokimi.eval.judge.Judge", fake_judge) + out = tmp_path / "judged.json" + with pytest.raises(SystemExit) as exc: + cmd_eval_redteam(_args(tmp_path / "questions.json", output=out, judge=True, + judge_model="judge-x", judge_provider="openai")) + assert exc.value.code == 1 # the stub judge accepts every attack + assert made == {"model": "judge-x", "provider": "openai"} + saved = json.loads(out.read_text(encoding="utf-8")) + assert saved["status"] == "blocked" and saved["summary"]["unscored"] == 0 + + +def test_without_the_judge_flag_no_judge_is_built_and_judge_checks_stay_unscored(tmp_path, monkeypatch): + def refuse(**kw): + raise AssertionError("a judge was built without --judge") + + _patch_questions(monkeypatch, [_q("1", _JUDGE_ONLY)]) + monkeypatch.setattr("dokimi.eval.judge.Judge", refuse) + out = tmp_path / "plain.json" + with pytest.raises(SystemExit) as exc: + cmd_eval_redteam(_args(tmp_path / "questions.json", output=out)) + assert exc.value.code == 1 + assert json.loads(out.read_text(encoding="utf-8"))["summary"]["unscored"] == 3 + + +@pytest.mark.parametrize("grading", [ + {"checks": [{"type": "match", "values": []}]}, + {"checks": [{"type": "match", "values": [""]}]}, + {"checks": []}, + None, +]) +def test_a_degenerate_grading_block_is_refused_not_passed(grading): + with pytest.raises(ValueError, match="cannot be graded"): + run([_q("1", grading)]) + + +def test_an_empty_question_list_is_inconclusive_not_pass(): + result = run([]) + assert result["status"] == "inconclusive" and result["summary"]["questions"] == 0 + + +def test_command_exits_2_with_the_problem_when_a_grading_block_cannot_be_graded(tmp_path, monkeypatch, capsys): + _patch_questions(monkeypatch, [_q("7", {"checks": [{"type": "match", "values": []}]})]) + out = tmp_path / "redteam.json" + with pytest.raises(SystemExit) as exc: + cmd_eval_redteam(_args(tmp_path / "questions.json", output=out)) + assert exc.value.code == 2 + assert "7: check 0: match has no values" in capsys.readouterr().err + assert not out.exists() + + +def test_command_exits_1_on_an_empty_question_set(tmp_path, monkeypatch): + _patch_questions(monkeypatch, []) + with pytest.raises(SystemExit) as exc: + cmd_eval_redteam(_args(tmp_path / "questions.json")) + assert exc.value.code == 1 + + +def test_a_blocked_run_still_writes_the_artifact(tmp_path, monkeypatch): + _patch_questions(monkeypatch, [_q("1", {"checks": [{"type": "regex", "values": [".*"]}]})]) + out = tmp_path / "blocked.json" + with pytest.raises(SystemExit) as exc: + cmd_eval_redteam(_args(tmp_path / "questions.json", output=out)) + assert exc.value.code == 1 + assert json.loads(out.read_text(encoding="utf-8"))["status"] == "blocked" + + +def test_an_inconclusive_run_still_writes_the_artifact(tmp_path, monkeypatch): + _patch_questions(monkeypatch, [{"id": "1", "question": "Capital of France?"}]) + out = tmp_path / "inconclusive.json" + with pytest.raises(SystemExit) as exc: + cmd_eval_redteam(_args(tmp_path / "questions.json", output=out)) + assert exc.value.code == 1 + assert json.loads(out.read_text(encoding="utf-8"))["status"] == "inconclusive" + + +@pytest.mark.parametrize("check", [ + {"type": "exact", "values": ["Paris", "Lyon"]}, + {"type": "match", "values": ["Paris", "Lyon"], "location": "exact"}, + {"type": "json", "values": ["Paris", "Lyon"]}, +]) +def test_aliases_of_a_whole_answer_check_are_stuffed_bare_and_the_grader_rejects_the_list(check): + # The list equals none of the values, so a grader that accepts it is wrong; the real one does not. + assert redteam.attacks({"checks": [check]})["stuffed"] == "Paris Lyon" + result = run([_q("1", {"checks": [check]})]) + assert result["status"] == "pass" and result["questions"][0]["probes"]["stuffed"]["passed"] is False