API reference

Generated by autodoc straight from the source docstrings; nothing on these pages is hand-maintained, so nothing on them can drift.

The module docstrings are the point. Most of moonbuggy’s design rationale lives in them — why the parent must never import the module under test, why the cache key is deliberately coarse, why a mutation that cannot be applied raises rather than returning — and it survives into these pages rather than being flattened into a list of signatures.

For the shape of the whole thing rather than the parts, read Architecture first.

Pipeline

Zero-config project layout discovery.

The whole point is the lowest possible floor: bare moonbuggy in a pytest project’s root should just work, with configuration available but never required.

Discovery is deliberately conservative and explains itself. Guessing wrong here means mutating the wrong files, and a tool that silently mutates a project’s test suite or its vendored dependencies wastes a long run and produces nonsense. When the layout cannot be identified, this raises with the flag to pass rather than picking something plausible.

exception moonbuggy.discover.LayoutError

Bases: RuntimeError

Raised when the project layout cannot be identified with confidence.

moonbuggy.discover.find_source_dir(project_dir: str | PathLike[str]) → Path

The directory whose Python files should be mutated.

Order matters: an explicit src/ layout is unambiguous, so it wins before any heuristic gets a chance to be clever.

moonbuggy.discover.find_source_files(source_dir: str | PathLike[str], project_dir: str | PathLike[str], include: Sequence[str] = (), exclude: Sequence[str] = ()) → list[str]

Every mutable file under source_dir, as paths relative to project_dir.

“Mutable” means: a .py file whose project-relative path has no directory component in EXCLUDED, and whose name is neither test_*.py nor conftest.py nor setup.py. Those exclusions are unconditional – an include fragment cannot bring one back.

Parameters:

• source_dir – the directory to walk, recursively.

• project_dir – the root the returned paths are relative to.

• include – keep only files whose project-relative path contains one of these. Plain substring matches, not globs and not path prefixes: models matches pkg/models.py and pkg/models/a.py alike, and *.py matches nothing.

• exclude – drop any file whose project-relative path contains one of these. Same substring rule, applied after include.

Returns:

The project-relative paths, sorted.

moonbuggy.discover.looks_like_pytest_project(project_dir: str | PathLike[str]) → bool

Whether this directory plausibly hosts a pytest suite.

Used to fail fast with a clear message rather than after a coverage pass that collects nothing.

Diff scoping: mutate only the lines a branch changed.

A full run is a session you schedule; a diff-scoped run is cheap enough to be a habit, which is the whole point of –since. On a typical pull request the changed lines carry a handful of mutants and seconds of runtime, so mutation testing can gate every PR rather than being an audit somebody remembers to do.

This is a filter on generation, not a second mutation engine. Generation produces exactly the mutants it always did; the scope is intersected with them afterwards, and with –include/–exclude rather than in place of them. That is what keeps a diff-scoped verdict identical to the verdict the same mutant would get in a full run – which is also why –since is deliberately absent from the cache fingerprint (see moonbuggy.cache.run_fingerprint()): how you reached a mutant cannot change its answer, so both kinds of run fill and read the same cache.

Two decisions worth stating, because both differ from the obvious reading:

• The diff is taken between the merge base and the working tree, not <ref>…HEAD. The merge base half is the familiar three-dot semantics – scope is this branch’s work, not whatever main has drifted to since. The working-tree half is a correctness requirement: moonbuggy mutates the files on disk, so a scope computed against HEAD would carry line numbers for a file that is not the one being read, and uncommitted work – the state you are in when you most want a fast run – would be scoped out.

• Untracked files are entirely in scope. git diff cannot see them, and a module you just wrote is the least-tested code in the tree; silently skipping it would be the worst possible way to be fast.

exception moonbuggy.diffscope.DiffScopeError

Bases: RuntimeError

Raised when a –since scope cannot be computed.

Always actionable and never a traceback: cli.main turns this into exit code 2 with the message alone, the same as every other anticipated failure.

class moonbuggy.diffscope.DiffScope(ref: str, merge_base: str, ranges: dict[str, tuple[tuple[int, int], ...]])

Bases: object

The changed lines of a working tree, relative to a git ref.

A plain frozen record with no I/O of its own, so a caller can build one in a test without a repository, and so the structured summary a –json run reports is the same object the run was filtered by rather than a second derivation of it.

ref: str

The ref as the user spelled it, e.g. origin/main.

merge_base: str

The full sha the diff was taken from.

ranges: dict[str, tuple[tuple[int, int], ...]]

Changed line ranges per file, as inclusive (first, last) pairs.

Keyed by path relative to the project directory, with / separators. Files whose only change was a deletion do not appear: there is nothing left on those lines to mutate.

property files: tuple[str, ...]

The changed files that have at least one mutable line, sorted.

property changed_lines: int

How many lines are in scope, across every file.

touches(module: str) → bool

Whether any line of this file is in scope.

Used to drop whole files before generation runs on them, which is the cheap half of the filter.

Parameters:

module – a path relative to the project directory.

Returns:

True if the file has changed lines.

contains(module: str, line: int) → bool

Whether one source line is in scope.

Parameters:

• module – a path relative to the project directory.

• line – a 1-based line number.

Returns:

True if that line falls in a changed range.

describe() → str

One line saying what was scoped away, for a report footer.

Returns:

A sentence naming the ref, the merge base and the size of the scope – so a 100% kill rate on three mutants can never be read as a clean full run.

moonbuggy.diffscope.scope_summary(scope: DiffScope | None) → dict[str, object]

The scope of a run as structured data, whether or not it was scoped.

Every run has an answer to “was this the whole codebase?”, so this takes None and reports it rather than leaving the caller to build the negative case itself. –json consumers read these keys; the human footer reads DiffScope.describe(). Both come from the same object.

Parameters:

scope – the run’s diff scope, or None for a full run.

Returns:

A JSON-serialisable mapping with diff_scoped, since, merge_base, files and changed_lines. The last four are None/0 for a full run.

moonbuggy.diffscope.scope_since(ref: str, project_dir: str | PathLike[str]) → DiffScope

The lines project_dir has changed since ref.

Parameters:

• ref – any git revision – a branch, a tag, a sha.

• project_dir – the project root, which may be a subdirectory of the repository. Only paths under it are reported, and they are reported relative to it, because that is how a mutant names its module.

Returns:

The DiffScope for this working tree.

Raises:

DiffScopeError – if this is not a git repository, ref does not resolve, the two have no merge base, or the git diff itself fails.

Mutant generation: source -> AST -> candidate mutants.

AST-based rather than text substitution, which is what keeps mutations out of string literals and comments for free, and gives exact line and column positions for reporting.

This module knows nothing about which operators exist. It asks the registry and walks the tree – adding an operator requires no change here.

exception moonbuggy.generate.GenerationError

Bases: RuntimeError

This module’s source cannot be turned into mutants.

Raised instead of letting SyntaxError or RecursionError escape, so the CLI can name the offending file and carry on with the rest.

moonbuggy.generate.generate_mutants(source: str, module: str, on_skip: Callable[[int, str], None] | None = None, logging_policy: LoggingPolicy | None = None, operators: Collection[str] | None = None) → list[Mutant]

Return every mutant the selected operators find in one module’s source.

In a stable order. The selection defaults to the default tier rather than every registered operator, so this is not every mutation moonbuggy knows how to make – see the operators argument below.

Parameters:

• source – the module’s full source text.

• module – the module’s path, used as the first field of each mutant id.

• on_skip – called as (lineno, reason) for each site that could not be turned into a mutant. Sites are skipped rather than silently dropped, so the caller can say how much of the file it gave up on.

• logging_policy – which calls count as logging calls, and whether mutants inside their arguments are suppressed. Defaults to LoggingPolicy’s own defaults, which tag those mutants and suppress them. The policy never changes which mutants exist, only how they are labelled – so mutant ids are the same whichever way it is set.

• operators – the operator names to run. The default, None, is the default tier – what a bare moonbuggy does – and not every registered operator. That distinction only started to matter when the first deep operator existed: before that the two sets were equal, and a caller who asked for “all of them” got a cheap run by accident rather than by contract. Filtering here rather than in the caller keeps the cost of a deep operator unpaid when it is not selected; it cannot move an id, because the occurrence index is counted per line and operator, so no operator’s index depends on which others ran.

Returns:

a list of Mutant, sorted by line then operator then

id, so ids are stable across runs.

Raises:

GenerationError – if the source does not parse, or is nested so deeply that the parser itself gives up.

The Mutant record.

Deliberately a plain frozen dataclass with no behaviour: it is pickled inside a moonbuggy.forkserver.Job and sent down a pipe to the warm host, and its fields are hashed into every cache key, so it stays trivially serialisable.

Two things that sound like they carry a Mutant do not. The cache stores a CacheRecord of {status, tests_run, nearest_test} under a digest of the mutant’s fields – nothing ever reconstructs a Mutant from disk. And what reaches an xdist worker is moonbuggy.plugin.MutantEnvPayload, a {path, line, mutated} JSON blob in an environment variable, not this record.

class moonbuggy.mutant.Mutant(id: str, module: str, line: int, operator: str, original: str, mutated: str, suppressed: bool = False, module_level: bool = False, logging_call: bool = False)

Bases: object

One candidate mutation of one source line.

id is stable across runs for unchanged source – the results cache keys on it, so a shifting id silently invalidates the cache. It includes an occurrence index because a single line can host several mutants: line 9 of the inventory fixture carries three, from three different operators, and a line with two + operators would carry two from the same one. A file and a line alone are not an identity.

id: str

module: str

line: int

operator: str

original: str

mutated: str

suppressed: bool = False

True when this mutant is settled as SKIPPED without ever being run: the line carries the # moonbuggy: skip marker, or it sits inside a logging call and the run did not ask for those. Suppressed mutants leave the score’s denominator – a mutant nobody could kill is not a test failure – but they are still reported, so the count stays honest.

module_level: bool = False

True when the mutated line runs at import time rather than inside a function body. Selection has to widen the test set for these: the line->test map is built from test-body execution, so a module-level line is attributed to no test, and an empty covering set reports a false NO_COVERAGE.

logging_call: bool = False

True when the mutation sits inside the arguments of a logging call, as moonbuggy.logging_policy recognises one. Nothing asserts on the contents of a debug line, so these survive whatever the tests do – by default they are also suppressed, but the tag is set either way so a project that does assert on log output can run them and still tell them apart.

moonbuggy.mutant.make_id(module: str, line: int, operator: str, index: int) → str

Build a mutant id from its four parts.

Parameters:

• module – the module’s path, as the mutant records it.

• line – the mutated line number.

• operator – the operator’s name.

• index – the occurrence index within that line and operator.

Returns:

The id, in the path:line:operator:index form the reports print.

moonbuggy.mutant.parse_id(mutant_id: str) → tuple[str, int, str, int] | None

Take a mutant id back apart.

The inverse of make_id(), and the reason both live here: an id is the only thing a user hands back to moonbuggy (moonbuggy show, moonbuggy accept, moonbuggy run), so the format is a contract and it is defined in exactly one place.

Split from the right, so a module path containing a colon – a Windows drive letter – survives the round trip.

Parameters:

mutant_id – an id as printed in id=….

Returns:

(module, line, operator, index), or None if the string is not an id at all. None rather than an exception: callers get this string from a human or a pipeline and have to say something helpful about it.

Which mutants sit inside a logging call, and what to do about them.

A mutation inside logger.debug(f”retrying in {delay * 2}s”) is unkillable by construction. Nothing asserts on the contents of a debug line, so the mutant survives every suite that will ever be written against that code, and it arrives in the survivor list looking exactly like a real finding. In one reported session two thirds of the survivors in a retry region were arithmetic inside logger.debug(…) arguments – pure triage tax on the one output the tool exists to produce.

So they are recognised as a class. Every such mutant is tagged (Mutant.logging_call), because a project that does assert on log output has a real finding here and the tool must not lie about what it generated; and by default it is also suppressed, reported SKIPPED and left out of the survivor list, because that is what almost everyone wants. –include-logging- mutants runs them and leaves the tag in place.

The boundary that defines the whole module: only the argument expressions of the call qualify. In

if attempts > 5:

logger.debug(“giving up after %d”, attempts * 2)

the > is a real finding – a branch nobody checked – and must survive untouched, while attempts * 2 is noise. One is inside the call’s arguments and one is around the call. Nothing else separates them, which is why this is decided from the mutation site’s enclosing-node chain rather than from the line it sits on.

class moonbuggy.logging_policy.LoggingPolicy(logger_names: frozenset[str] = frozenset({'LOG', 'LOGGER', '_LOG', '_LOGGER', '_log', '_logger', 'log', 'logger', 'logging'}), skip: bool = True)

Bases: object

What counts as a logging call, and whether its mutants are run.

Frozen and passed down into generation rather than read from a global, so that a caller mutating two projects with different logger names in one process gets two answers rather than whichever it configured last.

logger_names: frozenset[str]

Receiver names recognised as loggers. Compared against the last component of the attribute chain.

skip: bool

Whether a mutant inside a logging call is suppressed – settled as SKIPPED without being run. Tagging happens either way.

moonbuggy.logging_policy.policy_for(extra_names: Iterable[str] = (), *, include_logging_mutants: bool = False) → LoggingPolicy

The policy a run’s options ask for.

Parameters:

• extra_names – additional receiver names to recognise as loggers, for projects that wrap the stdlib logger under a name of their own. Added to the defaults rather than replacing them: a project with an audit logger almost certainly still has a logger too.

• include_logging_mutants – run them instead of suppressing them. They stay tagged either way, so a project that does assert on log output can opt in and see the findings.

Returns:

The LoggingPolicy for the run.

moonbuggy.logging_policy.is_logging_call(node: AST, logger_names: Iterable[str]) → bool

Whether this call looks like a call to a logger.

A heuristic, and knowingly one: there is no way to know from the AST alone that audit.info(…) reaches logging.Logger.info. It errs towards not recognising a call, because a missed logging call costs one noisy survivor and a wrongly-recognised one hides a real finding.

Parameters:

• node – any AST node; only an ast.Call can qualify.

• logger_names – the receiver names to accept.

Returns:

True if node is a call of a level method on something named like a logger.

moonbuggy.logging_policy.in_logging_call(context: Context, policy: LoggingPolicy) → bool

Whether the node in this context sits inside a logging call’s arguments.

Walks outwards to the nearest enclosing call and asks about that one only. A mutation inside logger.debug(“%s”, compute(n + 1)) is inside compute, not inside the logging call, and stays a finding: compute runs for real and a mutation to its argument can raise, be returned, or be observed. Only the arguments the logging call itself evaluates are unkillable by construction, and the nearest-call rule is exactly the line between the two.

Parameters:

• context – where the node sits, as generation walked it.

• policy – the run’s logging policy.

Returns:

True if the nearest enclosing call is a logging call and this node is somewhere in its arguments.

Baseline health: is the suite green, and is any of it flaky?

Two correctness properties meet in one mechanism, because they are answered by the same evidence.

An already-red suite. Mutation results are meaningless against a failing baseline. Every mutant covered by a test that already fails is reported KILLED, and the run looks like an excellent score. That is worse than no result, so moonbuggy refuses rather than reporting.

A flaky test. A test that fails at random makes some mutants KILLED for reasons that have nothing to do with the mutation, and can equally make a genuinely-killed mutant look SURVIVED. Neither status is honest, and SUSPICIOUS exists precisely for “we cannot tell”.

The distinguishing evidence is the same in both cases: run the unmutated suite more than once and compare per-test outcomes.

failed every time -> genuinely red. Refuse the run. outcome varied -> flaky. Every mutant that selects it is SUSPICIOUS. passed every time -> trustworthy.

One extra unmutated run is the default (--flaky-probe 1). It happens inside the warm host, where every test module is already imported, so it costs test execution time and nothing else. It can be turned off with --flaky-probe 0 for a project known to be deterministic, at the cost of the flakiness guarantee.

The probe cannot be perfect, and does not claim to be: a test that fails one run in a thousand will usually pass both. Detecting flakiness is a sampling problem, and more probes buy more confidence linearly. What the probe does guarantee is that a flake it observes never becomes a confident status.

This module doubles as a pytest plugin. It is loaded with -p moonbuggy.baseline and writes its outcomes to the path named by MOONBUGGY_OUTCOMES, which is what lets the in-process (forked) and out-of-process (subprocess) paths collect identical evidence.

exception moonbuggy.baseline.BaselineError

Bases: RuntimeError

The suite is not in a state where mutation results would mean anything.

class moonbuggy.baseline.OutcomesFile

Bases: TypedDict

The JSON written to OUTCOMES_ENV_VAR at session end.

Direction: child (this pytest process, in-process or forked) -> parent. Written once per run by pytest_sessionfinish() and read back with read_outcomes(), which only ever consults outcomes – but the file also carries exit_status for a human inspecting it by hand.

exit_status: int

outcomes: dict[str, str]

class moonbuggy.baseline.OutcomeRecorder

Bases: object

Collects one outcome per test node id from a pytest run.

Registered as a plugin rather than reading pytest’s exit code, because the exit code says only “something failed” – and knowing which tests failed is what separates a red baseline from a flake.

outcomes: dict[str, str]

pytest_runtest_logreport(report: TestReport) → None

Record the worst outcome seen for this test across its three phases.

pytest_collectreport(report: CollectReport) → None

A module that fails to import produces no tests at all.

Left unrecorded, a collection error would look like an empty but green suite, and moonbuggy would report on every line it covers as though no test existed – NO_COVERAGE, since an empty selection settles there.

moonbuggy.baseline.pytest_configure(config: Config) → None

Plugin hook: install a recorder when MOONBUGGY_OUTCOMES names a file.

moonbuggy.baseline.pytest_sessionfinish(session: Session, exitstatus: int | ExitCode) → None

Plugin hook: write the recorded outcomes where the parent can read them.

moonbuggy.baseline.classify(runs: list[dict[str, str]]) → tuple[set[str], set[str]]

Split tests into consistently-failing and inconsistent, given several runs.

Parameters:

runs – a list of {node_id: outcome} mappings, one per unmutated run of the suite. The first is authoritative for which tests exist.

Returns:

(failing, flaky), both sets of node ids. A test appears in at most one of

them.

moonbuggy.baseline.check(runs: list[dict[str, str]], allow_empty: bool = False) → set[str]

Raise if the baseline cannot support a mutation run.

Parameters:

• runs – as for classify().

• allow_empty – when True, a suite with no tests is permitted. Only used by callers that have already reported the emptiness themselves.

Returns:

the set of flaky test node ids, which callers use to mark affected mutants

SUSPICIOUS.

Raises:

BaselineError – if no tests ran, or if any test fails consistently.

moonbuggy.baseline.probe_env(path: str | PathLike[str]) → dict[str, str]

Environment additions that make a pytest run record its outcomes.

moonbuggy.baseline.read_outcomes(path: str | PathLike[str]) → dict[str, str]

Read one run’s recorded outcomes back.

Parameters:

path – the file named by OUTCOMES_ENV_VAR for that run.

Returns:

{node_id: outcome}, empty if the file was never written – which is itself

meaningful, since it means the run did not finish.

The coverage pass: one instrumented run producing a line -> covering-tests map.

This is the input to coverage-guided test selection, the largest single speed lever in the design. The mechanism is pytest-cov’s per-test contexts: the fastest of the candidates measured, and the only one recording real pytest node ids – and node ids are what selection has to hand back to pytest.

Correctness here is asymmetric. A map missing a covering test makes moonbuggy run too few tests and report a false SURVIVED, which looks exactly like a real finding. A map with a spurious extra test only costs time. Every judgement call below therefore favours the larger set.

class moonbuggy.coverage_pass.LineEntry

Bases: TypedDict

One source line’s coverage, as LineMap.to_dict() serialises it.

file: str

line: int

tests: list[str]

class moonbuggy.coverage_pass.LineMapDict

Bases: TypedDict

The whole map as plain data. See LineMap.to_dict().

tests: list[str]

lines: list[LineEntry]

class moonbuggy.coverage_pass.LineMap(mapping: dict[tuple[str, int], set[str]], tests: Iterable[str], project_dir: str | PathLike[str])

Bases: object

Which tests execute which lines.

tests_covering(module: str, line: int) → set[str]

The node ids of tests that executed line of module.

all_tests() → set[str]

Every test node id the instrumented run observed.

select_for(mutant: Mutant) → set[str]

The tests to run for one mutant.

Module-level mutants get the whole suite. Their line runs at import time, not inside any test body, so the map attributes it to nothing – and “no covering tests” is indistinguishable from “genuinely uncovered” without this. Running everything is wasteful for a rare case, and wrong-and-fast is not a trade worth making on a silent failure mode.

to_dict() → LineMapDict

The whole map as plain data, for serialising or inspecting.

moonbuggy.coverage_pass.run_coverage_pass(project_dir: str | PathLike[str], source_dir: str | PathLike[str], python: str | None = None, extra_args: Iterable[str] = (), use_fork: bool | None = None, timeout: int = 600) → LineMap

Run the suite once under coverage, returning a LineMap.

moonbuggy.coverage_pass.run_baseline_pass(project_dir: str | PathLike[str], source_dir: str | PathLike[str], probes: int = 1, python: str | None = None, timeout: int = 600, extra_args: Iterable[str] = ()) → tuple[LineMap, set[str]]

Coverage pass plus flakiness probe, for the paths that cannot fork.

Same evidence as the warm session gathers (see moonbuggy.forkserver.run_warm_session()), collected the slow way: the suite runs 1 + probes times as separate processes, each writing its per-test outcomes to a file the parent reads back.

Parameters:

• project_dir – project root.

• source_dir – directory to measure coverage of.

• probes – extra unmutated runs used to detect flaky tests.

• python – interpreter to run pytest with when forking is unavailable.

• timeout – seconds before one suite run is abandoned.

• extra_args – pytest arguments to add to every run.

Returns:

(linemap, flaky_test_ids).

Raises:

• BaselineError – if no tests ran or the suite is already failing.

• CoveragePassError – if pytest could not complete at all.

exception moonbuggy.coverage_pass.CoveragePassError(returncode: int, stdout: str, stderr: str)

Bases: RuntimeError

pytest could not complete the instrumented run at all.

moonbuggy.coverage_pass.prewarm_reader() → None

Import coverage in the parent, before the warm host is forked.

The warm-session path imports coverage twice, one after the other, and both are on the critical path. The host imports it when pytest-cov starts the instrumented run; the parent imports it again in read_coverage_data(), once the host has finished and handed back its evidence. That is ~36ms of identical work, done serially, twice.

Doing it here collapses the two into one. The host inherits the import across the fork, and the parent’s own read then needs nothing. Note the direction is a parent import added: it is work the parent was going to do anyway, just done earlier, and the host still imports coverage itself when pytest-cov starts.

Importing coverage starts no measurement and creates no Coverage object, so the host still builds its own from scratch. And coverage is not the project under mutation, so the rule the parent actually has to obey – never import the code being mutated – is untouched.

moonbuggy.coverage_pass.read_coverage_data(data_file: str | PathLike[str], project_dir: str | PathLike[str], known_tests: Iterable[str] = ()) → LineMap

Build a LineMap from a coverage data file.

Parameters:

• data_file – path to the coverage database the instrumented run wrote.

• project_dir – the project root, used to resolve module paths.

• known_tests – every test node id the run actually executed, from the outcome recorder. Coverage contexts only name tests that executed a measured line, so a module whose functions are never called during any test contributes no contexts at all – and then all_tests() is empty, and a module-level mutant that widens to “the whole suite” runs nothing and is reported NO_COVERAGE – a claim about the user’s suite that is really about ours. Found during a search of real-world open-source suites for module-level mis-attributions; see tests/test_module_level_aliases.py.

Returns:

A LineMap of which tests executed which lines.

The fast path: apply each mutant in memory, run only its covering tests.

Both speed levers of the design meet here. Coverage-guided selection decides WHICH tests run; the in-memory import hook decides HOW the mutation is applied. Neither writes a mutated file to disk.

One process per mutant. That is not merely convenient – it is what makes the xdist story work without any cross-process state, since the mutant’s identity travels in the environment and every worker installs it independently. On the warm path it is one host process plus one grandchild per mutant, which keeps the same isolation.

run_session is the primary entry point and the one the CLI calls for a normal run: it does the coverage pass and the mutant runs in a single warm process. run_mutants is the fallback for when forking is unavailable or xdist workers were asked for.

The whole reported status vocabulary is settled here: SKIPPED (suppressed), SUSPICIOUS (a flaky test in the selection) and NO_COVERAGE (nothing selected) without running anything, and KILLED, KILLED_BY_ERROR, SURVIVED and TIMEOUT from the mutant’s own process. See moonbuggy.forkserver.Status for what a process can decide and ResultStatus for what can reach a Result.

class moonbuggy.runner.Result(mutant: Mutant, status: Literal['SURVIVED', 'KILLED', 'KILLED_BY_ERROR', 'TIMEOUT', 'SUSPICIOUS', 'UNAPPLIED', 'SKIPPED', 'NO_COVERAGE'], tests_run: int, duration: float, nearest_test: str | None = None, from_cache: bool = False, killreason: KillReasonCode | None = None)

Bases: object

One mutant’s outcome, before it becomes a report record.

mutant: Mutant

status: Literal['SURVIVED', 'KILLED', 'KILLED_BY_ERROR', 'TIMEOUT', 'SUSPICIOUS', 'UNAPPLIED', 'SKIPPED', 'NO_COVERAGE']

tests_run: int

duration: float

nearest_test: str | None = None

from_cache: bool = False

killreason: KillReasonCode | None = None

Why this mutant was killed, or None when it was not. One of the KillReasonCode enumeration (assertion_failed, test_errored, execution_crash, flaky_probe), or None for any status where the reason does not apply – survivors, timeouts, uncovered lines and skipped mutants. A JSONL consumer comparing two records compares this token directly; it is never free-text.

Added with record schema 4. On a schema-3 record read from an older file it is None, because no older version could have written it.

class moonbuggy.runner.Plan

Bases: TypedDict

Mutants split into already-settled and still-needing-a-process.

See _plan().

results: dict[int, Result]

to_run: list[tuple[int, Mutant, list[str]]]

class moonbuggy.runner.RunSessionState

Bases: TypedDict

Shared state threaded through run_session’s warm-host callbacks.

total=False because it is built up field by field inside build_jobs rather than constructed all at once – every field is present by the time anything reads it, but the dict genuinely does not have all three keys for part of run_session’s body.

flaky: set[str]

linemap: LineMap

plan: Plan

moonbuggy.runner.run_mutants(project_dir: str | PathLike[str], mutants: list[Mutant], linemap: LineMap, timeout: float = 30, python: str | None = None, xdist_workers: int = 0, cache: ResultCache | None = None, use_fork: bool | None = None, jobs: int | None = None, flaky: Iterable[str] = (), on_result: Callable[[Result], None] | None = None, extra_args: Iterable[str] = ()) → list[Result]

Run every mutant against its selected tests.

Parameters:

• project_dir – the project root.

• mutants – every mutant to consider, in report order.

• linemap – the line to covering-tests map.

• timeout – seconds before one mutant is called TIMEOUT.

• python – interpreter to use for the subprocess path.

• xdist_workers – pytest-xdist workers within each mutant’s run; asking for any opts out of the warm session, which needs one process.

• cache – a ResultCache, or None.

• use_fork – force the fork path on or off; None picks the fast one.

• jobs – how many mutants to run concurrently.

• flaky – test node ids whose outcome is not reproducible; mutants selecting one are settled SUSPICIOUS rather than run.

• on_result – called with each Result as it is settled.

• extra_args – pytest arguments to add to every run.

Returns:

a list of Result, one per mutant, in the input order. Each status is one of KILLED, KILLED_BY_ERROR, SURVIVED, TIMEOUT, SUSPICIOUS, SKIPPED or NO_COVERAGE – plus UNAPPLIED on the warm-batch path, which unlike run_session does not scrub it (see ResultStatus).

moonbuggy.runner.run_one(project_dir: str | PathLike[str], mutant: Mutant, linemap: LineMap, timeout: float, python: str, xdist_workers: int = 0, cache: ResultCache | None = None, use_fork: bool = False, flaky: Iterable[str] = (), outcomes: str | PathLike[str] | None = None) → Result

Run one mutant against its selected tests.

The serial path, used when forking is unavailable or when xdist workers were requested. run_mutants is the entry point that chooses between this and the parallel one.

Parameters:

• project_dir – the project root.

• mutant – the mutant to run.

• linemap – the line to covering-tests map.

• timeout – seconds before this mutant is called TIMEOUT.

• python – interpreter to use for the subprocess path.

• xdist_workers – pytest-xdist workers within this mutant’s run.

• cache – a ResultCache, or None.

• use_fork – whether to fork rather than spawn a subprocess.

• flaky – test node ids whose outcome is not reproducible.

• outcomes – a file for this run’s per-test outcomes, or None. Only the subprocess path can produce them – a forked child reports one exit-code byte and nothing else – so asking for outcomes turns forking off, said here rather than left as a file that silently never appears.

Returns:

A Result, whose status is one of KILLED, KILLED_BY_ERROR, SURVIVED, TIMEOUT, SUSPICIOUS, SKIPPED or NO_COVERAGE.

moonbuggy.runner.run_session(project_dir: str | PathLike[str], mutants: list[Mutant], source_dir: str | PathLike[str], timeout: float = 30, cache: ResultCache | None = None, jobs: int | None = None, probes: int = 1, on_result: Callable[[Result], None] | None = None, extra_args: Iterable[str] = ()) → tuple[LineMap, list[Result]]

Coverage pass and mutant execution in a single warm process.

The two phases run the same test suite, so running them separately meant executing it twice – over half the wall clock on the benchmark workload. Here one forked host runs the suite under coverage, and the same host then forks a grandchild per mutant with every test module already imported.

Parameters:

• project_dir – project root.

• mutants – every mutant to consider, in report order.

• source_dir – directory to measure coverage of.

• timeout – seconds before one mutant is called TIMEOUT.

• cache – a ResultCache, or None.

• jobs – how many mutants to run concurrently.

• probes – extra unmutated suite runs used to detect flaky tests.

• on_result – called with each Result as it is settled, so a run killed mid-flight has already emitted what it knew.

• extra_args – pytest arguments to add to every run, baseline and mutant alike. A project whose real test command is not bare pytest – one that needs –doctest-modules, say – is otherwise measured against a suite smaller than the one it actually runs, and every mutant its doctests would catch is reported as a survivor.

Returns:

(linemap, results). Each result’s status is one of KILLED, KILLED_BY_ERROR, SURVIVED, TIMEOUT, SUSPICIOUS, SKIPPED or NO_COVERAGE; UNAPPLIED never reaches a caller here, because _rerun_unapplied re-runs those coldly first.

Raises:

BaselineError – if the suite is already failing or collects nothing. Falls back to the separate cold path when the warm host cannot complete, so a host failure costs time rather than correctness.

moonbuggy.runner.check_selection_is_runnable(project_dir: str | PathLike[str], selected: Collection[str]) → None

Verify pytest can resolve the tests selection is about to ask for.

Parameters:

• project_dir – the project root, which is also pytest’s working directory for every run moonbuggy makes.

• selected – pytest node ids from the coverage map.

Returns:

None. The check either passes silently or raises.

Raises:

BaselineError – if any node id names a file that does not exist, which means the whole map is expressed in the wrong frame of reference and no result from it would mean anything.

Reporting: canonical JSONL, plus a plaintext view derived from it.

The output is designed for a reader – human or agent – grepping the report rather than viewing a dashboard, which makes the format the feature:

• A fixed leading status keyword, so grep SURVIVED works with zero knowledge of the schema. The keyword is padded to a fixed-width column for the eye’s benefit; the padding is cosmetic, and a keyword longer than the column overflows it rather than widening it – NO_COVERAGE and KILLED_BY_ERROR both do. What a line guarantees is therefore the order of its fields under whitespace splitting (field 0 is the status, field 1 is file:line, and so on), never a column offset. Split, do not slice.

• key=value tokens rather than prose, so naive whitespace splitting parses it.

• Exactly one line per mutant, so grep and awk stay usable. The diff is deliberately NOT inlined – moonbuggy show <id> retrieves it.

The plaintext is derived from the JSONL rather than authored alongside it, so the two cannot drift apart by construction – which is why render_line takes a record dict rather than a Result.

class moonbuggy.report.Record

Bases: TypedDict

The canonical JSONL record for one mutant – record_for’s return shape.

Every key is required, including the ones added after the format shipped. That is what schema buys: read_jsonl() upgrades an older line to this shape as it reads it, so the promise holds for a file written by any version rather than only for a record made in this process.

schema: int

id: str

status: str

file: str

line: int

operator: str

category: str

nearest_test: str | None

tests_run: int

duration: float

module_level: bool

suppressed: bool

logging_call: bool

original: str

mutated: str

diff: str

accepted: bool

accept_reason: str | None

killreason: str | None

Why this mutant was killed, or None. One of the moonbuggy.killreason enumeration, stable across runs – a consumer comparing two records compares this token directly. None for any status where no reason applies (survivors, timeouts, uncovered lines, skipped mutants). Added with record schema 4.

moonbuggy.report.record_for(result: Result, reason: str | None = None) → Record

The canonical record for one mutant. This is the JSONL line’s content.

Parameters:

• result – the mutant’s verdict.

• reason – the accepted-equivalents ledger’s reason for this mutant, if a live acceptance covers it. Stamped onto the record rather than kept beside it, because a reader filtering results.jsonl for real findings needs the acceptance in the same object as the status – jq ‘select(.status==”SURVIVED” and .accepted|not)’ is the question, and it cannot be asked across two files.

Returns:

The record.

moonbuggy.report.write_jsonl(results: Iterable[Result], path: str | PathLike[str], reasons: Mapping[str, str] | None = None) → None

Stream records to disk, one complete line at a time.

Flushed per record so a run killed partway leaves only whole, parseable lines behind. A half-written final record would break every downstream reader, which matters more here than the cost of the flush.

Parameters:

• results – the verdicts to write, in the order to write them.

• path – the file to write.

• reasons – accepted-equivalents reasons by mutant id, as accepted.Resolution.reasons returns.

class moonbuggy.report.StreamingJSONL(path: str | PathLike[str], reasons: Mapping[str, str] | None = None)

Bases: object

Write records as they are produced, keeping the file valid throughout.

A run killed mid-flight has to leave something a later reader can parse, not a truncated final line. Every record is written and flushed whole, so the file is a valid JSONL document at every instant between writes – it is only ever incomplete, which readers can see for themselves by counting lines.

Records arrive in completion order rather than mutant order. That is fine for a partial file, and the caller rewrites the whole thing in canonical order once the run finishes.

write(result: Result) → None

Append one result. Safe to call from a runner callback.

moonbuggy.report.read_jsonl(path: str | PathLike[str]) → list[Record]

Read every record back from a JSONL file, in file order.

Records written by an older moonbuggy are upgraded to the current shape as they are read, so a caller never has to ask which version wrote the file.

Parameters:

path – the file to read.

Returns:

The records, each one a complete Record.

moonbuggy.report.render_line(record: Record) → str

One plaintext line for one record. Never contains a newline.

Whitespace-separated, in a fixed field order: status, file:line, category, line=, nearest_test=, tests_run=, killreason=, id=. The status is padded to nine columns for the eye only – NO_COVERAGE and KILLED_BY_ERROR are longer and push the rest of the line right, so parse by splitting rather than by column.

Parameters:

record – the record to render.

Returns:

The line.

moonbuggy.report.plaintext_from_records(records: Iterable[Record]) → str

The whole plaintext view: one line per record, no trailing newline.

moonbuggy.report.summarise(records: Iterable[Record]) → dict[str, int]

Counts per status, for the run’s final line.

moonbuggy.report.run_summary(records: Iterable[Record], *, elapsed: float, cached: int, config: Mapping[str, object], scope: Mapping[str, object], acceptance: Mapping[str, object], exit_code: int) → dict[str, object]

The whole run as one JSON-serialisable object.

A run has exactly one of these, which is why it is an object and not a line in results.jsonl: that file is one record per mutant, and a reader counting its lines is counting mutants. Keeping the summary out of it means no consumer has to learn a discriminator to avoid mistaking the two.

The config, scope and acceptance mappings are carried through verbatim from the components that own them – moonbuggy.diffscope.scope_summary() and moonbuggy.accepted.Acceptance.summary() – so the summary reports what those made of the run rather than a second opinion about it.

Parameters:

• records – the run’s records, as results.jsonl holds them.

• elapsed – wall time for the whole run, in seconds.

• cached – how many verdicts were served from the results cache.

• config – the run’s effective configuration.

• scope – the run’s diff scope, from scope_summary.

• acceptance – the ledger outcome, from Acceptance.summary.

• exit_code – the code the process is about to exit with, so a consumer reading the file afterwards need not re-derive the gate’s answer.

Returns:

schema (this document’s version), record_schema (the version of the records beside it), moonbuggy (the version string), total, cached, measured, elapsed, exit_code, counts (one lower-cased key per status keyword), acceptance, scope and config (the three mappings carried through verbatim).

Return type:

The summary, with twelve keys

moonbuggy.report.write_summary(summary: Mapping[str, object], path: str | PathLike[str]) → None

Write one run summary to disk, as a single JSON object.

Parameters:

• summary – the summary, as run_summary() returns it.

• path – the file to write.

moonbuggy.report.find_record(records: Iterable[Record], mutant_id: str) → Record | None

The record with this mutant id, or None if there is not one.

One named mutant: re-measured (moonbuggy run), or explained (moonbuggy why).

moonbuggy show <id> prints a mutant’s record and diff but cannot run it, so the fix-verify loop – “I think this test kills that mutant, let me check” – had no cheap path and ended in hand-applied mutations and a manual pytest invocation. This module is the missing half: the same coverage-guided selection and the same runner, pointed at one mutant instead of all of them, and generation widened to every operator so an id from a –operators deep run still resolves.

Three decisions worth stating, because each of them could reasonably have gone the other way.

The cache is written but never read – except for SKIPPED and SUSPICIOUS, which say nothing about the mutation and are not stored (see _NOT_CACHEABLE). The whole point of the command is to re-measure, so serving the previous verdict would answer the one question the user is asking. Refusing to store the fresh verdict would be a different mistake: the measurement is a real one, keyed on the same run_fingerprint() a full run uses, so the next full run can honour it instead of paying for it again. That is the payoff of the loop – verify a fix here, and the run in CI is already shorter.

Verdicts are fresh, artifacts are not. results.jsonl stays exactly as the last full run left it. It is the canonical record of a run, and rewriting one line of it from a single-mutant measurement would leave a file whose summary no longer describes its contents.

The coverage pass is not narrowed to the target’s module. The issue asked for it and coverage.py cannot do it: –cov=path/to/one.py reports module-not-imported and collects nothing, since coverage’s source list takes directories and importable names, not files. It would buy little anyway – the suite still has to run, and instrumentation scope is not what that costs.

explain() is the other half of the same machinery, and the one thing it does not do is run the mutant. “Selection never picked up my new test” and “this verdict came from the cache” produce the identical symptom – a survivor that will not die – and the only way to tell them apart used to be an experiment. So why reports the decision instead of the outcome: the same coverage pass and the same select_for(), plus a lookup of the cache key those inputs produce. Re-measuring is what run is for, and a why that also measured would be a slower run rather than a different command.

There is no line->test map on disk to read this out of, and this deliberately does not add one. Persisting the map would be a cache with all of a cache’s staleness problems and none of the results cache’s key discipline, and the answer a user wants from why is about the source as it stands now – which is exactly what a stored map cannot be.

exception moonbuggy.verify.VerifyError

Bases: RuntimeError

A named mutant cannot be run, and the user has to fix something.

Raised rather than reported, so the CLI’s single failure funnel turns it into an exit 2 with a message and never a traceback.

class moonbuggy.verify.Verification(result: Result, selected: tuple[str, ...], failed: tuple[str, ...], reason: str | None = None, prior: PriorFinding | None = None)

Bases: object

One mutant, re-measured, with the evidence behind the verdict.

The evidence is the reason this is not just a Result: a full run reports one line per mutant and cannot afford to name every selected test, but a user asking about a single mutant is asking exactly which tests were consulted and which of them objected.

result: Result

The runner’s verdict, unchanged.

selected: tuple[str, ...]

Every test node id selection chose, sorted.

failed: tuple[str, ...]

The selected tests that failed under the mutation, sorted. Empty for a survivor by definition, and empty for a TIMEOUT because the run was killed before it could report anything.

reason: str | None = None

The accepted-equivalents ledger’s reason for this mutant, if a live acceptance covers it.

prior: PriorFinding | None = None

The export record this re-measurement was made against, when the run named –against. None for a plain re-measurement. Kept separate from the verdict on purpose: the transition is a pairing of frozen export record and fresh trace, and only this field knows the prior side.

property mutant: Mutant

The mutant this verifies.

property status: str

The fresh verdict.

trace() → dict[str, object]

The verdict plus the exact evidence that produced it.

The persisted verdict-trace audit log’s per-mutant record, consumed by moonbuggy run <id> –trace-json. Like summary(), one place decides the keys, so the human block and the JSON trace cannot describe the same measurement differently – and like the agent line, the leading verdict.status is always a bare keyword from moonbuggy.report.STATUS_KEYWORDS, so jq .verdict.status selects the same mutants grep KILLED does.

The evidence discipline: every verdict names something. A KILLED or KILLED_BY_ERROR names the assertion (or erroring test) that died; a SURVIVED identifies the surviving mutation itself, since no test objected; a NO_COVERAGE says nothing ran; and a SUSPICIOUS names its cause, which the killreason already distinguishes (flaky_probe vs execution_crash) – at –flaky-probe 0 only the crash path can produce it, and the trace says which.

Returns:

A mapping with trace_schema, id, file, line, operator, original, mutated, selection (selected, failed) and verdict (status, killreason, assert, crash). When this verification was made against a survivor-export record (–against on the command line), a reinject key is added: the frozen prior verdict paired with the fresh one, under reinject_schema 1, with the closed transition token a consumer gates on.

summary() → dict[str, object]

The verification as JSON-serialisable data.

Follows moonbuggy.diffscope.scope_summary() and moonbuggy.accepted.Acceptance.summary(): one place that decides which keys a structured consumer sees, so the human view and a future –json cannot describe the same measurement differently.

Returns:

A mapping with id, status, file, line, operator, original, mutated, tests_run, selected, failed, nearest_test, duration, accepted and accept_reason.

class moonbuggy.verify.Explanation(mutant: Mutant, selected: tuple[str, ...], selection: str, flaky: tuple[str, ...] = (), cache_key: str | None = None, cached: CacheRecord | None = None, last_run: Record | None = None, fingerprint_inputs: dict[str, object] | None = None, reason: str | None = None)

Bases: object

Why one mutant would be handled the way it would be, without running it.

Everything here is a decision rather than a measurement: which tests selection picks and on what grounds, and whether the results cache already holds an answer for those inputs. Between them they settle the question why exists for – “is my new test being ignored, or am I being served a stale verdict?” – which no single field can answer on its own.

mutant: Mutant

The mutant, regenerated from the source as it stands now.

selected: tuple[str, ...]

Every test node id selection would choose, sorted. Empty means either that no test reaches the line (a run reports NO_COVERAGE) or that the mutant is suppressed and nothing would run (SKIPPED) – selection says which.

selection: str

coverage (tests the instrumented pass saw execute the line), module_level (an import-time line, which widens to the whole suite) or suppressed (a # moonbuggy: skip marker, or a mutation inside a logging call the run did not ask for, so nothing runs).

Type:

How that set was arrived at

flaky: tuple[str, ...] = ()

Selected tests that disagreed with themselves between unmutated runs. Always empty unless probes were requested – why measures nothing by default, and a flakiness probe is a measurement.

cache_key: str | None = None

The key_for() digest these inputs produce, or None when the caller asked for no cache.

cached: CacheRecord | None = None

The entry stored under cache_key, if there is one. Present means the next full run replays this instead of measuring.

last_run: Record | None = None

This mutant’s record in results.jsonl, if the last run reported it. A historical verdict, and deliberately separate from cached: the two disagree exactly when something has changed since, which is worth seeing.

fingerprint_inputs: dict[str, object] | None = None

The run inputs mixed into the cache key by run_fingerprint() – pytest_args, timeout and python. Named here because a key nobody can see the inputs of is not an explanation.

reason: str | None = None

The accepted-equivalents ledger’s reason for this mutant, if a live acceptance covers it.

property cache_covers: tuple[str, ...]

Every file whose contents change this mutant’s cache key.

The mutated module plus one entry per selected test file. Edit any of them and the key changes, which is precisely why a new test cannot be hidden behind a stale hit – provided selection picked it up, which is the other half of this report.

property next_run: str

What a full run would do with this mutant now, without doing it.

Mirrors moonbuggy.runner._plan()’s order of decisions, which is the only way this can be a prediction rather than a guess: skipped for a suppressed mutant, then suspicious for a flaky selection, then cache when an entry exists, then no_coverage for an empty selection, and measure when a process really would be spent on it.

summary() → dict[str, object]

The explanation as JSON-serialisable data.

Follows moonbuggy.diffscope.scope_summary() and Verification.summary(): one place decides which keys a structured consumer sees, so the human block and –json cannot describe the same mutant differently.

Returns:

A mapping with id, file, line, operator, original, mutated, selection, selected, tests_run, flaky, next_run, cache_key, cache_covers, cache_hit, cached_status, cached_tests_run, run_inputs, last_run_status, last_run_tests_run, accepted and accept_reason.

moonbuggy.verify.explain(project_dir: str | PathLike[str], mutants: list[Mutant], source_dir: str | PathLike[str], probes: int = 0, extra_args: Iterable[str] = (), python: str | None = None, cache: ResultCache | None = None, fingerprint_inputs: dict[str, object] | None = None, reasons: Mapping[str, str] | None = None, records: Mapping[str, Record] | None = None) → list[Explanation]

Work out how each mutant would be handled, and run none of them.

One coverage pass answers the selection question for every target at once, exactly as verify() does. What follows it is lookups: the cache key those inputs produce, and whether anything is stored under it.

probes defaults to 0 rather than 1 here, unlike verify. Each probe is another whole unmutated suite run, and it buys only the SUSPICIOUS prediction – a poor trade for a command whose selling point is that it answers without measuring. Pass a positive value to have flaky tests in the selection reported.

Parameters:

• project_dir – the project root.

• mutants – the targets, in the order to report them.

• source_dir – the directory to measure coverage of.

• probes – extra unmutated suite runs used to detect flaky tests. 0 skips the detection rather than skipping the coverage pass.

• extra_args – pytest arguments added to the coverage pass. The same ones the run being explained uses, or this explains a different suite.

• python – interpreter for the coverage pass; None uses this one.

• cache – the ResultCache a run would consult, or None. Only ever read from – why stores nothing, because it measures nothing worth storing.

• fingerprint_inputs – the pytest_args/timeout/python behind cache’s fingerprint, for reporting. Not used to derive anything.

• reasons – accepted-equivalents reasons by mutant id, as moonbuggy.accepted.Resolution.reasons() returns.

• records – the last run’s records by mutant id, as moonbuggy.report.read_jsonl() supplies them.

Returns:

One Explanation per mutant, in the input order.

Raises:

• BaselineError – if the suite is already failing or collects nothing. Selection derived from a red suite would not describe any run moonbuggy is willing to make.

• CoveragePassError – if the instrumented run could not complete.

moonbuggy.verify.resolve_targets(project_dir: str | PathLike[str], mutant_ids: Iterable[str], logging_policy: LoggingPolicy | None = None) → list[Mutant]

Find the mutant each id names, by regenerating the module it points at.

Regeneration rather than a lookup in results.jsonl, deliberately. The caller is in the middle of editing, and the mutant it needs is the one that stands in the source now – the previous run’s record carries the source line as it was, which is the one thing a fix-verify loop must not be measured against. It also means the command works before any run has happened.

Parameters:

• project_dir – the project root. Ids name modules relative to it.

• mutant_ids – ids as printed in id=…, in the order to run them.

• logging_policy – the run’s logging policy. Pass the same one the full run used: it decides whether a mutant inside a logging call comes back suppressed, and an explanation built under a different policy describes a run nobody made.

Returns:

One Mutant per id, in the same order.

Raises:

VerifyError – if an id is malformed, names a module that cannot be read, or names a mutant that module no longer produces.

moonbuggy.verify.verify(project_dir: str | PathLike[str], mutants: list[Mutant], source_dir: str | PathLike[str], timeout: float = 30.0, workers: int = 0, probes: int = 1, extra_args: Iterable[str] = (), python: str | None = None, cache: ResultCache | None = None, reasons: Mapping[str, str] | None = None, jobs: int = 1, priors: Mapping[str, PriorFinding] | None = None) → list[Verification]

Re-measure each mutant against the tests that cover it.

One coverage pass serves every target, so verifying ten survivors costs barely more than verifying one. Each mutant then runs in its own pytest subprocess rather than a fork, because a forked child reports a single exit-code byte and this command has to name the tests that failed. With jobs above 1 those subprocesses overlap; each is unchanged – same selection, same environment, same fresh verdict – so the verdicts are byte-identical to the serial ones and only the wall clock moves.

Parameters:

• project_dir – the project root.

• mutants – the targets, in the order to report them.

• source_dir – the directory to measure coverage of.

• timeout – seconds before a mutant is called TIMEOUT.

• workers – pytest-xdist workers within each mutant’s run.

• probes – extra unmutated suite runs used to detect flaky tests.

• extra_args – pytest arguments added to every run, baseline included.

• python – interpreter for the mutant runs; None uses this one.

• cache – a ResultCache to write fresh verdicts into, or None. Never read from – see the module docstring. SKIPPED and SUSPICIOUS verdicts are not written (_NOT_CACHEABLE); everything else is.

• reasons – accepted-equivalents reasons by mutant id, as moonbuggy.accepted.Resolution.reasons() returns.

• jobs – how many mutant runs to hold open at once. 1 is the serial behaviour; each run is an independent subprocess, so the value changes overlap and nothing else.

• priors – prior findings by mutant id, from the survivor-export the run was named against (–against). Each verification carries its prior so the trace can pair the frozen record with the fresh verdict; an id with no prior gets none, and a prior nobody re-measured is the caller’s gap to report, not this function’s.

Returns:

One Verification per mutant, in the input order.

Raises:

• BaselineError – if the suite is already failing or collects nothing.

• CoveragePassError – if the instrumented run could not complete.

The accepted-equivalents ledger: a triage decision that outlives the run.

A triage session’s real end state is “three survivors, reviewed, verified equivalent”. Without somewhere to put that, the next run reports the same three forever, every reviewer pays the same cost again, and a CI gate can never be green. This module is that somewhere: a small TOML file, checked into version control next to the code it makes claims about, holding one entry per mutant a human has decided is equivalent – and the reason they decided it.

Three decisions shape everything below.

Accepted mutants still run, and are still reported. Acceptance is an annotation, never a filter. A mutant dropped from generation would take its verdict with it, and the day the code changes underneath the acceptance there would be nothing left to notice. So the run is identical either way; only the counting and the presentation differ, and moonbuggy says how many acceptances it honoured every time it honours one.

Drift expires an acceptance. An entry stores a fingerprint of the mutation it was made for – the operator, the original line and the mutated line. If the source under it changes, the fingerprint no longer matches, the entry is stale, and the mutant is reported as unexplained. Silently honouring a stale acceptance is precisely how a real regression sneaks in behind a decision somebody made about different code last year.

Stale is the label drift gets when the entry’s id still names some mutant in this run. Edit the line and shift it – insert something above it – and the id no longer resolves, so the same drift is classified orphaned instead (step 4 below). The safety property is unaffected either way: neither is honoured, and the mutant is still reported unexplained. What changes is the reporting, since the human report names stale entries and says nothing about orphaned ones.

The fingerprint deliberately covers the mutated line, not the whole module. A module-wide hash is the stricter reading of drift, and it was rejected: it expires every acceptance in a file on any edit to that file, including a comment, so a working ledger would be a ledger of nothing within a week – and re-accepting fifty entries in bulk is not a review, it is a ritual. The cost of the narrower hash is stated rather than hidden: a change elsewhere in the module can invalidate an equivalence argument without expiring the entry. The reason field is what a reviewer re-reads when that matters, which is why writing one is mandatory.

Acceptance keys on the id first and on content second. Mutant ids are path:line:operator:index, so inserting a line above shifts every id below it, and a ledger keyed on the id alone would silently lose its entries to an unrelated edit – worse than no ledger, because the loss is invisible. A ledger keyed on content alone cannot distinguish two identical lines in one file, and would honour a decision about one of them for the other. So the match is:

1. the id resolves and the fingerprint agrees -> honoured;

2. otherwise, exactly one unclaimed mutant in the same file carries the same fingerprint -> honoured, and reported as relocated (the id shifted);

3. otherwise, if the id resolves but the fingerprint disagrees -> stale;

4. otherwise -> orphaned: no such mutant in this run, which is the normal state of most entries under –since and is usually not a finding. A heuristic rather than a rule – drift that also moved the id lands here too, as the drift paragraph above explains.

Ambiguity in step 2 – two identical candidates, no exact id – is refused rather than guessed. Equivalence is a judgement about a line in its context, and picking one of two candidates at random honours a decision nobody made about it.

The file is written by hand rather than through a TOML library: the schema is six flat string keys, tomllib reads it back on every supported Python, and adding a runtime dependency to emit six keys would be the more expensive mistake. tomllib is the authority on what a valid file is, so anything this module writes is read back by the same parser everyone else uses.

exception moonbuggy.accepted.AcceptError

Bases: RuntimeError

Raised when the ledger cannot be read, written, or trusted.

Always actionable and never a traceback: cli.main turns this into exit code 2 with the message alone, like every other anticipated failure.

class moonbuggy.accepted.Entry(id: str, file: str, operator: str, fingerprint: str, reason: str, accepted_at: str)

Bases: object

One accepted mutant, as stored in the ledger.

id: str

The mutant id at the moment of acceptance. A hint, not the identity – see the module docstring: the fingerprint outranks it when they disagree.

file: str

The module the mutant lives in, relative to the project root. Bounds a content match to one file, so two identical lines in different modules cannot borrow each other’s decision.

operator: str

The operator that produced the mutation.

fingerprint: str

A digest of (operator, original, mutated). The drift detector.

reason: str

an acceptance without one is a claim nobody can check and nobody will revisit.

Type:

Why a human decided this mutant is equivalent. Mandatory

accepted_at: str

The ISO date the entry was written, so a reviewer can see its age.

class moonbuggy.accepted.Resolution(live: dict[str, ~moonbuggy.accepted.Entry]=<factory>, relocated: dict[str, str]=<factory>, stale: tuple[~moonbuggy.accepted.Entry, ...]=(), ambiguous: tuple[~moonbuggy.accepted.Entry, ...]=(), orphaned: tuple[~moonbuggy.accepted.Entry, ...]=())

Bases: object

How each ledger entry lines up with the mutants this run generated.

live: dict[str, Entry]

Honoured acceptances, keyed by the mutant id in this run – which is not the entry’s stored id when the entry was relocated.

relocated: dict[str, str]

Stored id -> current id, for entries whose line number moved.

stale: tuple[Entry, ...] = ()

Entries whose mutation has changed under them. Reported, not honoured.

ambiguous: tuple[Entry, ...] = ()

Entries with more than one equally good candidate. Not honoured.

orphaned: tuple[Entry, ...] = ()

Entries matching no mutant in this run – normal under –since.

reasons() → dict[str, str]

The honoured reasons, keyed by this run’s mutant ids.

Returns:

A mapping suitable for report.record_for, which stamps the reason onto the mutant’s record so results.jsonl says why a survivor was not counted against the run.

class moonbuggy.accepted.Acceptance(path: str, accepted: tuple[str, ...], unexplained: tuple[str, ...], stale: tuple[Entry, ...], ambiguous: tuple[Entry, ...], orphaned: tuple[Entry, ...], relocated: dict[str, str], gating: bool)

Bases: object

A run’s ledger outcome: what was accepted, and what was not.

Structured rather than formatted, because the same numbers are wanted in three places – the human footer, the exit code, and the –json run summary – and a string is only usable by the first of them.

path: str

The ledger’s display path, for a message that names the file to edit.

accepted: tuple[str, ...]

Ids of findings covered by a live acceptance, in report order.

unexplained: tuple[str, ...]

Ids of findings that are not. This is what –fail-on-unexplained gates on.

stale: tuple[Entry, ...]

Acceptances that expired because their line changed.

ambiguous: tuple[Entry, ...]

Acceptances with no unambiguous candidate.

orphaned: tuple[Entry, ...]

Acceptances for mutants this run did not generate.

relocated: dict[str, str]

Acceptances honoured under a shifted id.

gating: bool

Whether –fail-on-unexplained decides this run’s exit code.

summary() → dict[str, object]

The counts as JSON-serialisable data.

Returns:

A mapping with accepted, unexplained, stale, ambiguous, orphaned and relocated counts, the ledger path, and whether the run was gated on unexplained findings.

moonbuggy.accepted.fingerprint(operator: str, original: str, mutated: str) → str

A digest of one mutation’s content, independent of where it sits.

Parameters:

• operator – the operator’s name.

• original – the source line before mutation, as generate stores it.

• mutated – the same line after mutation.

Returns:

A short hex digest. Short because this is a drift detector a human reads in a diff, not a security boundary – and because a mutation whose 64-bit digest collides with another’s, in the same file, is not a failure mode worth trading readability for.

moonbuggy.accepted.entry_for(mutant_id: str, module: str, operator: str, original: str, mutated: str, *, reason: str, at: str | None = None) → Entry

Build a ledger entry for one mutation.

Takes the mutation’s parts rather than a Mutant or a Record, because both callers have one of those and neither has the other: moonbuggy accept reads a record out of results.jsonl, and the run resolves against freshly generated mutants.

Parameters:

• mutant_id – the mutant’s id at the moment of acceptance.

• module – the file it lives in, relative to the project root.

• operator – the operator that produced it.

• original – the source line before mutation.

• mutated – the same line after mutation.

• reason – why it is equivalent. Mandatory.

• at – the ISO date to stamp, or None for today.

Returns:

The Entry to store.

moonbuggy.accepted.load(path: str | PathLike[str]) → tuple[Entry, ...]

Read the ledger, in file order.

Parameters:

path – the ledger file. A missing file is an empty ledger, not an error – every project starts without one.

Returns:

The entries, in the order they appear in the file.

Raises:

AcceptError – if the file is not valid TOML, is missing a required field, was written by a newer moonbuggy, or holds two entries for one mutant id.

moonbuggy.accepted.save(path: str | PathLike[str], entries: Iterable[Entry]) → None

Write the whole ledger, replacing whatever was there.

Entries are written in id order rather than in the order they were accepted, so two people accepting different mutants on two branches produce a diff a merge can resolve line by line instead of a reordered file.

Parameters:

• path – the ledger file. Its parent directory is created if needed.

• entries – the entries to store.

Raises:

AcceptError – if the file cannot be written.

moonbuggy.accepted.resolve(entries: Sequence[Entry], mutants: Sequence[Mutant]) → Resolution

Match ledger entries against the mutants this run generated.

The matching rules, and why they are these rules, are in the module docstring – they are the ledger’s answer to drift and to id stability, and they belong next to the format they justify.

Parameters:

• entries – the ledger, as load() returns it.

• mutants – every mutant this run will execute.

Returns:

The Resolution for this run.

moonbuggy.accepted.tally(records: Iterable[Record | Mapping[str, object]], resolution: Resolution, *, path: str, gating: bool) → Acceptance

Split this run’s findings into accepted and unexplained.

Only findings can be either: a KILLED mutant with an acceptance on file is simply killed, and counting it as accepted would report a decision that no longer does any work as if it were live.

Parameters:

• records – this run’s records, in report order.

• resolution – the ledger matched against this run’s mutants.

• path – the ledger’s display path, carried into the result so a message can name the file to edit.

• gating – whether –fail-on-unexplained decides the exit code.

Returns:

The run’s Acceptance.

moonbuggy.accepted.is_git_ignored(path: str | PathLike[str]) → bool

Whether git would ignore this path.

Asked once, when the ledger is first written, because .moonbuggy/ is a directory most projects ignore – moonbuggy’s own repository did – and a ledger nobody can commit is a ledger that vanishes on the next clone. A warning at the moment of writing is the only time the answer is actionable.

Parameters:

path – the ledger file.

Returns:

True if git reports the path as ignored. False for anything else, including “not a git repository” and “git is not installed”: neither is a reason to warn.

Mutation operators

Adding an operator means adding a file here and nothing else – no edit to the engine’s traversal, no import list to update, no registration call to remember. See Writing an operator.

Mutation operators, and the registry that discovers them.

Operators are discovered by importing every module in this package, so adding one means adding a file here and nothing else – no edit to the engine’s traversal, no import list to update, no registration call to remember. The engine asks for all_operators() and never learns their names.

An operator is any class decorated with @register that provides:

name str – stable identifier, matches the operator names in oracle.toml mutations() takes an AST node, yields replacement nodes (possibly none)

and may optionally declare, as plain class attributes:

tier str – “default” (the default) or “deep” cost str – “low” (the default), “medium” or “high” description str – one line for the listing, defaulting to the docstring

Tier lives on the operator for the same reason discovery does: a tier table in a central file would mean adding an operator required editing two places, and the one property this package is built around is that it requires editing none.

An operator whose decision depends on where the node sits implements mutations_in_context(node, context) instead, and is handed a Context describing its parent, the field it occupies, and the chain of enclosing nodes. Either method may yield a bare replacement for the node it was given, or a (target, replacement) pair naming a different node to rewrite.

Operators must not mutate the node they are given. replace_operator is the supported way to obey that without paying for a deep copy.

class moonbuggy.operators.Context(parent: AST | None, field: str | None, index: int | None, outer: Context | None)

Bases: object

Where a node sits in the tree it was walked from.

Handed to operators that implement mutations_in_context. Without it an operator sees a bare ast.Name and cannot tell the test of an if from the right-hand side of an assignment – and “negate every name” is absurd where “negate the ones in test position” is exactly right.

Frozen, and built lazily: outer links to the parent node’s own context rather than materialising an ancestor tuple per node, so a deeply nested expression costs one small object per node instead of one tuple per node whose length grows with depth.

parent: AST | None

The enclosing node, or None for the root of the walk.

field: str | None

The field of parent holding this node – “test”, “body”, “ifs”, “defaults”. None at the root.

index: int | None

This node’s position within field when that field is a list, and None when it holds a single node. “Where am I in the enclosing body?” is this, plus parent.

outer: Context | None

The parent node’s own context, or None at the root.

property ancestors: tuple[AST, ...]

Every enclosing node, outermost first.

Returns:

the chain from the root of the walk down to the immediate parent, empty at the root.

nearest(*types: type[AST]) → AST | None

The closest enclosing node of any of types.

“What is my nearest enclosing Call?” is the question this answers, and it walks outwards lazily rather than through ancestors, so the common case of a hit one or two levels up costs two comparisons.

Parameters:

*types – AST node classes to look for.

Returns:

the innermost enclosing node matching any of types, or None.

class moonbuggy.operators.NodeOperator(*args, **kwargs)

Bases: Protocol

The shape of an operator that needs nothing but the node.

A class decorated with @register needs a stable name and a mutations method that takes an AST node and yields its mutated replacements (or nothing). This is a structural type, not a base class – operator modules never import it and never subclass anything; they just happen to match it.

name: str

mutations(node: AST) → Iterator[AST | tuple[AST, AST]]

Yield replacements for node, or nothing if none apply.

class moonbuggy.operators.ContextualOperator(*args, **kwargs)

Bases: Protocol

The shape of an operator that needs to know where the node sits.

A second method name rather than a second parameter on mutations, so that an operator which does not need context does not have to accept it. The engine tells the two apart by which method is present; nothing else about registration or discovery changes.

name: str

mutations_in_context(node: AST, context: Context) → Iterator[AST | tuple[AST, AST]]

Yield replacements for node in context, or nothing if none apply.

moonbuggy.operators.replace_operator(node: NodeT, **changes: object) → NodeT

A shallow copy of node with some fields replaced.

Every operator needs the same thing: the original node with one field different. The obvious way to write that is copy.deepcopy(node) followed by an assignment, and every operator did – which makes mutating one expression cost a copy of its entire subtree. An expression with n nested operators then costs O(n^2) node copies, and a 6000-term expression took over a minute to generate.

A shallow copy is enough because nothing downstream writes to the tree. Generation unparses the replacement node and throws it away; the children it shares with the original are only ever read. What the deep copy actually bought was protection against an operator that mutated its input, and that protection is better provided by not writing such an operator – which this function makes the path of least resistance.

Parameters:

• node – the AST node being mutated. Not modified.

• **changes – field values to replace on the copy. Untyped because the fields vary per AST node type (op for a BinOp, ops for a Compare, args for a Call); ast.AST itself declares no common field set to check them against.

Returns:

a new node of the same type, sharing the untouched children.

moonbuggy.operators.DEFAULT_TIER = 'default'

The tier an operator lands in when it does not say. Every operator that predates tiers is cheap and high-signal, which is what default means, so silence and the truth coincide.

moonbuggy.operators.TIERS = ('default', 'deep')

The tiers an operator may declare itself into, cheapest first.

default is what a bare moonbuggy runs: cheap to run, high signal. deep is for operators that are expensive in wall-clock or noisy in output, and is opted into deliberately – statement_deletion is its first member, at roughly one extra mutant per statement.

moonbuggy.operators.ALL_TIER = 'all'

Not a tier an operator can declare, but a selector meaning every registered operator regardless of tier. Kept out of TIERS so that “which tier is this operator in?” and “what can I type after –operators?” stay two questions.

moonbuggy.operators.RESERVED_SELECTORS = frozenset({'all', 'deep', 'default'})

Selector words an operator may not take as its name. Enforced at registration rather than documented, because the failure it prevents is silent: an operator file named deep would shadow the tier, and every existing –operators deep would quietly start meaning something else.

moonbuggy.operators.COSTS = ('low', 'medium', 'high')

Rough wall-clock-and-noise cost, for the moonbuggy operators listing. Deliberately three coarse buckets rather than a number: the honest claim is an ordering, and a measured per-operator cost would be a property of the project under test, not of the operator.

moonbuggy.operators.DEFAULT_COST = 'low'

The cost an operator claims when it does not say.

exception moonbuggy.operators.SelectionError

Bases: Exception

A –operators selection that cannot be resolved to a set of operators.

Raised for an unknown name, a malformed + token, or a selection that resolves to no operators at all. The last one matters most: before tiers, a typo’d operator name produced a run with zero mutants that exited 0 and read exactly like a clean bill of health.

class moonbuggy.operators.OperatorInfo(name: str, tier: str, description: str, cost: str)

Bases: object

One operator, as moonbuggy operators reports it.

name: str

The name used in mutant ids, in results.jsonl, and in –operators.

tier: str

Which of TIERS it belongs to.

description: str

One line about what it mutates.

cost: str

One of COSTS.

moonbuggy.operators.register(cls: OperatorT) → OperatorT

Register an operator class. Used as a decorator.

Parameters:

cls – the operator class.

Returns:

cls, unchanged.

Raises:

ValueError – if the name collides with a reserved selector word, or the declared tier or cost is not one this version knows. A contributor error rather than a user error, so it is loud at import time.

moonbuggy.operators.describe_operators() → list[OperatorInfo]

Every registered operator’s name, tier, description and cost.

What moonbuggy operators prints, in both of its forms. The description falls back to the first line of the class docstring, so an operator author who writes the docstring they were going to write anyway gets a usable listing entry for free.

Returns:

One entry per operator, sorted by name – the same order all_operators() uses.

moonbuggy.operators.tier_members(tier: str) → tuple[str, ...]

The operator names in tier, sorted.

Parameters:

tier – one of TIERS, or ALL_TIER.

Returns:

The names, sorted. Empty is a legitimate answer – a tier with no members is a state this version has been in and can be in again.

moonbuggy.operators.resolve_operators(spec: str) → frozenset[str]

Turn a –operators selection into the set of names it selects.

The grammar is comma-separated tokens, each either a bare selector or one prefixed with +:

• A bare list of names is an exact set. comparison_swap,boundary meant exactly those two before tiers existed and means exactly those two now; everything else here is layered on top of that, never underneath it.

• A bare tier name expands to that tier’s members.

• A + token adds to a base rather than replacing it. The base is whatever the bare tokens named, or the default tier when there are none – so +statement_deletion is “the ordinary run plus this one”, which is the case that would otherwise mean typing out the default set by hand.

Parameters:

spec – the raw –operators value.

Returns:

The selected operator names.

Raises:

SelectionError – for an unknown selector, a bare +, or a selection that resolves to no operators at all.

moonbuggy.operators.all_operators() → list[NodeOperator | ContextualOperator]

Every registered operator, instantiated.

Sorted by name so mutant ordering – and therefore mutant ids – stay stable across runs regardless of filesystem iteration order.

Arithmetic operator swaps.

Applies to binary operations and to augmented assignment – += is a site in its own right, not merely a +. See operators.arithmetic_swap in oracle.toml.

class moonbuggy.operators.arithmetic.ArithmeticSwap

Bases: object

Swap one arithmetic operator for a plausibly-mistaken neighbour.

Fires on binary operations and on augmented assignment. += is a site in its own right rather than merely a +, because writing -= where += belongs is its own mistake.

name = 'arithmetic_swap'

mutations(node: AST) → Iterator[AST]

Yield the node with its arithmetic operator swapped.

Parameters:

node – any AST node; non-arithmetic nodes yield nothing.

Yields:

One replacement node, if this node has a swappable operator.

Boolean operator swaps.

Binary boolean operators only. not is deliberately not mutated – see operators.boolean_swap in oracle.toml.

class moonbuggy.operators.boolean.BooleanSwap

Bases: object

Swap and for or and back.

Binary boolean operators only. not is deliberately left alone – see operators.boolean_swap in oracle.toml.

name = 'boolean_swap'

mutations(node: AST) → Iterator[AST]

Yield the node with its boolean operator swapped.

Parameters:

node – any AST node; only BoolOp produces a mutation.

Yields:

One replacement node, if this node is a boolean operation.

Off-by-one boundary mutation.

range(x) -> range(x - 1), single-argument form only. See operators.boundary in oracle.toml.

class moonbuggy.operators.boundary.Boundary

Bases: object

Shift a single-argument range() by one.

range(x) becomes range(x - 1). Off-by-one in a loop bound is the mistake this exists to notice, and a test that never checks the last iteration will not.

name = 'boundary'

mutations(node: AST) → Iterator[AST]

Yield the call with its range shortened by one.

Parameters:

node – any AST node; only a one-argument range() call qualifies.

Yields:

One replacement node, if this node is such a call.

Comparison operator swaps.

Each comparison maps to exactly one replacement, chosen to probe the boundary rather than to invert the predicate – see the operators.comparison_swap section of tests/fixtures/oracle.toml.

class moonbuggy.operators.comparison.ComparisonSwap

Bases: object

Swap one comparison operator for the one next to it.

Each comparison maps to exactly one replacement, chosen to probe the boundary rather than to invert the predicate – an inverted predicate is caught by almost any test, and teaches nothing.

name = 'comparison_swap'

mutations(node: AST) → Iterator[AST]

Yield one replacement per comparison operator in the node.

A chained comparison is a single node carrying several operators, and each is a separate site: mutating two at once would be a different and weaker test of the same line.

Parameters:

node – any AST node; only Compare produces mutations.

Yields:

One replacement node per swappable operator.

Condition negation.

Wraps the test of an if, an elif, a conditional expression, or a comprehension guard in not (…). One mutant per condition. See operators.condition_negation in oracle.toml.

Two deliberate exclusions. A while test is never negated: the mutant either does nothing observable or loops forever and burns the whole –timeout. A literal test is never negated: if True: already gets if False: from constant_bool, and if not True: says the same thing twice. The reasoning for each is written at its site – above TEST_HOLDERS and in mutations_in_context – and both are enforced, not aspirational.

This is the operator that reaches conditions the other five cannot. They all need a specific node type to bite – a Compare, a BoolOp, a literal – so if is_valid(x):, if flag: and if not ready: were completely unmutated, and a suite entirely blind to is_valid produced no finding at all. Predicate helpers are the ordinary way people write conditionals, not an exotic shape.

It is the first operator to use the context seam: negating every ast.Name in a module would be absurd, and negating the ones in test position is exactly right, so the decision is about where the node sits rather than what it is.

class moonbuggy.operators.condition.ConditionNegation

Bases: object

Invert a condition, whatever shape it has.

Fires on the test of if/elif, on the test of a conditional expression, and on each if clause of a comprehension. A survivor is unambiguous: an entire branch was inverted and no test noticed.

Does not fire on a while test or on a literal test; see the module docstring for why.

name = 'condition_negation'

mutations_in_context(node: AST, context: Context) → Iterator[AST | tuple[AST, AST]]

Yield the condition wrapped in not, if this node is one.

Parameters:

• node – any AST node; only an expression in test position qualifies.

• context – where node sits, which is the whole of the decision.

Yields:

One replacement node, if this node is a condition.

Constant mutation.

Split into two operators because integers and booleans need different handling and, more importantly, because bool is a subclass of int in Python. A single operator that checked isinstance(value, int) would emit a nonsense True -> 2 mutant and double-count every boolean site.

Floats are deliberately not mutated: it invites float-comparison flakiness for no extra signal. Strings are never mutated, so no mutation ever lands inside a string literal – a guarantee that falls out of the AST approach for free.

class moonbuggy.operators.constant.ConstantInt

Bases: object

Increment an integer literal by one.

Floats are deliberately not mutated: it invites float-comparison flakiness for no extra signal.

name = 'constant_int'

mutations(node: AST) → Iterator[AST]

Yield the constant, one larger.

Parameters:

node – any AST node; only a non-boolean integer constant qualifies.

Yields:

One replacement node, if this node is such a constant.

class moonbuggy.operators.constant.ConstantBool

Bases: object

Flip a boolean literal.

Separate from ConstantInt because bool is a subclass of int in Python, so one operator checking isinstance(value, int) would emit a nonsense True -> 2 mutant and double-count every boolean site.

name = 'constant_bool'

mutations(node: AST) → Iterator[AST]

Yield the constant, negated.

Parameters:

node – any AST node; only a boolean constant qualifies.

Yields:

One replacement node, if this node is such a constant.

Statement deletion: replace one statement with pass.

The flagship of the deep tier, and the highest-yield operator in the mutation-testing literature. It subsumes others for free – return x becomes pass, which is return-value mutation without a separate operator – and it reaches lines no expression-level operator can touch, because plenty of statements contain no comparison, no arithmetic and no literal at all.

It fits the single-line splice model better than deletion might suggest:

• splicing at col_offset preserves indentation, so ` foo(x)` becomes ` pass` and not a dedented pass at column zero;

• pass keeps a one-statement if/for body syntactically valid, which literal removal of the line would not;

• multi-line statements self-exclude through _splice’s existing guard, so nothing here has to special-case a call spread over four lines.

It is genuinely expensive – roughly one extra mutant per statement – which is why it declares tier = “deep” and is opted into rather than run by default.

PROVE INERT, NOT IMPACTFUL

The instinct is to detect the statements that “have an effect” and mutate those. That is the wrong direction, and the reason is asymmetric cost. Proving that a statement matters is the hard half of the problem, and a wrong answer there loses a finding – the tool silently declines to mutate a line that a suite does not check, and reports a cleaner score than the truth. Proving that a statement is inert is the easy half, and a wrong answer costs one equivalent mutant: noise in the survivor list, which a human can see and dismiss, and which moonbuggy accept exists to retire.

So the exclusions below are a closed list of shapes that are provably free of observable effect (plus the name-establishing statements – imports and type aliases – excluded for a different reason given at DELETABLE), and everything else is mutated. The impactful set is arrived at by subtraction, not by enumeration.

class moonbuggy.operators.deletion.StatementDeletion

Bases: object

Replace a statement with pass.

A survivor means the statement can be removed from the program entirely and the suite still passes – either the line is dead, or nothing checks what it does. Statements that are provably inert are never mutated, so a survivor here is a claim about the tests rather than about the mutation.

name = 'statement_deletion'

tier = 'deep'

cost = 'high'

mutations_in_context(node: AST, context: Context) → Iterator[AST | tuple[AST, AST]]

Yield pass in place of this statement, unless it is provably inert.

Parameters:

• node – any AST node; only a simple statement in a body qualifies.

• context – where node sits. Both halves of the decision need it – whether this is a body position at all, and which function body the dead-store check should look at.

Yields:

One ast.Pass, if deleting this statement could be observable.

Applying mutations

Apply a mutation to an ALREADY-IMPORTED module, in place.

This is the “function-level swap” option from section 4.2, and it exists for a reason the design doc did not anticipate: it is what makes a warm process usable.

The import-hook approach in inmemory.py can only mutate a module that has not been imported yet, so every mutant needs a process where imports have not happened – and that means paying test-module import, assert rewriting and pytest collection per mutant. Measured at ~90ms per mutant against ~12ms for pytest.main() in a process where those imports are already done.

Swapping in place lifts that restriction. A test module that did from app.thing import compute holds a reference to the FUNCTION OBJECT, so replacing that object’s __code__ changes what the test calls, with no re-import and no reference chasing.

Two shapes of mutation, two mechanisms:

• inside a function: replace the function’s __code__ with the code compiled from mutated source.

• at module level: exec the mutated line in the module’s __dict__. Functions read their globals dynamically at call time, so a rebound module-level constant takes effect immediately for every already-imported caller.

Where neither applies (a decorator has replaced the function object with a wrapper, say), this reports failure rather than guessing, and the caller falls back to the import-hook path. A silently-unapplied mutation is a false SURVIVED.

exception moonbuggy.codeswap.SwapFailed

Bases: Exception

The mutation could not be applied in place. Caller should fall back.

moonbuggy.codeswap.index_modules(targets: Iterable[str] | None = None) → None

Record resolved __file__ to module for the modules that can be swapped.

Called in the warm host once, before any grandchild is forked, for the same reason as prewarm(): the work is identical for every mutant, and a process that does it before forking does it once instead of once per mutant. Finding the module to swap was a scan of sys.modules calling Path.resolve() on each entry – 4.7ms in a process with 250 modules loaded, which is more than a fast mutant’s tests take.

Indexing everything then became the expensive part: 11.8ms in a warm host holding ~700 modules, resolving all of them so that at most a few dozen can ever be looked up. targets is the set of paths that will actually be asked for, and everything outside it is skipped after a basename comparison, which costs nothing. Path(x).resolve() over 341 modules is 6.28ms and os.path.realpath is 4.11ms, so the few survivors use the cheaper one; os.path.abspath would be 60x faster again and is wrong, because a project under a symlink (/tmp on macOS) must resolve to the same path the interpreter reports for the loaded module.

Narrowing cannot produce a wrong answer, only a slower one: module_at() already rescans sys.modules on a miss.

First entry wins, matching the scan this replaces: sys.modules keeps insertion order, so the module an aliased path resolves to does not change.

Parameters:

targets – resolved paths that may be looked up, or None to index every imported module – which is what callers that do not know the mutant set (the run_warm_batch host) still need.

moonbuggy.codeswap.module_at(target: str) → ModuleType | None

The imported module whose file is target, or None.

Answers from the index built by index_modules() when it can, and falls back to scanning sys.modules when it cannot – a module imported after the index was built is a miss in the index and a hit in the scan, and the wrong answer here is a mutation that silently does not apply.

moonbuggy.codeswap.apply_in_place(module: ModuleType, path: str | PathLike[str], line: int, mutated_text: str) → None

Mutate an imported module object in place.

Parameters:

• module – the live module object, already imported.

• path – the module’s file path.

• line – 1-based line number to replace.

• mutated_text – the replacement line.

Raises:

SwapFailed – if the mutation cannot be applied in place, so the caller falls back to the import-hook path rather than reporting a status for a mutation that never took effect.

In-memory mutation application via a meta path import hook.

This is the cold path. The default on POSIX is the warm session, which mutates modules that are already imported by swapping code objects in place (moonbuggy.codeswap); it falls back here for the mutations codeswap refuses, and this is also the only mechanism when forking is unavailable or xdist workers were asked for. Nothing below is stale – it is just no longer the whole story.

This path carries the highest design uncertainty, with three named risks identified up front. What was found for each:

1. COMPOSING WITH PYTEST’S ASSERT-REWRITE HOOK. Not the collision it looked like. pytest’s hook rewrites test modules, conftest files, and registered plugins; this hook rewrites the module under test. Those sets are disjoint, so both can sit on sys.meta_path and each handles imports the other ignores. The finder here narrows to a single file path and returns None for every other module, which keeps it that way – a broader finder that returned a spec for test modules would displace assert rewriting and silently degrade every failure message in the run.

2. LINECACHE. Handled here rather than left to the reporting layer, because by the time a traceback is formatted the original source is what gets quoted. The entry is stored with mtime None, which makes linecache.checkcache treat it as not-from-a-file and leave it alone; a real mtime would be compared against the unmutated file on disk and evicted.

3. XDIST. Not solvable here – it is a process-lifecycle problem, so it lives in plugin.py where the worker’s own pytest_configure can call install().

moonbuggy.inmemory.mutated_source(path: str | PathLike[str], line: int, mutated_text: str) → str

The file’s source with one line replaced, keeping its indentation.

Parameters:

• path – the module file being mutated.

• line – 1-based line number to replace.

• mutated_text – the replacement line, stripped of indentation.

Returns:

the full mutated source as str.

Raises:

SourceError – if the file cannot be decoded (see moonbuggy.srcio).

moonbuggy.inmemory.install(path: str | PathLike[str], line: int, mutated_text: str) → str

Apply a mutation to path for every subsequent import in this process.

Safe to call before the module is imported, which is the normal case: pytest_configure runs before test collection imports anything under test. runner._apply_in_child also calls this directly inside a forked child, re-running a mutant the warm path reported UNAPPLIED, and the same before-import guarantee holds there.

Parameters:

• path – the module file to mutate.

• line – 1-based line number to replace.

• mutated_text – the replacement line.

Returns:

the mutated source as str.

Raises:

SourceError – if the file cannot be decoded.

moonbuggy.inmemory.uninstall_all() → None

Remove every installed mutation from this process.

Only meaningful in-process; the normal lifecycle is one mutant per process, which is what makes the xdist story work. Tests need it to avoid leaking a finder into later cases.

pytest plugin that applies the active mutant in every process that runs tests.

This is the answer to risk 2 in section 4.2: pytest-xdist workers are separate processes that re-import from disk, so a mutation applied only in the controller leaves workers running unmutated code and reporting a false SURVIVED. Silent, and on a common configuration.

The fix is that the mutant’s identity travels in the environment rather than in controller memory. execnet gives each xdist worker the controller’s environment, and every worker runs its own pytest_configure – so each one installs the same mutation independently, before collection imports anything under test. No xdist hook, no serialisation of our own state, and it degrades to the serial case for free because the controller does exactly the same thing.

class moonbuggy.plugin.MutantEnvPayload

Bases: TypedDict

The JSON carried by MUTANT_ENV_VAR.

Written by runner._env_for in the parent (or controller) process and read here by every process that runs tests for that mutant – the plain serial run, a forked child, and each pytest-xdist worker alike, since all of them inherit the same environment. Crosses via an environment variable rather than a pipe, which is what lets an xdist worker – spawned by execnet, not by moonbuggy – see it at all.

path: str

line: int

mutated: str

moonbuggy.plugin.pytest_configure(config: Config) → None

Install the active mutant, if this process was told about one.

Runs in every pytest process – controller and each xdist worker – which is what makes the xdist story work without any cross-process state.

Parameters:

config – the pytest config for this process.

Why a mutant died: a failed assertion, or a test that blew up.

A kill is not a kill. When a mutation makes the code raise NameError at the first line that touches it, every test near it errors out – and the tool reports the same KILLED it reports for a test that computed the wrong answer and said so. The two are different findings about the suite:

• a failed assertion proves a test checked the behaviour the mutation changed. That is the thing mutation testing measures.

• an errored test proves only that a test executes the line. The mutant broke the code badly enough that touching it explodes; nothing was checked.

Under the default tier the distinction is a curiosity, because most mutations there produce a program that still runs and merely computes something else. Under statement_deletion it is the difference between a meaningful kill rate and a meaningless one: delete a binding and everything downstream raises. A deep-tier run without this would report a higher score for a suite that checks less.

HOW THE ANSWER TRAVELS

Every path that runs a mutant’s tests – a forked child, a warm grandchild, a python -m pytest subprocess – reports back through pytest’s exit code and nothing else. So this plugin puts the answer there, by rewriting session.exitstatus from TESTS_FAILED to TESTS_ERRORED in pytest_sessionfinish. Each runner then learns one new code rather than a new channel, and the three of them cannot drift apart by reading different things.

The classification itself is made from call.excinfo in pytest_exception_interact, which is exact – the exception class, not a rendered string – and stamped onto the report rather than kept in an attribute of this object. That is what makes it survive pytest-xdist: the worker classifies, TestReport carries unknown attributes through its own JSON round trip, and the controller reads the flag off the deserialised report in pytest_runtest_logreport.

class moonbuggy.killreason.KillReasonCode(*values)

Bases: StrEnum

Stable per-kill reason — one token per verdict cause.

Each member is its own string value (StrEnum), so it compares equal to the string token it carries and serialises to that token in JSON. The .code property is an explicit alias for the string value, and .label is the human-readable form for documentation and the human trace.

ASSERTION_FAILED = 'assertion_failed'

A selected test’s assertion failed under the mutation – the test checked the mutated behaviour and objected.

TEST_ERRORED = 'test_errored'

A selected test errored out under the mutation – the test executed the line but did not check its result.

EXECUTION_CRASH = 'execution_crash'

collection error, internal error, usage error, or nothing collected. Not a statement about the mutation.

Type:

Pytest could not complete

FLAKY_PROBE = 'flaky_probe'

Test outcomes disagreed across unmutated runs – a selected test behaved inconsistently, so no confident verdict is possible.

property code: str

Machine-readable code, identical to the enum value (the JSONL token).

property label: str

Human-readable label, derived from the human trace.

class moonbuggy.killreason.KillReason

Bases: object

Records whether this session’s failures were assertions or errors.

One instance per process that runs one mutant’s tests. It carries no state between mutants because no two mutants ever share a process – see forkserver’s module docstring for why that is load-bearing rather than incidental.

reset() → None

Forget this session’s verdict.

The warm host builds one config before the first fork and every grandchild inherits a copy of this object with it, so each one starts by clearing whatever the host happened to leave in it.

property errored: bool

Whether this session’s kill was an error rather than an assertion.

One assertion failure anywhere is enough to call the kill ordinary: it is direct evidence that a test checked the mutated behaviour, and the errors beside it are then a consequence rather than the finding.

pytest_exception_interact(node: object, call: Any, report: Any) → None

Classify one failure, stamp its report, and fold it in.

This is where the classification happens in whichever process actually ran the test – the ordinary case, and the xdist worker.

Parameters:

• node – the item or collector that failed. Unused.

• call – the CallInfo, carrying the ExceptionInfo this reads.

• report – the report to stamp, so the answer survives xdist.

pytest_runtest_logreport(report: Any) → None

Fold in a report that already carries a classification.

Only an xdist controller reaches this with a flag set: in a single process call_and_report calls pytest_runtest_logreport before pytest_exception_interact, so an unflagged report here has not been classified yet rather than been classified as an assertion. Folding it in anyway is how the first version of this got every crash-kill wrong.

Parameters:

report – one test report, possibly deserialised from a worker.

pytest_sessionfinish(session: Any, exitstatus: Any) → None

Rewrite the exit code when the kill was an error.

Parameters:

• session – the pytest session, whose exitstatus is what every caller of pytest.main (and of pytest_cmdline_main) ends up returning.

• exitstatus – the code pytest settled on, before this.

moonbuggy.killreason.pytest_exception_interact(node: object, call: Any, report: Any) → None

Module-level hook, delegating to the process’s _ACTIVE recorder.

Parameters:

• node – the failing item or collector.

• call – the CallInfo for the failing phase.

• report – the report to stamp.

moonbuggy.killreason.pytest_runtest_logreport(report: Any) → None

Module-level hook, delegating to the process’s _ACTIVE recorder.

Parameters:

report – one test report.

moonbuggy.killreason.pytest_sessionfinish(session: Any, exitstatus: Any) → None

Module-level hook, delegating to the process’s _ACTIVE recorder.

Parameters:

• session – the pytest session.

• exitstatus – the code pytest settled on.

Reading and writing Python source that is not necessarily clean UTF-8.

Path.read_text() assumes UTF-8, so a file carrying a PEP 263 coding cookie (# -*- coding: latin-1 -*-) either raises a UnicodeDecodeError that surfaced as a traceback, or – worse – decodes without error into the wrong characters and gets mutated anyway. A mis-mutated file produces a status, and a status derived from the wrong bytes is a lie.

Two rules here:

• Decode the way the interpreter does. tokenize.detect_encoding implements PEP 263 exactly: BOM first, then a cookie in either of the first two lines, then UTF-8. Using it means moonbuggy sees the same characters Python does.

• Refuse rather than guess. A file whose declared encoding cannot decode it is not a file we can safely mutate, so it raises SourceError and the caller skips it with a message naming the file.

The cookie also has to be neutralised before the text reaches ast.parse or compile, which reject a coding declaration in an already-decoded str. That is what strip_coding_cookie is for, and why it substitutes a same-length replacement rather than deleting the line: every byte offset and line number in the rest of the file has to stay exactly where it was, because mutant splicing and line attribution are both offset-based.

exception moonbuggy.srcio.SourceError

Bases: RuntimeError

A source file could not be read as Python text.

moonbuggy.srcio.detect_encoding(path: str | PathLike[str]) → str

The encoding Python itself would use to read path.

Parameters:

path – filesystem path to a Python source file.

Returns:

an encoding name suitable for str.encode().

Raises:

SourceError – if the file cannot be opened or declares an encoding that does not exist.

moonbuggy.srcio.read_source(path: str | PathLike[str]) → str

Read a Python source file as text, honouring its declared encoding.

Parameters:

path – filesystem path to a Python source file.

Returns:

the file’s contents as str.

Raises:

SourceError – if the file cannot be read or cannot be decoded with the encoding it declares.

moonbuggy.srcio.prewarm(paths: Iterable[str | PathLike[str]]) → None

Fill the source cache for paths, ignoring any that cannot be read.

Called in the warm host once the job list is known, so the cost of reading each module is paid once for the whole run rather than once per mutant. Failures are ignored here on purpose: this is an optimisation, and the grandchild that actually needs the file will raise a real SourceError with a message about that file.

moonbuggy.srcio.encode_source(text: str, encoding: str) → bytes

Encode source text back to the bytes an importer will decode.

Parameters:

• text – the source as str.

• encoding – the encoding the file declares.

Returns:

bytes.

Raises:

SourceError – if the text cannot be represented in that encoding.

moonbuggy.srcio.replace_line(source: str, line: int, text: str) → str

Replace one line’s content, keeping everything around it byte for byte.

The single place mutation is applied to text. It was previously written twice – once for the in-memory path and once for the naive runner – and two copies of “preserve the indentation” is two chances to preserve slightly different things.

Indentation and trailing whitespace both come back from the line being replaced, because Mutant.original and Mutant.mutated are stripped for display. Restoring only the indentation loses any trailing whitespace, which breaks the round-trip property this module guarantees – and a mutation that quietly edits the whitespace it was not asked to edit is a mutation whose effect is not fully described by its own diff.

Parameters:

• source – the full source text.

• line – 1-based line number to replace.

• text – the replacement content, without indentation.

Returns:

the full source with that line replaced.

moonbuggy.srcio.strip_coding_cookie(text: str) → str

Neutralise a PEP 263 cookie so the text can be passed to ast.parse.

compile() and ast.parse() refuse a coding declaration inside a str, since the string is already decoded. Rather than delete the line, the word coding is replaced by an equal-length word that the cookie pattern does not match, so every subsequent line number, column offset and byte position is unchanged. Splicing and line attribution both depend on that.

Parameters:

text – source text, possibly carrying a cookie in its first two lines.

Returns:

source text with any cookie neutralised.

Process management

Run each mutant in a forked child instead of a fresh interpreter.

Why this exists: the first comparative benchmark had moonbuggy 12x SLOWER than mutmut on a workload where coverage-guided selection should have won easily (0.82s for 108 mutants against our 10.13s for 84). Selection was working perfectly – it was the per-mutant python -m pytest subprocess that cost everything. Interpreter startup plus importing pytest plus collection ran to roughly 120ms per mutant, and the selected tests themselves took a fraction of that. We were measuring process creation, not mutation testing.

Forking solves it because the expensive work is done once. The parent imports pytest and does nothing else; each child inherits that memory, applies its mutation, and runs only its own tests. A fork is a few milliseconds.

The parent must NEVER import the module under test. If it did, every child would inherit an already-imported unmutated module, install() would evict it from sys.modules but the parent’s copy would still be there for the next fork, and mutations would silently do nothing – a false SURVIVED, the same failure mode as the xdist bug from a third direction.

POSIX only. Windows has no fork, so runner.py keeps the subprocess path as a fallback rather than this being the only way to run.

class moonbuggy.forkserver.Job(mutant: Mutant, selected: list[str])

Bases: NamedTuple

One mutant queued to run, paired with the tests selected for it.

Direction: parent -> child. Every fork-per-mutant path (run_batch, run_warm_batch’s warm host, _fork_grandchildren) receives a list of these as a plain in-process argument – the child inherits it via fork, so nothing is actually serialised. run_warm_session is the one path where a Job list really does cross a pipe: build_jobs returns it in the parent, and it is pickled to the warm host.

Named because every call site destructures it as mutant, selected = job – the two fields already have names in the reader’s head, just not in the code before this.

mutant: Mutant

Alias for field number 0

selected: list[str]

Alias for field number 1

class moonbuggy.forkserver.WarmSessionEvidence

Bases: TypedDict

What the warm host learned from running the suite, sent back once.

Direction: child (the warm host) -> parent. Pickled across the status pipe by _warm_session_host right after its coverage (and probe) runs finish, and unpickled in run_warm_session before build_jobs is called with it as the parent’s only evidence about what happened in the child.

runs: list[dict[str, str]]

the coverage-pass run first, then one per flakiness probe. Same shape baseline.classify consumes.

Type:

One {node_id: outcome} mapping per unmutated run

startup: float

coverage_seconds: float

probe_seconds: float

class moonbuggy.forkserver.WarmSessionOutcome(jobs: list[Job], statuses: list[Literal['SURVIVED', 'KILLED', 'KILLED_BY_ERROR', 'TIMEOUT', 'SUSPICIOUS', 'UNAPPLIED']], child_seconds: float, child_wall_seconds: float)

Bases: NamedTuple

What run_warm_session hands back once every job has settled.

Not itself pickled – assembled in the parent from data that already crossed the pipe (the evidence above, and one streamed frame per job) – but named for the same reason as Job: run_session destructures it positionally as jobs, statuses, child_seconds, child_wall_seconds = outcome, and those four names belong on the type, not just at the call site.

jobs: list[Job]

Alias for field number 0

statuses: list[Literal['SURVIVED', 'KILLED', 'KILLED_BY_ERROR', 'TIMEOUT', 'SUSPICIOUS', 'UNAPPLIED']]

Alias for field number 1

child_seconds: float

Alias for field number 2

child_wall_seconds: float

Alias for field number 3

moonbuggy.forkserver.available() → bool

Whether this platform can fork. False on Windows, where the subprocess path in runner.py is used instead.

moonbuggy.forkserver.warm_up() → None

Import pytest in the parent so children inherit it already loaded.

Deliberately imports nothing from the project under test – see the module docstring for why that would be a correctness bug rather than a slow path.

moonbuggy.forkserver.run_in_fork(project_dir: str | PathLike[str], mutant: Mutant, selected: Iterable[str], timeout: float, install_mutation: Callable[[Mutant], None]) → Literal['SURVIVED', 'KILLED', 'KILLED_BY_ERROR', 'TIMEOUT', 'SUSPICIOUS', 'UNAPPLIED']

Fork, apply the mutation in the child, run its tests. Returns a status.

moonbuggy.forkserver.prebuild_mutant_config(extra_args: Iterable[str] = ()) → object | None

Build the pytest Config every mutant run needs, once, in the warm host.

pytest.main is two things: _prepareconfig, which parses the arguments, registers plugins, walks the entry points and loads the initial conftests; and then the session that actually collects and runs. Profiling one warm grandchild put the first at 4.4ms of an 11ms run – and its answer is identical for every mutant, because the only thing that differs between mutants is which node ids to run.

So it is built here, before the first fork, and each grandchild inherits a copy and points it at its own node ids. The move is the same as the other prebuilds: hoist a per-mutant constant into the host.

Why nothing is shared between mutants. Sharing a Config between mutants in one process would be the deferred hypothesis, and its failure mode – state from one mutant surviving into the next – is the thing the whole design exists to prevent. Nothing is shared between mutants here. Each grandchild is a separate process that gets its own copy-on-write copy of a config built before any mutation existed, uses it exactly once, and exits. Two mutants can no more see each other’s config than they can see each other’s sys.modules.

Built with no node ids: the ids are per-mutant, and everything expensive about the config is not.

Parameters:

extra_args – pytest arguments every mutant run shares.

Returns:

the Config, or None if it could not be built – in which case each grandchild falls back to a plain pytest.main, which is slower and identical in every other respect.

moonbuggy.forkserver.precollect(config: object, node_ids: Iterable[str], extra_args: Iterable[str] = ()) → object | None

Collect every test any mutant can select, once, in the warm host.

With the prebuilds landed, cProfile put perform_collect at 4ms of a 6.3ms warm grandchild – the largest thing left in the process that repeats. And its input is the same for every mutant: the union of the node ids selection can ask for is known before the first fork, and collecting that union once is collecting a superset of every mutant’s own selection.

So the host builds the Session, runs pytest_sessionstart and collects. Each grandchild inherits it, keeps the items its own node ids name, and runs those – 6.3ms to 2.3ms in the micro-benchmark, for 5-12ms paid once.

Why nothing is shared between mutants. The collection happens before any mutation exists. Each grandchild gets its own copy-on-write copy of it, filters it, runs it once and exits; no two mutants share a process, so neither can see the other’s items any more than it can see the other’s sys.modules. What makes that safe here specifically is that codeswap replaces a function’s __code__ in place rather than rebinding the name, so an item collected before the swap still reaches the mutated code – the same property that already makes the host’s own imports safe to inherit.

Why this gets its own config. Collecting configures the config it runs under, registers plugins (the session, the fixture manager, the terminal reporter) and mutates args. All of that is exactly what the prebuilt mutant config must NOT have picked up when the grandchild later runs it: a half-configured config makes the fallback grandchild fail in one of several ways – a registered Session under the name “session” makes wrap_session raise “Plugin name already registered”, a _do_configure`d config trips `assert not self._configured – and each failure is a CHILD_CRASHED per mutant, read as SUSPICIOUS execution_crash for the whole run. Found on humanize 4.16.0, where one unresolvable duplicate-named test id under pytest 9 rejected the whole precollect. So this builds a throwaway config for the collection and leaves the caller’s untouched: on failure the caller falls back to the pristine one, and on success the session’s config IS this throwaway and nothing else ever runs under it.

Parameters:

• config – the config from prebuild_mutant_config(), kept pristine for the fallback grandchild.

• node_ids – every test node id any mutant selects.

• extra_args – the pytest arguments every mutant run shares. The collected session runs under them, exactly as the grandchild would have: a project whose args change what fails (-W error being the sharpest case) must not be collected under arguments that cannot fail.

Returns:

the collected Session, or None to fall back to collecting inside

each grandchild. None on any failure, including a collection error: a session that could not collect cleanly must not be the one every mutant is judged against.

moonbuggy.forkserver.run_pytest_in_fork(cwd: str | PathLike[str], args: list[str], env_updates: dict[str, str], timeout: float) → int | None

Run pytest.main in a forked child. Returns its exit code, or None on timeout.

Used for the coverage pass. That pass was a python -m pytest subprocess and measured at 0.297s of an 0.88s run – a third of the total, most of it interpreter startup that the parent has already paid. Forking from a parent that has pytest imported skips it.

Safe despite the child importing the whole project: the child gets its own address space, so nothing it imports reaches the parent, and the parent stays clean for the mutation forks that follow.

moonbuggy.forkserver.run_warm_session(project_dir: str | PathLike[str], cov_args: list[str], timeout: float, concurrency: int, build_jobs: Callable[[WarmSessionEvidence], list[Job]], apply_swap: Callable[[Mutant], None], probe_args: list[str] | None = None, probes: int = 0, on_result: Callable[[int, Literal['SURVIVED', 'KILLED', 'KILLED_BY_ERROR', 'TIMEOUT', 'SUSPICIOUS', 'UNAPPLIED'], float], None] | None = None, extra_args: Iterable[str] = ()) → WarmSessionOutcome | None

One suite run that both builds the coverage map and warms the process.

Previously these were two separate full runs of the test suite: a coverage pass in one fork, then a priming run inside the warm host. On the benchmark workload that was 0.275s and 0.15s of an 0.79s total – over half the run, spent executing the same tests twice.

They are the same work, so this does it once. The host runs the suite under coverage, sends back what it observed, and waits. The parent reads the coverage data, checks the baseline is green, builds the line->test map, decides which tests each mutant needs, and sends the jobs back. The host then forks a grandchild per mutant from a process where every test module is already imported.

Parameters:

• project_dir – project root; the host chdirs here.

• cov_args – pytest arguments for the instrumented baseline run.

• timeout – seconds before one mutant is called TIMEOUT.

• concurrency – how many grandchildren run at once.

• build_jobs – called in the PARENT with the baseline evidence; returns the (mutant, selected_tests) jobs to run. May raise, in which case the host is torn down and the exception propagates.

• apply_swap – called in each grandchild to apply its mutation.

• probe_args – pytest arguments for the extra unmutated probe runs.

• probes – how many probe runs to make.

• on_result – called as (index, status, test_seconds) the moment each mutant finishes, so a run killed mid-flight has already reported what it knew.

• extra_args – pytest arguments every run shares, including each mutant’s.

Returns:

A WarmSessionOutcome, or None if the host could not complete –

so the caller falls back rather than reporting results it did not actually produce.

moonbuggy.forkserver.run_warm_batch(project_dir: str | PathLike[str], jobs: list[Job], timeout: float, concurrency: int, warm_args: list[str], apply_swap: Callable[[Mutant], None]) → list[Literal['SURVIVED', 'KILLED', 'KILLED_BY_ERROR', 'TIMEOUT', 'SUSPICIOUS', 'UNAPPLIED']] | None

Run mutants from a process that has already imported the test suite.

The expensive part of a mutant run is not the tests – it is importing the test modules, rewriting their asserts and collecting them, which a fork-per-mutant pays every single time (~90ms against ~12ms for pytest.main in a process where that work is done).

So this forks ONE warm host, has it run the suite once to get everything imported, and then forks a grandchild per mutant from that warm state. Each grandchild mutates in place with codeswap and runs only its own tests.

The warm host is a child, never the parent: it imports the whole project, and the parent has to stay clean so the import-hook fallback path still works for anything codeswap cannot handle.

moonbuggy.forkserver.run_batch(project_dir: str | PathLike[str], jobs: list[Job], timeout: float, install_mutation: Callable[[Mutant], None], concurrency: int, extra_args: Iterable[str] = ()) → list[Literal['SURVIVED', 'KILLED', 'KILLED_BY_ERROR', 'TIMEOUT', 'SUSPICIOUS', 'UNAPPLIED']]

Run many mutants concurrently, one forked child each.

Mutants are independent by construction – each child gets its own address space and its own mutation, and none of them write to the project – so they parallelise with no coordination at all. Serial forking left most of the machine idle.

Returns statuses in the order the jobs were given, regardless of the order children finish.

Wall-clock accounting by phase.

This module exists because performance work went wrong twice: two changes were implemented on a hunch and measured as noise, and a cheap profile would have caught both. Taking one is the mandatory first step before any performance work.

Design constraints, in order of importance:

1. The phases must add up. The named phases must account for at least 95% of measured wall clock, with the remainder reported explicitly as other. An “other” bucket quietly holding 30% is how the last round went wrong: it looks like a rounding error and is actually the answer.

2. Off by default, and free when off. A profiler that changes the thing it measures is worse than no profiler. When MOONBUGGY_PROFILE is unset every method here is a few attribute lookups on a disabled object.

3. Honest about concurrency. Mutants run in parallel, so the durations children report sum to more than the wall clock they occupied. Summing them into a wall-clock breakdown would produce a total over 100%. Instead the parent measures the wall clock of the whole mutant phase and splits it in the ratio the children reported – which is a real attribution of shared time, and is labelled as such rather than presented as a measurement.

class moonbuggy.profiling.Summary

Bases: TypedDict

The shape returned by Profiler.summary().

notes values are whatever Profiler.note() was called with – this module makes no claim about what a caller finds worth recording.

wall: float

phases: dict[str, float]

other: float

attributed: float

notes: dict[str, object]

class moonbuggy.profiling.Profiler(enabled: bool = False)

Bases: object

Accumulates time per named phase for one run.

totals: dict[str, float]

notes: dict[str, object]

span(name: str) → Iterator[None]

Time a block and add it to name.

Nested spans are not supported and not needed: a nested span’s time would be counted twice, and the phases are chosen to be disjoint.

add(name: str, seconds: float) → None

Add measured seconds to a phase.

split(name: str, seconds: float, weights: Mapping[str, float]) → None

Attribute seconds of wall clock across phases in a given ratio.

Used for the concurrent mutant phase, where per-child durations sum to more than the elapsed time. weights maps phase name to the measured (overlapping) time for that phase; each gets the same share of the real wall clock that it had of the measured total.

Parameters:

• name – the phase to record the remainder against if weights are empty, so the time is never lost.

• seconds – real wall clock elapsed for the whole phase.

• weights – {phase: measured_seconds}.

Returns:

None. The attribution is recorded on this profiler.

note(key: str, value: object) → None

Record a non-timing fact worth reading next to the numbers.

summary() → Summary

The report: per-phase seconds, wall clock, and what is unattributed.

Returns:

a dict with wall, phases, other, attributed (the fraction of

wall clock the named phases account for) and any recorded notes.

write(path: str | None = None) → Summary | None

Write the summary as JSON, if profiling is on.

moonbuggy.profiling.active() → Profiler

The profiler for this process.

moonbuggy.profiling.reset(enabled: bool | None = None) → Profiler

Start a fresh profiler. Used by tests and by the in-process harness.

Supporting

Persistent results cache, so repeat runs skip mutants nothing has changed for.

Two widely-seen precedents pointed at this: Hypothesis’s replay database and Ruff’s incremental re-linting. Both converge on the same idea, which is why the design promoted it into the MVP rather than deferring it.

The risk is not a cache miss, it is a stale hit. Serving a SURVIVED after the user added the test that kills it hides the gap they just closed and reports it as still outstanding – worse than no cache, because it is confidently wrong. The key therefore covers the inputs a mutation run can be expected to notice:

• the mutant’s identity and mutated text,

• the full source of the module being mutated,

• the contents of every test file selected for it,

• the conftest.py chain those test files pull in (a fixture edit is exactly what changes what a test does while its own bytes are unchanged),

• the mutated module’s first-order imports, resolved to files inside the project, and

• a fingerprint of the run itself – see run_fingerprint, which covers the parts of the command line that decide what pytest does.

Hashing the whole module rather than just the mutated function is deliberately coarse. A mutant’s behaviour can depend on anything else in its module – a helper it calls, an import, a module-level constant – and per-function hashing would miss those. Coarse and correct beats precise and occasionally stale; if this ever shows up in profiles, it is a safe thing to tighten with evidence.

What the key cannot see

Those bullets are not the whole of key_for’s inputs, but they are not the whole of reality either. Everything still outside the key yields a stale hit when edited, and the boundary is drawn where the reuse it would cost outweighs the correctness it buys. Known gaps, kept small on purpose:

• Transitive imports. Only the mutated module’s first-order imports are resolved, and only to files inside the project. A helper two levels deep, an installed dependency’s source, or a from .. import x ancestor-relative import is invisible unless it is the mutated module’s own direct import.

• pytest configuration and dependency versions. run_fingerprint covers the command line, not pytest.ini, [tool.pytest.ini_options], or the installed environment behind them. A dependency bump invalidating everything is arguably right rather than a cost, but it has not been measured, so it is not wired in yet.

• Which tests *inside* an unchanged file were selected. The key hashes the test file, not the node id list, so a change in selection within one file does not move it.

A key that covers everything is not the goal – it would cost the reuse the cache exists to buy. When one of the above has changed, –no-cache is the answer, as it always has been.

moonbuggy.cache.run_fingerprint(pytest_args: Iterable[str] = (), *, timeout: float | None = None, python: str | None = None) → str

A digest of the run inputs that can change a mutant’s verdict.

Mixed into every key by ResultCache, so a run whose command line differs cannot be served the previous run’s answers. Only inputs that can genuinely change an outcome belong here – each one folded in halves cache reuse for anyone who varies it, and a cache that never hits is its own failure mode.

What is in, and why:

• pytest_args (–pytest-arg): they reach the baseline and every mutant run, and they decide both which tests exist (–doctest-modules, -p) and whether they pass (-W error, -m). Hashed in order, because pytest’s argument order is meaningful – the last -p wins and -W filters match last-first – unlike test selection, which is sorted.

• timeout (–timeout): TIMEOUT is a verdict about the clock. The same mutant is TIMEOUT at five seconds and KILLED at sixty, so the clock is an input to the answer and not merely to how long it takes to get it.

• python: a different interpreter is a different language version and a different set of installed packages. A test that is skipped on one and fails on another produces a different verdict from identical source.

What is deliberately out:

• -n/–workers: xdist changes how the selected tests are distributed across processes, not which of them run or what they assert. A suite whose outcome depends on worker count is order-dependent, which is the flaky-probe’s problem (such tests make their mutants SUSPICIOUS) rather than something to encode in the key. Folding it in would cost every cache entry to anyone who flips between serial and parallel runs, and buy no correctness.

• –jobs: parallelism across mutants. Each mutant still runs alone, in its own process, against its own selection.

• –since: how you reached a mutant cannot change its answer. A scoped run generates a subset of the mutants a full run does, and each one is the same mutation run against the same selection either way, so the two deliberately share a cache. See moonbuggy.diffscope.

Parameters:

• pytest_args – –pytest-arg values, in the order they were given.

• timeout – seconds before a mutant is called TIMEOUT, or None.

• python – the resolved interpreter for mutant runs, or None.

Returns:

A hex digest, stable for identical inputs and different for any change to them.

class moonbuggy.cache.CacheRecord

Bases: TypedDict

One mutant’s stored outcome, as read back by runner.run_one.

status: str

tests_run: int

nearest_test: str | None

class moonbuggy.cache.ResultCache(path: str | PathLike[str], fingerprint: str = '')

Bases: object

Mutant outcomes from previous runs, keyed on everything they depend on.

__init__(path: str | PathLike[str], fingerprint: str = '') → None

Open the cache at path.

Parameters:

• path – the cache file. It need not exist yet.

• fingerprint – a run_fingerprint() digest for this run, mixed into every key. The default is empty, which keys purely on the code – correct only for a caller that never varies the run inputs.

key_for(mutant: Mutant, project_dir: str | PathLike[str], selected_tests: Iterable[str]) → str

The cache key for one mutant.

Parameters:

• mutant – the mutant whose outcome would be stored.

• project_dir – the project root, used to resolve relative paths.

• selected_tests – the pytest node ids selection chose for it.

Returns:

A hex digest covering the mutant, its module’s full source, the contents of every selected test file, the conftest.py chain each test file pulls in, the mutant’s first-order imports, and this run’s fingerprint.

get(key: str) → CacheRecord | None

The stored record for key, or None if there is not one.

put(key: str, record: CacheRecord) → None

Store record under key. Not written to disk until save.

save() → None

Persist the cache, atomically.

Written to a sibling temp file and renamed, so a run killed during the save leaves the previous cache intact rather than a half-written file. os.replace is atomic within a filesystem, and the temp file is deliberately a sibling so it is on the same one.

clear() → None

Forget everything, on disk and in memory.

CLI entry points and subcommand dispatch.

These names are re-exported on purpose: moonbuggy.cli is the public surface the tests and the [project.scripts] moonbuggy = “moonbuggy.cli:run” entry point import from, even though the implementation now lives in the submodules below.

moonbuggy.cli.main(argv: Sequence[str] | None = None) → int

Run moonbuggy.

Parameters:

argv – command-line arguments, or None to read sys.argv.

Returns:

0 for a run with no findings, 1 when there are findings (SURVIVED or NO_COVERAGE), 2 when the run could not happen at all, 130 when interrupted.

Return type:

The process exit code

moonbuggy.cli.run() → None

Entry point for the moonbuggy command: main, then exit at once.

main returns an exit code and this turns it into a process exit, which would normally be sys.exit. It is os._exit instead, because CPython’s finalisation of a process holding pytest, coverage and moonbuggy is 13ms — measured as 85ms to exit normally against 72ms to os._exit, on a process importing exactly what this one imports. That is a real part of what a user waits for, and none of it does any work: every object being torn down is about to stop existing anyway.

What os._exit skips is atexit, module teardown, and buffer flushing, and the last of those is the whole risk. moonbuggy’s two output files are written and closed inside _run before it returns, so the only thing left unflushed is the standard streams, which are flushed here explicitly. A missed flush would lose the report, which is worse than a slow exit – so this function stays this short, and anything that ever needs to happen at exit must happen before the flush rather than in an atexit hook that will not run.

main itself deliberately still returns rather than exiting, so the tests and any in-process caller get a value instead of a dead interpreter.

The naive reference implementation: mutate to disk, run the whole suite.

This is deliberately the slow, obvious way to do mutation testing, and it exists for two reasons:

1. It is the independent half of the oracle. Because it shares no code with the fast path – no import hook, no coverage-guided selection, no cache, no xdist – agreement between the two is real evidence rather than a tautology. The only thing they have in common is mutant generation, which is exactly the gap a hand-written inventory of the operators covers.

2. It is the naive baseline the benchmark measures against.

Nothing here should ever be optimised. If it grows a coverage map or an import hook it stops being an independent check and starts being a second copy of the thing under test.

moonbuggy.naive.run_naive(project_dir: Path, mutants: Iterable[Mutant], timeout: float = 30, python: str | None = None) → dict[Mutant, Literal['SKIPPED', 'SURVIVED', 'KILLED', 'TIMEOUT', 'SUSPICIOUS']]

Run every unsuppressed mutant against the full suite.

A suppressed mutant is settled SKIPPED without being run. Returns {Mutant: status}.