Prerequisites
This section synthesizes the test generation strategies from Section 18.1 and the property-based and mutation testing techniques from Section 18.2 into a complete automated system. You should be comfortable with the Anthropic Python SDK (used since Chapter 8), pytest fixtures, Hypothesis strategies, and mutmut's command-line interface. The system we build here extends the Discovery Workbench with a testing module that connects to the multi-agent architecture from Chapter 17.
An invariant discovery system closes the loop between AI code generation and AI test generation. When a developer writes a function (or an AI generates one), the system reads the source, infers behavioral invariants the function should satisfy, translates those invariants into executable Hypothesis property tests, runs mutation testing to verify the tests are strong enough to detect faults, and reports a composite confidence score. The result is a fully automated quality assurance (QA) pipeline that treats testing as a discovery process: searching for properties that the code satisfies, just as a scientist searches for laws that nature satisfies.
1. System Architecture
Production bugs in scientific software frequently trace back to properties that no one thought to test: a simulation that silently violates conservation of energy, an interpolation routine that returns plausible but wrong values near domain boundaries. Manually anticipating every such property is impractical once a codebase grows beyond a few functions, which is exactly the gap an automated invariant discovery system fills.
Imagine handing a function to a system that reads its source code, guesses every behavioral rule the function should obey, writes hundreds of tests for those rules, then deliberately breaks the code in dozens of places to confirm the tests would notice. That is what the invariant discovery system does, in four modular stages that follow the same architecture pattern used throughout the Discovery Workbench (introduced in Chapter 6), each with a well-defined input and output, communicating through structured data rather than shared state. The diagram below (Figure 18.3.1) shows how these four stages connect, with each stage's primary input and output labeled.
A behavioral invariant is a property that a function must satisfy for every valid input, not just for a handful of hand-picked test cases. Behavioral invariants transform testing from "check these five examples" into "prove this universal rule." A single property statement catches entire categories of bugs. The system expresses each invariant as a property test using Hypothesis (a Python library for property-based testing that verifies properties by generating hundreds of random inputs automatically) and confirms the property holds for all of them. Use behavioral invariants when a function has a large or continuous input domain where manually chosen examples cannot cover boundary conditions. Fall back to example-based tests when correct output depends on specific business rules that no general property can capture (such as "customer #42 gets a 10% discount"). In short: let the machine propose the universal laws your code must obey, then let mutations prove those laws have teeth.
- Invariant Inference: A large language model (LLM) reads the function's source code, docstring, and type annotations, then proposes a set of behavioral invariants from the five families (roundtrip, idempotence, preservation, metamorphic, oracle comparison).
- Test Generation: Each inferred invariant is translated into a Hypothesis property test with an appropriate custom strategy (a rule that tells Hypothesis how to generate valid random inputs) for the function's input types.
- Test Validation: The generated tests are compiled, executed, and filtered. Tests that fail on the current (presumably correct) code are discarded as false invariants. Tests that pass are retained as candidate properties.
- Mutation Analysis: mutmut (a Python mutation testing tool that injects small faults into source code) runs against the function with the validated property tests. The resulting mutation score (the fraction of injected faults the tests detect), combined with branch coverage (the fraction of conditional branches exercised by the tests) and property count, produces the composite confidence score.
"""
Invariant Discovery System: core data structures.
"""
from dataclasses import dataclass, field
from enum import Enum
class InvariantFamily(Enum):
"""The five families of behavioral invariants."""
ROUNDTRIP = "roundtrip"
IDEMPOTENCE = "idempotence"
PRESERVATION = "preservation"
METAMORPHIC = "metamorphic"
ORACLE = "oracle_comparison"
@dataclass
class InferredInvariant:
"""A behavioral invariant inferred by the LLM."""
family: InvariantFamily
description: str # human-readable statement
hypothesis_code: str # executable Hypothesis test code
confidence: float # LLM's self-assessed confidence (0 to 1)
validated: bool = False # True after passing on current code
killed_mutants: int = 0 # how many mutants this property catches
@dataclass
class DiscoveryReport:
"""Output of the invariant discovery pipeline."""
function_name: str
source_file: str
invariants_proposed: int
invariants_validated: int
branch_coverage: float # 0 to 1
mutation_score: float # 0 to 1
property_ratio: float # validated / target
confidence_score: float # composite metric
surviving_mutants: list[str] # descriptions of unkilled mutants
invariants: list[InferredInvariant] = field(default_factory=list)
def summary(self) -> str:
return (
f"Invariant Discovery Report: {self.function_name}\n"
f" Invariants: {self.invariants_validated}/{self.invariants_proposed} validated\n"
f" Branch coverage: {self.branch_coverage:.1%}\n"
f" Mutation score: {self.mutation_score:.1%}\n"
f" Confidence score: {self.confidence_score:.1%}\n"
f" Surviving mutants: {len(self.surviving_mutants)}"
)
2. Stage 1: Invariant Inference
The inference stage uses an LLM to read a function's source code and propose behavioral
invariants. The prompt is structured to elicit properties from each of the five families,
with the LLM returning structured JSON that can be parsed directly into
InferredInvariant objects. The key design decision is asking the LLM to
generate executable Hypothesis code for each invariant, not just a natural-language
description. This eliminates a translation step and ensures the invariant is testable.
"""
Stage 1: Invariant Inference using an LLM.
"""
import json
from anthropic import Anthropic
INVARIANT_INFERENCE_PROMPT = """You are a software testing expert. Analyze the following
Python function and infer behavioral invariants it should satisfy.
For each invariant, provide:
1. family: one of "roundtrip", "idempotence", "preservation", "metamorphic", "oracle_comparison"
2. description: a one-sentence natural-language statement of the property
3. hypothesis_code: a complete, runnable Hypothesis property test (with imports)
4. confidence: your confidence that this property holds (0.0 to 1.0)
Guidelines:
- Propose at least 2 invariants from each applicable family.
- Include Hypothesis strategies appropriate for the function's input types.
- For scientific functions, consider physical constraints (e.g., conservation laws,
dimensional consistency, monotonicity, symmetry).
- For data transformations, consider roundtrip properties with inverse operations.
- Be specific: "output length equals input length" is better than "output is correct."
Return a JSON array of objects. No other text.
Function to analyze:
```python
{function_source}
```"""
def infer_invariants(
function_source: str,
model: str = "claude-sonnet-4-20250514"
) -> list[InferredInvariant]:
"""Infer behavioral invariants from function source code.
Args:
function_source: Complete source of the function under test.
model: Claude model to use.
Returns:
List of inferred invariants with executable test code.
"""
client = Anthropic()
response = client.messages.create(
model=model,
max_tokens=8192,
messages=[{
"role": "user",
"content": INVARIANT_INFERENCE_PROMPT.format(
function_source=function_source
)
}]
)
raw_text = response.content[0].text
# Strip markdown fences if present
if raw_text.startswith("```"):
raw_text = raw_text.split("\n", 1)[1]
raw_text = raw_text.rsplit("```", 1)[0]
invariants_data = json.loads(raw_text)
return [
InferredInvariant(
family=InvariantFamily(item["family"]),
description=item["description"],
hypothesis_code=item["hypothesis_code"],
confidence=item["confidence"]
)
for item in invariants_data
]
# Example: infer invariants for a data transformation function
source = '''
def interpolate_spectrum(
wavelengths: list[float],
intensities: list[float],
target_wavelengths: list[float]
) -> list[float]:
"""Linearly interpolate a spectrum onto a new wavelength grid.
Args:
wavelengths: Original wavelength values (sorted ascending, at least 2).
intensities: Original intensity values (same length as wavelengths).
target_wavelengths: Desired output wavelength values (sorted ascending).
Returns:
Interpolated intensities at each target wavelength.
Values outside the original range are extrapolated linearly.
"""
result = []
for target in target_wavelengths:
# Find bracketing interval
if target <= wavelengths[0]:
# Extrapolate below
slope = (intensities[1] - intensities[0]) / (wavelengths[1] - wavelengths[0])
result.append(intensities[0] + slope * (target - wavelengths[0]))
elif target >= wavelengths[-1]:
# Extrapolate above
slope = (intensities[-1] - intensities[-2]) / (wavelengths[-1] - wavelengths[-2])
result.append(intensities[-1] + slope * (target - wavelengths[-1]))
else:
# Interpolate: find the interval
for i in range(len(wavelengths) - 1):
if wavelengths[i] <= target <= wavelengths[i + 1]:
t = (target - wavelengths[i]) / (wavelengths[i + 1] - wavelengths[i])
val = intensities[i] + t * (intensities[i + 1] - intensities[i])
result.append(val)
break
return result
'''
invariants = infer_invariants(source)
for inv in invariants:
print(f"[{inv.family.value}] {inv.description} (confidence: {inv.confidence})")
Step-Through: Confidence Score Computation
(This walkthrough previews the full confidence formula; the code behind each metric
appears in Stages 3 and 4 below.)
Trace the composite confidence score calculation with concrete numbers from a small
run. Suppose the system proposes 8 invariants for a function, and after validation
5 survive (3 are false invariants). The target property count is 10, so
property_ratio = min(1.0, 5 / 10) = 0.50. Branch coverage measured by
coverage.py is 0.88. Mutation testing with mutmut generates 40 mutants; the 5 validated
property tests kill 34 of them, so mutation_score = 34 / 40 = 0.85.
The composite confidence score is 0.88 * 0.85 * 0.50 = 0.374. This low
score (despite decent coverage and mutation score) signals that the system needs more
validated properties. After a second inference round adds 3 more validated invariants,
property_ratio rises to min(1.0, 8 / 10) = 0.80, the new
tests also kill 2 of the 6 surviving mutants (pushing mutation score to
36 / 40 = 0.90), and confidence jumps to 0.88 * 0.90 * 0.80 = 0.634.
A generic testing tool would propose only structural properties (length preservation, type preservation). The LLM, having been trained on scientific code, infers domain-specific properties: linear interpolation at the original grid points should reproduce the original values (a roundtrip property); interpolation should preserve monotonicity if the original spectrum is monotonic (a preservation property); scaling all intensities by a constant and then interpolating should give the same result as interpolating and then scaling (a metamorphic property). These domain-aware invariants catch bugs that structural tests miss entirely, such as an off-by-one error in the bracketing interval search that only manifests for certain wavelength spacings.
Mental Model
Think of invariant discovery like a building inspector checking a house. The inspector does not test whether the house looks nice (that would be an example-based test for a specific input). Instead, the inspector checks universal rules: every load-bearing wall must support its share of weight, every electrical outlet must be grounded, every staircase must have a railing. These rules hold regardless of the house's style, size, or color. The inspector proposes rules (invariants), verifies each one against the actual house (validation), and then deliberately introduces faults (removing a screw, loosening a wire) to confirm the inspection would catch them (mutation testing). A house that passes all structural rules under deliberate sabotage earns a high confidence score, just as a function that satisfies all behavioral invariants even when its source code is mutated earns a high composite confidence metric.
3. Stage 2: Test Generation and Validation
Domain-specific invariants are powerful precisely because they encode deep knowledge about a function's purpose, but that same depth means the LLM can also hallucinate plausible properties that the function does not actually satisfy.
The inferred invariants arrive as strings of Python code. The system compiles each one, runs it against the current implementation, and discards any that fail. A failing invariant either reflects a hallucinated property or exposes a genuine bug. Consistent failures (every run) signal real violations; intermittent failures signal flawed strategies or unstable numerical comparisons, and those invariants are discarded as flaky (a test is "flaky" when it passes on some runs and fails on others without any code change, typically due to nondeterminism or tight numerical tolerances) .
"""
Stage 2: Test Generation and Validation.
Compile, execute, and filter inferred invariants.
"""
import subprocess
import tempfile
import textwrap
from pathlib import Path
def validate_invariants(
invariants: list[InferredInvariant],
source_file: str,
max_examples: int = 200,
validation_runs: int = 3
) -> list[InferredInvariant]:
"""Validate inferred invariants against the current implementation.
Runs each invariant's Hypothesis test multiple times. Invariants that
pass consistently are marked as validated. Invariants that fail
consistently are flagged as potential bugs. Flaky invariants are
discarded.
Args:
invariants: List of inferred invariants with hypothesis_code.
source_file: Path to the module containing the function under test.
max_examples: Hypothesis max_examples setting for validation.
validation_runs: Number of times to run each test for flakiness check.
Returns:
Filtered list of validated invariants.
"""
validated = []
module_name = Path(source_file).stem
for inv in invariants:
# Build a standalone test file
test_code = textwrap.dedent(f"""
import sys
sys.path.insert(0, "{Path(source_file).parent}")
from {module_name} import *
from hypothesis import given, settings, assume
from hypothesis import strategies as st
from hypothesis.strategies import composite
import math
import numpy as np
settings.register_profile("validate",
max_examples={max_examples}, derandomize=True)
settings.load_profile("validate")
{inv.hypothesis_code}
""")
# Write to a temp file and run with pytest
pass_count = 0
fail_count = 0
with tempfile.NamedTemporaryFile(
mode="w", suffix=".py", prefix="test_inv_",
delete=False, dir=tempfile.gettempdir()
) as f:
f.write(test_code)
test_path = f.name
for run in range(validation_runs):
result = subprocess.run(
["pytest", test_path, "-x", "--tb=short", "-q"],
capture_output=True, text=True, timeout=120
)
if result.returncode == 0:
pass_count += 1
else:
fail_count += 1
# Classify the result
if pass_count == validation_runs:
inv.validated = True
validated.append(inv)
elif fail_count == validation_runs:
# Consistent failure: possible real bug
print(f" POTENTIAL BUG: {inv.description}")
print(f" Family: {inv.family.value}")
# Still include it so the report can flag it
inv.validated = False
validated.append(inv)
else:
# Flaky: discard
print(f" FLAKY (discarded): {inv.description}")
# Clean up
Path(test_path).unlink(missing_ok=True)
return validated
4. Stage 3: Mutation Analysis
With validated property tests in hand, we run mutation testing to measure their strength. This stage combines all validated tests into a single test file, runs mutmut against the function under test, and collects the mutation score along with details of each surviving mutant.
Common Misconception
Readers often assume that a high mutation score (say, 95%) alone proves a test suite is thorough. This is incorrect: mutation score measures only how many artificial faults the tests detect, not whether the tests cover the right behavioral properties. A test suite composed entirely of "output is not None" assertions can kill many mutants (any mutation that causes a crash) while missing subtle logic errors that still produce non-None but wrong results. The composite confidence score in this system multiplies mutation score by branch coverage and property ratio precisely to guard against this trap; all three factors must be high for the overall score to be meaningful.
"""
Stage 3: Mutation Analysis.
Run mutmut with the validated property tests and collect results.
"""
import subprocess
import json
import re
from pathlib import Path
def run_mutation_analysis(
source_file: str,
validated_invariants: list[InferredInvariant],
timeout_per_mutant: int = 30
) -> dict:
"""Run mutation testing with validated property tests.
Args:
source_file: Path to the module under test.
validated_invariants: List of validated invariants (only those
where validated=True will be used).
timeout_per_mutant: Timeout in seconds per mutant test run.
Returns:
Dict with mutation_score, killed, survived, total, and
surviving_mutant_descriptions.
"""
# Collect only validated invariants
valid_tests = [
inv for inv in validated_invariants if inv.validated
]
if not valid_tests:
return {
"mutation_score": 0.0,
"killed": 0,
"survived": 0,
"total": 0,
"surviving_mutants": []
}
# Build combined test file
module_name = Path(source_file).stem
test_dir = Path(source_file).parent / "tests"
test_dir.mkdir(exist_ok=True)
test_file = test_dir / f"test_{module_name}_properties.py"
header = f"""
import sys
sys.path.insert(0, "{Path(source_file).parent}")
from {module_name} import *
from hypothesis import given, settings, assume
from hypothesis import strategies as st
from hypothesis.strategies import composite
import math
import numpy as np
settings.register_profile("mutation",
max_examples=50, derandomize=True, deadline=None)
settings.load_profile("mutation")
"""
body = "\n\n".join(inv.hypothesis_code for inv in valid_tests)
test_file.write_text(header + "\n" + body)
# Run mutmut
result = subprocess.run(
[
"mutmut", "run",
f"--paths-to-mutate={source_file}",
f"--tests-dir={test_dir}",
f"--runner=pytest {test_file} -x --tb=no -q",
],
capture_output=True, text=True,
timeout=timeout_per_mutant * 200 # generous overall timeout
)
# Parse results
result_output = subprocess.run(
["mutmut", "results"],
capture_output=True, text=True
)
# Extract counts from mutmut output
output = result_output.stdout
killed = len(re.findall(r"Killed", output))
survived = len(re.findall(r"Survived", output))
timeout_count = len(re.findall(r"Timeout", output))
# Timed-out mutants are excluded: they may be "equivalent mutants"
# (mutations that change the source without changing observable behavior),
# so counting them as survived would unfairly penalize the score.
total_non_equivalent = killed + survived
mutation_score = (
killed / total_non_equivalent
if total_non_equivalent > 0 else 1.0
)
# Collect surviving mutant descriptions
surviving_descriptions = []
if survived > 0:
for i in range(1, killed + survived + timeout_count + 1):
show_result = subprocess.run(
["mutmut", "show", str(i)],
capture_output=True, text=True
)
if "Survived" in show_result.stdout or survived > 0:
surviving_descriptions.append(
show_result.stdout.strip()[:200]
)
return {
"mutation_score": mutation_score,
"killed": killed,
"survived": survived,
"total": killed + survived + timeout_count,
"surviving_mutants": surviving_descriptions[:10] # limit
}
5. Stage 4: Confidence Scoring and Reporting
The final stage computes the composite confidence score and generates the
DiscoveryReport. This is where all the metrics converge: branch coverage
from coverage.py (a Python tool that measures which lines and conditional branches the tests actually execute), mutation score from mutmut, and property count from the validation
stage.
"""
Stage 4: Confidence Scoring and Reporting.
Assemble the complete invariant discovery report.
"""
import subprocess
import json
from pathlib import Path
def compute_branch_coverage(
source_file: str,
test_file: str
) -> float:
"""Measure branch coverage of the property tests."""
module_name = Path(source_file).stem
subprocess.run(
[
"pytest", test_file,
f"--cov={module_name}",
"--cov-branch",
"--cov-report=json:cov_report.json",
"-q"
],
capture_output=True, text=True,
timeout=120
)
try:
with open("cov_report.json") as f:
cov_data = json.load(f)
file_cov = cov_data.get("files", {}).get(source_file, {})
return file_cov.get("summary", {}).get(
"percent_covered_branches", 0.0
) / 100.0
except (FileNotFoundError, json.JSONDecodeError, KeyError):
return 0.0
finally:
Path("cov_report.json").unlink(missing_ok=True)
def run_invariant_discovery(
source_file: str,
function_source: str,
target_properties: int = 10,
model: str = "claude-sonnet-4-20250514"
) -> DiscoveryReport:
"""Run the complete invariant discovery pipeline.
This is the main entry point that orchestrates all four stages.
Args:
source_file: Path to the Python source file.
function_source: Source code of the function to analyze.
target_properties: Target number of validated properties.
model: Claude model for invariant inference.
Returns:
A complete DiscoveryReport with confidence score.
"""
import re
# Extract function name
match = re.search(r"def (\w+)\(", function_source)
func_name = match.group(1) if match else "unknown"
print(f"Analyzing: {func_name}")
# Stage 1: Infer invariants
print(" Stage 1: Inferring invariants...")
invariants = infer_invariants(function_source, model=model)
print(f" Proposed {len(invariants)} invariants")
# Stage 2: Validate invariants
print(" Stage 2: Validating invariants...")
validated = validate_invariants(invariants, source_file)
valid_count = sum(1 for inv in validated if inv.validated)
print(f" Validated {valid_count} / {len(invariants)}")
# Stage 3: Mutation analysis
print(" Stage 3: Running mutation analysis...")
mutation_results = run_mutation_analysis(
source_file,
validated
)
print(f" Mutation score: {mutation_results['mutation_score']:.1%}")
# Stage 4: Compute confidence score
test_dir = Path(source_file).parent / "tests"
test_file = str(
test_dir / f"test_{Path(source_file).stem}_properties.py"
)
branch_cov = compute_branch_coverage(source_file, test_file)
property_ratio = min(1.0, valid_count / target_properties)
# Multiplication (rather than, say, a weighted average) is deliberate:
# if ANY single factor is near zero the overall score collapses,
# reflecting the principle that high coverage with weak properties
# (or strong properties with low coverage) is not trustworthy.
confidence = (
branch_cov
* mutation_results["mutation_score"]
* property_ratio
)
report = DiscoveryReport(
function_name=func_name,
source_file=source_file,
invariants_proposed=len(invariants),
invariants_validated=valid_count,
branch_coverage=branch_cov,
mutation_score=mutation_results["mutation_score"],
property_ratio=property_ratio,
confidence_score=confidence,
surviving_mutants=mutation_results["surviving_mutants"],
invariants=validated
)
print(f"\n{report.summary()}")
return report
The Re-Inference Loop
The dashed feedback arrow in Figure 18.3.1 represents the most important part of the pipeline: when surviving mutants indicate weak property coverage, the system re-runs Stage 1 with an augmented prompt that includes the surviving mutant diffs. This gives the LLM concrete evidence of what the current properties miss, so it can propose targeted invariants that address those gaps. The function below implements one iteration of this loop.
def reinfer_for_survivors(
function_source: str,
surviving_mutants: list[str],
existing_invariants: list[InferredInvariant],
model: str = "claude-sonnet-4-20250514"
) -> list[InferredInvariant]:
"""Re-run invariant inference targeting surviving mutants.
Args:
function_source: Source code of the function under test.
surviving_mutants: Diffs of mutants that the current tests
failed to detect.
existing_invariants: Already-validated invariants (to avoid
duplicates).
model: Claude model for inference.
Returns:
New invariants specifically targeting the surviving mutants.
"""
existing_descriptions = [
inv.description for inv in existing_invariants
]
survivors_text = "\n---\n".join(surviving_mutants[:5])
prompt = f"""You are a software testing expert. The following function
has surviving mutants that the current property tests do not detect.
Function:
```python
{function_source}
```
Surviving mutant diffs (each is a small code change that goes undetected):
{survivors_text}
Existing properties (do NOT repeat these):
{chr(10).join(f'- {d}' for d in existing_descriptions)}
Propose new behavioral invariants that would detect these specific mutations.
Return a JSON array (same schema as before)."""
client = Anthropic()
response = client.messages.create(
model=model, max_tokens=4096,
messages=[{"role": "user", "content": prompt}]
)
raw = response.content[0].text
if raw.startswith("```"):
raw = raw.split("\n", 1)[1].rsplit("```", 1)[0]
import json
return [
InferredInvariant(
family=InvariantFamily(item["family"]),
description=item["description"],
hypothesis_code=item["hypothesis_code"],
confidence=item["confidence"]
)
for item in json.loads(raw)
]
A drug discovery team applied the invariant discovery system to their molecular docking score function, a 120-line Python function that computes binding affinity from atomic coordinates and partial charges. The system inferred 14 invariants, including: "translating all atoms by the same vector does not change the score" (metamorphic, translation invariance), "swapping the protein and ligand produces the same score" (metamorphic, symmetry), and "doubling all partial charges quadruples the electrostatic component" (metamorphic, quadratic scaling). After validation, 11 invariants survived. Mutation testing with these 11 properties achieved a 91% mutation score. The three surviving mutants all involved the van der Waals distance cutoff: the tests did not generate atom pairs at exactly the cutoff distance. Adding a targeted property test for boundary distances killed all remaining mutants, bringing the confidence score to \(0.97 \times 0.97 \times 1.0 = 0.94\).
6. Discovery Workbench Integration
A pipeline that produces high confidence scores is useful only if developers and automated workflows can invoke it without manual setup at every step.
The invariant discovery system integrates into the Discovery Workbench as a testing service that can be triggered from the experiment dashboard, the command-line interface (CLI), or the continuous integration (CI) pipeline. The integration follows the same service pattern established in Chapter 6 and extended through Chapter 12: Building MCP Servers.
"""
Discovery Workbench integration: the invariant discovery service.
Exposes the pipeline as a FastAPI endpoint and an MCP tool.
"""
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, Field
from pathlib import Path
import asyncio
app = FastAPI(title="Invariant Discovery Service")
class DiscoveryRequest(BaseModel):
"""Request to run invariant discovery on a function."""
source_file: str = Field(
description="Path to the Python source file"
)
function_name: str = Field(
description="Name of the function to analyze"
)
target_properties: int = Field(
default=10,
description="Target number of validated properties"
)
model: str = Field(
default="claude-sonnet-4-20250514",
description="Claude model for invariant inference"
)
class DiscoveryResponse(BaseModel):
"""Response from the invariant discovery pipeline."""
function_name: str
invariants_proposed: int
invariants_validated: int
branch_coverage: float
mutation_score: float
confidence_score: float
surviving_mutant_count: int
status: str # "complete", "running", "failed"
# In-memory store for async results (use a database in production)
_results: dict[str, DiscoveryResponse] = {}
async def _run_discovery_async(
request_id: str,
request: DiscoveryRequest
):
"""Run the discovery pipeline in the background."""
try:
source_path = Path(request.source_file)
source_code = source_path.read_text()
# Extract the target function's source
import ast
tree = ast.parse(source_code)
func_source = None
for node in ast.walk(tree):
if (isinstance(node, ast.FunctionDef)
and node.name == request.function_name):
func_source = ast.get_source_segment(
source_code, node
)
break
if func_source is None:
_results[request_id] = DiscoveryResponse(
function_name=request.function_name,
invariants_proposed=0,
invariants_validated=0,
branch_coverage=0.0,
mutation_score=0.0,
confidence_score=0.0,
surviving_mutant_count=0,
status="failed"
)
return
report = run_invariant_discovery(
source_file=request.source_file,
function_source=func_source,
target_properties=request.target_properties,
model=request.model
)
_results[request_id] = DiscoveryResponse(
function_name=report.function_name,
invariants_proposed=report.invariants_proposed,
invariants_validated=report.invariants_validated,
branch_coverage=report.branch_coverage,
mutation_score=report.mutation_score,
confidence_score=report.confidence_score,
surviving_mutant_count=len(report.surviving_mutants),
status="complete"
)
except Exception as e:
_results[request_id] = DiscoveryResponse(
function_name=request.function_name,
invariants_proposed=0,
invariants_validated=0,
branch_coverage=0.0,
mutation_score=0.0,
confidence_score=0.0,
surviving_mutant_count=0,
status=f"failed: {str(e)[:200]}"
)
@app.post("/discover", response_model=dict)
async def start_discovery(
request: DiscoveryRequest,
background_tasks: BackgroundTasks
):
"""Start an invariant discovery analysis (runs in background)."""
import uuid
request_id = str(uuid.uuid4())
_results[request_id] = DiscoveryResponse(
function_name=request.function_name,
invariants_proposed=0,
invariants_validated=0,
branch_coverage=0.0,
mutation_score=0.0,
confidence_score=0.0,
surviving_mutant_count=0,
status="running"
)
background_tasks.add_task(
_run_discovery_async, request_id, request
)
return {"request_id": request_id, "status": "running"}
@app.get("/discover/{request_id}", response_model=DiscoveryResponse)
async def get_discovery_result(request_id: str):
"""Check the status of a discovery analysis."""
if request_id not in _results:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Not found")
return _results[request_id]
Hypothesis includes a built-in hypothesis.extra.ghostwriter module that
generates property tests automatically from function signatures and type annotations.
Running hypothesis write normalize_spectrum on the command line produces
a test file with roundtrip, idempotence, and equivalence tests. In practice, the ghostwriter
typically handles a substantial fraction of what our invariant discovery system does, in a single
command with zero LLM cost . The LLM adds value for the rest: domain-specific
metamorphic relations, physically meaningful invariants, and properties that require
understanding the function's purpose rather than just its type signature. Use the
ghostwriter as a fast baseline and the LLM for the deeper analysis.
7. The Testing-as-Discovery Feedback Loop
Whether invoked through a web endpoint, an MCP tool, or the ghostwriter shortcut, every run of the pipeline generates the same structured report, and that report carries a deeper significance than its numeric scores suggest.
The invariant discovery system creates a feedback loop that mirrors the scientific method. The LLM proposes hypotheses (invariants). The test framework designs experiments (Hypothesis examples). The mutation engine simulates faulty implementations (mutants). The confidence score quantifies how well the hypotheses explain the observed behavior. When the confidence score is low, the system can be re-run with an expanded prompt asking the LLM to target the specific mutant survival patterns, just as a scientist designs follow-up experiments to address gaps in their theory.
This feedback loop connects directly to the broader discovery themes of the book. The search framework from Chapter 1 treats discovery as navigating a search space. Here the search space is the set of behavioral invariants, and the objective is to find properties that are both true (validated) and strong (high mutation score). The hypothesis generation patterns from Chapter 39 apply directly: the LLM generates candidate hypotheses, and testing either confirms or refutes them. The evaluation metrics from Chapter 56 will revisit this confidence score as a component of overall system trustworthiness.
Checkpoint
So far: the invariant discovery system infers properties from source code, validates them against the running implementation, measures their fault-detection strength via mutation testing, and produces a composite confidence score; this same loop mirrors the scientific method of proposing hypotheses, running experiments, and refining theories based on gaps.
Real-World Application: Netflix's Chaos Engineering Meets Property Testing
Netflix's ChAP (Chaos Automation Platform) uses invariant-style properties to decide whether a fault-injection experiment is safe to run in production. Before injecting a failure (killing a service instance, adding latency), ChAP verifies that baseline behavioral invariants hold: request error rates stay below a threshold, latency percentiles remain stable, and no downstream service exceeds its circuit-breaker budget. These invariants function exactly like the validated properties in our discovery system, except the "function under test" is a distributed microservice and the "mutants" are real infrastructure faults.
Lab: Invariant Discovery on a Matrix Library
Goal: Use Hypothesis to discover and validate behavioral invariants for
NumPy matrix operations, then measure how well those invariants detect injected faults.
Tools needed: Python 3.10+, numpy, hypothesis,
mutmut (install all three via pip). Setup (5 min): Write a
small module matlib.py with three functions: mat_multiply(A, B)
wrapping np.dot, mat_inverse(A) wrapping np.linalg.inv,
and mat_transpose(A) wrapping A.T. Discover invariants
(10 min): Write Hypothesis property tests for at least six invariants across
multiple families: roundtrip (inverse(inverse(A)) ≈ A), idempotence
(transpose(transpose(A)) == A), preservation (multiply preserves dimensions),
metamorphic (transpose(A * B) == transpose(B) * transpose(A)), and oracle
(compare your multiply against np.matmul). Use
st.floats(min_value=-100, max_value=100, allow_nan=False, allow_infinity=False)
with np.testing.assert_allclose for tolerance. Vary and observe
(10 min): Run mutmut run --paths-to-mutate=matlib.py and record the
mutation score. Then remove one invariant at a time and re-run to see which property
contributes the most unique mutant kills. What to observe: Which invariant
family catches the most mutants? Does any single property kill more than half the mutants
on its own?
Beyond the invariant discovery pattern presented here, recent systems push toward fully autonomous specification mining and test maintenance. MuTAP (Dakhel et al., 2024) demonstrated that LLMs can simultaneously generate tests and apply mutation analysis in a single feedback loop, achieving mutation scores 20-30% higher than standalone LLM-generated tests by iteratively refining properties based on surviving mutants. Meanwhile, Meta's TestGen-LLM (2024) showed that LLM-generated tests can be integrated into industrial CI pipelines at scale, automatically filtering for tests that improve line and branch coverage on the existing production codebase. The next frontier combines these threads: systems like CoverUp (Pizzorno and Berger, 2024) close the loop entirely by running coverage analysis after every code change, identifying uncovered branches, and generating targeted tests without human intervention, while also pruning redundant tests that kill no unique mutants. These advances point toward test suites that evolve alongside the code they protect, paralleling the self-driving laboratory concept explored in Chapter 55.
There is a delightful recursive quality to mutation testing: you can run mutmut on mutmut's own source code to check whether mutmut's tests are strong enough to detect faults in mutmut itself. The mutmut project does exactly this, and as of early versions their mutation score reportedly hovered around 85% . The remaining 15% of surviving mutants are mostly in logging and display formatting code, where the "correct" behavior is a matter of taste rather than correctness. This self-referential quality is fitting: a tool that measures test quality should itself have measurably high-quality tests.
Exercise 18.3.1
A function normalize(values: list[float]) -> list[float] scales every
element so the output sums to 1.0. Write down one invariant from each of three
different families (preservation, idempotence, and metamorphic) that this function
should satisfy. For each invariant, explain whether a mutation that replaces division
with multiplication would be caught.
Hint
Preservation: the output length must equal the input length. Idempotence: normalizing an already-normalized list should return the same list (within floating-point tolerance). Metamorphic: multiplying every input element by a positive constant should not change the output. The division-to-multiplication mutation breaks the sum-to-one property, so the preservation invariant alone will not catch it (the length is unchanged), but both the idempotence and metamorphic invariants will fail because the output values will be wildly wrong.
Try It: Invariant Discovery on a Sorting Function
Build a minimal invariant discovery pipeline for a custom sorting function using only standard Python libraries and Hypothesis. (1) Write a file my_sort.py containing a function merge_sort(lst: list[int]) -> list[int] that implements merge sort from scratch (no calls to sorted() or list.sort()). (2) Manually write five Hypothesis property tests covering the five invariant families: roundtrip (sorting a sorted list returns the same list), idempotence (sorting twice equals sorting once), preservation (output has the same length and same elements as the input), metamorphic (appending a value smaller than all elements places it first after sorting), and oracle comparison (output matches Python's built-in sorted()). (3) Run pytest test_my_sort.py -v and confirm all five properties pass. (4) Install mutmut (pip install mutmut) and run mutmut run --paths-to-mutate=my_sort.py --tests-dir=. to measure the mutation score. Record how many mutants survive. (5) For each surviving mutant, run mutmut show <id> to inspect the mutation, then write a targeted property test that kills it. Re-run mutmut and confirm the mutation score increases. This exercise takes about 30 minutes and requires only pytest, hypothesis, and mutmut.
Exercises
Exercise 18.3.1 (Conceptual): The invariant discovery system uses the LLM's self-assessed confidence score for each proposed invariant. Describe two failure modes of this self-assessment: a case where the LLM assigns high confidence to a false invariant, and a case where it assigns low confidence to a true (and important) invariant. How does the validation stage (Stage 2) mitigate these failures?
Exercise 18.3.2 (Coding): Implement the complete invariant discovery pipeline for a function of your choice (at least 30 lines, with multiple branches and at least two input parameters). Run all four stages and report the final confidence score. Then, intentionally introduce a bug and verify that the confidence score drops. What is the minimum bug severity (single character change, operator replacement, constant change) that your discovered invariants can detect?
Exercise 18.3.3 (Analysis):
Compare the invariants generated by Hypothesis's ghostwriter
(hypothesis write your_function) with those generated by the LLM-based
Stage 1 of the invariant discovery system. For a scientific function of your choice,
count the invariants unique to each approach, classify them by family, and measure
the mutation score achieved by each set independently and combined. What does this
tell you about the complementary strengths of type-driven and semantics-driven
invariant inference?