diff --git a/scripts/evolution_flip_gate.py b/scripts/evolution_flip_gate.py new file mode 100644 index 0000000000..191851a55c --- /dev/null +++ b/scripts/evolution_flip_gate.py @@ -0,0 +1,302 @@ +#!/usr/bin/env python3 +"""Example-level flip gate for skill promotion (issue #1446, Child B of #1308). + +The promotion decision used to be "did the aggregate score go up?". AgentDevel +(arXiv:2601.04620) shows that is the question that ships regressions: an +aggregate can rise while specific previously-working cases break. Their ablation +puts the regression rate at **14.8% with 4 bad releases** without example-level +flip gating, and **3.1% with 0** with the full release-engineering pipeline. + +This module answers the reliability question instead — *did we break anything +that worked?* — by classifying every probe into one of four cells: + + P->F regression the alarming one: it passed before, it fails now + F->P fix the gain + P->P held unchanged pass + F->F still broken unchanged fail + +and blocking promotion when P->F exceeds a threshold, no matter how many F->P +gains sit beside it. Gains do not buy regressions. + +Slice A (#1355, ``evolution_probe_set.py``) defines the frozen probe sets this +consumes. Wiring the verdict into merge verification is Child C (#1447), and +version metadata on promoted skills is Child D (#1448) — this module only +produces the table and the verdict. + +Deterministic: no LLM, no network. Pure functions plus a thin CLI, matching the +``evolution_*.py`` idiom (JSON to stdout, distinct exit codes). +""" + +from __future__ import annotations + +import json +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Optional + +SCHEMA_VERSION = "1" + +#: The four outcome cells. +REGRESSION = "P->F" +FIX = "F->P" +HELD = "P->P" +STILL_BROKEN = "F->F" + +#: Probes present in one run but not the other. Never silently dropped: a probe +#: that vanished between versions may be *hiding* a regression, and one that +#: appeared has no baseline to be judged against. +MISSING_AFTER = "missing-after" +MISSING_BEFORE = "missing-before" + +#: Default tolerance. Zero on purpose — the entire premise is that a single +#: broken previously-working example is worth a human look. Callers that need +#: slack must ask for it explicitly. +DEFAULT_MAX_REGRESSIONS = 0 + +PROMOTE = "promote" +BLOCK = "block" + + +@dataclass +class FlipTable: + """Per-probe outcome deltas between two runs of the same probe set.""" + + cells: Dict[str, str] = field(default_factory=dict) # probe id -> cell + + def ids_in(self, cell: str) -> List[str]: + """Probe ids landing in ``cell``, sorted for stable output.""" + return sorted(pid for pid, c in self.cells.items() if c == cell) + + def count(self, cell: str) -> int: + return sum(1 for c in self.cells.values() if c == cell) + + @property + def regressions(self) -> List[str]: + return self.ids_in(REGRESSION) + + @property + def fixes(self) -> List[str]: + return self.ids_in(FIX) + + def to_dict(self) -> Dict[str, Any]: + return { + "schema_version": SCHEMA_VERSION, + "cells": dict(sorted(self.cells.items())), + "counts": { + cell: self.count(cell) + for cell in (REGRESSION, FIX, HELD, STILL_BROKEN, + MISSING_AFTER, MISSING_BEFORE) + }, + } + + +@dataclass +class FlipVerdict: + """The promotion decision plus the reason it was reached.""" + + verdict: str + reason: str + regressions: List[str] = field(default_factory=list) + fixes: List[str] = field(default_factory=list) + + @property + def blocked(self) -> bool: + return self.verdict == BLOCK + + def to_dict(self) -> Dict[str, Any]: + return { + "verdict": self.verdict, + "reason": self.reason, + "regressions": list(self.regressions), + "fixes": list(self.fixes), + } + + +def _passed(value: Any) -> bool: + """Coerce a probe result into pass/fail. + + Accepts the shapes a grader is likely to emit: a bare bool, a dict with + ``passed``/``pass``/``ok``/``success``, or a status string. Anything + unrecognised is treated as a FAIL — an unreadable result must not be able + to look like a pass and let a regression through. + """ + if isinstance(value, bool): + return value + if isinstance(value, dict): + for key in ("passed", "pass", "ok", "success"): + if key in value: + return bool(value[key]) + status = str(value.get("status", "")).strip().lower() + if status: + return status in ("pass", "passed", "ok", "success") + return False + if isinstance(value, str): + return value.strip().lower() in ("pass", "passed", "ok", "success", "true") + return False + + +def compare_runs(before: Dict[str, Any], after: Dict[str, Any]) -> FlipTable: + """Classify every probe id into one of the four cells (or a missing marker). + + ``before`` and ``after`` map probe id -> result, in any of the shapes + :func:`_passed` accepts. Ids present in only one run are recorded as + ``missing-after`` / ``missing-before`` rather than dropped. + """ + table = FlipTable() + for pid in set(before) | set(after): + in_before, in_after = pid in before, pid in after + if in_before and not in_after: + table.cells[pid] = MISSING_AFTER + continue + if in_after and not in_before: + table.cells[pid] = MISSING_BEFORE + continue + was, now = _passed(before[pid]), _passed(after[pid]) + if was and not now: + table.cells[pid] = REGRESSION + elif not was and now: + table.cells[pid] = FIX + elif was: + table.cells[pid] = HELD + else: + table.cells[pid] = STILL_BROKEN + return table + + +def flip_verdict( + table: FlipTable, + max_regressions: int = DEFAULT_MAX_REGRESSIONS, + *, + block_on_missing: bool = True, +) -> FlipVerdict: + """Decide promote/block from a :class:`FlipTable`. + + Blocks when P->F exceeds ``max_regressions``, regardless of how many F->P + gains accompany them — the point of the gate is that gains do not buy + regressions. + + ``block_on_missing`` also blocks when a probe disappeared between runs. A + probe set is frozen by contract (#1355); one going missing means either the + set was mutated or the run did not complete, and both make the comparison + unsound rather than merely incomplete. + """ + regressions = table.regressions + fixes = table.fixes + n_reg = len(regressions) + + if block_on_missing: + missing = table.ids_in(MISSING_AFTER) + if missing: + return FlipVerdict( + verdict=BLOCK, + reason=( + f"{len(missing)} probe(s) present before and absent after " + f"({', '.join(missing[:5])}) — the probe set is frozen by " + f"contract, so a comparison missing probes is unsound" + ), + regressions=regressions, + fixes=fixes, + ) + + if n_reg > max_regressions: + return FlipVerdict( + verdict=BLOCK, + reason=( + f"{n_reg} regression(s) ({', '.join(regressions[:5])}) exceed the " + f"allowed {max_regressions}; {len(fixes)} fix(es) do not offset " + f"breaking what previously worked" + ), + regressions=regressions, + fixes=fixes, + ) + + return FlipVerdict( + verdict=PROMOTE, + reason=( + f"{n_reg} regression(s) within the allowed {max_regressions}; " + f"{len(fixes)} fix(es), {table.count(HELD)} held" + ), + regressions=regressions, + fixes=fixes, + ) + + +def format_table(table: FlipTable, verdict: FlipVerdict) -> str: + """Human-readable summary for a PR comment or a log line.""" + lines = [ + f"[flip-gate] {verdict.verdict.upper()}: {verdict.reason}", + f" {REGRESSION} {table.count(REGRESSION)} " + f"{FIX} {table.count(FIX)} " + f"{HELD} {table.count(HELD)} " + f"{STILL_BROKEN} {table.count(STILL_BROKEN)}", + ] + for cell, label in ((REGRESSION, "regressed"), (FIX, "fixed")): + ids = table.ids_in(cell) + if ids: + lines.append(f" {label}: {', '.join(ids)}") + for cell in (MISSING_AFTER, MISSING_BEFORE): + ids = table.ids_in(cell) + if ids: + lines.append(f" {cell}: {', '.join(ids)}") + return "\n".join(lines) + + +def _load(path: Path) -> Dict[str, Any]: + data = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(data, dict): + raise ValueError(f"{path}: expected an object mapping probe id -> result") + return data + + +def _usage() -> str: + return ( + "usage: evolution_flip_gate.py " + "[--max-regressions N] [--allow-missing]\n" + " Each file maps probe id -> result (bool, status string, or an object\n" + " with passed/ok/success/status).\n" + " Exit 0 = promote, 1 = block, 2 = bad input." + ) + + +def main(argv: Optional[List[str]] = None) -> int: + args = list(argv if argv is not None else sys.argv[1:]) + if "--help" in args or "-h" in args: + print(_usage()) + return 0 + + max_regressions = DEFAULT_MAX_REGRESSIONS + if "--max-regressions" in args: + i = args.index("--max-regressions") + try: + max_regressions = int(args[i + 1]) + del args[i : i + 2] + except (IndexError, ValueError): + print(_usage(), file=sys.stderr) + return 2 + + block_on_missing = True + if "--allow-missing" in args: + block_on_missing = False + args.remove("--allow-missing") + + if len(args) != 2: + print(_usage(), file=sys.stderr) + return 2 + + try: + before, after = _load(Path(args[0])), _load(Path(args[1])) + except (OSError, ValueError) as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + table = compare_runs(before, after) + verdict = flip_verdict(table, max_regressions, block_on_missing=block_on_missing) + + print(json.dumps({**table.to_dict(), **verdict.to_dict()}, indent=2)) + print(format_table(table, verdict), file=sys.stderr) + return 1 if verdict.blocked else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/scripts/test_evolution_flip_gate.py b/tests/scripts/test_evolution_flip_gate.py new file mode 100644 index 0000000000..fbfda6a40a --- /dev/null +++ b/tests/scripts/test_evolution_flip_gate.py @@ -0,0 +1,195 @@ +# -*- coding: utf-8 -*- +"""Tests for the example-level flip gate (issue #1446, Child B of #1308).""" + +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "scripts")) + +from evolution_flip_gate import ( # noqa: E402 + BLOCK, + FIX, + HELD, + MISSING_AFTER, + MISSING_BEFORE, + PROMOTE, + REGRESSION, + STILL_BROKEN, + _passed, + compare_runs, + flip_verdict, + format_table, + main, +) + + +class TestPassedCoercion: + """A grader may emit any of several shapes; an unreadable one must not + look like a pass, or a regression slips through as P->P.""" + + def test_bare_bool(self): + assert _passed(True) is True + assert _passed(False) is False + + def test_dict_variants(self): + for key in ("passed", "pass", "ok", "success"): + assert _passed({key: True}) is True + assert _passed({key: False}) is False + + def test_status_string_in_dict(self): + assert _passed({"status": "pass"}) is True + assert _passed({"status": "PASSED"}) is True + assert _passed({"status": "error"}) is False + + def test_bare_string(self): + assert _passed("pass") is True + assert _passed("fail") is False + + def test_unrecognised_is_a_fail(self): + """Fail closed — an unparseable result must never read as a pass.""" + assert _passed(None) is False + assert _passed(42) is False + assert _passed({}) is False + assert _passed({"weird": "shape"}) is False + + +class TestCompareRuns: + def test_four_cells(self): + table = compare_runs( + before={"a": True, "b": False, "c": True, "d": False}, + after={"a": False, "b": True, "c": True, "d": False}, + ) + assert table.cells["a"] == REGRESSION + assert table.cells["b"] == FIX + assert table.cells["c"] == HELD + assert table.cells["d"] == STILL_BROKEN + + def test_counts_and_id_lists(self): + table = compare_runs( + before={"a": True, "b": True, "c": False}, + after={"a": False, "b": False, "c": True}, + ) + assert table.count(REGRESSION) == 2 + assert table.regressions == ["a", "b"] + assert table.fixes == ["c"] + + def test_missing_probes_are_marked_not_dropped(self): + table = compare_runs(before={"a": True, "gone": True}, after={"a": True, "new": True}) + assert table.cells["gone"] == MISSING_AFTER + assert table.cells["new"] == MISSING_BEFORE + assert len(table.cells) == 3 + + def test_empty_runs(self): + assert compare_runs({}, {}).cells == {} + + +class TestFlipVerdict: + def test_clean_run_promotes(self): + table = compare_runs({"a": True, "b": False}, {"a": True, "b": True}) + v = flip_verdict(table) + assert v.verdict == PROMOTE + assert v.blocked is False + assert v.fixes == ["b"] + + def test_single_regression_blocks_by_default(self): + table = compare_runs({"a": True}, {"a": False}) + v = flip_verdict(table) + assert v.verdict == BLOCK + assert v.regressions == ["a"] + + def test_gains_do_not_offset_regressions(self): + """The whole premise: an aggregate can rise while something breaks.""" + table = compare_runs( + before={"a": True, "b": False, "c": False, "d": False}, + after={"a": False, "b": True, "c": True, "d": True}, + ) + v = flip_verdict(table) + assert v.verdict == BLOCK + assert len(v.fixes) == 3 and len(v.regressions) == 1 + + def test_threshold_allows_slack_when_asked(self): + table = compare_runs({"a": True, "b": True}, {"a": False, "b": False}) + assert flip_verdict(table, max_regressions=2).verdict == PROMOTE + assert flip_verdict(table, max_regressions=1).verdict == BLOCK + + def test_missing_probe_blocks(self): + """A frozen set losing a probe makes the comparison unsound, not merely + incomplete — it could be hiding the regression.""" + table = compare_runs({"a": True, "gone": True}, {"a": True}) + v = flip_verdict(table) + assert v.verdict == BLOCK + assert "gone" in v.reason + + def test_missing_probe_can_be_allowed_explicitly(self): + table = compare_runs({"a": True, "gone": True}, {"a": True}) + assert flip_verdict(table, block_on_missing=False).verdict == PROMOTE + + def test_new_probe_alone_does_not_block(self): + """An added probe has no baseline, but nothing regressed.""" + table = compare_runs({"a": True}, {"a": True, "new": False}) + assert flip_verdict(table).verdict == PROMOTE + + def test_reason_names_the_regressed_probes(self): + table = compare_runs({"x": True}, {"x": False}) + assert "x" in flip_verdict(table).reason + + +class TestFormatTable: + def test_summary_names_verdict_and_ids(self): + table = compare_runs({"a": True, "b": False}, {"a": False, "b": True}) + out = format_table(table, flip_verdict(table)) + assert "BLOCK" in out + assert "regressed: a" in out + assert "fixed: b" in out + + +class TestCli: + @staticmethod + def _write(tmp_path, name, data): + p = tmp_path / name + p.write_text(json.dumps(data), encoding="utf-8") + return str(p) + + def test_exit_zero_on_promote(self, tmp_path, capsys): + b = self._write(tmp_path, "b.json", {"a": True}) + a = self._write(tmp_path, "a.json", {"a": True}) + assert main([b, a]) == 0 + assert json.loads(capsys.readouterr().out)["verdict"] == PROMOTE + + def test_exit_one_on_block(self, tmp_path, capsys): + b = self._write(tmp_path, "b.json", {"a": True}) + a = self._write(tmp_path, "a.json", {"a": False}) + assert main([b, a]) == 1 + assert json.loads(capsys.readouterr().out)["verdict"] == BLOCK + + def test_max_regressions_flag(self, tmp_path, capsys): + b = self._write(tmp_path, "b.json", {"a": True}) + a = self._write(tmp_path, "a.json", {"a": False}) + assert main([b, a, "--max-regressions", "1"]) == 0 + capsys.readouterr() + + def test_allow_missing_flag(self, tmp_path, capsys): + b = self._write(tmp_path, "b.json", {"a": True, "gone": True}) + a = self._write(tmp_path, "a.json", {"a": True}) + assert main([b, a]) == 1 + capsys.readouterr() + assert main([b, a, "--allow-missing"]) == 0 + capsys.readouterr() + + def test_bad_input_exits_two(self, tmp_path, capsys): + assert main(["nope.json", "also-nope.json"]) == 2 + capsys.readouterr() + + def test_wrong_arg_count_exits_two(self, capsys): + assert main(["only-one.json"]) == 2 + capsys.readouterr() + + def test_non_object_payload_exits_two(self, tmp_path, capsys): + p = self._write(tmp_path, "list.json", ["not", "an", "object"]) + assert main([p, p]) == 2 + capsys.readouterr() + + def test_help_exits_zero(self, capsys): + assert main(["--help"]) == 0 + assert "usage" in capsys.readouterr().out