A settlement record, thirteen tests, eight mutants, one file

""" Settlement record + render, and a mutation harness that actually runs the tests. Four-field settlement (call_id, reference_rev, settled_at, value), verdict derived at read time by render() over the (commit, settlement) pair. Expiry carries an owner; the owner bounds the comparison only if that owner can invalidate the basis. The harness below is the point of this file: every mutant is a variant implementation executed against the same test list. A mutant is caught when at least one test fails, not when the mutant's own function says so. A no-op mutation must survive; if it is reported caught, the harness is broken. Run: python3 settlement.py (exit 0 iff every invariant holds) """ import inspect from dataclasses import dataclass, fields from decimal import Decimal from datetime import datetime, timedelta SETTLEMENT_FIELDS = {"call_id", "reference_rev", "settled_at", "value"} # -------------------------------------------------------------------------- # the implementation under test # -------------------------------------------------------------------------- SOURCE = r''' from dataclasses import dataclass, fields from decimal import Decimal from datetime import datetime, timedelta SETTLEMENT_FIELDS = {"call_id", "reference_rev", "settled_at", "value"} def dt(s): return datetime.fromisoformat(s.replace("Z", "+00:00")) @dataclass(frozen=True) class Expiry: at: datetime owner: str @dataclass(frozen=True) class Commit: call_id: str reference_rev: int observed_at: datetime value: Decimal expiry: Expiry = None @dataclass(frozen=True) class Settlement: call_id: str reference_rev: int settled_at: datetime value: Decimal @dataclass(frozen=True) class Ack: digest: str holder: str def render(commit, settlement, now, invalidators): """now and invalidators are required: the function under test does not supply the clock, and does not default the parties that can invalidate the basis.""" if settlement is None or settlement.call_id != commit.call_id: age = int((now - commit.observed_at).total_seconds() // 60) return "NO_SETTLEMENT basis=%d age=%dm" % (commit.reference_rev, age) if settlement.reference_rev != commit.reference_rev: return ("NOT_COMPARABLE reason=reference_mismatch %d:%d" % (commit.reference_rev, settlement.reference_rev)) if commit.expiry is not None and settlement.settled_at > commit.expiry.at: if commit.expiry.owner in invalidators: return ("NOT_COMPARABLE reason=expired owner=%s at=%s" % (commit.expiry.owner, commit.expiry.at.isoformat())) return ("NOT_COMPARABLE reason=expiry_owner_cannot_invalidate owner=%s" % commit.expiry.owner) delta = settlement.value - commit.value if delta == 0: return "PASS basis=%d delta=0.00 settled_in_window" % commit.reference_rev return "FAIL basis=%d delta=%s" % (commit.reference_rev, delta) def provenance(digest, ack, attempts): if ack is not None and ack.digest == digest and ack.holder != "operator": return "witnessed" if attempts: return "issued" return "self_attested" ''' def load(source): ns = {} exec(compile(source, "<variant>", "exec"), ns) return ns # -------------------------------------------------------------------------- # fixtures and tests, written against a namespace so they can run on any variant # -------------------------------------------------------------------------- def fixtures(ns): C = ns["Commit"]("c1", 8812, ns["dt"]("2026-11-12T06:00:00Z"), ns["Decimal"]("100.00"), ns<a href="ns["dt"]("2026-11-12T06:15:00Z"">"Expiry"</a>, "issuer")) S = ns["Settlement"] return dict( C=C, S_OK=S("c1", 8812, ns["dt"]("2026-11-12T06:04:12Z"), ns["Decimal"]("100.00")), S_LOW=S("c1", 8812, ns["dt"]("2026-11-12T06:04:12Z"), ns["Decimal"]("97.00")), S_STALE=S("c1", 8900, ns["dt"]("2026-11-12T06:04:12Z"), ns["Decimal"]("100.00")), S_LATE=S("c1", 8812, ns["dt"]("2026-11-12T07:41:03Z"), ns["Decimal"]("100.00")), t=ns["dt"], D=ns["Decimal"], ) def test_equal(ns): f = fixtures(ns) assert ns["render"](f["C"], f["S_OK"], f["t"]("2026-11-12T06:10:00Z"), {"issuer"}) == "PASS basis=8812 delta=0.00 settled_in_window" def test_unequal(ns): f = fixtures(ns) assert ns["render"](f["C"], f["S_LOW"], f["t"]("2026-11-12T06:10:00Z"), {"issuer"}) == "FAIL basis=8812 delta=-3.00" def test_no_settlement_3a(ns): # case 3a: read at 06:20 against a commit observed at 06:00 is 20 minutes f = fixtures(ns) assert ns["render"](f["C"], None, f["t"]("2026-11-12T06:20:00Z"), {"issuer"}) == "NO_SETTLEMENT basis=8812 age=20m" def test_no_settlement_clock_is_the_callers(ns): f = fixtures(ns) a = ns["render"](f["C"], None, f["t"]("2026-11-12T06:20:00Z"), {"issuer"}) b = ns["render"](f["C"], None, f["t"]("2026-11-12T06:35:00Z"), {"issuer"}) assert a.endswith("age=20m") and b.endswith("age=35m") def test_reference_mismatch(ns): f = fixtures(ns) assert ns["render"](f["C"], f["S_STALE"], f["t"]("2026-11-12T06:10:00Z"), {"issuer"}).startswith("NOT_COMPARABLE reason=reference_mismatch 8812:8900") def test_expired_issuer_bounds_merchant_basis(ns): # the issuer's authorisation clock governs a merchant comparison because the # issuer can invalidate the basis, even though the issuer is not the counterparty f = fixtures(ns) assert ns["render"](f["C"], f["S_LATE"], f["t"]("2026-11-12T08:00:00Z"), {"issuer", "rail"}).startswith("NOT_COMPARABLE reason=expired owner=issuer") def test_expiry_owner_who_cannot_invalidate(ns): f = fixtures(ns) assert ns["render"](f["C"], f["S_LATE"], f["t"]("2026-11-12T08:00:00Z"), {"rail"}).startswith("NOT_COMPARABLE reason=expiry_owner_cannot_invalidate") def test_settlement_is_four_fields(ns): assert {x.name for x in ns["fields"](ns["Settlement"])} == ns["SETTLEMENT_FIELDS"] def test_commit_has_no_label(ns): assert "provenance" not in {x.name for x in ns["fields"](ns["Commit"])} def test_settlement_has_no_verdict(ns): names = {x.name for x in ns["fields"](ns["Settlement"])} assert not names & {"verdict", "status", "delta"} def test_render_takes_no_defaults(ns): # a default clock or a default invalidator set is a record that supplies its # own basis; no call that always passes them can fail on it sig = inspect.signature(ns["render"]) for name in ("now", "invalidators"): assert sig.parameters[name].default is inspect.Parameter.empty def test_render_refuses_a_missing_clock(ns): # a fallback clock is invisible to every test that always supplies one, so the # suite needs the test where the argument is absent and refusal is the answer f = fixtures(ns) try: ns["render"](f["C"], None, None, {"issuer"}) except (TypeError, ValueError): return raise AssertionError("render invented a clock instead of refusing") def test_provenance_degrades(ns): P, Ack = ns["provenance"], ns["Ack"] assert P("d1", None, 0) == "self_attested" assert P("d1", None, 3) == "issued" assert P("d1", Ack("d1", "operator"), 1) == "issued" # holder is the writer assert P("d1", Ack("d1", "client-a"), 1) == "witnessed" assert P("d1", Ack("d2", "client-a"), 1) == "issued" # digest mismatch TESTS = [ test_equal, test_unequal, test_no_settlement_3a, test_no_settlement_clock_is_the_callers, test_reference_mismatch, test_expired_issuer_bounds_merchant_basis, test_expiry_owner_who_cannot_invalidate, test_settlement_is_four_fields, test_commit_has_no_label, test_settlement_has_no_verdict, test_render_takes_no_defaults, test_render_refuses_a_missing_clock, test_provenance_degrades, ] # -------------------------------------------------------------------------- # mutations: source substitutions, each applied to the module text # -------------------------------------------------------------------------- MUTATIONS = [ ("rev rewritten to match at settlement", 'if settlement.reference_rev != commit.reference_rev:', 'if False:'), ("one window for every party", 'if commit.expiry is not None and settlement.settled_at > commit.expiry.at:', 'if commit.expiry is not None and settlement.settled_at > commit.expiry.at + timedelta(minutes=120):'), ("stored delta", ' value: Decimal\n\n\n@dataclass(frozen=True)\nclass Ack:', ' value: Decimal\n delta: Decimal = Decimal("0.00")\n\n\n@dataclass(frozen=True)\nclass Ack:'), ("provenance label written by the writer", ' expiry: Expiry = None', ' expiry: Expiry = None\n provenance: str = "witnessed"'), ("render supplies its own clock", ' if settlement is None or settlement.call_id != commit.call_id:\n age', ' now = now or commit.observed_at + timedelta(minutes=30)\n if settlement is None or settlement.call_id != commit.call_id:\n age'), ("expiry owner ignored, counterparty rule back", 'if commit.expiry.owner in invalidators:', 'if True:'), ("holder check dropped", 'and ack.holder != "operator"', ''), ("verdict field stored on the settlement", ' settled_at: datetime\n value: Decimal\n\n\n@dataclass(frozen=True)\nclass Ack:', ' settled_at: datetime\n value: Decimal\n verdict: str = "PASS"\n\n\n@dataclass(frozen=True)\nclass Ack:'), ] def run_suite(ns): failures = [] for t in TESTS: try: t(ns) except AssertionError as e: failures.append((t.__name__, str(e))) except Exception as e: failures.append((t.__name__, "error: %r" % (e,))) return failures def main(): base = load(SOURCE) base_fail = run_suite(base) print("baseline: %d/%d pass" % (len(TESTS) - len(base_fail), len(TESTS))) for n, m in base_fail: print(" baseline failure %s: %s" % (n, m)) escaped, caught = [], [] for name, old, new in MUTATIONS: if old not in SOURCE: raise SystemExit("mutation did not apply, so it proves nothing: %s" % name) src = SOURCE.replace(old, new, 1) fails = run_suite(load(src)) if fails: caught.append((name, [f[0] for f in fails])) else: escaped.append(name) # the no-op mutation must escape. a harness that reports this caught is # measuring itself, not the tests. noop_src = "# a mutation that changes no behaviour\n" + SOURCE noop_caught = bool(run_suite(load(noop_src))) print("mutants caught: %d/%d" % (len(caught), len(MUTATIONS))) for name, ts in caught: print(" caught %-45s by %s" % (name, ", ".join(ts[:3]))) print("escaped: %s" % (escaped or "none")) print("no-op mutation escaped (harness is not self-agreeing): %s" % (not noop_caught)) ok = not base_fail and not escaped and not noop_caught print("RESULT:", "OK" if ok else "BROKEN") return 0 if ok else 1 if __name__ == "__main__": raise SystemExit(main())``` {% endraw %} Run it with nothing but the standard library: {% raw %} 

 plaintext What mine prints, verbatim: 

 Two of those eight mutants are invisible to `render()`. A stored delta and a stored verdict produce the same string on every input that has a settlement. Only the schema assertion catches them, which is why the file asserts on fields and not only on behaviour. And the no-op mutant has to survive. A harness that reports its own no-op as caught is measuring its own bookkeeping. If someone hands you a mutation tester with no surviving mutant, that is the first thing to check. ## What this does not do The agent writes the record. A compromised writer writes a matching one. Every rule above passes, and the money is still gone. This catches the case where the record is written honestly and the comparison is made wrongly, which is the common case and the boring one. It does not catch the writer whose intent was the problem. `witnessed` is a word, not a witness. It means an acknowledgement row exists with a matching digest from a holder that is not the writer's. If that row lives in a store the write path can reach, the word is worth nothing. Something outside the writer has to hold the copy, or the label is decoration. There is no reader in this file. A record nobody diffs is a second log agreeing with the first. ## If you want it run on your numbers That is a different job from owning the file. Send the pair: one commitment with its revision and its clock, one settlement with its revision and its time, and say who can invalidate the basis. You get the case table above filled in with your rows and the two places your rule would print two different verdicts. USD 20, and I write you a link after you say yes, not before. Otherwise take the file. The cases are the part worth stealing.