When a Claude Code Refactor Passes Every Test but Behaves Differently in Production — Catching Silent Contract Drift with a Behavior Diff Harness

# probes.py — enumerate representative scenarios where contracts concentrate
from dataclasses import dataclass
from typing import Any, Callable
 
@dataclass
class Probe:
    name: str
    run: Callable[[], Any]   # hit the target code path once; value may be un-normalized
 
def build_probes(order_service) -> list[Probe]:
    return [
        Probe("normal_single_item",
              lambda: order_service.total(items=[{"sku": "A", "qty": 1, "price": 100}])),
        # boundary: zero quantity. rounding and empty-set handling is the contract
        Probe("boundary_zero_qty",
              lambda: order_service.total(items=[{"sku": "A", "qty": 0, "price": 100}])),
        # failure: missing inventory. throw vs. default is the heart of the contract
        Probe("missing_inventory",
              lambda: order_service.total(items=[{"sku": "GHOST", "qty": 1, "price": 100}])),
        # failure: malformed input. verifies the caller's swallow assumption
        Probe("malformed_input",
              lambda: order_service.total(items=[{"sku": "A"}])),
    ]

# contract_harness.py — capture contract snapshots and diff them
import json
import logging
import sys
from contextlib import contextmanager
 
def _normalize_value(v):
    """Into a comparable canonical form. Fix rounding and ordering on purpose."""
    if isinstance(v, float):
        # rounding direction is a contract, so sign it at a fixed precision
        return round(v, 4)
    if isinstance(v, dict):
        return {k: _normalize_value(v[k]) for k in sorted(v)}
    if isinstance(v, (list, tuple)):
        return [_normalize_value(x) for x in v]
    return v
 
@contextmanager
def _capture_logs():
    """Collect only the 'shape' of emitted log lines (logger, level, key set)."""
    buf = []
    handler = logging.Handler()
    handler.emit = lambda r: buf.append(
        (r.name, r.levelname, tuple(sorted((r.__dict__.get("extra_keys") or []))))
    )
    root = logging.getLogger()
    root.addHandler(handler)
    try:
        yield buf
    finally:
        root.removeHandler(handler)
 
def take_record(probe):
    """Run one probe; sign value, channel, and log shape into a record."""
    with _capture_logs() as logs:
        try:
            value = probe.run()
            channel = "return"
            exc_type = None
        except Exception as e:          # noqa: BLE001 — detecting channel drift is the goal
            value = None
            channel = "raise"
            exc_type = type(e).__name__
    return {
        "probe": probe.name,
        "channel": channel,             # return / raise — this itself is a contract
        "exc_type": exc_type,           # a changed exception type is contract drift too
        "value": _normalize_value(value),
        "log_shape": sorted(set(logs)), # sign the skeleton, not the values
    }
 
def take_snapshot(probes):
    return {r["probe"]: r for r in (take_record(p) for p in probes)}
 
def diff_snapshots(baseline: dict, current: dict) -> list[dict]:
    drifts = []
    for name, base in baseline.items():
        cur = current.get(name)
        if cur is None:
            drifts.append({"probe": name, "kind": "probe_missing"})
            continue
        if base["channel"] != cur["channel"]:
            drifts.append({"probe": name, "kind": "channel_drift",
                           "from": base["channel"], "to": cur["channel"]})
        if base["exc_type"] != cur["exc_type"]:
            drifts.append({"probe": name, "kind": "exception_drift",
                           "from": base["exc_type"], "to": cur["exc_type"]})
        if base["value"] != cur["value"]:
            drifts.append({"probe": name, "kind": "value_drift",
                           "from": base["value"], "to": cur["value"]})
        if base["log_shape"] != cur["log_shape"]:
            drifts.append({"probe": name, "kind": "log_shape_drift"})
    return drifts
 
def main():
    from probes import build_probes
    from app import build_order_service     # your app's assembly entry point
    probes = build_probes(build_order_service())
    snapshot = take_snapshot(probes)
 
    mode = sys.argv[1] if len(sys.argv) > 1 else "check"
    path = ".contract/baseline.json"
    if mode == "record":
        with open(path, "w") as f:
            json.dump(snapshot, f, ensure_ascii=False, indent=2, default=str)
        print(f"baseline recorded: {len(snapshot)} probes")
        return 0
 
    with open(path) as f:
        baseline = json.load(f)
    drifts = diff_snapshots(baseline, snapshot)
    if drifts:
        print("contract drift detected:")
        print(json.dumps(drifts, ensure_ascii=False, indent=2, default=str))
        return 1
    print("no contract drift")
    return 0
 
if __name__ == "__main__":
    raise SystemExit(main())

# run once at the commit that was definitely working, before the refactor
mkdir -p .contract
python contract_harness.py record
git add .contract/baseline.json
git commit -m "chore: pin the contract snapshot baseline"

# .git/hooks/pre-push — stop the push if contract drift appears
#!/usr/bin/env bash
set -euo pipefail
if ! python contract_harness.py check; then
  echo ""
  echo "Contract drift detected. If the change is intended, update the baseline:"
  echo "  python contract_harness.py record && git add .contract/baseline.json"
  exit 1
fi

We are doing a large refactor in this repo. Please follow these constraints.
 
1. Split changes into one-commit-equals-one-reversible-unit, and each commit
   must build and start on its own.
2. After each commit I will run: python contract_harness.py check.
   If drift appears in channel / exception / value / log_shape, resolve it
   within that commit, or stop and state why it cannot be resolved.
3. Do not modify .contract/baseline.json or probes.py.
   I manage the definition of the contract.