混亂程式碼庫重構中，如何用互動日誌固定呼叫順序，而不只檢查回傳值

A return-value assertion will not catch a moved webhook.

In a messy repo, the cheapest safety net is an ordered log of collaborator calls. Record it once, diff it after every small change. When the log matches, the change is mechanical. When it diverges, you have a decision to make, not a mystery.

This post is the recording-and-diffing half of that loop. It assumes you already decided what to refactor and only need proof that the change stayed small.

Why order is the part your tests miss

Characterization tests usually pin outputs. Outputs survive most extractions, so most extractions look green.

Effects have order. Write the row, then send the receipt, then emit the event. Move one call and the total is unchanged.

A multiset check says "same calls, same counts". It cannot say "same calls, different order". Order is exactly where the money bugs live.

Step 1 — Census the effects before you touch anything

Start with a grep over the target module. One line per call that leaves the process.

grep -nE '\b(execute|executemany|commit|send|publish|emit|open|requests\.|httpx\.|os\.environ|time\.time|datetime\.now|random\.)' billing/settle.py

Write the hits into three columns. Keep it short.

The rule is simple. Anything that leaves the process is an effect. Anything that returns a fresh value every call gets normalized instead.

Step 2 — Record the ledger without editing production code

Do not add logging to the function you are about to refactor. Instrument the collaborators at the test seam. A side_effect wrapper gives you global ordering across mocks.

# ledger.py
import json
from pathlib import Path


def norm(value, depth=0):
    """Make values comparable across runs."""
    if depth > 4:
        return "<depth-limit>"
    if isinstance(value, float):
        return round(value, 6)
    if isinstance(value, dict):
        return {str(k): norm(v, depth + 1) for k, v in sorted(value.items())}
    if isinstance(value, (list, tuple)):
        return [norm(v, depth + 1) for v in value]
    if isinstance(value, (str, int, bool)) or value is None:
        return value
    return f"<{type(value).__name__}>"


class Ledger:
    def __init__(self):
        self.events = []

    def record(self, kind, target, **fields):
        event = {"seq": len(self.events), "kind": kind, "target": target}
        event.update({k: norm(v) for k, v in fields.items()})
        self.events.append(event)

    def wrap(self, target, fn):
        def side_effect(*args, **kwargs):
            self.record("call", target, args=list(args), kwargs=kwargs)
            result = fn(*args, **kwargs)
            self.record("return", target, result=result)
            return result

        return side_effect

    def write(self, path):
        Path(path).parent.mkdir(parents=True, exist_ok=True)
        with open(path, "w", encoding="utf-8") as fh:
            for event in self.events:
                fh.write(json.dumps(event, sort_keys=True) + "\n")
        return Path(path)

The function under test stays untouched. Here is the module and the recording test.

# billing/settle.py
def settle_invoice(order, db, mailer):
    total = sum(line["qty"] * line["unit_cents"] for line in order["lines"])
    if order.get("discount_pct"):
        total -= total * order["discount_pct"] // 100
    db.execute("INSERT INTO charges (order_id, cents) VALUES (?, ?)", (order["id"], total))
    mailer.send(order["email"], f"Charged {total}")
    return total

# tests/test_settle_ledger.py
import os
from pathlib import Path
from unittest.mock import Mock

from ledger import Ledger
from billing.settle import settle_invoice

GOLDEN = Path("tests/golden/settle_invoice.jsonl")

ORDER = {
    "id": "ord_1",
    "email": "buyer@example.test",
    "discount_pct": 10,
    "lines": [{"qty": 2, "unit_cents": 1500}, {"qty": 1, "unit_cents": 500}],
}


def run_settle(ledger):
    db, mailer = Mock(), Mock()
    db.execute = Mock(side_effect=ledger.wrap("db.execute", lambda *a, **k: None))
    mailer.send = Mock(side_effect=ledger.wrap("mailer.send", lambda *a, **k: "msg_1"))
    return settle_invoice(ORDER, db=db, mailer=mailer)


def test_settle_invoice_order_is_stable(tmp_path):
    ledger = Ledger()
    assert run_settle(ledger) == 3150

    out = ledger.write(tmp_path / "ledger.jsonl")
    if os.environ.get("REGEN"):
        GOLDEN.parent.mkdir(parents=True, exist_ok=True)
        GOLDEN.write_text(out.read_text())

    assert out.read_text() == GOLDEN.read_text()

Four lines come out. call db.execute, return db.execute, call mailer.send, return mailer.send. That is the whole contract you are protecting.

Step 3 — Normalize fixtures, not the recorder

Volatile values ruin the comparison. Timestamps, ids, and floats change every run.

Fix them at the fixture, not in norm(). Use buyer@example.test and ord_1. Only round floats and sort dict keys in code.

If a value is genuinely random, patch the source of randomness. Do not patch the ledger.

Step 4 — Classify the diff instead of eyeballing it

A line diff on JSONL is noisy. Sort each change into one of four classes.

# diff_ledger.py
import json
import sys
from collections import Counter


def load(path):
    with open(path, encoding="utf-8") as fh:
        return [json.loads(line) for line in fh if line.strip()]


def key(event):
    return (event["kind"], event["target"])


def classify(old, new, allow):
    allowed = {(a["kind"], a["target"]) for a in allow}
    deltas = []
    for i in range(max(len(old), len(new))):
        a = old[i] if i < len(old) else None
        b = new[i] if i < len(new) else None
        if a == b:
            continue
        if a is None or b is None:
            kind = "extra" if a is None else "missing"
            target = (b or a)["target"]
        elif key(a) == key(b):
            kind, target = "value", a["target"]
        else:
            kind, target = "reorder", a["target"]
        deltas.append({"index": i, "kind": kind, "target": target, "old": a, "new": b})

    counts_match = Counter(map(key, old)) == Counter(map(key, new))
    reorder_only = counts_match and all(d["kind"] == "reorder" for d in deltas)
    undeclared = [d for d in deltas if (d["kind"], d["target"]) not in allowed]
    return {"deltas": deltas, "reorder_only": reorder_only, "undeclared": undeclared}


def main(argv):
    allow = []
    if len(argv) > 3:
        with open(argv[3], encoding="utf-8") as fh:
            allow = json.load(fh).get("allow", [])

    report = classify(load(argv[1]), load(argv[2]), allow)
    print(
        f"deltas={len(report['deltas'])} "
        f"reorder_only={report['reorder_only']} "
        f"undeclared={len(report['undeclared'])}"
    )
    for d in report["undeclared"]:
        print(f"  #{d['index']} {d['kind']} {d['target']}")
        print(f"    old={json.dumps(d['old'], sort_keys=True)}")
        print(f"    new={json.dumps(d['new'], sort_keys=True)}")
    return 1 if report["undeclared"] else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))

reorder_only is the interesting flag. Counts match, values match, order does not. Most naive checks stop right there and pass.

Step 5 — Declare intended deltas in a reviewed allowlist

Some deltas are the point of the refactor. Those get written down, with a ticket.

{
  "allow": [
    {"kind": "value", "target": "db.execute", "reason": "add currency column, ticket #418"}
  ]
}

Run it in CI.

python diff_ledger.py tests/golden/settle_invoice.before.jsonl \
  tests/golden/settle_invoice.after.jsonl allow.json

An allowlist entry is a reviewed decision. It is not a scratch pad for making CI green.

Step 6 — Pick the smallest safe change with the table

Record the before ledger on the current code. Apply one candidate change. Record after. Classify.

I ran this on two candidates for the same extraction. Numbers from the same fixture above.

Candidate A extracted the total into invoice_total(order). Ledger comparison: deltas=0. Return value 3150 both times. Merged.

Candidate B also hoisted mailer.send above the write, "to keep the billing text together". Ledger comparison: deltas=4 reorder_only=True undeclared=4. Return value still 3150. The receipt now fires before the charge row exists.

That is the failure mode this gate exists for. The output test passed, and the customer gets a receipt for money you have not recorded.

Step 7 — Where a free model and a free server help

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

I keep the model away from the assertions and away from the allowlist decision. Both are mine to defend in review.

The model writes the boring prose. Paste the raw diff and ask for one sentence per delta, in the format the allowlist needs. Then check every line against the JSONL. Free model access makes that cheap enough to do per change instead of per week.

The free server option solves a smaller annoyance. Golden files churn when they are recorded on a laptop with a different locale, user, or installed package set. Record them on a disposable Linux box instead, and keep fixtures synthetic. No customer rows, no real addresses, no tokens. The server is not your CI runner, only a stable recording surface.

Limitations, and who should skip this

This ledger only sees collaborators you instrument. In-process global state, thread interleavings, and database triggers stay invisible.

reorder_only is a heuristic. It compares positions, not causally paired events. Treat it as a red flag, not a proof.

Golden files rot when fixtures change. Regenerate with REGEN=1, then read the diff by hand.

Do not use this on intentionally nondeterministic paths such as async fan-out or retry storms. Do not use it where effects hide inside a compiled dependency you cannot wrap. And skip it if no reviewer will actually read allowlist entries, because a rubber-stamped allowlist is worse than no gate.

Checklist before you touch the function

1. Grep the target module for outbound calls.
2. Write the three-row effect table.
3. Wrap each collaborator with ledger.wrap in a test only.
4. Fix volatile values in fixtures, not in norm.
5. Commit the golden ledger next to the test.
6. Apply one candidate change, then classify.
7. Merge only on deltas=0, or on a reviewed allowlist entry.

If you want to try the recording step somewhere disposable, the free server option is the cheapest path. Keep the fixtures fake and the allowlist short.