Claude Code に大規模リファクタを任せたら本番だけ挙動が変わったとき — 暗黙の契約ドリフトを差分で計測する運用メモ

# probes.py — 契約が集中する代表シナリオを列挙する
from dataclasses import dataclass
from typing import Any, Callable
 
@dataclass
class Probe:
    name: str
    run: Callable[[], Any]   # 対象コードパスを1回叩く。戻り値は正規化前でよい
 
def build_probes(order_service) -> list[Probe]:
    return [
        Probe("normal_single_item",
              lambda: order_service.total(items=[{"sku": "A", "qty": 1, "price": 100}])),
        # 境界: 数量ゼロ。丸めと空集合の扱いが契約
        Probe("boundary_zero_qty",
              lambda: order_service.total(items=[{"sku": "A", "qty": 0, "price": 100}])),
        # 失敗系: 在庫欠品。例外を投げるか既定値かが契約の核心
        Probe("missing_inventory",
              lambda: order_service.total(items=[{"sku": "GHOST", "qty": 1, "price": 100}])),
        # 失敗系: 不正入力。呼び出し側の握りつぶし前提を検証する
        Probe("malformed_input",
              lambda: order_service.total(items=[{"sku": "A"}])),
    ]

# contract_harness.py — 契約スナップショットの採取と差分
import io
import json
import logging
import sys
from contextlib import contextmanager
 
def _normalize_value(v):
    """比較可能な正規形へ。丸めや順序を意図的に固定する。"""
    if isinstance(v, float):
        # 丸めの向きは契約なので、あえて固定桁で署名する
        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():
    """発生したログ行の『形』（logger名・レベル・キー集合）だけを集める。"""
    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):
    """1プローブを実行し、値・チャネル・ログ形状を署名レコードにする。"""
    with _capture_logs() as logs:
        try:
            value = probe.run()
            channel = "return"
            exc_type = None
        except Exception as e:          # noqa: BLE001 — チャネル差分の検出が目的
            value = None
            channel = "raise"
            exc_type = type(e).__name__
    return {
        "probe": probe.name,
        "channel": channel,             # return / raise — これ自体が契約
        "exc_type": exc_type,           # 例外型が変わるのも契約ドリフト
        "value": _normalize_value(value),
        "log_shape": sorted(set(logs)), # 値ではなく骨格だけを署名化
    }
 
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     # 対象アプリの組み立て口
    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("契約ドリフトを検出しました:")
        print(json.dumps(drifts, ensure_ascii=False, indent=2, default=str))
        return 1
    print("契約ドリフトなし")
    return 0
 
if __name__ == "__main__":
    raise SystemExit(main())

# リファクタ開始前の、確実に動いていたコミットで一度だけ実行する
mkdir -p .contract
python contract_harness.py record
git add .contract/baseline.json
git commit -m "chore: 契約スナップショットのベースラインを固定"

# .git/hooks/pre-push — 契約ドリフトが出たら push を止める
#!/usr/bin/env bash
set -euo pipefail
if ! python contract_harness.py check; then
  echo ""
  echo "契約ドリフトが検出されました。意図した変更ならベースラインを更新してください:"
  echo "  python contract_harness.py record && git add .contract/baseline.json"
  exit 1
fi

このリポジトリで大規模リファクタを行います。次の制約を必ず守ってください。
 
1. 変更は1コミット＝1可逆な単位に分割し、各コミット単体でビルド・起動できること。
2. 各コミットの後、私が python contract_harness.py check を回します。
   channel / exception / value / log_shape のいずれかにドリフトが出たら、
   そのコミット内で解消するか、解消できない理由を明記して止まってください。
3. .contract/baseline.json と probes.py は変更しないでください。
   契約の定義はこちら側が管理します。