From 3857013c728e13d0fc33fdb80bdf363daa0a4be5 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 3 Nov 2026 17:29:37 -0500 Subject: [PATCH 1/5] feat(flow): verifier audit, a fault-injection test of an environment's own scorer `python -m openadapt_flow.verifier_audit run --target module:factory --out DIR` replays each task's reference trajectory with mechanical faults and asks the environment's own scorer for a verdict. Families: unauthorized side effect, wrong record, wrong value, duplicate, partial write, late write (after the success claim), optimistic success, and an incidental-change positive control. Ground truth never comes from the scorer. The target reads the end state through its own path, and a fault whose effect matches the reference is degenerate and excluded. A family never sees a verdict; the runner picks the first candidate whose replayed effect differs, then scores it. The report gives per-family false-accept and false-reject rates with Wilson 95% intervals and counterexamples with effect diffs; an append-only ledger makes reruns resume. Adapters: a self-contained toy target used by the tests, AgentDojo (in process) and AppWorld (a worker in AppWorld's own interpreter, which pins pydantic 1). Both benchmarks are imported at run time; no benchmark code is copied. Private family sets plug in with --families-module. This generalizes the method of "Finding Blind Spots in AppWorld and WorkArena Task Verifiers" (Abrich, 2026). --- packages/flow/CHANGELOG.md | 2 + packages/flow/docs/VERIFIER_AUDIT.md | 109 +++++ .../openadapt_flow/verifier_audit/__init__.py | 57 +++ .../openadapt_flow/verifier_audit/__main__.py | 88 ++++ .../openadapt_flow/verifier_audit/families.py | 400 ++++++++++++++++++ .../openadapt_flow/verifier_audit/model.py | 119 ++++++ .../openadapt_flow/verifier_audit/report.py | 163 +++++++ .../openadapt_flow/verifier_audit/runner.py | 342 +++++++++++++++ .../verifier_audit/targets/__init__.py | 1 + .../verifier_audit/targets/agentdojo.py | 175 ++++++++ .../verifier_audit/targets/appworld.py | 165 ++++++++ .../verifier_audit/targets/appworld_worker.py | 214 ++++++++++ .../verifier_audit/targets/toy.py | 96 +++++ packages/flow/tests/test_verifier_audit.py | 160 +++++++ 14 files changed, 2091 insertions(+) create mode 100644 packages/flow/docs/VERIFIER_AUDIT.md create mode 100644 packages/flow/openadapt_flow/verifier_audit/__init__.py create mode 100644 packages/flow/openadapt_flow/verifier_audit/__main__.py create mode 100644 packages/flow/openadapt_flow/verifier_audit/families.py create mode 100644 packages/flow/openadapt_flow/verifier_audit/model.py create mode 100644 packages/flow/openadapt_flow/verifier_audit/report.py create mode 100644 packages/flow/openadapt_flow/verifier_audit/runner.py create mode 100644 packages/flow/openadapt_flow/verifier_audit/targets/__init__.py create mode 100644 packages/flow/openadapt_flow/verifier_audit/targets/agentdojo.py create mode 100644 packages/flow/openadapt_flow/verifier_audit/targets/appworld.py create mode 100644 packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py create mode 100644 packages/flow/openadapt_flow/verifier_audit/targets/toy.py create mode 100644 packages/flow/tests/test_verifier_audit.py diff --git a/packages/flow/CHANGELOG.md b/packages/flow/CHANGELOG.md index ec21cf1c0c..975b5c2426 100644 --- a/packages/flow/CHANGELOG.md +++ b/packages/flow/CHANGELOG.md @@ -19,6 +19,8 @@ ### 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`. + - `run --self-repair` (and repair by default) now works outside a browser. `repair.session` in `deploy.yaml` says how to start the app for each probe, fault probe and resumed run (an argv, no shell, kept running while the app is up; `fault_commands`, `close_command`, `ready_s`). The run's configured backend must bind to a window owned by that command's process group, so a session never binds to the halted session or a copy already open; Flow ends the group afterwards, and closes the halted session before each probe. This works where the bound window is on Flow's machine (macOS, and the local RDP or Citrix client window); Windows over its agent and FreeRDP are refused for now. Without `repair.session`, a desktop run refuses repair with that reason, as before. - Signed write receipts. A verifier verdict, with its scope (stores read, time window, strength) and a plain list of what it did not check, is signed as a COSE_Sign1 in the IETF SCITT shape and appended to a local transparency log. `openadapt-flow receipt verify FILE --issuer-key --log-key` checks one offline and prints the not-checked list every time; a receipt that claims more than its scope is neither issued nor accepted. `verify_write(..., write_receipt=True)`, `issue_write_receipt(result)` and `openadapt-flow shadow --receipts` issue them; `receipt anchor` puts the digest in Rekor. Extra: `openadapt-flow[receipts]`. Design: `docs/WRITE_RECEIPTS.md`. diff --git a/packages/flow/docs/VERIFIER_AUDIT.md b/packages/flow/docs/VERIFIER_AUDIT.md new file mode 100644 index 0000000000..d127356bd2 --- /dev/null +++ b/packages/flow/docs/VERIFIER_AUDIT.md @@ -0,0 +1,109 @@ +# 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), `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 and effect rows with digests. + +## 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..c9ca773c60 --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/families.py @@ -0,0 +1,400 @@ +"""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 :] + return [ + Candidate( + tuple(keep_first), + f"only the first of {len(writes)} writes is sent (call {first})", + ), + Candidate( + tuple(drop_last), + f"the last of {len(writes)} writes (call {last}) is dropped", + ), + ] + + +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, + ), +) + + +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]} + + +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..1ec3b55f48 --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/runner.py @@ -0,0 +1,342 @@ +"""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" + ) + 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) + 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" + ) + 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["edited_call_accepted"] for a in attempts) + and not any(a["effect_differs"] for a in attempts) + ): + status = "DEGENERATE" + else: + status = "CANDIDATES_EXHAUSTED" + return _cell( + task, family, status, edit=None, verdict=None, census_attempts=attempts + ) + + +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 False + return re.search(rf"(? dict[str, Any]: + cell = dict(cell) + 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 ledger_path and ledger_path.is_file(): + for line in ledger_path.read_text().splitlines(): + if line.strip(): + cell = json.loads(line) + done[(cell["task"], cell["family"])] = cell + if cfg.out_dir: + cfg.out_dir.mkdir(parents=True, exist_ok=True) + + 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, + "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..84d0a05d26 --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py @@ -0,0 +1,165 @@ +"""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 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, args = [], {} + 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, + ) -> 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._proc = subprocess.Popen( + [python, "-I", str(WORKER), appworld_root, cache_dir], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + 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 + req = { + "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() + line = self._proc.stdout.readline() + if not line: + raise RuntimeError("AppWorld worker exited") + resp = json.loads(line) + 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 close(self) -> None: + if self._proc.stdin: + self._proc.stdin.close() + self._proc.wait(timeout=60) + + +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..fae19e9c67 --- /dev/null +++ b/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py @@ -0,0 +1,214 @@ +"""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") + module.ic = lambda *a: a[0] if len(a) == 1 else (a or None) + module.ic.configureOutput = lambda *a, **k: None + 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) + worker = Worker(appworld_root, cache) + real_stdout = sys.stdout + sys.stdout = sys.stderr # AppWorld prints; keep the protocol channel clean + for line in sys.stdin: + if not line.strip(): + continue + resp = worker.handle(json.loads(line)) + 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..7617728039 --- /dev/null +++ b/packages/flow/tests/test_verifier_audit.py @@ -0,0 +1,160 @@ +"""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, + 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_and_effects(): + report = run_audit(ToyPayments(), AuditConfig(redact=True)) + text = json.dumps(report["cells"]) + assert "lunch" not in text and "sha256:" in text + + +@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", + } + json.dumps(spec) From 760557b9f1367de2c56f3f152393bf202d55b203 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 3 Nov 2026 17:32:05 -0500 Subject: [PATCH 2/5] feat(flow): map MockMed/MockLoan fault-server modes to verifier-audit families --- .../openadapt_flow/verifier_audit/families.py | 21 ++++++++++++++++++- packages/flow/tests/test_verifier_audit.py | 2 ++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/packages/flow/openadapt_flow/verifier_audit/families.py b/packages/flow/openadapt_flow/verifier_audit/families.py index c9ca773c60..a204b60339 100644 --- a/packages/flow/openadapt_flow/verifier_audit/families.py +++ b/packages/flow/openadapt_flow/verifier_audit/families.py @@ -378,9 +378,28 @@ def incidental_change(ctx: FamilyContext) -> Union[list[Candidate], Refusal]: ) +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", +} +"""The MockMed and MockLoan fault-server modes (``?fault=``) 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]} + 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, ...]: diff --git a/packages/flow/tests/test_verifier_audit.py b/packages/flow/tests/test_verifier_audit.py index 7617728039..1e3bf24c60 100644 --- a/packages/flow/tests/test_verifier_audit.py +++ b/packages/flow/tests/test_verifier_audit.py @@ -157,4 +157,6 @@ def test_family_spec_is_shareable_json(): "late_write", "incidental_change", } + ids = {f["id"] for f in spec["families"]} + assert set(spec["server_fault_modes"].values()) <= ids json.dumps(spec) From 5dc4549343338e9c1a9f9e15807b6c64d501fd46 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 3 Nov 2026 20:18:16 -0500 Subject: [PATCH 3/5] fix(flow): satisfy mypy in the verifier audit kit --- .../flow/openadapt_flow/verifier_audit/runner.py | 16 ++++++++++++---- .../verifier_audit/targets/appworld.py | 3 ++- .../verifier_audit/targets/appworld_worker.py | 8 ++++++-- 3 files changed, 20 insertions(+), 7 deletions(-) diff --git a/packages/flow/openadapt_flow/verifier_audit/runner.py b/packages/flow/openadapt_flow/verifier_audit/runner.py index 1ec3b55f48..3151a86b37 100644 --- a/packages/flow/openadapt_flow/verifier_audit/runner.py +++ b/packages/flow/openadapt_flow/verifier_audit/runner.py @@ -108,8 +108,12 @@ def _audit_family( 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" + status = ( + {Verdict.PASS: "ACCEPTED", Verdict.FAIL: "FALSE_REJECT"}.get( + ep.verdict, "SCORER_ERROR" + ) + if ep.verdict + else "SCORER_ERROR" ) return _cell( task, @@ -164,8 +168,12 @@ def _audit_family( if not reproducible: status = "EFFECT_NOT_REPRODUCIBLE" else: - status = {Verdict.PASS: "FALSE_ACCEPT", Verdict.FAIL: "CAUGHT"}.get( - ep.verdict, "SCORER_ERROR" + status = ( + {Verdict.PASS: "FALSE_ACCEPT", Verdict.FAIL: "CAUGHT"}.get( + ep.verdict, "SCORER_ERROR" + ) + if ep.verdict + else "SCORER_ERROR" ) return _cell( task, diff --git a/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py b/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py index 84d0a05d26..5238ca8603 100644 --- a/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py +++ b/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py @@ -49,7 +49,8 @@ def to_call(raw: dict[str, Any]) -> Call: raw["method"].lower(), raw["url"] if raw["url"].startswith("/") else "/" + raw["url"], ) - parts, args = [], {} + parts: list[str] = [] + args: dict[str, Any] = {} for part in url.split("/"): if _SEGMENT_ID.match(part): args[f"_path_{len(args)}"] = part diff --git a/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py b/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py index fae19e9c67..d00d0c3a59 100644 --- a/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py +++ b/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py @@ -32,8 +32,12 @@ def _shim_icecream() -> None: import icecream # noqa: F401 except ModuleNotFoundError: module = types.ModuleType("icecream") - module.ic = lambda *a: a[0] if len(a) == 1 else (a or None) - module.ic.configureOutput = lambda *a, **k: None + + 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 From 17ca17158c9b0cc6236b06dc126a0344d2c9c979 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 3 Nov 2026 20:26:11 -0500 Subject: [PATCH 4/5] fix(flow): close verifier-audit review findings Redaction now digests edit text, call and harness errors and excluded-task details. A crashed census replay moves on to the next candidate instead of counting as an effect. The ledger records its target, families and redaction mode and refuses a mismatched resume; harness and scorer errors are retried. The AppWorld worker tags responses with a request id, moves fd 1 to stderr, logs stderr and times out. partial_write no longer emits a duplicate candidate. --- packages/flow/docs/VERIFIER_AUDIT.md | 8 ++- .../openadapt_flow/verifier_audit/families.py | 16 +++-- .../openadapt_flow/verifier_audit/runner.py | 71 ++++++++++++++++--- .../verifier_audit/targets/appworld.py | 38 ++++++++-- .../verifier_audit/targets/appworld_worker.py | 11 ++- packages/flow/tests/test_verifier_audit.py | 59 +++++++++++++-- 6 files changed, 175 insertions(+), 28 deletions(-) diff --git a/packages/flow/docs/VERIFIER_AUDIT.md b/packages/flow/docs/VERIFIER_AUDIT.md index d127356bd2..b9c7615e72 100644 --- a/packages/flow/docs/VERIFIER_AUDIT.md +++ b/packages/flow/docs/VERIFIER_AUDIT.md @@ -89,13 +89,17 @@ python -m openadapt_flow.verifier_audit run \ ``` The output directory holds `cells.jsonl` (an append-only ledger; a rerun -resumes and does not rescore finished cells), `report.json` and `REPORT.md`. +resumes and does not rescore finished cells, but retries harness and scorer +errors), `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 and effect rows with digests. +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 diff --git a/packages/flow/openadapt_flow/verifier_audit/families.py b/packages/flow/openadapt_flow/verifier_audit/families.py index a204b60339..32db06e98b 100644 --- a/packages/flow/openadapt_flow/verifier_audit/families.py +++ b/packages/flow/openadapt_flow/verifier_audit/families.py @@ -257,16 +257,20 @@ def partial_write(ctx: FamilyContext) -> Union[list[Candidate], Refusal]: 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 :] - return [ + out = [ Candidate( tuple(keep_first), f"only the first of {len(writes)} writes is sent (call {first})", - ), - Candidate( - tuple(drop_last), - f"the last of {len(writes)} writes (call {last}) is dropped", - ), + ) ] + 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]: diff --git a/packages/flow/openadapt_flow/verifier_audit/runner.py b/packages/flow/openadapt_flow/verifier_audit/runner.py index 3151a86b37..4f3a29179f 100644 --- a/packages/flow/openadapt_flow/verifier_audit/runner.py +++ b/packages/flow/openadapt_flow/verifier_audit/runner.py @@ -148,6 +148,11 @@ def _audit_family( 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( { @@ -188,10 +193,12 @@ def _audit_family( scorer_detail=ep.scorer_detail, census_attempts=attempts, ) - if ( + if attempts and all(a.get("harness_error") for a in attempts): + status = "HARNESS_ERROR" + elif ( attempts - and all(a["edited_call_accepted"] for a in attempts) - and not any(a["effect_differs"] for a in 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: @@ -201,6 +208,35 @@ def _audit_family( ) +_RETRY = ("HARNESS_ERROR", "SCORER_ERROR") + + +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]: @@ -219,13 +255,27 @@ def _value_named( if after["args"].get(key) != value and isinstance(value, (str, int, float)): needle = str(value).lower().strip() if len(needle) < 3: - return False + 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"] = [ { @@ -251,13 +301,17 @@ def run_audit(target: AuditTarget, cfg: Optional[AuditConfig] = None) -> dict[st 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) - done[(cell["task"], cell["family"])] = cell - if cfg.out_dir: - cfg.out_dir.mkdir(parents=True, exist_ok=True) + key = (cell["task"], cell["family"]) + if cell.get("status") in _RETRY: + 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) @@ -297,8 +351,9 @@ def record(cell: dict[str, Any]) -> None: "task": task, "family": "__control__", "reason": reason, - "scorer_detail": control.scorer_detail, } + if not cfg.redact: + row["scorer_detail"] = control.scorer_detail excluded.append(row) if ledger_path: with ledger_path.open("a") as fh: diff --git a/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py b/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py index 5238ca8603..b0dbb90af5 100644 --- a/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py +++ b/packages/flow/openadapt_flow/verifier_audit/targets/appworld.py @@ -16,6 +16,7 @@ import json import re +import select import subprocess from pathlib import Path from typing import Any, Optional, Sequence @@ -87,6 +88,7 @@ def __init__( appworld_root: str, references: str, cache: Optional[str] = None, + timeout: str = "1800", ) -> None: self.appworld_root = appworld_root self.references = Path(references) @@ -94,11 +96,15 @@ def __init__( 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=subprocess.DEVNULL, + stderr=self._log, text=True, bufsize=1, ) @@ -133,7 +139,9 @@ def run(self, task: str, calls: Sequence[Call], *, score: bool) -> Episode: 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, @@ -142,10 +150,7 @@ def run(self, task: str, calls: Sequence[Call], *, score: bool) -> Episode: assert self._proc.stdin and self._proc.stdout self._proc.stdin.write(json.dumps(req) + "\n") self._proc.stdin.flush() - line = self._proc.stdout.readline() - if not line: - raise RuntimeError("AppWorld worker exited") - resp = json.loads(line) + resp = self._read(self._next_id) return Episode( final_effect=canonical_effect(resp["final_rows"]), claim_effect=canonical_effect(resp["claim_rows"]) @@ -156,10 +161,33 @@ def run(self, task: str, calls: Sequence[Call], *, score: bool) -> Episode: 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: diff --git a/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py b/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py index d00d0c3a59..5546cfa7a1 100644 --- a/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py +++ b/packages/flow/openadapt_flow/verifier_audit/targets/appworld_worker.py @@ -202,13 +202,18 @@ def handle(self, req: dict) -> dict: 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) - real_stdout = sys.stdout - sys.stdout = sys.stderr # AppWorld prints; keep the protocol channel clean for line in sys.stdin: if not line.strip(): continue - resp = worker.handle(json.loads(line)) + 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 diff --git a/packages/flow/tests/test_verifier_audit.py b/packages/flow/tests/test_verifier_audit.py index 1e3bf24c60..f6a1c940d8 100644 --- a/packages/flow/tests/test_verifier_audit.py +++ b/packages/flow/tests/test_verifier_audit.py @@ -11,6 +11,8 @@ AuditConfig, Call, CallKind, + Episode, + Verdict, families_spec, run_audit, wilson_interval, @@ -132,10 +134,59 @@ def run(self, task, calls_, *, score): assert scored.count(False) == 3 # only the reference reproducibility replays -def test_redaction_digests_arguments_and_effects(): - report = run_audit(ToyPayments(), AuditConfig(redact=True)) - text = json.dumps(report["cells"]) - assert "lunch" not in text and "sha256:" in text +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"]) @pytest.mark.parametrize( From d6df929ecd47a8c0a2382eeb45febd42356aa4a7 Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 02:06:45 -0500 Subject: [PATCH 5/5] fix(verifier-audit): generic fault-mode docstring; retry errored controls on resume --- packages/flow/docs/VERIFIER_AUDIT.md | 2 +- .../openadapt_flow/verifier_audit/families.py | 2 +- .../openadapt_flow/verifier_audit/runner.py | 14 +++++++++++++- packages/flow/tests/test_verifier_audit.py | 17 +++++++++++++++++ 4 files changed, 32 insertions(+), 3 deletions(-) diff --git a/packages/flow/docs/VERIFIER_AUDIT.md b/packages/flow/docs/VERIFIER_AUDIT.md index b9c7615e72..699b4e9999 100644 --- a/packages/flow/docs/VERIFIER_AUDIT.md +++ b/packages/flow/docs/VERIFIER_AUDIT.md @@ -90,7 +90,7 @@ python -m openadapt_flow.verifier_audit run \ 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), `ledger.json` (the target, kit version, families and redaction mode; +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 diff --git a/packages/flow/openadapt_flow/verifier_audit/families.py b/packages/flow/openadapt_flow/verifier_audit/families.py index 32db06e98b..0285ed1fac 100644 --- a/packages/flow/openadapt_flow/verifier_audit/families.py +++ b/packages/flow/openadapt_flow/verifier_audit/families.py @@ -391,7 +391,7 @@ def incidental_change(ctx: FamilyContext) -> Union[list[Candidate], Refusal]: "no_persist": "optimistic_success", "collateral": "unauthorized_side_effect", } -"""The MockMed and MockLoan fault-server modes (``?fault=``) that produce the +"""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.""" diff --git a/packages/flow/openadapt_flow/verifier_audit/runner.py b/packages/flow/openadapt_flow/verifier_audit/runner.py index 4f3a29179f..bc1987bc11 100644 --- a/packages/flow/openadapt_flow/verifier_audit/runner.py +++ b/packages/flow/openadapt_flow/verifier_audit/runner.py @@ -211,6 +211,18 @@ def _audit_family( _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).""" @@ -308,7 +320,7 @@ def run_audit(target: AuditTarget, cfg: Optional[AuditConfig] = None) -> dict[st if line.strip(): cell = json.loads(line) key = (cell["task"], cell["family"]) - if cell.get("status") in _RETRY: + 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 diff --git a/packages/flow/tests/test_verifier_audit.py b/packages/flow/tests/test_verifier_audit.py index f6a1c940d8..51be011d83 100644 --- a/packages/flow/tests/test_verifier_audit.py +++ b/packages/flow/tests/test_verifier_audit.py @@ -189,6 +189,23 @@ def test_resume_retries_harness_errors(tmp_path): 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)],