null: crdtkit -- OR-Set/PN-Counter/LWW-Register with merge laws proven by simulation and by mutation

wizard-work/null/crdtkit/README-crdtkit.md

@@ -0,0 +1,84 @@
+# crdtkit — state-based CRDTs with proven merge laws
+
+Three dependency-free CvRDTs in `orcrdt.py`, and a test suite that checks the merge
+laws two ways: by random gossip simulation, and by **mutation** — deliberately
+broken merges must be *caught* by the same check, or the check is decoration.
+
+    crdtkit/orcrdt.py        ORSet, PNCounter, LWWRegister, CRDTError   (stdlib only)
+    crdtkit/test_crdt.py     29 unittest cases: laws, semantics, serialization,
+                             convergence, mutation teeth, differential oracle
+    crdtkit/evidence_run.py  prints the evidence matrix; exit 0 iff every claim holds
+    crdtkit/evidence.txt     captured output of the run below
+
+Re-run (any cwd):
+
+    python3 /root/troupe-standup/wizard-work/null/crdtkit/test_crdt.py
+    python3 /root/troupe-standup/wizard-work/null/crdtkit/evidence_run.py
+
+## Semantics
+
+| type | state | merge | guarantee |
+|---|---|---|---|
+| `ORSet` | `elem -> {tag}`, tombstone set of tags; tag = `(replica, counter)` | union of both sets | add-wins: a `remove` kills only the tags it observed |
+| `PNCounter` | per-replica `inc`/`dec` vectors | pointwise **max** | each replica writes only its own slot, so max is a join |
+| `LWWRegister` | `(value, stamp)`; stamp = `(timestamp:int, replica:str)` | max stamp | winner independent of merge order; equal stamp + differing value is refused |
+
+`merge` is pure — it never mutates `self` or its arguments — and every state has a
+canonical `state_key()` plus a `to_state()/dumps()/loads()` JSON round-trip.
+
+## Measured evidence (verbatim `evidence.txt`)
+
+    orcrdt evidence run
+    python 3.14.7 on Linux
+    seeds 0..15, 3 replicas x 80 ops x 4 gossip rounds (5x150 for the big one)
+
+    [A] convergence: every replica equals the join of all pre-gossip states
+        ORSet        16/16 seeds  5-replica ensemble: converged
+        PNCounter    16/16 seeds  5-replica ensemble: converged
+        LWWRegister  16/16 seeds  5-replica ensemble: converged
+
+    [B] mutation teeth: a broken merge must be caught by the same check
+        ORSetDropsForeignTombstones      caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes
+        PNCounterDropsDecrements         caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes
+        LWWLastArgumentWins              caught at 10/16 seeds; semantic divergence: yes; real class green on those seeds: yes
+        ORSetNoOpMerge                   caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes
+        PNCounterSumMerge                caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes
+
+    [C] differential vs independent reference model (OR-Set)
+        16/16 seeds agree (both implementations, all replicas)
+
+    FAILURES: none
+
+Additional runs (measured, not claimed): suite exits 0 under `PYTHONHASHSEED` in
+{0, 1, 7, 42, 12345} (no dict-iteration-order dependence) and under `python3 -O`.
+
+Negative control (measured): a copy of this kit whose `ORSet.merge` drops foreign
+tombstones — a one-line change in the merge — exits **1** with 7 failures, while the
+unmodified kit exits 0. The CHECK is therefore a real test, not a tautology.
+
+## What each block proves, and what it does not
+
+* `[A]` — replicas that gossip **serialized snapshots** in random order, with
+  duplicate deliveries, all end at the exact join of their pre-gossip states, in
+  both state and semantic value. 3 replicas x 80 ops x 16 seeds, plus a 5-replica
+  x 150-op ensemble. Not proven: any real transport, reordering under partitions,
+  or Byzantine behaviour.
+* `[B]` — five wrong merges (drop foreign tombstones, drop decrements, last-argument
+  wins, ignore the argument, sum instead of max) all go red on the same check, while
+  the real class stays green on exactly those seeds. This is the part that makes
+  `[A]` evidence rather than a green light that would also shine on broken code.
+  `LWWLastArgumentWins` is caught on 10/16 seeds, not 16 — reported as measured.
+* `[C]` — `ORSet` agrees with `RefORSet`, a separately written naive model with a
+  different state layout (flat `(elem, replica, counter)` tag set). Agreement is
+  cross-implementation, not self-confirmation. Not proven: agreement on workloads
+  outside the harness (only 5 distinct elements, 3 replicas, one op mix).
+
+## Honest limits
+
+* Elements must be JSON scalars (`str/int/float/bool/None`) — deliberate, so
+  `to_state()` is lossless rather than `repr`-based.
+* `LWWRegister` assumes replica ids are unique per node; two nodes sharing an id
+  and writing at the same timestamp raise `CRDTError` instead of silently diverging.
+* Tombstones and per-replica vector slots are never garbage-collected: state grows
+  with history. No delta-CRDT or anti-entropy protocol is provided.
+* Merge is whole-state (`O(state)`), not incremental.

wizard-work/null/crdtkit/evidence.txt

@@ -0,0 +1,20 @@
+orcrdt evidence run
+python 3.14.7 on Linux
+seeds 0..15, 3 replicas x 80 ops x 4 gossip rounds (5x150 for the big one)
+
+[A] convergence: every replica equals the join of all pre-gossip states
+    ORSet        16/16 seeds  5-replica ensemble: converged
+    PNCounter    16/16 seeds  5-replica ensemble: converged
+    LWWRegister  16/16 seeds  5-replica ensemble: converged
+
+[B] mutation teeth: a broken merge must be caught by the same check
+    ORSetDropsForeignTombstones      caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes
+    PNCounterDropsDecrements         caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes
+    LWWLastArgumentWins              caught at 10/16 seeds; semantic divergence: yes; real class green on those seeds: yes
+    ORSetNoOpMerge                   caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes
+    PNCounterSumMerge                caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes
+
+[C] differential vs independent reference model (OR-Set)
+    16/16 seeds agree (both implementations, all replicas)
+
+FAILURES: none

wizard-work/null/crdtkit/evidence_run.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python3
+"""evidence_run.py -- reproducible evidence for orcrdt; exit 0 iff all claims hold.
+
+Prints, in one pass:
+  [A] random-gossip convergence over every seed, per CRDT family
+  [B] mutation teeth: each deliberately broken merge must be caught, and the real
+      class must pass exactly the seeds where its mutant failed
+  [C] differential agreement with an independently written reference model
+
+Nothing here trusts the library: every line is recomputed from the simulation
+harness in test_crdt.py, and the script's own exit code is the verdict.
+"""
+
+from __future__ import annotations
+
+import os
+import platform
+import random
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import test_crdt as T  # noqa: E402
+from orcrdt import LWWRegister, ORSet, PNCounter  # noqa: E402
+
+FAMILIES = ((ORSet, T.ORSetDropsForeignTombstones),
+            (PNCounter, T.PNCounterDropsDecrements),
+            (LWWRegister, T.LWWLastArgumentWins),
+            (ORSet, T.ORSetNoOpMerge),
+            (PNCounter, T.PNCounterSumMerge))
+
+
+def main():
+    failures = []
+    print("orcrdt evidence run")
+    print("python %s on %s" % (platform.python_version(), platform.system()))
+    print("seeds 0..%d, 3 replicas x 80 ops x 4 gossip rounds (5x150 for the big one)"
+          % (max(T.SEEDS),))
+    print()
+
+    print("[A] convergence: every replica equals the join of all pre-gossip states")
+    for real in dict.fromkeys(real for real, _ in FAMILIES):
+        passed = sum(1 for s in T.SEEDS if T.check_convergence(real, s))
+        big = all(T.check_convergence(real, s, n=5, ops=150, rounds=6)
+                  for s in (0, 1, 2))
+        print("    %-12s %2d/%d seeds  5-replica ensemble: %s"
+              % (real.__name__, passed, len(list(T.SEEDS)),
+                 "converged" if big else "DIVERGED"))
+        if passed != len(list(T.SEEDS)) or not big:
+            failures.append("convergence %s" % real.__name__)
+    print()
+
+    print("[B] mutation teeth: a broken merge must be caught by the same check")
+    for real, mutant in FAMILIES:
+        hits = T.mutant_detected(mutant)
+        real_ok = all(T.check_convergence(real, s) for s in hits)
+        value_wrong = any(not T.check_convergence_detail(mutant, s)["value_ok"]
+                          for s in hits)
+        print("    %-32s caught at %2d/%d seeds; semantic divergence: %-3s; "
+              "real class green on those seeds: %s"
+              % (mutant.__name__, len(hits), len(list(T.SEEDS)),
+                 "yes" if value_wrong else "no", "yes" if real_ok else "NO"))
+        if not hits or not real_ok:
+            failures.append("teeth %s" % mutant.__name__)
+    print()
+
+    print("[C] differential vs independent reference model (OR-Set)")
+    agree = 0
+    for seed in range(16):
+        rng = random.Random(seed * 31 + 5)
+        live = [ORSet("r%d" % i) for i in range(3)]
+        ref = [T.RefORSet("r%d" % i) for i in range(3)]
+        for _ in range(120):
+            i, elem = rng.randrange(3), rng.choice(T.ELEMS)
+            if rng.random() < 0.55:
+                live[i].add(elem)
+                ref[i].add(elem)
+            else:
+                live[i].remove(elem)
+                ref[i].remove(elem)
+        for dst, src in [(rng.randrange(3), rng.randrange(3)) for _ in range(60)]:
+            if dst == src:
+                continue
+            live[dst] = live[dst].merge(live[src])
+            ref[dst] = ref[dst].merge(ref[src])
+        if live[0].elements() == ref[0].elements() == live[1].elements() == live[2].elements():
+            agree += 1
+    print("    %d/16 seeds agree (both implementations, all replicas)" % agree)
+    if agree != 16:
+        failures.append("differential")
+
+    print()
+    print("FAILURES: %s" % (", ".join(failures) if failures else "none"))
+    return 1 if failures else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

wizard-work/null/crdtkit/orcrdt.py

@@ -0,0 +1,412 @@
+#!/usr/bin/env python3
+"""orcrdt -- dependency-free state-based CRDT kit (OR-Set, PN-Counter, LWW-Register).
+
+Design rules
+------------
+* State-based (CvRDT): every replica exposes ``merge(other, ...) -> new`` implementing
+  a join over a semilattice. ``merge`` is PURE -- it never mutates ``self`` or its
+  arguments, so a replica can re-deliver a snapshot any number of times.
+* A merge that is not commutative / associative / idempotent makes replicas diverge.
+  That is not a matter of opinion, so ``test_crdt.py`` checks it two ways: random
+  gossip simulation over many seeds, and mutation -- deliberately broken merges must
+  be *detected* by the same check.
+* Elements of ``ORSet`` must be JSON scalars (str/int/float/bool/None). This is a
+  deliberate constraint: it makes ``to_state()`` lossless instead of repr-based.
+* No third-party imports. Python 3.9+.
+
+Public surface
+--------------
+ORSet         add-wins observed-remove set
+PNCounter     increment/decrement counter (per-replica vectors, pointwise max)
+LWWRegister   last-writer-wins register, stamp = (timestamp, replica)
+CRDTError     raised on malformed state or a broken local operation
+"""
+
+from __future__ import annotations
+
+import json
+
+__all__ = ["ORSet", "PNCounter", "LWWRegister", "CRDTError"]
+
+_SCALARS = (str, int, float, bool, type(None))
+
+
+class CRDTError(ValueError):
+    """Malformed state, non-scalar element, or a non-monotone local operation."""
+
+
+def _check_scalar(elem, where):
+    if isinstance(elem, _SCALARS):
+        return elem
+    raise CRDTError(
+        "%s: element must be a JSON scalar (str/int/float/bool/None), got %r"
+        % (where, type(elem).__name__)
+    )
+
+
+def _canon(elem):
+    """Canonical, order-stable, comparable form of a scalar element."""
+    return json.dumps(elem, sort_keys=True)
+
+
+def _sorted_tags(tags):
+    return sorted(tags, key=lambda t: (t[0], t[1]))
+
+
+def _pmax(a, b):
+    """Pointwise max over the union of two non-negative-int vectors."""
+    out = dict(a)
+    for k, v in b.items():
+        if v > out.get(k, 0):
+            out[k] = v
+    return out
+
+
+# ---------------------------------------------------------------------------
+# OR-Set
+# ---------------------------------------------------------------------------
+class ORSet:
+    """Observed-Remove Set with add-wins semantics.
+
+    State = (``_adds``: elem -> set of unique tags, ``_removed``: set of tags).
+    A tag is ``(replica, counter)`` and is unique per local ``add``.
+
+    ``contains(e)`` is true iff ``e`` has at least one tag that no ``remove`` has
+    tombstoned. ``remove(e)`` tombstones only the tags the caller had *observed*,
+    so a concurrent ``add`` survives the merge -- that is the add-wins rule.
+    """
+
+    __slots__ = ("replica", "_adds", "_removed", "_clock")
+
+    def __init__(self, replica="r0"):
+        if not isinstance(replica, str) or not replica:
+            raise CRDTError("replica id must be a non-empty string")
+        self.replica = replica
+        self._adds = {}        # elem -> set(tag)
+        self._removed = set()  # set(tag)
+        self._clock = 0
+
+    # -- local mutations ----------------------------------------------------
+    def add(self, elem):
+        _check_scalar(elem, "ORSet.add")
+        self._clock += 1
+        tag = (self.replica, self._clock)
+        self._adds.setdefault(elem, set()).add(tag)
+        return tag
+
+    def remove(self, elem):
+        """Tombstone every tag of ``elem`` observed locally. Returns tags removed."""
+        _check_scalar(elem, "ORSet.remove")
+        tags = self._adds.get(elem)
+        if not tags:
+            return 0
+        fresh = [t for t in tags if t not in self._removed]
+        self._removed.update(fresh)
+        return len(fresh)
+
+    # -- queries ------------------------------------------------------------
+    def contains(self, elem):
+        return any(t not in self._removed for t in self._adds.get(elem, ()))
+
+    def elements(self):
+        return frozenset(e for e in self._adds if self.contains(e))
+
+    def value_key(self):
+        return tuple(sorted(_canon(e) for e in self.elements()))
+
+    def __contains__(self, elem):
+        return self.contains(elem)
+
+    def __len__(self):
+        return len(self.elements())
+
+    def __iter__(self):
+        return iter(sorted(self.elements(), key=_canon))
+
+    # -- merge / state ------------------------------------------------------
+    def merge(self, *others):
+        out = ORSet(self.replica)
+        states = (self,) + others
+        for st in states:
+            if not isinstance(st, ORSet):
+                raise CRDTError("ORSet.merge: expected ORSet, got %r" % type(st).__name__)
+            for e, tags in st._adds.items():
+                out._adds.setdefault(e, set()).update(tags)
+            out._removed.update(st._removed)
+        out._clock = max(st._clock for st in states)
+        return out
+
+    def state_key(self):
+        adds = tuple(
+            sorted(
+                (_canon(e), tuple(_sorted_tags(t))) for e, t in self._adds.items()
+            )
+        )
+        return (adds, tuple(_sorted_tags(self._removed)))
+
+    def __eq__(self, other):
+        return isinstance(other, ORSet) and self.state_key() == other.state_key()
+
+    def __hash__(self):
+        return hash(self.state_key())
+
+    def to_state(self):
+        return {
+            "type": "ORSet",
+            "replica": self.replica,
+            "clock": self._clock,
+            "adds": [
+                [_canon(e), [[t[0], t[1]] for t in _sorted_tags(tags)]]
+                for e, tags in sorted(self._adds.items(), key=lambda kv: _canon(kv[0]))
+            ],
+            "removed": [[t[0], t[1]] for t in _sorted_tags(self._removed)],
+        }
+
+    @classmethod
+    def from_state(cls, d):
+        try:
+            if d["type"] != "ORSet":
+                raise CRDTError("ORSet.from_state: wrong type tag %r" % (d["type"],))
+            out = cls(d["replica"])
+            out._clock = int(d["clock"])
+            for raw_e, raw_tags in d["adds"]:
+                elem = json.loads(raw_e)
+                _check_scalar(elem, "ORSet.from_state")
+                out._adds.setdefault(elem, set()).update(
+                    (t[0], int(t[1])) for t in raw_tags
+                )
+            out._removed = {(t[0], int(t[1])) for t in d["removed"]}
+            if out._clock < 0 or any(c <= 0 for _, c in out._removed):
+                raise CRDTError("ORSet.from_state: non-positive clock/counter")
+            return out
+        except (KeyError, TypeError, ValueError) as exc:
+            if isinstance(exc, CRDTError):
+                raise
+            raise CRDTError("ORSet.from_state: malformed state: %s" % exc) from exc
+
+    def dumps(self):
+        return json.dumps(self.to_state(), sort_keys=True)
+
+    @classmethod
+    def loads(cls, s):
+        try:
+            d = json.loads(s)
+        except ValueError as exc:
+            raise CRDTError("ORSet.loads: not JSON: %s" % exc) from exc
+        if not isinstance(d, dict):
+            raise CRDTError("ORSet.loads: state must be a JSON object")
+        return cls.from_state(d)
+
+
+# ---------------------------------------------------------------------------
+# PN-Counter
+# ---------------------------------------------------------------------------
+class PNCounter:
+    """Increment/decrement counter.
+
+    State = two per-replica vectors (``inc``, ``dec``); merge = pointwise max.
+    Each replica writes only its own slot, which is what makes pointwise max a
+    correct join (a replica's own slot only ever grows).
+    """
+
+    __slots__ = ("replica", "inc", "dec")
+
+    def __init__(self, replica="r0"):
+        if not isinstance(replica, str) or not replica:
+            raise CRDTError("replica id must be a non-empty string")
+        self.replica = replica
+        self.inc = {}
+        self.dec = {}
+
+    def _bump(self, vec, n):
+        if not isinstance(n, int) or isinstance(n, bool) or n < 0:
+            raise CRDTError("PNCounter: delta must be a non-negative int, got %r" % (n,))
+        vec[self.replica] = vec.get(self.replica, 0) + n
+
+    def increment(self, n=1):
+        self._bump(self.inc, n)
+
+    def decrement(self, n=1):
+        self._bump(self.dec, n)
+
+    def value(self):
+        return sum(self.inc.values()) - sum(self.dec.values())
+
+    def value_key(self):
+        return self.value()
+
+    def merge(self, *others):
+        out = PNCounter(self.replica)
+        for st in (self,) + others:
+            if not isinstance(st, PNCounter):
+                raise CRDTError("PNCounter.merge: expected PNCounter, got %r" % type(st).__name__)
+            out.inc = _pmax(out.inc, st.inc)
+            out.dec = _pmax(out.dec, st.dec)
+        return out
+
+    def state_key(self):
+        return (tuple(sorted(self.inc.items())), tuple(sorted(self.dec.items())))
+
+    def __eq__(self, other):
+        return isinstance(other, PNCounter) and self.state_key() == other.state_key()
+
+    def __hash__(self):
+        return hash(self.state_key())
+
+    def to_state(self):
+        return {
+            "type": "PNCounter",
+            "replica": self.replica,
+            "inc": sorted(self.inc.items()),
+            "dec": sorted(self.dec.items()),
+        }
+
+    @classmethod
+    def from_state(cls, d):
+        try:
+            if d["type"] != "PNCounter":
+                raise CRDTError("PNCounter.from_state: wrong type tag %r" % (d["type"],))
+            out = cls(d["replica"])
+            for key in ("inc", "dec"):
+                vec = {}
+                for k, v in d[key]:
+                    if not isinstance(k, str) or not isinstance(v, int) or v < 0:
+                        raise CRDTError("PNCounter.from_state: bad entry in %s" % key)
+                    vec[k] = v
+                setattr(out, key, vec)
+            return out
+        except (KeyError, TypeError, ValueError) as exc:
+            if isinstance(exc, CRDTError):
+                raise
+            raise CRDTError("PNCounter.from_state: malformed state: %s" % exc) from exc
+
+    def dumps(self):
+        return json.dumps(self.to_state(), sort_keys=True)
+
+    @classmethod
+    def loads(cls, s):
+        try:
+            d = json.loads(s)
+        except ValueError as exc:
+            raise CRDTError("PNCounter.loads: not JSON: %s" % exc) from exc
+        if not isinstance(d, dict):
+            raise CRDTError("PNCounter.loads: state must be a JSON object")
+        return cls.from_state(d)
+
+
+# ---------------------------------------------------------------------------
+# LWW-Register
+# ---------------------------------------------------------------------------
+class LWWRegister:
+    """Last-writer-wins register.
+
+    Stamp = ``(timestamp: int, replica: str)``. Merge takes the maximum stamp, so
+    the winner is total-order-determined and independent of argument order. Local
+    writes must advance the stamp strictly (``set`` raises otherwise), and two
+    different values at an *identical* stamp are refused rather than silently
+    broken -- that only happens if two nodes reuse one replica id.
+    """
+
+    __slots__ = ("replica", "_value", "_stamp")
+
+    def __init__(self, replica="r0"):
+        if not isinstance(replica, str) or not replica:
+            raise CRDTError("replica id must be a non-empty string")
+        self.replica = replica
+        self._value = None
+        self._stamp = (-1, "")
+
+    def set(self, value, timestamp):
+        if not isinstance(timestamp, int) or isinstance(timestamp, bool):
+            raise CRDTError("LWWRegister.set: timestamp must be an int")
+        stamp = (timestamp, self.replica)
+        if stamp <= self._stamp:
+            raise CRDTError(
+                "LWWRegister.set: non-monotone stamp %r <= %r" % (stamp, self._stamp)
+            )
+        self._value = value
+        self._stamp = stamp
+        return stamp
+
+    def get(self):
+        return self._value
+
+    @property
+    def stamp(self):
+        return self._stamp
+
+    def value_key(self):
+        return self._value
+
+    def merge(self, *others):
+        out = LWWRegister(self.replica)
+        out._value, out._stamp = self._value, self._stamp
+        for st in (self,) + others:
+            if not isinstance(st, LWWRegister):
+                raise CRDTError(
+                    "LWWRegister.merge: expected LWWRegister, got %r" % type(st).__name__
+                )
+            if st._stamp > out._stamp:
+                out._value, out._stamp = st._value, st._stamp
+            elif st._stamp == out._stamp and st._value != out._value:
+                raise CRDTError(
+                    "LWWRegister.merge: conflicting values at identical stamp %r"
+                    % (st._stamp,)
+                )
+        return out
+
+    def state_key(self):
+        return (self._stamp, _canon(self._value) if isinstance(self._value, _SCALARS) else repr(self._value))
+
+    def __eq__(self, other):
+        return isinstance(other, LWWRegister) and self.state_key() == other.state_key()
+
+    def __hash__(self):
+        return hash(self.state_key())
+
+    def to_state(self):
+        return {
+            "type": "LWWRegister",
+            "replica": self.replica,
+            "stamp": [self._stamp[0], self._stamp[1]],
+            "value": self._value,
+        }
+
+    @classmethod
+    def from_state(cls, d):
+        try:
+            if d["type"] != "LWWRegister":
+                raise CRDTError("LWWRegister.from_state: wrong type tag %r" % (d["type"],))
+            out = cls(d["replica"])
+            out._stamp = (int(d["stamp"][0]), d["stamp"][1])
+            out._value = d["value"]
+            return out
+        except (KeyError, TypeError, ValueError, IndexError) as exc:
+            if isinstance(exc, CRDTError):
+                raise
+            raise CRDTError("LWWRegister.from_state: malformed state: %s" % exc) from exc
+
+    def dumps(self):
+        return json.dumps(self.to_state(), sort_keys=True)
+
+    @classmethod
+    def loads(cls, s):
+        try:
+            d = json.loads(s)
+        except ValueError as exc:
+            raise CRDTError("LWWRegister.loads: not JSON: %s" % exc) from exc
+        if not isinstance(d, dict):
+            raise CRDTError("LWWRegister.loads: state must be a JSON object")
+        return cls.from_state(d)
+
+
+if __name__ == "__main__":
+    # 12-line smoke illustration: concurrent remove loses to concurrent add.
+    a, b = ORSet("a"), ORSet("b")
+    a.add("x")
+    b = ORSet.loads(a.dumps())
+    a.remove("x")
+    b.add("x")
+    m = a.merge(b)
+    print("orcrdt smoke: a.elements=%r b.elements=%r merged=%r"
+          % (sorted(a), sorted(b), sorted(m)))

wizard-work/null/crdtkit/test_crdt.py

@@ -0,0 +1,506 @@
+#!/usr/bin/env python3
+"""test_crdt.py -- property + mutation tests for orcrdt.
+
+Run:  python3 test_crdt.py          (exit 0 iff every test passes)
+      python3 test_crdt.py -v
+
+Two kinds of evidence:
+1. SIMULATION -- N replicas take random local ops, then gossip their serialized
+   snapshots in random order with duplicate deliveries. Every replica must end at
+   the exact join of all pre-gossip states (state AND semantic value).
+2. MUTATION TEETH -- the same convergence check is pointed at deliberately broken
+   merge implementations; if it does not go red for them, the check is worthless
+   and this suite fails.
+"""
+
+from __future__ import annotations
+
+import os
+import random
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from orcrdt import CRDTError, LWWRegister, ORSet, PNCounter  # noqa: E402
+
+ELEMS = ["alpha", "beta", "gamma", "delta", "epsilon"]
+SEEDS = range(16)
+
+
+# ---------------------------------------------------------------------------
+# simulation harness
+# ---------------------------------------------------------------------------
+def _family(kind):
+    """Map a CRDT class (or a mutant subclass of one) to its base family."""
+    for base in (ORSet, PNCounter, LWWRegister):
+        if issubclass(kind, base):
+            return base
+    raise TypeError("unknown CRDT family: %r" % (kind,))
+
+
+def _local_op(kind, rng, rep):
+    fam = _family(kind)
+    if fam is ORSet:
+        elem = rng.choice(ELEMS)
+        if rng.random() < 0.55:
+            rep.add(elem)
+        else:
+            rep.remove(elem)
+    elif fam is PNCounter:
+        d = rng.choice([1, 1, 1, 2, 3, -1, -2])
+        if d > 0:
+            rep.increment(d)
+        else:
+            rep.decrement(-d)
+    else:
+        ts = rep.stamp[0] + 1 + rng.randrange(3)
+        rep.set(rng.randrange(50), ts)
+
+
+def make_replicas(kind, n, ops, seed):
+    rng = random.Random(seed)
+    reps = [kind("r%d" % i) for i in range(n)]
+    for _ in range(ops):
+        _local_op(kind, rng, rng.choice(reps))
+    return reps
+
+
+def gossip(kind, reps, seed, rounds=4):
+    """Random pairwise merges with duplicate deliveries, over serialized copies."""
+    rng = random.Random(seed * 7919 + 13)
+    n = len(reps)
+    live = [kind.loads(r.dumps()) for r in reps]
+    for _ in range(rounds * n * 3):
+        dst, src = rng.randrange(n), rng.randrange(n)
+        if dst == src:
+            continue
+        live[dst] = live[dst].merge(live[src])
+    return live
+
+
+def check_convergence(kind, seed, n=3, ops=80, rounds=4):
+    """True iff gossip converges to the join of the pre-gossip replica states."""
+    reps = make_replicas(kind, n, ops, seed)
+    snaps = [kind.loads(r.dumps()) for r in reps]
+
+    full = snaps[0]
+    for s in snaps[1:]:
+        full = full.merge(s)
+
+    live = gossip(kind, reps, seed, rounds)
+    state_ok = all(x.state_key() == full.state_key() for x in live)
+    value_ok = all(x.value_key() == full.value_key() for x in live)
+    join_ok = full.state_key() == full.merge(*snaps).state_key()
+    return state_ok and value_ok and join_ok
+
+
+def check_convergence_detail(kind, seed, **kw):
+    reps = make_replicas(kind, kw.get("n", 3), kw.get("ops", 80), seed)
+    snaps = [kind.loads(r.dumps()) for r in reps]
+    full = snaps[0]
+    for s in snaps[1:]:
+        full = full.merge(s)
+    live = gossip(kind, reps, seed, kw.get("rounds", 4))
+    return {
+        "state_ok": all(x.state_key() == full.state_key() for x in live),
+        "value_ok": all(x.value_key() == full.value_key() for x in live),
+    }
+
+
+# ---------------------------------------------------------------------------
+# deliberately broken merges (mutation testing of the harness above)
+# ---------------------------------------------------------------------------
+class ORSetDropsForeignTombstones(ORSet):
+    """BUG: merge keeps only self's tombstone set -> removed elements resurrect."""
+
+    def merge(self, *others):
+        out = super().merge(*others)
+        out._removed = set(self._removed)
+        return out
+
+
+class PNCounterDropsDecrements(PNCounter):
+    """BUG: merge folds inc by max but drops the others' dec vectors."""
+
+    def merge(self, *others):
+        out = PNCounter(self.replica)
+        out.inc, out.dec = dict(self.inc), dict(self.dec)
+        for st in others:
+            for k, v in st.inc.items():
+                if v > out.inc.get(k, 0):
+                    out.inc[k] = v
+        return out
+
+
+class LWWLastArgumentWins(LWWRegister):
+    """BUG: merge always takes the last argument -> order dependent."""
+
+    def merge(self, *others):
+        out = LWWRegister(self.replica)
+        out._value, out._stamp = self._value, self._stamp
+        for st in others:
+            out._value, out._stamp = st._value, st._stamp
+        return out
+
+
+class ORSetNoOpMerge(ORSet):
+    """BUG: merge ignores its argument (no propagation) -> not commutative."""
+
+    def merge(self, *others):
+        out = ORSet(self.replica)
+        out._adds = {e: set(t) for e, t in self._adds.items()}
+        out._removed = set(self._removed)
+        out._clock = self._clock
+        return out
+
+
+class PNCounterSumMerge(PNCounter):
+    """BUG: merge adds the vectors instead of taking pointwise max.
+
+    Commutative and associative, so naive gossip tests can miss it -- it fails
+    IDEMPOTENCE, and any duplicate delivery then inflates the value.
+    """
+
+    def merge(self, *others):
+        out = PNCounter(self.replica)
+        out.inc, out.dec = dict(self.inc), dict(self.dec)
+        for st in others:
+            for k, v in st.inc.items():
+                out.inc[k] = out.inc.get(k, 0) + v
+            for k, v in st.dec.items():
+                out.dec[k] = out.dec.get(k, 0) + v
+        return out
+
+
+def mutant_detected(mutant_cls, seeds=SEEDS):
+    """True iff the convergence check goes red (or raises) for >=1 seed."""
+    hits = []
+    for seed in seeds:
+        try:
+            ok = check_convergence(mutant_cls, seed)
+        except CRDTError:
+            ok = False
+        if not ok:
+            hits.append(seed)
+    return hits
+
+
+class RefORSet:
+    """Independent naive add-wins OR-Set, written from the paper semantics.
+
+    Different layout on purpose: a flat set of ``(elem, replica, counter)`` tags
+    plus a tombstone set, so agreeing with ``orcrdt.ORSet`` is evidence and not
+    self-confirmation. Test-only; not exported by the library.
+    """
+
+    def __init__(self, replica):
+        self.replica = replica
+        self.adds = set()
+        self.removed = set()
+        self.clock = 0
+
+    def add(self, elem):
+        self.clock += 1
+        self.adds.add((elem, self.replica, self.clock))
+
+    def remove(self, elem):
+        for tag in [t for t in self.adds if t[0] == elem]:
+            self.removed.add(tag)
+
+    def elements(self):
+        return frozenset(t[0] for t in self.adds if t not in self.removed)
+
+    def merge(self, *others):
+        out = RefORSet(self.replica)
+        for o in (self,) + others:
+            out.adds |= o.adds
+            out.removed |= o.removed
+            out.clock = max(out.clock, o.clock)
+        return out
+
+
+# ---------------------------------------------------------------------------
+# tests
+# ---------------------------------------------------------------------------
+class TestMergeLaws(unittest.TestCase):
+    def _sample(self, kind, seed):
+        reps = make_replicas(kind, 3, 40, seed)
+        return [kind.loads(r.dumps()) for r in reps]
+
+    def test_commutative_and_associative(self):
+        for kind in (ORSet, PNCounter, LWWRegister):
+            for seed in range(8):
+                a, b, c = self._sample(kind, seed)
+                self.assertEqual(a.merge(b).state_key(), b.merge(a).state_key(),
+                                 "%s not commutative (seed %d)" % (kind.__name__, seed))
+                self.assertEqual(a.merge(b).merge(c).state_key(),
+                                 a.merge(b.merge(c)).state_key(),
+                                 "%s not associative (seed %d)" % (kind.__name__, seed))
+
+    def test_idempotent(self):
+        for kind in (ORSet, PNCounter, LWWRegister):
+            for seed in range(8):
+                for x in self._sample(kind, seed):
+                    self.assertEqual(x.merge(x).state_key(), x.state_key(),
+                                     "%s not idempotent (seed %d)" % (kind.__name__, seed))
+                    self.assertEqual(x.merge(x, x, x).state_key(), x.state_key())
+
+    def test_merge_is_pure(self):
+        for kind in (ORSet, PNCounter, LWWRegister):
+            a, b, _ = self._sample(kind, 3)
+            before = (a.state_key(), b.state_key())
+            a.merge(b)
+            self.assertEqual((a.state_key(), b.state_key()), before,
+                             "%s.merge mutated an input" % kind.__name__)
+
+    def test_duplicate_delivery_harmless(self):
+        for kind in (ORSet, PNCounter, LWWRegister):
+            a, b, _ = self._sample(kind, 5)
+            once = a.merge(b)
+            many = a
+            for _ in range(5):
+                many = many.merge(b)
+            self.assertEqual(many.state_key(), once.state_key())
+
+    def test_type_confusion_refused(self):
+        self.assertRaises(CRDTError, ORSet("a").merge, PNCounter("b"))
+        self.assertRaises(CRDTError, PNCounter("a").merge, LWWRegister("b"))
+        self.assertRaises(CRDTError, LWWRegister("a").merge, ORSet("b"))
+
+
+class TestORSetSemantics(unittest.TestCase):
+    def test_add_wins_on_concurrent_add_remove(self):
+        a = ORSet("a")
+        a.add("x")
+        b = ORSet.loads(a.dumps())
+        a.remove("x")
+        b.add("x")
+        self.assertFalse(a.contains("x"))
+        self.assertTrue(b.contains("x"))
+        self.assertTrue(a.merge(b).contains("x"), "add-wins violated")
+        self.assertTrue(b.merge(a).contains("x"))
+
+    def test_remove_only_kills_observed_tags(self):
+        a = ORSet("a")
+        a.add("x")
+        b = ORSet.loads(a.dumps())
+        b.remove("x")
+        a.add("x")  # new tag, not observed by b
+        self.assertTrue(a.merge(b).contains("x"))
+        self.assertFalse(b.merge(b).contains("x"))
+
+    def test_remove_then_merge_then_still_absent(self):
+        a = ORSet("a")
+        a.add("x")
+        b = a.merge(ORSet("b"))
+        b.remove("x")
+        self.assertFalse(a.merge(b).contains("x"))
+
+    def test_elements_and_iteration(self):
+        a = ORSet("a")
+        for e in ("delta", "alpha", "gamma"):
+            a.add(e)
+        self.assertEqual(sorted(a), ["alpha", "delta", "gamma"])
+        self.assertEqual(len(a), 3)
+        self.assertIn("alpha", a)
+        a.remove("alpha")
+        self.assertEqual(sorted(a), ["delta", "gamma"])
+
+    def test_rejects_non_scalar_elements(self):
+        a = ORSet("a")
+        self.assertRaises(CRDTError, a.add, ["list"])
+        self.assertRaises(CRDTError, a.add, {"k": 1})
+        self.assertRaises(CRDTError, ORSet, "")
+        self.assertRaises(CRDTError, a.remove, ["list"])
+
+
+class TestPNCounter(unittest.TestCase):
+    def test_value_and_merge(self):
+        a, b = PNCounter("a"), PNCounter("b")
+        a.increment(5)
+        a.decrement(2)
+        b.increment(3)
+        self.assertEqual(a.value(), 3)
+        self.assertEqual(b.value(), 3)
+        m = a.merge(b)
+        self.assertEqual(m.value(), 6)
+        self.assertEqual(m.state_key(), b.merge(a).state_key())
+
+    def test_pointwise_max_not_sum(self):
+        a = PNCounter("a")
+        a.increment(4)
+        again = PNCounter.loads(a.dumps())
+        self.assertEqual(a.merge(again).value(), 4, "merge double-counted a slot")
+
+    def test_rejects_negative_delta(self):
+        a = PNCounter("a")
+        self.assertRaises(CRDTError, a.increment, -1)
+        self.assertRaises(CRDTError, a.decrement, -3)
+
+
+class TestLWWRegister(unittest.TestCase):
+    def test_max_stamp_wins_regardless_of_order(self):
+        a, b = LWWRegister("a"), LWWRegister("b")
+        a.set("old", 1)
+        b.set("new", 2)
+        self.assertEqual(a.merge(b).get(), "new")
+        self.assertEqual(b.merge(a).get(), "new")
+
+    def test_replica_tiebreak(self):
+        a, b = LWWRegister("a"), LWWRegister("b")
+        a.set("from-a", 7)
+        b.set("from-b", 7)
+        self.assertEqual(a.merge(b).get(), "from-b")  # ("b" > "a") at equal ts
+        self.assertEqual(b.merge(a).get(), "from-b")
+
+    def test_non_monotone_write_refused(self):
+        a = LWWRegister("a")
+        a.set("v1", 5)
+        self.assertRaises(CRDTError, a.set, "v2", 5)
+        self.assertRaises(CRDTError, a.set, "v2", 4)
+
+    def test_identical_stamp_conflict_refused(self):
+        a = LWWRegister("a")
+        a.set("v1", 3)
+        impostor = LWWRegister.loads(a.dumps())
+        impostor._value = "v2"
+        self.assertRaises(CRDTError, a.merge, impostor)
+
+
+class TestSerialization(unittest.TestCase):
+    def test_roundtrip_preserves_state(self):
+        for kind in (ORSet, PNCounter, LWWRegister):
+            for seed in range(6):
+                for r in make_replicas(kind, 3, 30, seed):
+                    self.assertEqual(kind.loads(r.dumps()).state_key(), r.state_key(),
+                                     "%s roundtrip lost state" % kind.__name__)
+
+    def test_merge_across_serialization_boundary(self):
+        for kind in (ORSet, PNCounter, LWWRegister):
+            reps = make_replicas(kind, 3, 50, 11)
+            direct = reps[0].merge(reps[1])
+            over_wire = kind.loads(reps[0].dumps()).merge(kind.loads(reps[1].dumps()))
+            self.assertEqual(over_wire.state_key(), direct.state_key())
+            self.assertEqual(over_wire.value_key(), direct.value_key())
+
+    def test_malformed_state_refused(self):
+        for kind in (ORSet, PNCounter, LWWRegister):
+            self.assertRaises(CRDTError, kind.loads, "not json")
+            self.assertRaises(CRDTError, kind.loads, "[1, 2, 3]")
+            self.assertRaises(CRDTError, kind.loads, '{"type": "Other"}')
+
+
+class TestDifferentialReference(unittest.TestCase):
+    """Compare OR-Set against an independently written naive reference model."""
+
+    def test_orset_matches_independent_reference(self):
+        for seed in range(16):
+            rng = random.Random(seed * 31 + 5)
+            live = [ORSet("r%d" % i) for i in range(3)]
+            ref = [RefORSet("r%d" % i) for i in range(3)]
+            for _ in range(120):
+                i, elem = rng.randrange(3), rng.choice(ELEMS)
+                if rng.random() < 0.55:
+                    live[i].add(elem)
+                    ref[i].add(elem)
+                else:
+                    live[i].remove(elem)
+                    ref[i].remove(elem)
+            schedule = [(rng.randrange(3), rng.randrange(3)) for _ in range(60)]
+            for dst, src in schedule:
+                if dst == src:
+                    continue
+                live[dst] = live[dst].merge(live[src])
+                ref[dst] = ref[dst].merge(ref[src])
+            self.assertEqual(live[0].elements(), ref[0].elements(),
+                             "differential mismatch at seed %d" % seed)
+            for x in live[1:]:
+                self.assertEqual(x.elements(), live[0].elements(),
+                                 "replica split at seed %d" % seed)
+
+
+class TestConvergence(unittest.TestCase):
+    def test_random_gossip_converges_all_kinds(self):
+        for kind in (ORSet, PNCounter, LWWRegister):
+            for seed in SEEDS:
+                detail = check_convergence_detail(kind, seed)
+                self.assertTrue(
+                    detail["state_ok"] and detail["value_ok"],
+                    "%s diverged at seed %d: %r" % (kind.__name__, seed, detail),
+                )
+                self.assertTrue(check_convergence(kind, seed),
+                                "%s failed join check at seed %d" % (kind.__name__, seed))
+
+    def test_larger_ensemble_converges(self):
+        for kind in (ORSet, PNCounter, LWWRegister):
+            for seed in (0, 1, 2):
+                self.assertTrue(
+                    check_convergence(kind, seed, n=5, ops=150, rounds=6),
+                    "%s diverged with 5 replicas at seed %d" % (kind.__name__, seed),
+                )
+
+
+class TestMutantsBreakNamedLaws(unittest.TestCase):
+    """Each mutant is tied to the specific law it violates."""
+
+    def test_sum_merge_is_not_idempotent(self):
+        a = PNCounterSumMerge("a")
+        a.increment(3)
+        self.assertNotEqual(a.merge(a).state_key(), a.state_key())
+        self.assertEqual(a.merge(a).value(), 6)
+
+    def test_noop_merge_is_not_commutative(self):
+        a, b = ORSetNoOpMerge("a"), ORSetNoOpMerge("b")
+        a.add("x")
+        b.add("y")
+        self.assertNotEqual(a.merge(b).state_key(), b.merge(a).state_key())
+
+    def test_last_argument_wins_is_not_commutative(self):
+        a, b = LWWLastArgumentWins("a"), LWWLastArgumentWins("b")
+        a.set("A", 1)
+        b.set("B", 2)
+        self.assertNotEqual(a.merge(b).value_key(), b.merge(a).value_key())
+
+
+class TestMutationTeeth(unittest.TestCase):
+    """If a broken merge survives the check, the check is decoration."""
+
+    MUTANTS = (ORSetDropsForeignTombstones, PNCounterDropsDecrements,
+               LWWLastArgumentWins, ORSetNoOpMerge, PNCounterSumMerge)
+
+    def test_broken_merges_are_caught(self):
+        for mutant in self.MUTANTS:
+            hits = mutant_detected(mutant)
+            self.assertTrue(hits, "%s survived every seed -- check has no teeth"
+                            % mutant.__name__)
+            for seed in hits:
+                self.assertFalse(check_convergence(mutant, seed))
+
+    def test_real_classes_pass_the_same_seeds(self):
+        pairs = ((ORSet, ORSetDropsForeignTombstones),
+                 (PNCounter, PNCounterDropsDecrements),
+                 (LWWRegister, LWWLastArgumentWins),
+                 (ORSet, ORSetNoOpMerge),
+                 (PNCounter, PNCounterSumMerge))
+        for real, mutant in pairs:
+            hits = mutant_detected(mutant)
+            self.assertTrue(hits)
+            for seed in hits:
+                self.assertTrue(check_convergence(real, seed),
+                                "%s failed where its mutant was caught (seed %d)"
+                                % (real.__name__, seed))
+
+    def test_orset_mutant_is_semantically_wrong_not_just_state_wrong(self):
+        hits = mutant_detected(ORSetDropsForeignTombstones)
+        self.assertTrue(hits)
+        wrong_value = False
+        for seed in hits:
+            d = check_convergence_detail(ORSetDropsForeignTombstones, seed)
+            if not d["value_ok"]:
+                wrong_value = True
+        self.assertTrue(wrong_value, "OR-Set mutant only differed in bookkeeping")
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2 if "-v" in sys.argv else 1)