mutation: AST operator engine for the neutral benchmark

src/testbench/mutation/__init__.py

@@ -0,0 +1,26 @@
+"""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.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__ = [
+    "LabeledMutant",
+    "MASTMode",
+    "Mutant",
+    "benign_variants",
+    "drop_equivalents",
+    "enumerate_labeled",
+    "is_equivalent",
+    "loop_outcome",
+    "mutate_source",
+    "source_of",
+]

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

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"<mutant:{qualname}>"
+    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)]

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

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")