Differential Testing Is the Proof Your Extraction Is Missing

Most extractions break at merge, not at edit time. The build passes. The tests pass. The function still misbehaves in production.

The missing layer is differential testing. Run the old function and the new function on identical recorded inputs. Use separate processes with identical starting state. Diff the outputs. Diff the state. A match means the extraction changed nothing. A difference shows you exactly where.

This article shows the full workflow: record the tape, build the rig, extract in slices. The harness below is self-contained. Adapt it to your module.

The core rule

Do not move code until the tape agrees.

One slice. One run of the rig. One commit. If the rig fails, revert the slice and look again. Extraction becomes a repeatable measurement instead of a leap of faith.

Why characterization tests are not enough

Characterization tests pin a handful of examples. Differential tests replay the recorded tape until the logic stops changing. They answer different questions:

Question	Tool that answers it
Does the known behavior still hold?	Characterization tests
Does the new code match the old on the real tape?	Differential testing
Does the logic hold for inputs you never recorded?	Property-based tests

For a 400-line god function, the tape question is the one that protects your Friday. Characterization alone leaves the gap wide open.

Phase 1: Find the tentacles

Record the state surface before you record calls. The dis module shows every global read and write at the bytecode level.

import dis

dis.dis(calculate_invoice)

Look for LOAD_GLOBAL and STORE_GLOBAL. Those names are your state surface: rate tables, caches, config dicts. State you forget to record is state you cannot compare.

Phase 2: Record the tape

Wrap the function. For every call, save the arguments, the state before the call, the state after the call, and the result. Append each record to a JSONL file.

import copy
import json

_RECORDING = "snapshot.jsonl"

def recorder(fn, state_names):
    def wrapped(*args, **kwargs):
        state_before = {
            name: copy.deepcopy(globals()[name])
            for name in state_names
        }
        result = fn(*args, **kwargs)
        state_after = {
            name: copy.deepcopy(globals()[name])
            for name in state_names
        }
        with open(_RECORDING, "a") as fh:
            fh.write(json.dumps({
                "args": args,
                "kwargs": kwargs,
                "state_before": state_before,
                "state_after": state_after,
                "result": result,
            }, default=str) + "\n")
        return result
    return wrapped

calculate_invoice = recorder(
    calculate_invoice, ["_RATE_TABLE", "_cached_total"]
)

Record for a few days if you can. Record for a few hours if you cannot. More traffic means more branch coverage. Every recorded case is a vote against silent regression.

Phase 3: Build the differential rig

Each candidate runs in its own subprocess. Both receive the same starting state. Compare the normalized outputs.

import json
import subprocess
import sys

RECORDING = "snapshot.jsonl"
ENTRY = "calculate_invoice"

def run_candidate(module_name, record):
    seed = record["state_before"]
    script = f"""
import json, sys
import importlib
mod = importlib.import_module({module_name!r})
for k, v in {seed!r}.items():
    setattr(mod, k, v)
args = {record['args']!r}
kwargs = {record['kwargs']!r}
out = mod.{ENTRY}(*args, **kwargs)
print(json.dumps({{"result": out, "state": {{
    k: getattr(mod, k) for k in {seed!r}
}}}}, default=str))
"""
    return subprocess.run(
        [sys.executable, "-c", script],
        capture_output=True, text=True, check=True
    ).stdout

def normalize(payload):
    return json.dumps(json.loads(payload), sort_keys=True, default=str)

def main():
    total = 0
    mismatches = 0
    for line in open(RECORDING):
        record = json.loads(line)
        old_out = normalize(run_candidate("legacy_billing", record))
        new_out = normalize(run_candidate("new_billing", record))
        total += 1
        if old_out != new_out:
            mismatches += 1
            print(f"case {total}: MISMATCH")
            print(f"old: {old_out}")
            print(f"new: {new_out}")
    print(f"{mismatches} mismatches out of {total} cases")
    sys.exit(1 if mismatches else 0)

if __name__ == "__main__":
    main()

Why a subprocess? Because state leaks are the point. In a shared process, the old run's cache can mask a regression in the new run. Fresh processes keep the comparison fair.

This is also where free tooling earns its place. MonkeyCode offers free model access and a free server option. Disclosure: This article was prepared as part of MonkeyCode's product outreach. That fits the economics of throwaway scaffolding: the rig is boilerplate you delete after the extraction. Spending paid quota on it is the wrong trade. Generate a draft with any free endpoint. Then treat that draft as suspect until it passes on real recordings. The rig, not the generator, owns the truth.

Phase 4: Extract in slices

Create new_billing.py as a copy of legacy_billing.py. Apply only one change: extract a pure pricing function and route the old body through it.

# new_billing.py — slice 1
def compute_price(rate, usage_minutes, discount, tax_rate):
    subtotal = rate * usage_minutes
    return subtotal * (1 - discount) * (1 + tax_rate)

def calculate_invoice(user_id, plan, usage_minutes):
    if user_id in _cached_total:
        return _cached_total[user_id]
    rate = _RATE_TABLE.get(plan, _DEFAULT_RATE)
    tax_rate = _load_tax_rate(user_id)
    discount = _discount_for(user_id)
    total = compute_price(rate, usage_minutes, discount, tax_rate)
    _cached_total[user_id] = total
    return total

Run the rig. It passes. Commit. That is slice one.

Slice two moves the cache into a small class. Slice three moves the tax lookup behind an interface. Slice four deletes the dead code. Every slice follows the same loop: change, run the rig, commit or revert.

Speed matters here. If a full run is slow, trim the tape to one case per input signature. Fifty representative cases beat a thousand near-duplicates.

When the old behavior is the bug

The tape records behavior, not intent. If the extraction includes a deliberate bug fix, the differential test will veto it. That is correct. Handle it with a controlled exception.

Run the approved fix against the tape. Collect the cases that differ. Review that diff line by line. If the changes are exactly the intended fix, pin the new outputs as golden and let the rig skip them. I described this as intentional drift in an earlier post. The drift must be explicit, reviewed, and committed as its own change. Never hide it inside a mechanical extraction.

Limitations

• Coverage bias. Recorded traffic reflects production, not rare branches. Add hand-built cases for error paths.
• Nondeterminism. Timestamps, request IDs, and randomness break naive diffs. Normalize or seed them first.
• Hidden state. Class attributes, open files, and database rows leak into comparisons. Enumerate the surface with dis and include everything.
• Speed. One subprocess per case is slow at scale. Shard the tape and run the rig in parallel.

Who should not use this

Skip this workflow if you have no recorded traffic and no staging environment to record from. The rig needs a real tape, not manufactured examples.

Skip it when the old behavior is dangerously wrong. Differential tests lock in bugs. If the contract itself is broken, write specification tests first. Treat that work as a rewrite, not a refactor.

Skip it for one-off migrations with no rollback path. A harness that runs once never pays for its own maintenance.

The takeaway

Extraction is a measurement problem. Record the tape. Build the rig. Move code in slices small enough to prove.

A refactor you cannot measure is a refactor you cannot trust. Free models generate more boilerplate than ever. Measurement is the only edge you control.

Next time a model hands you a 200-line extraction, ask it for the recording. Then run the rig. Let the tape decide.