diff --git a/packages/flow/CHANGELOG.md b/packages/flow/CHANGELOG.md index 75a582a84b..98bb850f66 100644 --- a/packages/flow/CHANGELOG.md +++ b/packages/flow/CHANGELOG.md @@ -25,6 +25,10 @@ - The OpenEMR contracts (`benchmark/lifecycle_proof/verify.yaml`, `benchmark/openemr_local/verify.yaml`) now declare how the New Patient form stores what was typed: street and city in capitals, names with each word capitalized. Without it the exact record check refuted 19 of 20 correct saves on the stand-in suites (#1811). A contract value's `stored` form gains `capitalize_words` (OpenEMR's "C" fields); the set stays closed and the check stays exact. +### Features + +- Verifier audit: `python -m openadapt_flow.verifier_audit run --target module:factory --out DIR` tests an environment's own scorer rather than an agent. It replays a reference trajectory with mechanical faults (an unauthorized side effect, the wrong record, a wrong value, a duplicate write, a partial write, a write after the success claim, a claim with no write) plus an incidental-change positive control. It reads the end state independently and reports, per family, where the scorer accepted a wrong end state or rejected a legitimate one, with counterexamples and Wilson 95% intervals. Adapters for AgentDojo and AppWorld ship as examples; private family sets plug in with `--families-module`. Design: `docs/VERIFIER_AUDIT.md`. + ## v2.6.0 (2026-11-04) Flow adds signed teach authorizations, encrypted local artifacts, durable diff --git a/packages/flow/docs/VERIFIER_AUDIT.md b/packages/flow/docs/VERIFIER_AUDIT.md new file mode 100644 index 0000000000..699b4e9999 --- /dev/null +++ b/packages/flow/docs/VERIFIER_AUDIT.md @@ -0,0 +1,113 @@ +# Verifier audit + +A benchmark or RL environment turns a finished episode into a reward with its +own scorer. If the scorer accepts a wrong end state, an agent trained or ranked +on it learns or gets credit for that wrong behavior. The verifier audit tests +the scorer, not an agent. + +The method comes from *Finding Blind Spots in AppWorld and WorkArena Task +Verifiers* (Abrich, 2026). This module generalizes it to any environment that +can reset, replay a trajectory, score, and read back its own state. + +## How a cell is built + +1. The target gives a **reference trajectory** for each task: steps the + environment's own scorer accepts. Typical sources are the shipped ground + truth or a recorded solution. +2. The kit replays the reference twice from a fresh reset. A task is excluded + if the scorer rejects the reference, if the reference effect differs + between the two replays, or if it has no effect at all (an answer-only + task). +3. Each **fault family** edits the reference mechanically. A family never sees + a scorer verdict. +4. For each edit, the kit first runs a **census replay without scoring**. It + keeps the edit only if the environment accepted the edited call and the end + state differs from the reference. An edit whose effect matches the + reference (an idempotent duplicate, a field the app ignores) is + **degenerate** and is listed, never counted. +5. The kept edit is replayed again from a fresh reset and scored once. + +Ground truth comes from the target's **effect read**: a diff of the end state +against the start state, taken through a path that does not run the scorer +(direct SQLite reads for AppWorld, a model dump for AgentDojo). A harmful-family +cell whose effect differs from the reference and that the scorer accepts is a +**false accept**. An incidental-change cell that the scorer rejects is a +**false reject**. + +## Families + +| Family | Polarity | Edit | +|---|---|---| +| `unauthorized_side_effect` | harmful | Keep every task write, and add a copy of one aimed at a record the task never touches | +| `wrong_record` | harmful | One task write lands on a different record of the same type | +| `wrong_value` | harmful | One task write carries a changed non-identity value | +| `duplicate` | harmful | One task write is sent twice | +| `partial_write` | harmful | A multi-write task sends only its first write, or drops its last | +| `late_write` | harmful | The last write is sent after the success claim, so the claim was false when made | +| `optimistic_success` | harmful | Reads and the claim, no writes | +| `incidental_change` | legitimate | Permitted extra calls (the reference's reads, repeated) before the claim | + +Identity arguments are recognized by name (`*_id`, `email`, `recipient`, +`iban`, URL path ids and so on). Retargets draw from values that other tasks in +the suite use for the same argument, skipping any value the task's own +reference uses. `wrong_value` and `wrong_record` cells also record whether the +original value appears as a whole word in the task instruction, so a reader can +set aside edits to fields the task never specified. + +`python -m openadapt_flow.verifier_audit families` prints the set as JSON +(`openadapt.verifier-audit.families/v1`). Add your own families with +`--families-module your.module`; the module exports a `FAMILIES` tuple of +`FaultFamily`. + +## Writing an adapter + +A target implements `AuditTarget`: + +```python +class MyEnv: + name = "my-env" + post_claim_steps = True # False if the episode ends at the final answer + + def tasks(self) -> list[str]: ... + def reference(self, task) -> list[Call]: ... # each Call has kind READ, WRITE, CLAIM or AUTH + def run(self, task, calls, *, score: bool) -> Episode: ... + def instruction(self, task) -> str: ... # optional +``` + +`run` must reset, replay `calls` in order, record which calls the environment +refused, read the end state (and, if it can, the state at the claim step), and +call the scorer only when `score` is true. `targets/toy.py` is the smallest +example. `targets/agentdojo.py` runs in process. `targets/appworld.py` drives a +worker in AppWorld's own interpreter, because AppWorld pins pydantic 1. + +## Running + +```bash +python -m openadapt_flow.verifier_audit run \ + --target openadapt_flow.verifier_audit.targets.agentdojo:make_target \ + --target-arg suite=banking --out audit/agentdojo-banking +``` + +The output directory holds `cells.jsonl` (an append-only ledger; a rerun +resumes and does not rescore finished cells, but retries harness and scorer +errors, including a task excluded only because the reference scorer errored), `ledger.json` (the target, kit version, families and redaction mode; +a rerun with a different target or mode refuses), `report.json` and +`REPORT.md`. +Rates are per family, with Wilson 95% intervals, and are not pooled. The report +also gives the share of tasks with at least one false accept. + +Use `--redact` when a target's task data may not be redistributed in the clear. +AppWorld's protected data, for example, may be shared only encrypted. Redaction +replaces call arguments, effect rows, edit descriptions and error text with +digests and drops scorer details. Task ids and family names stay readable. + +## What a result does and does not show + +- A false accept shows that the scorer returned PASS on an end state that + differs from the reference effect in the named way. The reference is one + correct effect, not the only one. Read the counterexample's effect diff + before you call it a bug. +- The audit replays scripted trajectories and makes no model calls. It shows + what a policy could get credit for, not how often an agent does it. +- Benchmarks own their scorers. Report findings to their maintainers before + you publish them. diff --git a/packages/flow/openadapt_flow/verifier_audit/__init__.py b/packages/flow/openadapt_flow/verifier_audit/__init__.py new file mode 100644 index 0000000000..9d52afbc79 --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/__init__.py @@ -0,0 +1,57 @@ +"""Verifier audit: does an environment's own scorer accept wrong end states? + +A benchmark or RL environment ships a scorer that turns a finished episode +into a reward. This kit tests that scorer, not an agent. It starts from a +reference trajectory that the environment itself counts as correct, applies +mechanical fault families to it (an unauthorized side effect, the wrong +record, a duplicate write, a partial write, a write after the success claim, +a success claim with no write), replays each edited trajectory from a fresh +reset, and asks the environment's own scorer for a verdict. + +Ground truth never comes from the scorer. The target adapter reads the end +state through its own path (a database diff, a model dump, a Table API read) +and the kit compares that effect with the effect of the unedited reference. +A fault whose effect matches the reference is degenerate and is excluded. A +fault whose effect differs and that the scorer still accepts is a false +accept, reported as a counterexample with the effect diff. A legitimate +incidental change (an extra read, a repeated lookup) that the scorer rejects +is a false reject. + +The kit is the method from "Finding Blind Spots in AppWorld and WorkArena +Task Verifiers" (Abrich, 2026), generalized to any environment that can +reset, replay a trajectory, score, and read back its state. See +``docs/VERIFIER_AUDIT.md``. +""" + +from __future__ import annotations + +from openadapt_flow.verifier_audit.families import ( + DEFAULT_FAMILIES, + FaultFamily, + Polarity, + families_spec, +) +from openadapt_flow.verifier_audit.model import ( + AuditTarget, + Call, + CallKind, + Episode, + Verdict, +) +from openadapt_flow.verifier_audit.report import wilson_interval +from openadapt_flow.verifier_audit.runner import AuditConfig, run_audit + +__all__ = [ + "AuditConfig", + "AuditTarget", + "Call", + "CallKind", + "DEFAULT_FAMILIES", + "Episode", + "FaultFamily", + "Polarity", + "Verdict", + "families_spec", + "run_audit", + "wilson_interval", +] diff --git a/packages/flow/openadapt_flow/verifier_audit/__main__.py b/packages/flow/openadapt_flow/verifier_audit/__main__.py new file mode 100644 index 0000000000..6aec703055 --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/__main__.py @@ -0,0 +1,88 @@ +"""``python -m openadapt_flow.verifier_audit``: audit an environment's scorer. + + python -m openadapt_flow.verifier_audit families + python -m openadapt_flow.verifier_audit run \\ + --target openadapt_flow.verifier_audit.targets.agentdojo:make_target \\ + --target-arg suite=banking --out audit-out/agentdojo-banking + +``--target`` names a factory ``module:function`` that returns an +:class:`~openadapt_flow.verifier_audit.model.AuditTarget`; each +``--target-arg key=value`` is passed to it as a keyword argument. +``--families-module`` adds a private family set (its ``FAMILIES`` tuple). +""" + +from __future__ import annotations + +import argparse +import importlib +import json +import sys +from pathlib import Path + +from openadapt_flow.verifier_audit.families import ( + DEFAULT_FAMILIES, + families_spec, + load_families, +) +from openadapt_flow.verifier_audit.runner import AuditConfig, run_audit + + +def _factory(spec: str): + module, _, attr = spec.partition(":") + if not attr: + raise SystemExit(f"--target must be module:function, got {spec!r}") + return getattr(importlib.import_module(module), attr) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="python -m openadapt_flow.verifier_audit") + sub = parser.add_subparsers(dest="cmd", required=True) + fam = sub.add_parser("families", help="print the family set as JSON") + fam.add_argument("--families-module") + run = sub.add_parser("run", help="audit a target") + run.add_argument("--target", required=True) + run.add_argument("--target-arg", action="append", default=[]) + run.add_argument("--out", type=Path, required=True) + run.add_argument("--tasks", help="comma-separated task ids (default: all)") + run.add_argument( + "--families-module", help="module whose FAMILIES are added to the defaults" + ) + run.add_argument("--only-families", help="comma-separated family ids to run") + run.add_argument("--max-candidates", type=int, default=6) + run.add_argument( + "--single-replay", + action="store_true", + help="score the census replay instead of replaying the chosen edit again", + ) + run.add_argument( + "--redact", action="store_true", help="digest call arguments and effect rows" + ) + args = parser.parse_args(argv) + + families = tuple(DEFAULT_FAMILIES) + if args.families_module: + families += load_families(args.families_module) + if args.cmd == "families": + print(json.dumps(families_spec(families), indent=2)) + return 0 + if args.only_families: + wanted = set(args.only_families.split(",")) + families = tuple(f for f in families if f.id in wanted) + kwargs = dict(kv.split("=", 1) for kv in args.target_arg) + target = _factory(args.target)(**kwargs) + cfg = AuditConfig( + families=families, + tasks=args.tasks.split(",") if args.tasks else None, + max_candidates=args.max_candidates, + single_replay=args.single_replay, + redact=args.redact, + out_dir=args.out, + log=lambda msg: print(msg, file=sys.stderr, flush=True), + ) + report = run_audit(target, cfg) + print(json.dumps(report["summary"], indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/flow/openadapt_flow/verifier_audit/families.py b/packages/flow/openadapt_flow/verifier_audit/families.py new file mode 100644 index 0000000000..0285ed1fac --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/families.py @@ -0,0 +1,423 @@ +"""Fault families: mechanical edits of a reference trajectory. + +Every family is a pure function of the reference trajectory and a pool of +identity values seen across the suite. It returns candidate edits, newest +write first, or a refusal that names the missing precondition. A family +never looks at a scorer verdict. The runner picks the first candidate whose +replayed effect actually differs from the reference (the census), and only +then asks the scorer, so the choice of cell never depends on the verdict. + +``harmful`` families make the end state wrong in a named way; the scorer +should reject them. ``legitimate`` families make a change the task permits; +the scorer should accept them. The incidental-change family is the positive +control: a scorer that rejects everything would otherwise look perfect. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Callable, Mapping, Optional, Sequence, Union + +from openadapt_flow.verifier_audit.model import Call, CallKind + +FORMAT_ID = "openadapt.verifier-audit.families/v1" + + +class Polarity(str, Enum): + HARMFUL = "harmful" + LEGITIMATE = "legitimate" + + +IDENTITY_KEY = re.compile( + r"(^id_?$|_id$|_ids$|^_path_\d+$|email|phone|recipient|receiver|^to$|^cc$|^bcc$" + r"|iban|account|user_?name$|^user$|^owner$|^file_?(path|name)?$|^path$|participants)", + re.IGNORECASE, +) +AUTH_KEY = re.compile( + r"(access_token|password|passwd|secret|^token$|api_key)", re.IGNORECASE +) + + +@dataclass(frozen=True) +class Candidate: + calls: tuple[Call, ...] + description: str + must_succeed: tuple[int, ...] = () + """Indices of edited calls that the environment must accept; a candidate + whose edited call was refused is not the named fault (a refused + retarget is an omission), so the runner moves to the next one.""" + ordering_only: bool = False + """The harm is about when the effect lands, not what it is: judged on the + claim-time effect.""" + + +@dataclass(frozen=True) +class Refusal: + reason: str + + +@dataclass +class FamilyContext: + task: str + reference: Sequence[Call] + identity_pool: Mapping[str, Sequence[Any]] = field(default_factory=dict) + """Argument name -> distinct values seen for it across all reference + trajectories in the suite.""" + incidental: Optional[Sequence[Call]] = None + + +Build = Callable[[FamilyContext], Union[list[Candidate], Refusal]] + + +@dataclass(frozen=True) +class FaultFamily: + id: str + polarity: Polarity + description: str + build: Build + public: bool = True + + def spec(self) -> dict[str, Any]: + return { + "id": self.id, + "polarity": self.polarity.value, + "description": self.description, + "public": self.public, + } + + +def write_indices(calls: Sequence[Call]) -> list[int]: + return [i for i, c in enumerate(calls) if c.kind is CallKind.WRITE] + + +def claim_index(calls: Sequence[Call]) -> Optional[int]: + for i in range(len(calls) - 1, -1, -1): + if calls[i].kind is CallKind.CLAIM: + return i + return None + + +def _alternates( + key: str, + value: Any, + pool: Mapping[str, Sequence[Any]], + own: set[str], + limit: int = 3, +) -> list[Any]: + """Other plausible values for an identity argument: values the suite + uses for the same argument, then numeric neighbours. Values that appear + anywhere in this task's own reference are skipped, so a retarget lands + on a record the task never touches rather than on another of its own.""" + out: list[Any] = [] + for other in pool.get(key, ()): + if str(other) in own: + continue + if other != value and type(other) is type(value) and other not in out: + out.append(other) + if len(out) >= limit: + return out + as_int: Optional[int] = None + if isinstance(value, int) and not isinstance(value, bool): + as_int = value + elif isinstance(value, str) and value.isdigit(): + as_int = int(value) + if as_int is not None: + for delta in (1, -1, 2): + n = as_int + delta + if n < 0: + continue + alt: Any = n if isinstance(value, int) else str(n).zfill(len(value)) + if alt not in out and str(alt) not in own: + out.append(alt) + return out[:limit] + + +def _own_values(calls: Sequence[Call]) -> set[str]: + return { + str(v) + for c in calls + for v in c.args.values() + if not isinstance(v, (list, dict)) + } + + +def _retargets( + call: Call, pool: Mapping[str, Sequence[Any]], own: set[str] +) -> list[tuple[str, Any, Call]]: + out = [] + for key, value in call.args.items(): + if AUTH_KEY.search(key) or not IDENTITY_KEY.search(key): + continue + if isinstance(value, (list, dict)) or value in (None, ""): + continue + for alt in _alternates(key, value, pool, own): + out.append((key, alt, call.with_args({**call.args, key: alt}))) + return out + + +def _distinct_writes_newest_first(calls: Sequence[Call]) -> list[int]: + seen: set[str] = set() + out = [] + for i in reversed(write_indices(calls)): + if calls[i].name not in seen: + seen.add(calls[i].name) + out.append(i) + return out + + +def unauthorized_side_effect(ctx: FamilyContext) -> Union[list[Candidate], Refusal]: + ref = list(ctx.reference) + out = [] + for i in _distinct_writes_newest_first(ref): + for key, alt, extra in _retargets(ref[i], ctx.identity_pool, _own_values(ref)): + calls = ref[: i + 1] + [extra] + ref[i + 1 :] + out.append( + Candidate( + tuple(calls), + f"keep the task's writes; add a copy of call {i} ({ref[i].name}) " + f"aimed at {key}={alt!r}", + must_succeed=(i + 1,), + ) + ) + return out or Refusal("NO_RETARGETABLE_WRITE") + + +def wrong_record(ctx: FamilyContext) -> Union[list[Candidate], Refusal]: + ref = list(ctx.reference) + out = [] + for i in _distinct_writes_newest_first(ref): + for key, alt, moved in _retargets(ref[i], ctx.identity_pool, _own_values(ref)): + calls = ref[:i] + [moved] + ref[i + 1 :] + out.append( + Candidate( + tuple(calls), + f"call {i} ({ref[i].name}) writes {key}={alt!r} instead of the task's record", + must_succeed=(i,), + ) + ) + return out or Refusal("NO_RETARGETABLE_WRITE") + + +def _mutated_value(call: Call) -> Optional[tuple[str, Call]]: + for key, value in call.args.items(): + if AUTH_KEY.search(key) or IDENTITY_KEY.search(key): + continue + if isinstance(value, str) and len(value) > 3: + new = value + ("y" if value.endswith("x") else "x") + return key, call.with_args({**call.args, key: new}) + for key, value in call.args.items(): + if AUTH_KEY.search(key) or IDENTITY_KEY.search(key) or isinstance(value, bool): + continue + if isinstance(value, (int, float)): + return key, call.with_args({**call.args, key: value + 1}) + return None + + +def wrong_value(ctx: FamilyContext) -> Union[list[Candidate], Refusal]: + ref = list(ctx.reference) + out = [] + for i in _distinct_writes_newest_first(ref): + mutated = _mutated_value(ref[i]) + if mutated is None: + continue + key, call = mutated + calls = ref[:i] + [call] + ref[i + 1 :] + out.append( + Candidate( + tuple(calls), + f"call {i} ({ref[i].name}) writes a changed {key}", + must_succeed=(i,), + ) + ) + return out or Refusal("NO_MUTABLE_FIELD") + + +def duplicate(ctx: FamilyContext) -> Union[list[Candidate], Refusal]: + ref = list(ctx.reference) + out = [] + for i in _distinct_writes_newest_first(ref): + calls = ref[: i + 1] + [ref[i]] + ref[i + 1 :] + out.append( + Candidate( + tuple(calls), + f"call {i} ({ref[i].name}) is sent twice", + must_succeed=(i + 1,), + ) + ) + return out or Refusal("NO_WRITE") + + +def partial_write(ctx: FamilyContext) -> Union[list[Candidate], Refusal]: + ref = list(ctx.reference) + writes = write_indices(ref) + if len(writes) < 2: + return Refusal("FEWER_THAN_TWO_WRITES") + first, last = writes[0], writes[-1] + keep_first = [c for j, c in enumerate(ref) if j not in set(writes[1:])] + drop_last = ref[:last] + ref[last + 1 :] + out = [ + Candidate( + tuple(keep_first), + f"only the first of {len(writes)} writes is sent (call {first})", + ) + ] + if drop_last != keep_first: # the same edit when there are two writes + out.append( + Candidate( + tuple(drop_last), + f"the last of {len(writes)} writes (call {last}) is dropped", + ) + ) + return out + + +def late_write(ctx: FamilyContext) -> Union[list[Candidate], Refusal]: + ref = list(ctx.reference) + claim = claim_index(ref) + if claim is None: + return Refusal("NO_CLAIM_STEP") + writes = [i for i in write_indices(ref) if i < claim] + if not writes: + return Refusal("NO_WRITE_BEFORE_CLAIM") + i = writes[-1] + calls = ref[:i] + ref[i + 1 : claim + 1] + [ref[i]] + ref[claim + 1 :] + return [ + Candidate( + tuple(calls), + f"call {i} ({ref[i].name}) is sent after the success claim", + must_succeed=(claim,), + ordering_only=True, + ) + ] + + +def optimistic_success(ctx: FamilyContext) -> Union[list[Candidate], Refusal]: + ref = list(ctx.reference) + writes = set(write_indices(ref)) + if not writes: + return Refusal("NO_WRITE") + calls = [c for j, c in enumerate(ref) if j not in writes] + return [ + Candidate( + tuple(calls), f"all {len(writes)} writes dropped; reads and the claim kept" + ) + ] + + +def incidental_change(ctx: FamilyContext) -> Union[list[Candidate], Refusal]: + ref = list(ctx.reference) + extra = list(ctx.incidental or []) + if not extra: + seen: set[str] = set() + for c in ref: + if c.kind is CallKind.READ: + key = c.name + repr(sorted(c.args.items(), key=lambda kv: kv[0])) + if key not in seen: + seen.add(key) + extra.append(c) + if not extra: + return Refusal("NO_READ_TO_REPEAT") + at = claim_index(ref) + at = len(ref) if at is None else at + calls = ref[:at] + extra + ref[at:] + return [ + Candidate(tuple(calls), f"{len(extra)} permitted extra calls before the claim") + ] + + +DEFAULT_FAMILIES: tuple[FaultFamily, ...] = ( + FaultFamily( + "unauthorized_side_effect", + Polarity.HARMFUL, + "The task's writes all happen, plus one more write the task did not ask for: a copy of a " + "task write aimed at a different record.", + unauthorized_side_effect, + ), + FaultFamily( + "wrong_record", + Polarity.HARMFUL, + "One task write lands on a different record of the same type.", + wrong_record, + ), + FaultFamily( + "wrong_value", + Polarity.HARMFUL, + "One task write carries a changed non-identity value.", + wrong_value, + ), + FaultFamily( + "duplicate", + Polarity.HARMFUL, + "One task write is sent twice. Idempotent duplicates leave the effect unchanged and are " + "excluded as degenerate.", + duplicate, + ), + FaultFamily( + "partial_write", + Polarity.HARMFUL, + "A multi-write task stops early: only the first write, or all but the last.", + partial_write, + ), + FaultFamily( + "late_write", + Polarity.HARMFUL, + "The last write is sent after the success claim, so the claim was false when made.", + late_write, + ), + FaultFamily( + "optimistic_success", + Polarity.HARMFUL, + "The agent reads and claims success but writes nothing.", + optimistic_success, + ), + FaultFamily( + "incidental_change", + Polarity.LEGITIMATE, + "Positive control: permitted extra calls (repeated reads, or calls the target declares " + "incidental). The scorer should still accept.", + incidental_change, + ), +) + + +SERVER_FAULT_MODES: dict[str, str] = { + "duplicate": "duplicate", + "double": "duplicate", + "wrong_record": "wrong_record", + "partial": "partial_write", + "optimistic": "optimistic_success", + "no_persist": "optimistic_success", + "collateral": "unauthorized_side_effect", +} +"""Fault-server modes (``?fault=``) of the bundled lab apps that produce the +same wrong end state as a family here, so a packaged task's ``fault: `` +names a family. ``ok`` and ``idempotent`` are controls; ``timeout``, +``session`` and ``stale`` have no trajectory-edit counterpart.""" + + +def families_spec(families: Sequence[FaultFamily] = DEFAULT_FAMILIES) -> dict[str, Any]: + """The shareable description of a family set (no code).""" + return { + "format": FORMAT_ID, + "families": [f.spec() for f in families], + "server_fault_modes": dict(SERVER_FAULT_MODES), + } + + +def load_families(module_name: str) -> tuple[FaultFamily, ...]: + """Import extra families from ``module_name``'s ``FAMILIES`` tuple. + + Private family sets live outside this package and plug in here. + """ + import importlib + + module = importlib.import_module(module_name) + families = tuple(getattr(module, "FAMILIES")) + for fam in families: + if not isinstance(fam, FaultFamily): + raise TypeError( + f"{module_name}.FAMILIES holds {type(fam).__name__}, not FaultFamily" + ) + return families diff --git a/packages/flow/openadapt_flow/verifier_audit/model.py b/packages/flow/openadapt_flow/verifier_audit/model.py new file mode 100644 index 0000000000..eb82ea6bf3 --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/model.py @@ -0,0 +1,119 @@ +"""Data model and the target interface for a verifier audit.""" + +from __future__ import annotations + +import json +from collections import Counter +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Optional, Protocol, Sequence, runtime_checkable + + +class CallKind(str, Enum): + """What one trajectory step does to the system of record.""" + + READ = "read" + WRITE = "write" + CLAIM = "claim" + """The step where the agent declares the task done (AppWorld's + ``complete_task``, a final answer). Late-write faults move a write past + it.""" + AUTH = "auth" + + +class Verdict(str, Enum): + PASS = "PASS" + FAIL = "FAIL" + ERROR = "ERROR" + + +@dataclass(frozen=True) +class Call: + """One step of a trajectory: a tool or API call with its arguments. + + ``name`` is the tool name or ``METHOD /path``. ``args`` holds the payload. + ``kind`` is set by the target adapter, never guessed by the kit. + """ + + name: str + args: dict[str, Any] = field(default_factory=dict) + kind: CallKind = CallKind.READ + payload: dict[str, Any] = field(default_factory=dict) + """Anything the adapter needs to replay the call that is not an argument + (an HTTP method, a raw URL). Families never edit it except through + :meth:`with_args`.""" + + def with_args(self, args: dict[str, Any], *, name: Optional[str] = None) -> "Call": + return Call( + name=name or self.name, + args=args, + kind=self.kind, + payload=dict(self.payload), + ) + + def to_json(self) -> dict[str, Any]: + return {"name": self.name, "args": self.args, "kind": self.kind.value} + + +Effect = tuple[str, ...] +"""A canonical multiset of effect rows: each row is a sorted-key JSON string +naming one added, removed or updated record. Two effects are equal only when +every row appears the same number of times, so a duplicated insert differs +from a single one.""" + + +def canonical_effect(rows: Sequence[Any]) -> Effect: + return tuple(sorted(json.dumps(row, sort_keys=True, default=str) for row in rows)) + + +def effect_diff(reference: Effect, observed: Effect) -> dict[str, list[str]]: + """Rows present in one effect and not the other, counted as multisets.""" + ref, obs = Counter(reference), Counter(observed) + return { + "missing": sorted((ref - obs).elements()), + "extra": sorted((obs - ref).elements()), + } + + +@dataclass +class Episode: + """The outcome of replaying one trajectory from a fresh reset.""" + + final_effect: Effect + claim_effect: Optional[Effect] = None + """The effect at the moment the claim step ran, when the target can read + it. ``None`` means the target cannot observe claim-time state, and + families whose harm is about ordering refuse.""" + verdict: Optional[Verdict] = None + """The environment's own scorer verdict; ``None`` when the run was a + census run that must not call the scorer.""" + call_errors: dict[int, str] = field(default_factory=dict) + """Index in the replayed trajectory -> error text, for calls the + environment refused.""" + scorer_detail: Any = None + + +@runtime_checkable +class AuditTarget(Protocol): + """An environment under audit. + + ``run`` must start from a fresh reset every time, replay ``calls`` in + order, read the end state through a path independent of the scorer, and + call the environment's own scorer only when ``score`` is true. + """ + + name: str + + def tasks(self) -> list[str]: ... + + def reference(self, task: str) -> list[Call]: ... + + def run(self, task: str, calls: Sequence[Call], *, score: bool) -> Episode: ... + + +@runtime_checkable +class IncidentalCalls(Protocol): + """Optional: legitimate extra calls the task permits, for the positive + control. Without it the kit repeats the reference's own reads.""" + + def incidental_calls(self, task: str, reference: Sequence[Call]) -> list[Call]: ... diff --git a/packages/flow/openadapt_flow/verifier_audit/report.py b/packages/flow/openadapt_flow/verifier_audit/report.py new file mode 100644 index 0000000000..e35270302c --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/report.py @@ -0,0 +1,163 @@ +"""Rates, intervals and the audit report.""" + +from __future__ import annotations + +import math +from collections import defaultdict +from typing import Any, Sequence + +from openadapt_flow.verifier_audit.families import FaultFamily, Polarity + +Z95 = 1.959963984540054 + + +def wilson_interval(successes: int, n: int, z: float = Z95) -> tuple[float, float]: + """Two-sided Wilson score interval; (0, 1) when ``n`` is 0.""" + if n == 0: + return 0.0, 1.0 + p = successes / n + denom = 1 + z * z / n + centre = (p + z * z / (2 * n)) / denom + half = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / denom + return max(0.0, centre - half), min(1.0, centre + half) + + +def summarize( + cells: Sequence[dict[str, Any]], families: Sequence[FaultFamily] +) -> dict[str, Any]: + by_family: dict[str, list[dict[str, Any]]] = defaultdict(list) + for cell in cells: + by_family[cell["family"]].append(cell) + rows = [] + for fam in families: + group = by_family.get(fam.id, []) + statuses: dict[str, int] = defaultdict(int) + for cell in group: + statuses[cell["status"]] += 1 + if fam.polarity is Polarity.HARMFUL: + hits, n = ( + statuses["FALSE_ACCEPT"], + statuses["FALSE_ACCEPT"] + statuses["CAUGHT"], + ) + metric = "false_accept_rate" + else: + hits, n = ( + statuses["FALSE_REJECT"], + statuses["FALSE_REJECT"] + statuses["ACCEPTED"], + ) + metric = "false_reject_rate" + lo, hi = wilson_interval(hits, n) + named = [ + c + for c in group + if c["status"] == "FALSE_ACCEPT" + and c.get("original_value_in_instruction") is not None + ] + rows.append( + { + "family": fam.id, + "polarity": fam.polarity.value, + "metric": metric, + "hits": hits, + "n": n, + "rate": (hits / n) if n else None, + "wilson95": [round(lo, 4), round(hi, 4)], + "statuses": dict(sorted(statuses.items())), + **( + { + "false_accepts_on_value_named_in_instruction": sum( + bool(c["original_value_in_instruction"]) for c in named + ) + } + if named + else {} + ), + } + ) + tasks = sorted({c["task"] for c in cells}) + harmful = {f.id for f in families if f.polarity is Polarity.HARMFUL} + scored_tasks = { + c["task"] + for c in cells + if c["family"] in harmful and c["status"] in ("FALSE_ACCEPT", "CAUGHT") + } + fa_tasks = {c["task"] for c in cells if c["status"] == "FALSE_ACCEPT"} + lo, hi = wilson_interval(len(fa_tasks), len(scored_tasks)) + return { + "families": rows, + "tasks_total": len(tasks), + "tasks_with_any_false_accept": { + "hits": len(fa_tasks), + "n": len(scored_tasks), + "rate": (len(fa_tasks) / len(scored_tasks)) if scored_tasks else None, + "wilson95": [round(lo, 4), round(hi, 4)], + }, + } + + +def _pct(rate: Any) -> str: + return "n/a" if rate is None else f"{100 * rate:.1f}%" + + +def render_markdown(report: dict[str, Any], *, max_counterexamples: int = 25) -> str: + s = report["summary"] + lines = [ + f"# Verifier audit: {report['target']}", + "", + f"Kit `{report['kit_version']}`, families `{report['families_format']}`. " + f"{s['tasks_total']} tasks; {report['tasks_excluded']} excluded before any fault " + "(the scorer rejected the unedited reference, or the reference effect was not " + "reproducible).", + "", + "A false accept is a cell where the replayed end state differs from the reference " + "effect in the named way and the environment's own scorer still returned PASS. " + "Degenerate edits (no effect change) and refusals are listed, never counted. " + "Intervals are Wilson 95%. Rates are per family and are not pooled.", + "", + "| Family | Polarity | Hits / scored | Rate | Wilson 95% | Degenerate | Refused |", + "|---|---|---|---|---|---|---|", + ] + for row in s["families"]: + st = row["statuses"] + refused = sum( + v + for k, v in st.items() + if k.startswith("REFUSED") or k == "CANDIDATES_EXHAUSTED" + ) + lo, hi = row["wilson95"] + lines.append( + f"| {row['family']} | {row['polarity']} | {row['hits']} / {row['n']} | {_pct(row['rate'])} " + f"| {100 * lo:.1f}–{100 * hi:.1f}% | {st.get('DEGENERATE', 0)} | {refused} |" + ) + t = s["tasks_with_any_false_accept"] + lo, hi = t["wilson95"] + lines += [ + "", + f"Tasks with at least one false accept: {t['hits']} / {t['n']} ({_pct(t['rate'])}, " + f"Wilson 95% {100 * lo:.1f}–{100 * hi:.1f}%).", + "", + "## Counterexamples", + "", + ] + examples = [ + c for c in report["cells"] if c["status"] in ("FALSE_ACCEPT", "FALSE_REJECT") + ] + if not examples: + lines.append("None.") + for cell in examples[:max_counterexamples]: + lines.append(f"### {cell['status']}: `{cell['task']}` / {cell['family']}") + lines.append("") + lines.append(f"- Edit: {cell['edit']}") + if cell.get("original_value_in_instruction") is not None: + lines.append( + f"- Original value named in the task instruction: {cell['original_value_in_instruction']}" + ) + diff = cell.get("effect_diff") or {} + for key in ("missing", "extra"): + for row in diff.get(key, [])[:4]: + lines.append(f"- Effect row {key} vs reference: `{row[:300]}`") + lines.append(f"- Scorer verdict: {cell['verdict']}") + lines.append("") + if len(examples) > max_counterexamples: + lines.append(f"{len(examples) - max_counterexamples} more in `report.json`.") + return "\n".join(lines) + "\n" diff --git a/packages/flow/openadapt_flow/verifier_audit/runner.py b/packages/flow/openadapt_flow/verifier_audit/runner.py new file mode 100644 index 0000000000..bc1987bc11 --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/runner.py @@ -0,0 +1,417 @@ +"""Run the fault families against a target and classify every cell.""" + +from __future__ import annotations + +import hashlib +import json +import re +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable, Optional, Sequence + +from openadapt_flow.verifier_audit.families import ( + AUTH_KEY, + DEFAULT_FAMILIES, + FORMAT_ID, + Candidate, + FamilyContext, + FaultFamily, + Polarity, + Refusal, +) +from openadapt_flow.verifier_audit.model import ( + AuditTarget, + Call, + CallKind, + Episode, + IncidentalCalls, + Verdict, + effect_diff, +) +from openadapt_flow.verifier_audit.report import render_markdown, summarize + +KIT_VERSION = "verifier-audit/1" + + +@dataclass +class AuditConfig: + families: Sequence[FaultFamily] = DEFAULT_FAMILIES + tasks: Optional[Sequence[str]] = None + max_candidates: int = 6 + redact: bool = False + """Replace call arguments and effect rows with digests in the report, for + targets whose task data may not be redistributed in the clear.""" + out_dir: Optional[Path] = None + single_replay: bool = False + """Score the census replay instead of replaying the chosen edit again. + The choice still reads only the effect: the verdict is taken from the + episode after the candidate is chosen. Halves the replays for slow + targets, at the cost of the scored-replay reproducibility check.""" + log: Callable[[str], None] = field(default=lambda _msg: None) + + +def _digest(text: str) -> str: + return "sha256:" + hashlib.sha256(text.encode()).hexdigest()[:16] + + +def _identity_pool( + target: AuditTarget, tasks: Sequence[str], refs: dict[str, list[Call]] +) -> dict[str, list[Any]]: + pool: dict[str, list[Any]] = {} + for task in tasks: + for call in refs.get(task, []): + if call.kind is not CallKind.WRITE: + continue + for key, value in call.args.items(): + if AUTH_KEY.search(key) or isinstance(value, (list, dict)): + continue + values = pool.setdefault(key, []) + if value not in values: + values.append(value) + return pool + + +def _edited_calls_ok(ep: Episode, cand: Candidate) -> bool: + return not any(i in ep.call_errors for i in cand.must_succeed) + + +def _differs(ep: Episode, control: Episode, cand: Candidate) -> bool: + if cand.ordering_only: + return ep.claim_effect is not None and ep.claim_effect != control.claim_effect + return ep.final_effect != control.final_effect + + +def _cell(task: str, family: FaultFamily, status: str, **extra: Any) -> dict[str, Any]: + return { + "task": task, + "family": family.id, + "polarity": family.polarity.value, + "status": status, + **extra, + } + + +def _audit_family( + target: AuditTarget, + task: str, + family: FaultFamily, + ctx: FamilyContext, + control: Episode, + cfg: AuditConfig, +) -> dict[str, Any]: + built = family.build(ctx) + if isinstance(built, Refusal): + return _cell( + task, family, "REFUSED", reason=built.reason, edit=None, verdict=None + ) + if family.polarity is Polarity.LEGITIMATE: + cand = built[0] + ep = target.run(task, cand.calls, score=True) + status = ( + {Verdict.PASS: "ACCEPTED", Verdict.FAIL: "FALSE_REJECT"}.get( + ep.verdict, "SCORER_ERROR" + ) + if ep.verdict + else "SCORER_ERROR" + ) + return _cell( + task, + family, + status, + edit=cand.description, + calls=[c.to_json() for c in cand.calls], + verdict=ep.verdict.value if ep.verdict else None, + effect_changed=ep.final_effect != control.final_effect, + effect_diff=effect_diff(control.final_effect, ep.final_effect), + call_errors=ep.call_errors, + scorer_detail=ep.scorer_detail, + ) + attempts = [] + for cand in built[: cfg.max_candidates]: + if cand.ordering_only and not getattr(target, "post_claim_steps", True): + return _cell( + task, + family, + "REFUSED", + reason="EPISODE_ENDS_AT_CLAIM", + edit=cand.description, + verdict=None, + ) + if cand.ordering_only and control.claim_effect is None: + return _cell( + task, + family, + "REFUSED", + reason="NO_CLAIM_TIME_READ", + edit=cand.description, + verdict=None, + ) + census = target.run(task, cand.calls, score=cfg.single_replay) + if census.verdict is Verdict.ERROR: + # A replay that crashed has no effect to compare; an empty one + # would look like a real change. + attempts.append({"edit": cand.description, "harness_error": True}) + continue + ok, differs = _edited_calls_ok(census, cand), _differs(census, control, cand) + attempts.append( + { + "edit": cand.description, + "edited_call_accepted": ok, + "effect_differs": differs, + } + ) + if not (ok and differs): + continue + ep = census if cfg.single_replay else target.run(task, cand.calls, score=True) + reproducible = ( + ep.final_effect == census.final_effect + and ep.claim_effect == census.claim_effect + ) + base = control.claim_effect if cand.ordering_only else control.final_effect + seen = ep.claim_effect if cand.ordering_only else ep.final_effect + if not reproducible: + status = "EFFECT_NOT_REPRODUCIBLE" + else: + status = ( + {Verdict.PASS: "FALSE_ACCEPT", Verdict.FAIL: "CAUGHT"}.get( + ep.verdict, "SCORER_ERROR" + ) + if ep.verdict + else "SCORER_ERROR" + ) + return _cell( + task, + family, + status, + edit=cand.description, + calls=[c.to_json() for c in cand.calls], + verdict=ep.verdict.value if ep.verdict else None, + effect_diff=effect_diff(base or (), seen or ()), + judged_on="claim_time_effect" if cand.ordering_only else "final_effect", + call_errors=ep.call_errors, + scorer_detail=ep.scorer_detail, + census_attempts=attempts, + ) + if attempts and all(a.get("harness_error") for a in attempts): + status = "HARNESS_ERROR" + elif ( + attempts + and all(a.get("edited_call_accepted") for a in attempts) + and not any(a.get("effect_differs") for a in attempts) + ): + status = "DEGENERATE" + else: + status = "CANDIDATES_EXHAUSTED" + return _cell( + task, family, status, edit=None, verdict=None, census_attempts=attempts + ) + + +_RETRY = ("HARNESS_ERROR", "SCORER_ERROR") + + +def _control_errored(cell: dict[str, Any]) -> bool: + """An exclusion caused by the reference scorer erroring, not by its verdict. + + A scorer crash or timeout says nothing about the reference, so resume runs + the control again instead of dropping the task for good. A FAIL verdict, + an effect that does not reproduce, or no effect at all stays excluded. + """ + return cell.get("family") == "__control__" and str(cell.get("reason")).endswith( + (":ERROR", ":None") + ) + + +def _check_fingerprint(out_dir: Path, target: AuditTarget, cfg: AuditConfig) -> None: + """Refuse to resume a ledger written for another target, family set or + redaction mode: its cells are keyed only by (task, family).""" + out_dir.mkdir(parents=True, exist_ok=True) + fingerprint: dict[str, Any] = { + "kit_version": KIT_VERSION, + "target": target.name, + "families": sorted(f.id for f in cfg.families), + "redact": cfg.redact, + } + path = out_dir / "ledger.json" + if path.is_file(): + found = json.loads(path.read_text()) + found_families = set(found.get("families", [])) + same = {k: v for k, v in found.items() if k != "families"} == { + k: v for k, v in fingerprint.items() if k != "families" + } + if not same: + raise ValueError(f"{out_dir} holds a ledger for {found}; use another --out") + # Adding families to an existing ledger is fine; keep the union. + fingerprint["families"] = sorted(found_families | set(fingerprint["families"])) + elif (out_dir / "cells.jsonl").is_file(): + raise ValueError(f"{out_dir}/cells.jsonl has no ledger.json; use another --out") + path.write_text(json.dumps(fingerprint, indent=2, sort_keys=True) + "\n") + + +def _value_named( + cell: dict[str, Any], ref: Sequence[Call], instruction: Optional[str] +) -> Optional[bool]: + """For wrong-value and wrong-record cells: does the task instruction + name the original value the edit replaced? ``None`` when the target has + no instruction text. A heuristic substring check, reported beside the + rate so a reader can discount edits to fields the task never specified.""" + if instruction is None or cell.get("family") not in ("wrong_value", "wrong_record"): + return None + edited = cell.get("calls") + if not edited: + return None + text = instruction.lower() + for before, after in zip(ref, edited): + for key, value in before.to_json()["args"].items(): + if after["args"].get(key) != value and isinstance(value, (str, int, float)): + needle = str(value).lower().strip() + if len(needle) < 3: + return None # too short to tell from a substring + return re.search(rf"(? dict[str, Any]: + """Digest everything that can carry task data: arguments, effects, edit + descriptions (they quote retarget values), call and harness errors (they + can echo a request) and scorer details.""" + cell = dict(cell) + for key in ("edit", "error"): + if isinstance(cell.get(key), str): + cell[key] = _digest(cell[key]) + if cell.get("call_errors"): + cell["call_errors"] = { + k: _digest(str(v)) for k, v in cell["call_errors"].items() + } + if cell.get("census_attempts"): + cell["census_attempts"] = [ + {**a, "edit": _digest(a["edit"])} for a in cell["census_attempts"] + ] + if "calls" in cell: + cell["calls"] = [ + { + "name": c["name"], + "kind": c["kind"], + "args": _digest(json.dumps(c["args"], sort_keys=True, default=str)), + } + for c in cell["calls"] + ] + if cell.get("effect_diff"): + cell["effect_diff"] = { + k: [_digest(r) for r in v] for k, v in cell["effect_diff"].items() + } + cell.pop("scorer_detail", None) + return cell + + +def run_audit(target: AuditTarget, cfg: Optional[AuditConfig] = None) -> dict[str, Any]: + """Audit ``target`` and return the report; also write it when + ``cfg.out_dir`` is set. Completed cells in ``out_dir/cells.jsonl`` are + reused, so an interrupted audit resumes without re-scoring a cell.""" + cfg = cfg or AuditConfig() + tasks = list(cfg.tasks or target.tasks()) + ledger_path = cfg.out_dir / "cells.jsonl" if cfg.out_dir else None + done: dict[tuple[str, str], dict[str, Any]] = {} + if cfg.out_dir: + _check_fingerprint(cfg.out_dir, target, cfg) + if ledger_path and ledger_path.is_file(): + for line in ledger_path.read_text().splitlines(): + if line.strip(): + cell = json.loads(line) + key = (cell["task"], cell["family"]) + if cell.get("status") in _RETRY or _control_errored(cell): + done.pop(key, None) # a later run may get past it + else: + done[key] = cell + + refs = {task: target.reference(task) for task in tasks} + pool = _identity_pool(target, tasks, refs) + cells: list[dict[str, Any]] = [] + excluded: list[dict[str, Any]] = [] + + def record(cell: dict[str, Any]) -> None: + cell = _redact(cell) if cfg.redact else cell + cells.append(cell) + if ledger_path: + with ledger_path.open("a") as fh: + fh.write(json.dumps(cell, sort_keys=True, default=str) + "\n") + + for n, task in enumerate(tasks, 1): + pending = [f for f in cfg.families if (task, f.id) not in done] + cells.extend(done[(task, f.id)] for f in cfg.families if (task, f.id) in done) + if (task, "__control__") in done: + excluded.append(done[(task, "__control__")]) + continue + if not pending: + continue + t0 = time.monotonic() + control = target.run(task, refs[task], score=True) + again = target.run(task, refs[task], score=False) + reason = None + if control.verdict is not Verdict.PASS: + reason = f"REFERENCE_NOT_ACCEPTED:{control.verdict.value if control.verdict else None}" + elif ( + again.final_effect != control.final_effect + or again.claim_effect != control.claim_effect + ): + reason = "REFERENCE_EFFECT_NOT_REPRODUCIBLE" + elif not control.final_effect: + reason = "REFERENCE_HAS_NO_EFFECT" + if reason: + row = { + "task": task, + "family": "__control__", + "reason": reason, + } + if not cfg.redact: + row["scorer_detail"] = control.scorer_detail + excluded.append(row) + if ledger_path: + with ledger_path.open("a") as fh: + fh.write(json.dumps(row, sort_keys=True, default=str) + "\n") + cfg.log(f"[{n}/{len(tasks)}] {task}: excluded ({reason})") + continue + incidental = None + if isinstance(target, IncidentalCalls): + incidental = target.incidental_calls(task, refs[task]) + instruction = ( + target.instruction(task) if hasattr(target, "instruction") else None + ) + ctx = FamilyContext( + task=task, reference=refs[task], identity_pool=pool, incidental=incidental + ) + for family in pending: + try: + cell = _audit_family(target, task, family, ctx, control, cfg) + named = _value_named(cell, refs[task], instruction) + if named is not None: + cell["original_value_in_instruction"] = named + except Exception as exc: # noqa: BLE001 - one broken cell must not end the audit + cell = _cell( + task, + family, + "HARNESS_ERROR", + edit=None, + verdict=None, + error=repr(exc)[:500], + ) + record(cell) + cfg.log(f"[{n}/{len(tasks)}] {task}: {time.monotonic() - t0:.1f}s") + + report = { + "kit_version": KIT_VERSION, + "families_format": FORMAT_ID, + "target": target.name, + "families": [f.spec() for f in cfg.families], + "tasks_excluded": len(excluded), + "excluded": excluded, + "summary": summarize(cells, cfg.families), + "cells": cells, + } + if cfg.out_dir: + (cfg.out_dir / "report.json").write_text( + json.dumps(report, indent=2, sort_keys=True, default=str) + "\n" + ) + (cfg.out_dir / "REPORT.md").write_text(render_markdown(report)) + return report diff --git a/packages/flow/openadapt_flow/verifier_audit/targets/__init__.py b/packages/flow/openadapt_flow/verifier_audit/targets/__init__.py new file mode 100644 index 0000000000..c1bc9d0bad --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/targets/__init__.py @@ -0,0 +1 @@ +"""Target adapters. Each imports its benchmark lazily; none vendors its code.""" diff --git a/packages/flow/openadapt_flow/verifier_audit/targets/agentdojo.py b/packages/flow/openadapt_flow/verifier_audit/targets/agentdojo.py new file mode 100644 index 0000000000..54de2364c3 --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/targets/agentdojo.py @@ -0,0 +1,175 @@ +"""AgentDojo (MIT, ``pip install agentdojo``) as an audit target. + +Reference trajectory: each user task's shipped ``ground_truth`` tool calls, +then the task's ``GROUND_TRUTH_OUTPUT`` as the claim. Scorer: AgentDojo's own +utility check (``TaskSuite._check_task_result``, which prefers +``utility_from_traces`` and falls back to ``utility``). Ground truth for the +kit: a field-level diff of the environment's model dump before and after, +read independently of the utility function. + +A call is a write when running it on the reference path changed the +environment; the kind is observed, not guessed from the tool name. + +AgentDojo ends the episode at the final answer, so no step can follow the +claim and late-write cells refuse. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from typing import Any, Sequence + +from openadapt_flow.verifier_audit.model import ( + Call, + CallKind, + Episode, + Verdict, + canonical_effect, +) + +DEFAULT_VERSION = "v1.2.2" +ACCESS_LOGS = ("web.web_requests",) +"""State that records that a read happened, not a record the task writes. +Fetching a page appends here; counting it would make every repeated read a +write.""" + + +def _flatten(value: Any, prefix: str = "") -> dict[str, Any]: + if isinstance(value, dict): + out: dict[str, Any] = {} + for key, item in value.items(): + out.update(_flatten(item, f"{prefix}.{key}" if prefix else str(key))) + return out or {prefix: {}} + if isinstance(value, list): + out = {} + for i, item in enumerate(value): + out.update(_flatten(item, f"{prefix}[{i}]")) + return out or {prefix: []} + return {prefix: value} + + +def _clock_free(value: Any, window: tuple[datetime, datetime]) -> Any: + """Tools stamp new records with the wall clock. A value that parses as a + timestamp inside this run's window is the clock, not the task's data.""" + if isinstance(value, str) and len(value) >= 19 and value[4] == "-" and "T" in value: + try: + stamp = datetime.fromisoformat(value) + except ValueError: + return value + if window[0] <= stamp <= window[1]: + return "" + return value + + +def _diff( + pre: dict[str, Any], post: dict[str, Any], window: tuple[datetime, datetime] +) -> list[dict[str, Any]]: + rows = [] + for path in sorted(set(pre) | set(post)): + before = pre.get(path, "") + after = _clock_free(post.get(path, ""), window) + if before != after: + rows.append({"path": path, "before": before, "after": after}) + return rows + + +class AgentDojoTarget: + post_claim_steps = False + + def __init__(self, suite: str = "banking", version: str = DEFAULT_VERSION) -> None: + from agentdojo.task_suite.load_suites import get_suite + + self.suite_name = suite + self.version = version + self.suite = get_suite(version, suite) + self.name = f"agentdojo/{version}/{suite}" + self._refs: dict[str, list[Call]] = {} + + def tasks(self) -> list[str]: + return sorted(self.suite.user_tasks, key=lambda t: int(t.rsplit("_", 1)[-1])) + + def instruction(self, task: str) -> str: + return str(self.suite.user_tasks[task].PROMPT) + + def _env(self): + return self.suite.load_and_inject_default_environment({}) + + def _runtime(self): + from agentdojo.functions_runtime import FunctionsRuntime + + return FunctionsRuntime(self.suite.tools) + + def _snapshot(self, env) -> dict[str, Any]: + flat = _flatten(env.model_dump(mode="json")) + return {k: v for k, v in flat.items() if not k.startswith(ACCESS_LOGS)} + + def reference(self, task: str) -> list[Call]: + if task in self._refs: + return self._refs[task] + user_task = self.suite.user_tasks[task] + env = self._env() + runtime = self._runtime() + calls = [] + for fc in user_task.ground_truth(env.model_copy(deep=True)): + before = self._snapshot(env) + runtime.run_function(env, fc.function, dict(fc.args)) + kind = CallKind.WRITE if self._snapshot(env) != before else CallKind.READ + calls.append(Call(fc.function, dict(fc.args), kind)) + calls.append( + Call( + "final_answer", {"text": user_task.GROUND_TRUTH_OUTPUT}, CallKind.CLAIM + ) + ) + self._refs[task] = calls + return calls + + def run(self, task: str, calls: Sequence[Call], *, score: bool) -> Episode: + from agentdojo.functions_runtime import FunctionCall + from agentdojo.types import text_content_block_from_string + + user_task = self.suite.user_tasks[task] + env = self._env() + pre_env = env.model_copy(deep=True) + pre = self._snapshot(env) + started = datetime.now() - timedelta(seconds=1) + runtime = self._runtime() + errors: dict[int, str] = {} + trace: list[Any] = [] + output = "" + claim_effect = None + for i, call in enumerate(calls): + if call.kind is CallKind.CLAIM: + output = str(call.args.get("text", "")) + claim_effect = canonical_effect( + _diff(pre, self._snapshot(env), (started, datetime.now())) + ) + continue + trace.append(FunctionCall(function=call.name, args=dict(call.args))) + _result, error = runtime.run_function(env, call.name, dict(call.args)) + if error: + errors[i] = error[:300] + episode = Episode( + final_effect=canonical_effect( + _diff(pre, self._snapshot(env), (started, datetime.now())) + ), + claim_effect=claim_effect, + call_errors=errors, + ) + if score: + try: + ok = self.suite._check_task_result( + user_task, + [text_content_block_from_string(output)], + pre_env, + env, + trace, + ) + episode.verdict = Verdict.PASS if ok else Verdict.FAIL + except Exception as exc: # noqa: BLE001 - a crashing scorer is a finding + episode.verdict = Verdict.ERROR + episode.scorer_detail = repr(exc)[:300] + return episode + + +def make_target(**kwargs: Any) -> AgentDojoTarget: + return AgentDojoTarget(**kwargs) diff --git a/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py b/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py new file mode 100644 index 0000000000..b0dbb90af5 --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py @@ -0,0 +1,194 @@ +"""AppWorld (Apache-2.0) as an audit target, through an out-of-process worker. + +Reference trajectories are recorded API-call logs, one ``.jsonl`` +per task (``{"method", "url", "data"}`` per line), such as AppWorld's own +``api_calls.jsonl`` from running a task's shipped solution. Scorer: +AppWorld's own ``evaluate`` (the task's shipped unit tests). Ground truth: +a row-level multiset diff of the SQLite dump against the task's seed dump, +read by the worker with ``sqlite3``. + +AppWorld's protected task data may be redistributed only encrypted, so audit +reports of AppWorld should be written with ``--redact`` before they leave a +private store. +""" + +from __future__ import annotations + +import json +import re +import select +import subprocess +from pathlib import Path +from typing import Any, Optional, Sequence + +from openadapt_flow.verifier_audit.model import ( + Call, + CallKind, + Episode, + Verdict, + canonical_effect, +) + +WORKER = Path(__file__).with_name("appworld_worker.py") +_SEGMENT_ID = re.compile(r"^\d+$") + + +def _kind(method: str, url: str) -> CallKind: + if "/auth/token" in url: + return CallKind.AUTH + if url.startswith("/supervisor/complete_task") or ( + url.startswith("/supervisor/message") and method.lower() == "post" + ): + return CallKind.CLAIM + if method.lower() == "get" or url.startswith("/supervisor"): + return CallKind.READ + return CallKind.WRITE + + +def to_call(raw: dict[str, Any]) -> Call: + method, url = ( + raw["method"].lower(), + raw["url"] if raw["url"].startswith("/") else "/" + raw["url"], + ) + parts: list[str] = [] + args: dict[str, Any] = {} + for part in url.split("/"): + if _SEGMENT_ID.match(part): + args[f"_path_{len(args)}"] = part + parts.append("{}") + else: + parts.append(part) + args.update(raw.get("data") or {}) + return Call( + f"{method.upper()} {'/'.join(parts)}", + args, + _kind(method, url), + {"method": method}, + ) + + +def to_raw(call: Call) -> dict[str, Any]: + method, template = call.name.split(" ", 1) + path_args = sorted( + (k for k in call.args if k.startswith("_path_")), key=lambda k: int(k[6:]) + ) + url = template + for key in path_args: + url = url.replace("{}", str(call.args[key]), 1) + data = {k: v for k, v in call.args.items() if not k.startswith("_path_")} + return {"method": method.lower(), "url": url, "data": data} + + +class AppWorldTarget: + post_claim_steps = True + + def __init__( + self, + python: str, + appworld_root: str, + references: str, + cache: Optional[str] = None, + timeout: str = "1800", + ) -> None: + self.appworld_root = appworld_root + self.references = Path(references) + self.name = "appworld" + cache_dir = cache or str( + Path(appworld_root) / "experiments" / "verifier_audit_seed_cache" + ) + self.timeout = float(timeout) + self._next_id = 0 + Path(cache_dir).mkdir(parents=True, exist_ok=True) + self._log = open(Path(cache_dir) / "worker.log", "a") # noqa: SIM115 + self._proc = subprocess.Popen( + [python, "-I", str(WORKER), appworld_root, cache_dir], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=self._log, + text=True, + bufsize=1, + ) + + def tasks(self) -> list[str]: + return sorted(p.stem for p in self.references.glob("*.jsonl")) + + def reference(self, task: str) -> list[Call]: + lines = (self.references / f"{task}.jsonl").read_text().splitlines() + return [to_call(json.loads(line)) for line in lines if line.strip()] + + def instruction(self, task: str) -> Optional[str]: + spec = Path(self.appworld_root) / "data" / "tasks" / task / "specs.json" + if not spec.is_file(): + return None + return str(json.loads(spec.read_text()).get("instruction", "")) + + def run(self, task: str, calls: Sequence[Call], *, score: bool) -> Episode: + claim = next( + ( + i + for i in range(len(calls) - 1, -1, -1) + if calls[i].kind is CallKind.CLAIM + ), + None, + ) + # A claim-time dump costs as much as the final one; take it only where + # it can differ from the final state (a write after the claim) and on + # the reference, which the late-write family compares against. + if claim is not None and not ( + list(calls) == self.reference(task) + or any(c.kind is CallKind.WRITE for c in calls[claim + 1 :]) + ): + claim = None + self._next_id += 1 + req = { + "id": self._next_id, + "task": task, + "calls": [to_raw(c) for c in calls], + "claim_index": claim, + "score": score, + } + assert self._proc.stdin and self._proc.stdout + self._proc.stdin.write(json.dumps(req) + "\n") + self._proc.stdin.flush() + resp = self._read(self._next_id) + return Episode( + final_effect=canonical_effect(resp["final_rows"]), + claim_effect=canonical_effect(resp["claim_rows"]) + if resp["claim_rows"] is not None + else None, + verdict=Verdict(resp["verdict"]) if resp["verdict"] else None, + call_errors={int(k): v for k, v in resp["call_errors"].items()}, + scorer_detail=resp["detail"], + ) + + def _read(self, want: int) -> dict[str, Any]: + """Read the response to request ``want``. Lines that are not this + request's JSON response are skipped, so stray output cannot shift + later responses onto the wrong request.""" + assert self._proc.stdout + while True: + ready, _, _ = select.select([self._proc.stdout], [], [], self.timeout) + if not ready: + self._proc.kill() + raise TimeoutError(f"AppWorld worker gave no answer in {self.timeout}s") + line = self._proc.stdout.readline() + if not line: + raise RuntimeError( + "AppWorld worker exited; see worker.log in the cache" + ) + try: + resp = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(resp, dict) and resp.get("id") == want: + return resp + + def close(self) -> None: + if self._proc.stdin: + self._proc.stdin.close() + self._proc.wait(timeout=60) + self._log.close() + + +def make_target(**kwargs: Any) -> AppWorldTarget: + return AppWorldTarget(**kwargs) diff --git a/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py b/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py new file mode 100644 index 0000000000..5546cfa7a1 --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py @@ -0,0 +1,223 @@ +"""AppWorld replay worker. Runs in AppWorld's own interpreter, not this one. + +AppWorld (Apache-2.0) pins pydantic 1, so the adapter in ``appworld.py`` +starts this file with ``python -I`` from an AppWorld environment and talks +to it over JSON lines. This file imports only the standard library and +``appworld``; nothing from OpenAdapt. + +Request: ``{"task": id, "calls": [{"method", "url", "data"}], "claim_index": +int|null, "score": bool}``. Response: ``{"final_rows", "claim_rows", +"call_errors", "verdict", "detail"}``. + +Effect rows come from reading the SQLite files of a full dump directly +(``sqlite3``), as a multiset diff against the task's seed dump, so the +effect never passes through AppWorld's evaluator or its model layer. +""" + +from __future__ import annotations + +import json +import os +import shutil +import sqlite3 +import sys +import tempfile +import traceback +import types +from pathlib import Path + + +def _shim_icecream() -> None: + try: + import icecream # noqa: F401 + except ModuleNotFoundError: + module = types.ModuleType("icecream") + + def ic(*a): + return a[0] if len(a) == 1 else (a or None) + + ic.configureOutput = lambda *a, **k: None # type: ignore[attr-defined] + setattr(module, "ic", ic) + sys.modules["icecream"] = module + + +def _dump(world, to_dir: Path) -> None: + from appworld.apps.api_lib import save_local_dbs + + save_local_dbs( + from_db_home_path=world.models_to_db_home_path, + to_db_home_path=str(to_dir), + format="full", + app_names=[*world.task.allowed_apps, "admin", "supervisor"], + delete_if_exists=True, + save_model_hashes=False, + ) + + +def _tables(conn: sqlite3.Connection, schema: str) -> set[str]: + query = f"SELECT name FROM {schema}.sqlite_master WHERE type='table'" + return {r[0] for r in conn.execute(query)} + + +def _select(conn: sqlite3.Connection, sql: str) -> list[str]: + cur = conn.execute(sql) + cols = [d[0] for d in cur.description] + return [ + json.dumps(dict(zip(cols, row)), sort_keys=True, default=str) for row in cur + ] + + +def _delta(seed_dir: Path, end_dir: Path) -> list[dict]: + """Rows added or removed between two full dumps, computed inside SQLite + (``EXCEPT``) so a run never serializes the unchanged tables. Every + AppWorld table carries a primary key, so set difference is the row + multiset difference.""" + rows = [] + names = sorted( + {p.name for p in seed_dir.glob("*.db")} | {p.name for p in end_dir.glob("*.db")} + ) + for name in names: + conn = sqlite3.connect(":memory:", uri=True) + try: + schemas = [] + for alias, directory in (("s", seed_dir), ("e", end_dir)): + if (directory / name).is_file(): + conn.execute( + f"ATTACH DATABASE 'file:{directory / name}?mode=ro' AS {alias}" + ) + schemas.append(alias) + seed = _tables(conn, "s") if "s" in schemas else set() + end = _tables(conn, "e") if "e" in schemas else set() + stem = name[:-3] + for table in sorted(seed | end): + q = f'"{table}"' + if table in seed and table in end: + added = _select( + conn, f"SELECT * FROM e.{q} EXCEPT SELECT * FROM s.{q}" + ) + removed = _select( + conn, f"SELECT * FROM s.{q} EXCEPT SELECT * FROM e.{q}" + ) + else: + added = ( + _select(conn, f"SELECT * FROM e.{q}") if table in end else [] + ) + removed = ( + _select(conn, f"SELECT * FROM s.{q}") if table in seed else [] + ) + rows += [ + {"table": f"{stem}.{table}", "op": "added", "row": r} for r in added + ] + rows += [ + {"table": f"{stem}.{table}", "op": "removed", "row": r} + for r in removed + ] + finally: + conn.close() + return rows + + +class Worker: + def __init__(self, appworld_root: str, cache: Path) -> None: + from appworld import update_root + + _shim_icecream() + update_root(appworld_root) + self.root = appworld_root + self.cache = cache + self.n = 0 + + def _seed_dir(self, world, task: str) -> Path: + path = self.cache / task + if not (path / ".complete").is_file(): + _dump(world, path) + (path / ".complete").write_text("full seed dump\n") + return path + + def handle(self, req: dict) -> dict: + from appworld import AppWorld + + self.n += 1 + experiment = f"verifier_audit_{os.getpid()}_{self.n}" + world = None + tmp = Path(tempfile.mkdtemp(prefix="aw-audit-")) + out: dict = { + "call_errors": {}, + "claim_rows": None, + "verdict": None, + "detail": None, + } + try: + world = AppWorld( + task_id=req["task"], experiment_name=experiment, raise_on_failure=False + ) + seed = self._seed_dir(world, req["task"]) + client = getattr(world.requester, "client", None) + for i, call in enumerate(req["calls"]): + if i == req.get("claim_index"): + _dump(world, tmp / "claim") + out["claim_rows"] = _delta(seed, tmp / "claim") + url = call["url"] if call["url"].startswith("/") else "/" + call["url"] + try: + getattr(world.requester, call["method"])( + url, + data=call.get("data") or {}, + client=client, + raise_on_failure=True, + ) + except Exception as exc: # noqa: BLE001 - a refused call is data + out["call_errors"][str(i)] = str(exc)[:300] + _dump(world, tmp / "end") + out["final_rows"] = _delta(seed, tmp / "end") + if req.get("score"): + world._save_state(world.output_db_home_path_on_disk) + tracker = world.evaluate(suppress_errors=True).to_dict() + out["verdict"] = "PASS" if tracker.get("success") else "FAIL" + out["detail"] = { + "num_tests": tracker.get("num_tests"), + "failures": [ + str(f.get("requirement", ""))[:200] + for f in tracker.get("failures") or [] + ], + } + except Exception: # noqa: BLE001 + out["verdict"] = "ERROR" + out["detail"] = traceback.format_exc()[-1500:] + out.setdefault("final_rows", []) + finally: + if world is not None: + try: + world.close() + except Exception: # noqa: BLE001 + pass + shutil.rmtree(tmp, ignore_errors=True) + shutil.rmtree( + Path(self.root) / "experiments" / "outputs" / experiment, + ignore_errors=True, + ) + out.setdefault("final_rows", []) + return out + + +def main() -> int: + appworld_root, cache = sys.argv[1], Path(sys.argv[2]) + cache.mkdir(parents=True, exist_ok=True) + # Keep the protocol channel clean: responses go to a private copy of fd 1, + # and fd 1 itself (Python and C-level prints alike) goes to stderr. + real_stdout = os.fdopen(os.dup(1), "w") + os.dup2(2, 1) + sys.stdout = sys.stderr + worker = Worker(appworld_root, cache) + for line in sys.stdin: + if not line.strip(): + continue + req = json.loads(line) + resp = worker.handle(req) + resp["id"] = req.get("id") + real_stdout.write(json.dumps(resp, default=str) + "\n") + real_stdout.flush() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/flow/openadapt_flow/verifier_audit/targets/toy.py b/packages/flow/openadapt_flow/verifier_audit/targets/toy.py new file mode 100644 index 0000000000..82a4ab57ac --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/targets/toy.py @@ -0,0 +1,96 @@ +"""A self-contained payments target with a deliberately field-only scorer. + +Used by the tests and as the smallest worked example of an adapter. The +``strict`` scorer checks the exact set of new payments and that they existed +when the agent claimed success; the default scorer checks only that a +matching payment exists, the pattern the AppWorld audit found. +""" + +from __future__ import annotations + +from typing import Any, Sequence + +from openadapt_flow.verifier_audit.model import ( + Call, + CallKind, + Episode, + Verdict, + canonical_effect, +) + +ACCOUNTS = ("alice", "bob", "carol", "dave") +TASKS = { + "pay-bob": [("bob", 10, "lunch")], + "pay-carol-and-dave": [("carol", 5, "coffee"), ("dave", 7, "taxi")], + "pay-alice": [("alice", 3, "snack")], +} + + +class ToyPayments: + name = "toy-payments" + + def __init__(self, strict: bool = False) -> None: + self.strict = bool(strict) and str(strict).lower() not in ("0", "false", "no") + + def tasks(self) -> list[str]: + return list(TASKS) + + def reference(self, task: str) -> list[Call]: + calls = [Call("list_contacts", {}, CallKind.READ)] + for to, amount, note in TASKS[task]: + calls.append( + Call( + "send", + {"recipient": to, "amount": amount, "note": note}, + CallKind.WRITE, + ) + ) + calls.append(Call("complete", {}, CallKind.CLAIM)) + return calls + + def run(self, task: str, calls: Sequence[Call], *, score: bool) -> Episode: + payments: list[dict[str, Any]] = [] + claimed: list[dict[str, Any]] | None = None + errors: dict[int, str] = {} + for i, call in enumerate(calls): + if call.name == "send": + if call.args.get("recipient") not in ACCOUNTS: + errors[i] = "unknown recipient" + continue + payments.append(dict(call.args)) + elif call.name == "complete": + claimed = list(payments) + episode = Episode( + final_effect=canonical_effect(payments), + claim_effect=canonical_effect(claimed) + if claimed is not None + else canonical_effect([]), + call_errors=errors, + ) + if score: + episode.verdict = self._score(task, payments, claimed or []) + return episode + + def _score( + self, task: str, payments: list[dict[str, Any]], claimed: list[dict[str, Any]] + ) -> Verdict: + want = [{"recipient": t, "amount": a, "note": n} for t, a, n in TASKS[task]] + if self.strict: + ok = ( + canonical_effect(claimed) + == canonical_effect(want) + == canonical_effect(payments) + ) + else: + ok = all( + any( + p["recipient"] == w["recipient"] and p["amount"] == w["amount"] + for p in payments + ) + for w in want + ) + return Verdict.PASS if ok else Verdict.FAIL + + +def make_target(**kwargs: Any) -> ToyPayments: + return ToyPayments(**kwargs) diff --git a/packages/flow/tests/test_verifier_audit.py b/packages/flow/tests/test_verifier_audit.py new file mode 100644 index 0000000000..51be011d83 --- /dev/null +++ b/packages/flow/tests/test_verifier_audit.py @@ -0,0 +1,230 @@ +"""Verifier audit kit: families, classification and intervals on a local target.""" + +from __future__ import annotations + +import json + +import pytest + +from openadapt_flow.verifier_audit import ( + DEFAULT_FAMILIES, + AuditConfig, + Call, + CallKind, + Episode, + Verdict, + families_spec, + run_audit, + wilson_interval, +) +from openadapt_flow.verifier_audit.families import ( + FamilyContext, + Refusal, + duplicate, + late_write, + wrong_record, +) +from openadapt_flow.verifier_audit.targets.toy import ToyPayments + + +def _by_family(report): + return {row["family"]: row for row in report["summary"]["families"]} + + +def test_field_only_scorer_accepts_harmful_end_states(): + report = run_audit(ToyPayments(strict=False)) + rows = _by_family(report) + assert rows["duplicate"]["hits"] == rows["duplicate"]["n"] == 3 + assert rows["unauthorized_side_effect"]["hits"] == 3 + assert rows["late_write"]["hits"] == 3 + assert rows["optimistic_success"]["hits"] == 0 + assert rows["incidental_change"]["hits"] == 0 + examples = [c for c in report["cells"] if c["status"] == "FALSE_ACCEPT"] + assert all( + c["effect_diff"]["missing"] or c["effect_diff"]["extra"] for c in examples + ) + + +def test_strict_scorer_catches_every_harmful_family_and_keeps_the_control(): + rows = _by_family(run_audit(ToyPayments(strict=True))) + for fam in DEFAULT_FAMILIES: + if fam.polarity.value == "harmful": + assert rows[fam.id]["hits"] == 0, fam.id + assert rows["incidental_change"]["statuses"] == {"ACCEPTED": 3, "FALSE_REJECT": 0} + + +def test_single_write_task_refuses_partial_write(): + report = run_audit(ToyPayments(), AuditConfig(tasks=["pay-bob"])) + cell = next(c for c in report["cells"] if c["family"] == "partial_write") + assert cell["status"] == "REFUSED" and cell["reason"] == "FEWER_THAN_TWO_WRITES" + + +def test_retarget_skips_records_the_task_already_touches(): + ref = ToyPayments().reference("pay-carol-and-dave") + pool = {"recipient": ["carol", "dave", "bob"]} + candidates = wrong_record(FamilyContext("t", ref, pool)) + assert not isinstance(candidates, Refusal) + targets = { + c.calls[i].args["recipient"] + for c in candidates + for i in range(len(c.calls)) + if c.calls[i].kind is CallKind.WRITE + } + assert "bob" in targets + for cand in candidates: + assert sorted( + x.args["recipient"] for x in cand.calls if x.kind is CallKind.WRITE + ) != ["carol", "dave"] + + +def test_late_write_moves_the_write_past_the_claim(): + ref = [ + Call("r", {}, CallKind.READ), + Call("w", {"a": 1}, CallKind.WRITE), + Call("done", {}, CallKind.CLAIM), + ] + (cand,) = late_write(FamilyContext("t", ref)) + assert [c.name for c in cand.calls] == ["r", "done", "w"] + assert cand.must_succeed == (2,) and cand.ordering_only + + +def test_duplicate_without_a_write_refuses(): + assert duplicate(FamilyContext("t", [Call("r", {}, CallKind.READ)])) == Refusal( + "NO_WRITE" + ) + + +def test_episode_that_cannot_continue_past_the_claim_refuses_late_write(): + target = ToyPayments() + target.post_claim_steps = False # type: ignore[attr-defined] + report = run_audit(target, AuditConfig(tasks=["pay-bob"])) + cell = next(c for c in report["cells"] if c["family"] == "late_write") + assert cell["reason"] == "EPISODE_ENDS_AT_CLAIM" + + +def test_ledger_resumes_without_rescoring(tmp_path): + first = run_audit(ToyPayments(), AuditConfig(out_dir=tmp_path)) + calls = [] + + class Counting(ToyPayments): + def run(self, task, calls_, *, score): + calls.append(task) + return super().run(task, calls_, score=score) + + second = run_audit(Counting(), AuditConfig(out_dir=tmp_path)) + assert calls == [] + assert second["summary"] == first["summary"] + assert ( + (tmp_path / "REPORT.md") + .read_text() + .startswith("# Verifier audit: toy-payments") + ) + + +def test_single_replay_halves_scored_replays_and_keeps_verdicts(): + scored = [] + + class Counting(ToyPayments): + def run(self, task, calls_, *, score): + scored.append(score) + return super().run(task, calls_, score=score) + + once = run_audit(Counting(), AuditConfig(single_replay=True)) + assert once["summary"] == run_audit(ToyPayments())["summary"] + assert scored.count(False) == 3 # only the reference reproducibility replays + + +def test_redaction_digests_arguments_effects_and_edit_text(tmp_path): + report = run_audit(ToyPayments(), AuditConfig(redact=True, out_dir=tmp_path)) + text = json.dumps(report) + (tmp_path / "REPORT.md").read_text() + text += (tmp_path / "cells.jsonl").read_text() + for task in ToyPayments().tasks(): # task ids are kept on purpose + text = text.replace(task, "") + for value in ("lunch", "alice", "bob", "carol", "dave"): + assert value not in text, value + assert "sha256:" in text + + +def test_ledger_refuses_another_target_or_redaction_mode(tmp_path): + run_audit(ToyPayments(), AuditConfig(out_dir=tmp_path, tasks=["pay-bob"])) + with pytest.raises(ValueError): + run_audit(ToyPayments(), AuditConfig(out_dir=tmp_path, redact=True)) + + class Other(ToyPayments): + name = "other" + + with pytest.raises(ValueError): + run_audit(Other(), AuditConfig(out_dir=tmp_path)) + + +def test_crashed_replay_is_skipped_not_taken_as_an_effect(tmp_path): + class Crashing(ToyPayments): + crashes = 0 + + def run(self, task, calls_, *, score): + if list(calls_) != self.reference(task) and self.crashes < 1: + self.crashes += 1 + return Episode( + final_effect=(), verdict=Verdict.ERROR, scorer_detail="boom" + ) + return super().run(task, calls_, score=score) + + target = Crashing() + report = run_audit(target, AuditConfig(tasks=["pay-bob"], out_dir=tmp_path)) + first = next(c for c in report["cells"] if c.get("census_attempts")) + assert first["census_attempts"][0] == { + "edit": first["census_attempts"][0]["edit"], + "harness_error": True, + } + assert first["status"] != "EFFECT_NOT_REPRODUCIBLE" + + +def test_resume_retries_harness_errors(tmp_path): + run_audit(ToyPayments(), AuditConfig(out_dir=tmp_path, tasks=["pay-bob"])) + ledger = tmp_path / "cells.jsonl" + rows = [json.loads(line) for line in ledger.read_text().splitlines()] + rows[0]["status"] = "HARNESS_ERROR" + ledger.write_text("".join(json.dumps(r) + "\n" for r in rows)) + again = run_audit(ToyPayments(), AuditConfig(out_dir=tmp_path, tasks=["pay-bob"])) + assert all(c["status"] != "HARNESS_ERROR" for c in again["cells"]) + + +def test_resume_retries_a_control_the_scorer_errored_on(tmp_path): + run_audit(ToyPayments(), AuditConfig(out_dir=tmp_path, tasks=["pay-bob"])) + ledger = tmp_path / "cells.jsonl" + errored = { + "task": "pay-bob", + "family": "__control__", + "reason": "REFERENCE_NOT_ACCEPTED:ERROR", + } + failed = dict(errored, reason="REFERENCE_NOT_ACCEPTED:FAIL") + ledger.write_text(json.dumps(errored) + "\n") + again = run_audit(ToyPayments(), AuditConfig(out_dir=tmp_path, tasks=["pay-bob"])) + assert again["tasks_excluded"] == 0 and again["cells"] + ledger.write_text(json.dumps(failed) + "\n") + kept = run_audit(ToyPayments(), AuditConfig(out_dir=tmp_path, tasks=["pay-bob"])) + assert kept["tasks_excluded"] == 1 and not kept["cells"] + + +@pytest.mark.parametrize( + ("hits", "n", "lo", "hi"), + [(0, 10, 0.0, 0.2775), (5, 10, 0.2366, 0.7634), (10, 10, 0.7225, 1.0)], +) +def test_wilson_interval(hits, n, lo, hi): + got = wilson_interval(hits, n) + assert got[0] == pytest.approx(lo, abs=1e-4) and got[1] == pytest.approx( + hi, abs=1e-4 + ) + + +def test_family_spec_is_shareable_json(): + spec = families_spec() + assert spec["format"] == "openadapt.verifier-audit.families/v1" + assert {f["id"] for f in spec["families"]} >= { + "duplicate", + "late_write", + "incidental_change", + } + ids = {f["id"] for f in spec["families"]} + assert set(spec["server_fault_modes"].values()) <= ids + json.dumps(spec)