#!/usr/bin/env python3 """ marks_gc_check -- task #110 phase 1: Python validation of two attacks on the Peritext MARKS-layer GC problem (machine-checked verdicts, no Lean here). Background. runtime/src/datatypes/peritext.js keeps text deletes LOGICAL (shadow of every birth + a deleted set) because the document-order mark resolver rehomes a DEAD boundary anchor to its nearest surviving neighbour in reading order, which needs the dead anchor's birth position; consequently compactStable REFUSES for peritext. This file tests two ways to remove that refusal soundly, against a never-compacted twin as the specification. H-A (retention roots): compact with keep-set = live ids UNION boundary anchor ids of present marks (dead anchors retained as dead records, re-coded). CLAIM: renderMarksDoc reads identical to the twin for EVERY beyond-cut continuation at every epoch. Sub-claims: A1 (retained coords keep order under the stable-prefix relabeling, machine-checked inside compact_map), A2 (dropped interior dead nodes are irrelevant to the resolver), A3 (add+remove pairs and their retention roots droppable under a guard). Cost claim: retained extra records <= 2 per mark record. Measured. H-B (early re-anchoring): at the cut rewrite each dead mark boundary to the resolver's own cut-time rehoming target (side kept), then compact with NO mark retention. CLAIM (expected FALSE): reads identical to the twin. The weaker true statement (compacted replicas converge among themselves; divergence is only vs the uncompacted twin) is machine-checked separately and never presented as H-B succeeding. Model notes (verified in selfcheck against embed_tree.py + peritext_read_model.py): * A coordinate is the tuple of birth-chain deltas (delta = id - anchor_id, ids are Lamport timestamps). Reading order = birth-chain lexicographic: ancestor before its subtree; among children of one anchor NEWER timestamps (larger deltas) sort FIRST. key() below realizes this. * The resolver is a transliteration of peritext_read_model.py's DocumentOrderResolver + render (the authoritative model); fidelity is cross-checked on the Ex1/3/5/7/8 + gravity + trilemma histories AND on randomized histories, implementation vs implementation. * Compaction is modeled abstractly (per the task): settled cut = a barrier state every replica has; drop settled-dead records outside the keep-set; order-preserving ordinal relabeling per sibling group with the in-flight freeze guard; beyond-cut mints re-derive coordinates from the (re-coded) anchor record, which IS the stable-prefix map's extension law H3. PASS+FAIL convention (project rule): every claim carries a companion pinning the tempting degenerate behavior; expected values in directed cases are hand-derived in comments, never generated by running the model under test. The twin comparison in the PBT uses the never-compacted twin as the spec, which is not the implementation under test. Run: python3 marks_gc_check.py # full (>=2000 trials per attack) python3 marks_gc_check.py --fast # reduced trial counts (smoke) """ from __future__ import annotations from dataclasses import dataclass, replace import sys # ========================================================================= # Coordinates and reading order. # key(): negate deltas so ascending tuple sort gives (prefix first = ancestor # before subtree, larger delta first = newer sibling first). # ========================================================================= def key(coord): return tuple(-d for d in coord) # ========================================================================= # The peritext state: text shadow (id -> coord, insert-only) + logical # deleted set + codepoints + marks (grow-only map mid -> Mark). # ========================================================================= @dataclass(frozen=True) class Mark: mid: int # mark id = its Lamport timestamp (LWW + growth) mtype: str op: str # 'add' | 'remove' start_id: object # char id, or ('sent', kind) after H-B rewriting end_id: object startSide: str = 'before' # 'before' (inner) | 'after' (outer) endSide: str = 'after' # 'after' (inner) | 'before' (outer) value: object = None class PState: __slots__ = ('shadow', 'anchor', 'deleted', 'codept', 'marks') def __init__(self): self.shadow = {} # id -> coord tuple (birth constant within epoch) self.anchor = {} # id -> anchor id at mint (0 = document start) self.deleted = set() self.codept = {} self.marks = {} # mid -> Mark def copy(self): s = PState() s.shadow = dict(self.shadow) s.anchor = dict(self.anchor) s.deleted = set(self.deleted) s.codept = dict(self.codept) s.marks = dict(self.marks) return s def ins(self, i, cp, a): """Insert char i after anchor a (0 = doc start). Coord derived from THIS state's anchor record: on a compacted state this is exactly the stable-prefix map's extension law H3 (translate(pi ++ d) = translate(pi) ++ d).""" assert i > a, 'Lamport violation: op id must exceed anchor id' base = () if a == 0 else self.shadow[a] self.shadow[i] = base + (i - a,) self.anchor[i] = a self.codept[i] = cp def delete(self, i): self.deleted.add(i) def add_mark(self, m): self.marks[m.mid] = m def birth_order(self): return sorted(self.shadow, key=lambda i: key(self.shadow[i])) def live_order(self): return [c for c in self.birth_order() if c not in self.deleted] def live(self, c): return c in self.shadow and c not in self.deleted def merge(x, y): """State union (all components grow-only; records are epoch-consistent). The coord-agreement assert is the cross-replica consistency check: it fires if the same event ever carries two coordinates in one epoch.""" out = x.copy() for i, c in y.shadow.items(): if i in out.shadow: assert out.shadow[i] == c, f'coord divergence at {i}: {out.shadow[i]} vs {c}' else: out.shadow[i] = c out.anchor[i] = y.anchor[i] out.codept[i] = y.codept[i] out.deleted |= y.deleted for mid, m in y.marks.items(): if mid in out.marks: assert out.marks[mid] == m, f'mark divergence at {mid}' else: out.marks[mid] = m return out # ========================================================================= # The document-order resolver, transliterated from peritext_read_model.py # (DocumentOrderResolver._start_index/_end_index/covered + render). One # addition: a mark anchor with NO record in the shadow (possible only in the # no-retention negative control, or for H-B sentinels) degrades to the side's # collapse branch -- the defined degenerate behavior whose read flips justify # the runtime's current refusal. # ========================================================================= def _scan(birth, deleted, i, step): j = i + step while 0 <= j < len(birth): if birth[j] not in deleted: return birth[j] j += step return None def _start_index(st, m, live, pos, birth, bpos): n = len(live) a, side = m.start_id, m.startSide if isinstance(a, tuple): # H-B collapse sentinel return None if a[1] == 'collapse' else 0 # 'docstart' if not st.live(a): if a not in bpos: # dropped without retention return None if side == 'before' else 0 a = _scan(birth, st.deleted, bpos[a], +1 if side == 'before' else -1) if a is None: return None if side == 'before' else 0 i = pos[a] if side == 'before': # inner start: stable return i first = i + 1 # outer: skip newer run while first < n and live[first] > m.mid: first += 1 return first def _end_index(st, m, live, pos, birth, bpos): n = len(live) a, side = m.end_id, m.endSide if isinstance(a, tuple): return None if a[1] == 'collapse' else n - 1 # 'docend' if not st.live(a): if a not in bpos: return None if side == 'after' else n - 1 a = _scan(birth, st.deleted, bpos[a], -1 if side == 'after' else +1) if a is None: return None if side == 'after' else n - 1 i = pos[a] if side == 'after': # inner end: grows right last = i while last + 1 < n and live[last + 1] > m.mid: last += 1 return last last = i - 1 # outer: step back newer run while last >= 0 and live[last] > m.mid: last -= 1 return last def render(st): """Per live char -> active (mtype, value) set; LWW per (char, mtype) by mid; 'remove' suppresses. Returns [(id, codepoint, frozenset)].""" birth = st.birth_order() live = [c for c in birth if c not in st.deleted] pos = {c: i for i, c in enumerate(live)} bpos = {c: i for i, c in enumerate(birth)} best = {c: {} for c in live} for m in st.marks.values(): first = _start_index(st, m, live, pos, birth, bpos) last = _end_index(st, m, live, pos, birth, bpos) if first is None or last is None or first > last: continue for c in live[first:last + 1]: cur = best[c].get(m.mtype) if cur is None or m.mid > cur[0]: best[c][m.mtype] = (m.mid, m.op, m.value) return [(c, st.codept[c], frozenset((mt, v) for mt, (_i, op, v) in best[c].items() if op == 'add')) for c in live] def rview(st): """Canonical render view for twin comparison.""" return [(c, cp, tuple(sorted(s))) for (c, cp, s) in render(st)] def flags(st, mtype): """[(codepoint, is-mtype)] projection, the paper's rendered view.""" return [(cp, any(t == mtype for (t, _v) in s)) for (_c, cp, s) in render(st)] # ========================================================================= # The abstract compactor. # # Settled cut = a barrier state every replica has (the PBT enforces this by # a full exchange before each cut; the directed cases assert it by # construction). Declared in-flight ops (stragglers minted before the cut, # delivered after) are given as (id, anchor_id) for text and as Mark records # for marks; their anchors are retained and the anchor's child sibling group # is FROZEN (kept deltas verbatim) because the straggler's already-minted # delta must keep its order against its siblings -- renumbering that group # can flip an order (siblings with deltas 2 < 5(in flight) < 9 renumber to # 1,2 and the frozen 5 jumps the ex-9), the same guard as runtime compact.js. # # compact_map: order-preserving relabeling of the kept coordinate tree. # Machine-checks A1 inline: the comparator is preserved on every kept pair # (retained-dead vs live, retained-dead vs retained-dead). # ========================================================================= def compact_map(kept, frozen_parents, a1_pair_cap=45): """kept: {id: coord}. frozen_parents: set of parent coords whose child group must keep its deltas. Returns {id: new_coord}.""" children = {} for c in kept.values(): for k in range(len(c)): children.setdefault(c[:k], set()).add(c[k]) newpfx = {(): ()} for p in sorted(children, key=len): # parents strictly before children ds = sorted(children[p]) renumber = p not in frozen_parents for idx, d in enumerate(ds): nd = (idx + 1) if renumber else d newpfx[p + (d,)] = newpfx[p] + (nd,) cmap = {i: newpfx[c] for i, c in kept.items()} # --- A1, machine-checked: comparator preserved on the kept set. ids = list(kept) if len(ids) <= a1_pair_cap: pairs = [(a, b) for x, a in enumerate(ids) for b in ids[x + 1:]] else: # subsample on big states import random as _r rng = _r.Random(0xA1) pairs = [(rng.choice(ids), rng.choice(ids)) for _ in range(1200)] pairs = [(a, b) for a, b in pairs if a != b] for a, b in pairs: assert ((key(kept[a]) < key(kept[b])) == (key(cmap[a]) < key(cmap[b]))), \ f'A1 violation: order of {a},{b} not preserved' return cmap def rewrite_marks_early(barrier): """H-B: rewrite each DEAD mark boundary to the resolver's own cut-time rehoming target (nearest surviving neighbour on the side's scan direction, side kept; exhausted scan becomes the side's collapse sentinel). Computed once from the barrier, common to all replicas.""" birth = barrier.birth_order() bpos = {c: i for i, c in enumerate(birth)} out = {} for m in barrier.marks.values(): s, e = m.start_id, m.end_id if isinstance(s, int) and not barrier.live(s): t = _scan(birth, barrier.deleted, bpos[s], +1 if m.startSide == 'before' else -1) s = t if t is not None else \ ('sent', 'collapse' if m.startSide == 'before' else 'docstart') if isinstance(e, int) and not barrier.live(e): t = _scan(birth, barrier.deleted, bpos[e], -1 if m.endSide == 'after' else +1) e = t if t is not None else \ ('sent', 'collapse' if m.endSide == 'after' else 'docend') out[m.mid] = replace(m, start_id=s, end_id=e) return out def a3_pairs(barrier, declared_mark_mids, guard=True): """A3: (add, remove) pairs droppable with their retention roots. A pair is (add m, remove r): same mtype, same boundaries AND sides, r.mid > m.mid. GUARDS (all required when guard=True): (i) settledness: every concurrent mark is visible (the barrier is a settled cut) and none is declared in flight for this mtype; (iii) no OTHER same-mtype mark with mid < r.mid (else dropping r resurrects the older mark on any present or future overlap); (iv) the growth-window guard (a finding of this file): no char id strictly between m.mid and r.mid exists among settled ids -- such a char is grabbed by the add's growth but NOT the remove's, so the pair is not render-neutral. Future mints are Lamport-fresh (> r.mid), so checking the settled ids suffices. guard=False drops every syntactic pair (negative control).""" marks = list(barrier.marks.values()) pairs = [] for r in marks: if r.op != 'remove': continue for m in marks: if (m.op == 'add' and m.mtype == r.mtype and m.mid < r.mid and m.start_id == r.start_id and m.end_id == r.end_id and m.startSide == r.startSide and m.endSide == r.endSide): if guard: others = [o for o in marks if o.mtype == r.mtype and o.mid < r.mid and o.mid != m.mid] if others: continue # (iii) if any(o.mtype == r.mtype and o.mid < r.mid for o in declared_mark_mids): continue # (i) in-flight mark if any(m.mid < i < r.mid for i in barrier.shadow): continue # (iv) growth window pairs.append((m, r)) return pairs def build_cut(barrier, attack, declared_ins=(), declared_marks=(), a3_guard=True): """Compute the cut artifacts from the (settled) barrier state. attack: 'A' retention roots | 'A3' A plus guarded pair-drop | 'B' early re-anchoring, no mark retention | 'NONE' no retention, no rewrite (negative control). Returns (settled, keep, cmap, marks_override, cutstats).""" settled = set(barrier.shadow) live = {i for i in settled if i not in barrier.deleted} infl_anchor = {a for (_i, a) in declared_ins if a != 0} for m in declared_marks: for a in (m.start_id, m.end_id): if isinstance(a, int): infl_anchor.add(a) infl_anchor &= settled marks = dict(barrier.marks) dropped_pairs = 0 if attack == 'A3': for (am, rm) in a3_pairs(barrier, declared_marks, guard=a3_guard): if am.mid in marks and rm.mid in marks: del marks[am.mid] del marks[rm.mid] dropped_pairs += 1 if attack == 'B': marks = rewrite_marks_early(barrier) if attack in ('A', 'A3'): mark_anchors = {a for m in marks.values() for a in (m.start_id, m.end_id) if isinstance(a, int)} for m in declared_marks: # straggler mark anchors for a in (m.start_id, m.end_id): if isinstance(a, int): mark_anchors.add(a) mark_anchors &= settled keep = live | mark_anchors | infl_anchor else: # 'B' | 'NONE' keep = live | infl_anchor frozen = set() for (_i, a) in declared_ins: if a == 0: frozen.add(()) elif a in barrier.shadow: frozen.add(barrier.shadow[a]) cmap = compact_map({i: barrier.shadow[i] for i in keep}, frozen) dead = settled - live cutstats = { 'settled': len(settled), 'live': len(live), 'dead': len(dead), 'dropped': len(dead - keep), 'retained_dead': len(keep & dead), 'retained_for_marks': len((keep & dead) - infl_anchor), 'mark_records': len(marks), 'dropped_pairs': dropped_pairs, } return settled, keep, cmap, marks, cutstats def apply_compaction(st, settled, keep, cmap, marks_override): """One replica's state through the cut: drop settled-dead outside keep, re-code kept records, re-derive any unsettled record from its (already re-coded) anchor, restrict the deleted set, install the cut's marks.""" out = PState() for i in sorted(st.shadow): # anchors before children if i in settled and i not in keep: continue if i in cmap: out.shadow[i] = cmap[i] else: # unsettled own record a = st.anchor[i] assert a == 0 or a in out.shadow, \ f'unsettled record {i} lost its anchor {a} (undeclared in-flight)' out.shadow[i] = (() if a == 0 else out.shadow[a]) + (i - a,) out.anchor[i] = st.anchor[i] out.codept[i] = st.codept[i] out.deleted = {i for i in st.deleted if i in out.shadow} out.marks = dict(marks_override) return out # ========================================================================= # SELFCHECK: fidelity of this file's model against the authoritative ones. # S1 reading order vs embed_tree.EmbedTree (fraction geometry): sequential # and randomized DAG (union merge here vs ET.merge there). This also # machine-verifies the task's order description: among children of one # anchor NEWER timestamps sort FIRST, ancestor before subtree. # S2 resolver/render vs peritext_read_model.py on its own Ex1/3/5/7/8 + # gravity + trilemma histories (expected literals are PRM's hand-derived # ones, restated here) AND on 200 randomized histories, implementation # vs implementation. # ========================================================================= def selfcheck(): import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import embed_tree as ET import peritext_read_model as PRM from random import Random msgs = [] ok = True # --- S1a directed: ins 1@0, 2@0, 3@1. Children of root: 2 newer than 1 # so 2 first; 3 is 1's child so after 1. Hand-derived order: [2, 1, 3]. E = ET.EmbedTree() s = E.init() for (i, a) in ((1, 0), (2, 0), (3, 1)): s = E.apply(s, ('ins', i, a)) mine = PState() for (i, a) in ((1, 0), (2, 0), (3, 1)): mine.ins(i, chr(96 + i), a) ok1 = E.read(s) == [2, 1, 3] == mine.birth_order() ok &= ok1 msgs.append(('S1a newest-first order (hand: [2,1,3])', ok1, f'ET={E.read(s)} mine={mine.birth_order()}')) # --- S1b randomized sequential + delete: my live order vs ET refold. bad = 0 for e in range(120): rng = Random(1000 + e) E2, s2, m2, ids = ET.EmbedTree(), ET.EmbedTree().init(), PState(), [] nid = 1 for _ in range(rng.randint(3, 14)): if ids and rng.random() < 0.3: x = rng.choice([i for i in ids if m2.live(i)] or ids) s2 = E2.apply(s2, ('del', x)) m2.delete(x) else: a = rng.choice([0] + [i for i in ids if m2.live(i)]) s2 = E2.apply(s2, ('ins', nid, a)) m2.ins(nid, '?', a) ids.append(nid) nid += 1 if E2.read(s2) != m2.live_order(): bad += 1 ok &= bad == 0 msgs.append(('S1b 120 randomized histories vs embed_tree refold', bad == 0, f'{bad} mismatches')) # --- S1c randomized merge: my union merge vs ET.merge (3-way, empty LCA). bad = 0 for e in range(80): rng = Random(5000 + e) E3 = ET.EmbedTree() branches = [[E3.init(), PState()], [E3.init(), PState()]] nid = 1 for br in branches: for _ in range(rng.randint(1, 6)): a = rng.choice([0] + list(br[1].shadow)) br[0] = E3.apply(br[0], ('ins', nid, a)) br[1].ins(nid, '?', a) nid += 1 sM = E3.merge(E3.init(), E3.copy(branches[0][0]), E3.copy(branches[1][0])) mM = merge(branches[0][1], branches[1][1]) if E3.read(sM) != mM.birth_order(): bad += 1 ok &= bad == 0 msgs.append(('S1c 80 randomized 2-branch merges vs embed_tree.merge', bad == 0, f'{bad} mismatches')) # --- S2a: PRM's own directed histories, expected literals restated from # peritext_read_model.py (hand-derived there). Build both models with the # same ids and compare BOTH to the literal AND to each other. def both(recs, marks_spec, dels_mid=(), post=(), dels_post=()): d = PRM.Doc() mm = PState() for (i, cp, a) in recs: d.ins(i, cp, a) mm.ins(i, cp, a) prm_marks = [PRM.make_mark(d, *spec[:5], startSide=spec[5], endSide=spec[6], op=spec[7]) for spec in marks_spec] for (mid, mt, val, s0, e0, ss, es, op) in marks_spec: mm.add_mark(Mark(mid, mt, op, s0, e0, ss, es, val)) for x in dels_mid: d.delete(x) mm.delete(x) for (i, cp, a) in post: d.ins(i, cp, a) mm.ins(i, cp, a) for x in dels_post: d.delete(x) mm.delete(x) return d, prm_marks, mm cases = [] # Ex1: bold [a,c], insert b within -> all bold. d, pm, mm = both([(1, 'a', 0), (2, 'c', 1)], [(3, 'bold', None, 1, 2, 'before', 'after', 'add')], post=[(4, 'b', 1)]) cases.append(('Ex1', d, pm, mm, 'bold', [('a', True), ('b', True), ('c', True)])) # Ex3: delete whole span, reinsert -> fresh text plain. d, pm, mm = both([(1, 'a', 0), (2, 'b', 1), (3, 'c', 2)], [(10, 'bold', None, 1, 3, 'before', 'after', 'add')], dels_mid=(1, 2, 3), post=[(4, 'd', 0)]) cases.append(('Ex3', d, pm, mm, 'bold', [('d', False)])) # Ex7: typing at bold end extends (insert newer than mark). d, pm, mm = both([(1, 'a', 0), (2, 'b', 1)], [(3, 'bold', None, 1, 2, 'before', 'after', 'add')], post=[(4, 'x', 2)]) cases.append(('Ex7', d, pm, mm, 'bold', [('a', True), ('b', True), ('x', True)])) # Ex8 link half: endSide=before does not extend. d, pm, mm = both([(1, 'a', 0), (2, 'b', 1), (3, 'z', 2)], [(4, 'link', 'u', 1, 3, 'before', 'before', 'add')], post=[(5, 'x', 2)]) cases.append(('Ex8', d, pm, mm, 'link', [('a', True), ('b', True), ('x', False), ('z', False)])) # Trilemma re-span: delete separating C, newer D grabbed by growth. d, pm, mm = both([(1, 'A', 0), (2, 'B', 1), (3, 'C', 2)], [(4, 'bold', None, 1, 2, 'before', 'after', 'add')], post=[(5, 'D', 3)], dels_post=(3,)) cases.append(('Trilemma', d, pm, mm, 'bold', [('A', True), ('B', True), ('D', True)])) for name, d, pm, mm, mt, expected in cases: prm = PRM.render_flags(d, pm, PRM.DocumentOrderResolver(), mt) mineF = flags(mm, mt) good = prm == expected == mineF ok &= good msgs.append((f'S2a {name} (hand literal)', good, f'prm={prm} mine={mineF} expected={expected}')) # --- S2b: 200 randomized histories, PRM render vs mine, exact. bad = 0 first = None for e in range(200): rng = Random(9000 + e) d = PRM.Doc() mm = PState() pm = [] nid = 1 for cp in ('p', 'q'): d.ins(nid, cp, nid - 1) mm.ins(nid, cp, nid - 1) nid += 1 for _ in range(rng.randint(4, 16)): live = [c for c in mm.live_order()] r = rng.random() if r < 0.40 or not live: a = rng.choice([0] + live) d.ins(nid, chr(97 + nid % 26), a) mm.ins(nid, chr(97 + nid % 26), a) nid += 1 elif r < 0.60 and live: x = rng.choice(live) d.delete(x) mm.delete(x) else: birth = mm.birth_order() i = rng.randrange(len(birth)) j = rng.randrange(i, len(birth)) mt = rng.choice(['bold', 'italic']) op = 'add' if rng.random() < 0.7 else 'remove' ss = rng.choice(['before', 'after']) es = rng.choice(['after', 'before']) pm.append(PRM.make_mark(d, nid, mt, None, birth[i], birth[j], ss, es, op=op)) mm.add_mark(Mark(nid, mt, op, birth[i], birth[j], ss, es, None)) nid += 1 prm = [(c, cp, frozenset(s)) for (c, cp, s) in PRM.render(d, pm, PRM.DocumentOrderResolver())] mineR = [(c, cp, s) for (c, cp, s) in render(mm)] if prm != mineR: bad += 1 if first is None: first = (e, prm, mineR) ok &= bad == 0 msgs.append(('S2b 200 randomized histories, PRM render vs mine', bad == 0, f'{bad} mismatches' + (f' first={first}' if first else ''))) return ok, msgs # ========================================================================= # Directed cases. Geometry D1 (the H-B countermodel family): # ins R=1 at root, L=2 at root (concurrent mints; L newer, reads first), # d=3 child of L. Reading order: L d R. Mark mid 5. Delete d. Cut. # Continuations: # (a) beyond-cut ins n=10 anchored at L (n newer than the mark) # (b) in-flight straggler ins n=4 anchored at L (n OLDER than the mark), # declared at the cut, delivered after compaction # (c) in-flight straggler ins n=4 anchored at d (lands right of d) # Hand-derived expected covered sets (worked in whiteboard/marks-gc-note.md; # twin birth orders: (a),(b) L n d R; (c) L d n R): # combo mark (a) twin/early (b) twin : early (c) twin : early # start-inner [s=d before, e=R after] {R} = {R} {R} : {R} {n,R} : {R} DIVERGES (c) # start-outer [s=d after, e=R after] {R} = {R} {R} : {n,R} {n,R} : {n,R} DIVERGES (b) # end-inner [s=L before, e=d after] {L,n} = {L,n} {L,n} : {L} {L} : {L} DIVERGES (b) # end-outer [s=L before, e=d before] {L} = {L} {L,n} : {L,n} {L} : {L,n} DIVERGES (c) # Finding (a): with a beyond-cut NEWER insert the growth/skip rules # compensate and ALL four combos agree; the task's directed candidate as # narrated does not fire there. H-A must equal the twin in every cell. # ========================================================================= D1_COMBOS = { 'start-inner': dict(start_id=3, end_id=1, startSide='before', endSide='after'), 'start-outer': dict(start_id=3, end_id=1, startSide='after', endSide='after'), 'end-inner': dict(start_id=2, end_id=3, startSide='before', endSide='after'), 'end-outer': dict(start_id=2, end_id=3, startSide='before', endSide='before'), } D1_EXPECT = { # combo -> {variant: (twin covered cps, subjectB covered cps)} 'start-inner': {'a': ('R', 'R'), 'b': ('R', 'R'), 'c': ('nR', 'R')}, 'start-outer': {'a': ('R', 'R'), 'b': ('R', 'nR'), 'c': ('nR', 'nR')}, 'end-inner': {'a': ('Ln', 'Ln'), 'b': ('Ln', 'L'), 'c': ('L', 'L')}, 'end-outer': {'a': ('L', 'L'), 'b': ('Ln', 'Ln'), 'c': ('L', 'Ln')}, } def _d1_base(combo): st = PState() st.ins(1, 'R', 0) st.ins(2, 'L', 0) st.ins(3, 'd', 2) st.add_mark(Mark(5, 'bold', 'add', value=None, **D1_COMBOS[combo])) st.delete(3) return st def _bold_cps(st): return ''.join(cp for cp, f in flags(st, 'bold') if f) def d1_family(): """Runs every (combo, variant) cell for attacks A and B against the twin. Returns list of (name, ok, detail).""" out = [] for combo in D1_COMBOS: for variant in ('a', 'b', 'c'): barrier = _d1_base(combo) if variant == 'a': decl_ins = [] pending = [(10, 'n', 2)] elif variant == 'b': decl_ins = [(4, 2)] pending = [(4, 'n', 2)] else: decl_ins = [(4, 3)] pending = [(4, 'n', 3)] exp_twin, exp_early = D1_EXPECT[combo][variant] twin = barrier.copy() for (i, cp, a) in pending: twin.ins(i, cp, a) got_twin = _bold_cps(twin) row_ok = got_twin == exp_twin det = [f'twin={got_twin!r} (hand: {exp_twin!r})'] for attack, exp_subj in (('A', exp_twin), ('B', exp_early)): settled, keep, cmap, mo, _cs = build_cut( barrier, attack, declared_ins=decl_ins) subj = apply_compaction(barrier, settled, keep, cmap, mo) for (i, cp, a) in pending: subj.ins(i, cp, a) got = _bold_cps(subj) row_ok &= got == exp_subj if attack == 'A': row_ok &= rview(subj) == rview(twin) det.append(f'{attack}={got!r} (hand: {exp_subj!r})') tag = 'agree' if exp_twin == exp_early else 'H-B DIVERGES' out.append((f'D1 {combo} variant {variant} [{tag}]', row_ok, ' '.join(det))) return out def d1_double_death(): """D1(d): H-B divergence with NO straggler. Chain L=1 root, d=2 child of L; mark [s=L before, e=d after] mid 5; delete d; cut; beyond the cut: ins n=10 child of L, then delete L. Twin (hand): birth L n d, live [n]; start L dead scans right -> n (index 0); end d dead scans left -> n (index 0); covered {n}: n BOLD. H-B (hand): end rewritten at the cut to L (nearest survivor left of d). At read: start L dead -> n; end L dead, endSide=after scans LEFT from L, nothing lives left of L -> collapse -> covered {}: n PLAIN. DIVERGES. H-A (hand): d retained, twin-identical: n BOLD.""" barrier = PState() barrier.ins(1, 'L', 0) barrier.ins(2, 'd', 1) barrier.add_mark(Mark(5, 'bold', 'add', 1, 2, 'before', 'after')) barrier.delete(2) twin = barrier.copy() twin.ins(10, 'n', 1) twin.delete(1) ok = _bold_cps(twin) == 'n' det = [f'twin={_bold_cps(twin)!r} (hand: n)'] for attack, exp in (('A', 'n'), ('B', '')): settled, keep, cmap, mo, _cs = build_cut(barrier, attack) subj = apply_compaction(barrier, settled, keep, cmap, mo) subj.ins(10, 'n', 1) subj.delete(1) got = _bold_cps(subj) ok &= got == exp det.append(f'{attack}={got!r} (hand: {exp!r})') return ('D1(d) double-death, pure beyond-cut [H-B DIVERGES]', ok, ' '.join(det)) def d2_respan(): """D2 respan corner (doc_delete_can_respan analog, across the cut) + A2. Chars A=1 root, B=2 child of A, C=3 child of B; mark [A..C] inner/inner mid 5; delete interior B; cut (B settled dead, NOT a boundary: dropped by every attack); beyond the cut ins n=10 child of A (inside the span). Twin (hand): birth A n B C, live A n C; start=A live idx 0; end=C live, growth: nothing after C; covered {A,n,C} -- insert within span formatted. Subjects (hand): B dropped; same live order; same covered {A,n,C}. A2: the dropped interior dead is invisible to the resolver. Companion (whole-span death): delete A,B,C before the cut, reinsert n=10 at root after it. Twin (hand): start A dead scans right, all dead, exhausted -> collapse; covered {}: n PLAIN (Ex3 across the cut). H-A retains A,C (boundaries) and reads {} too; H-B rewrote both boundaries to collapse sentinels: {}.""" out = [] barrier = PState() barrier.ins(1, 'A', 0) barrier.ins(2, 'B', 1) barrier.ins(3, 'C', 2) barrier.add_mark(Mark(5, 'bold', 'add', 1, 3, 'before', 'after')) barrier.delete(2) twin = barrier.copy() twin.ins(10, 'n', 1) ok = _bold_cps(twin) == 'AnC' det = [f'twin={_bold_cps(twin)!r} (hand: AnC)'] for attack in ('A', 'B'): settled, keep, cmap, mo, cs = build_cut(barrier, attack) subj = apply_compaction(barrier, settled, keep, cmap, mo) assert 2 not in subj.shadow, 'interior dead B must be dropped' subj.ins(10, 'n', 1) got = _bold_cps(subj) ok &= got == 'AnC' and rview(subj) == rview(twin) det.append(f'{attack}={got!r}') out.append(('D2 respan across cut + A2 interior drop', ok, ' '.join(det))) barrier2 = PState() barrier2.ins(1, 'A', 0) barrier2.ins(2, 'B', 1) barrier2.ins(3, 'C', 2) barrier2.add_mark(Mark(5, 'bold', 'add', 1, 3, 'before', 'after')) for x in (1, 2, 3): barrier2.delete(x) twin2 = barrier2.copy() twin2.ins(10, 'n', 0) ok2 = _bold_cps(twin2) == '' det2 = [f'twin={_bold_cps(twin2)!r} (hand: empty, Ex3)'] for attack in ('A', 'B'): settled, keep, cmap, mo, _cs = build_cut(barrier2, attack) subj = apply_compaction(barrier2, settled, keep, cmap, mo) subj.ins(10, 'n', 0) got = _bold_cps(subj) ok2 &= got == '' and rview(subj) == rview(twin2) det2.append(f'{attack}={got!r}') out.append(('D2 whole-span death, fresh text stays plain', ok2, ' '.join(det2))) return out def d3_same_dead_run(): """D3: two marks with boundaries at two DIFFERENT dead ids inside the SAME dead run; a straggler gap insert lands between them. Chars A=1 root, d1=2 child of A, d2=3 child of d1, B=4 child of d2. Marks m1 [A..d1] mid 10, m2 [A..d2] mid 11, both inner/inner. Delete d1, d2. Straggler g=5 child of d1 (declared): children of d1 newest first, so g(delta 3) sorts before d2(delta 1): birth A d1 g d2 B. Twin (hand): live A g B. m1 end=d1 dead scans left -> A, growth: g=5>10 false -> covered {A}. m2 end=d2 dead scans left -> g (live), growth: B=4>11 false -> covered {A,g}. So g carries m2 only. H-A (hand): d1, d2 retained with their relative order (A1); reads twin-identical, g carries m2 only. H-B (hand): at the cut BOTH ends rewrite to A (nearest survivor left of either dead id: the run's left edge). After g arrives: m1 covered {A}, m2 covered {A}: g LOSES m2. The two boundaries' relative order inside the dead run is erased. DIVERGES.""" barrier = PState() barrier.ins(1, 'A', 0) barrier.ins(2, 'x', 1) barrier.ins(3, 'y', 2) barrier.ins(4, 'B', 3) barrier.add_mark(Mark(10, 'bold', 'add', 1, 2, 'before', 'after')) barrier.add_mark(Mark(11, 'ital', 'add', 1, 3, 'before', 'after')) barrier.delete(2) barrier.delete(3) decl = [(5, 2)] twin = barrier.copy() twin.ins(5, 'g', 2) tw_b, tw_i = _bold_cps(twin), ''.join( cp for cp, f in flags(twin, 'ital') if f) ok = (tw_b, tw_i) == ('A', 'Ag') det = [f'twin bold={tw_b!r} ital={tw_i!r} (hand: A / Ag)'] for attack, exp in (('A', ('A', 'Ag')), ('B', ('A', 'A'))): settled, keep, cmap, mo, _cs = build_cut(barrier, attack, declared_ins=decl) subj = apply_compaction(barrier, settled, keep, cmap, mo) subj.ins(5, 'g', 2) got = (_bold_cps(subj), ''.join(cp for cp, f in flags(subj, 'ital') if f)) ok &= got == exp if attack == 'A': ok &= rview(subj) == rview(twin) det.append(f'{attack} bold={got[0]!r} ital={got[1]!r} (hand: {exp[0]}/{exp[1]})') return ('D3 two boundaries in one dead run + gap straggler [H-B DIVERGES]', ok, ' '.join(det)) def d4_growth_across_cut(): """D4: growth end-side across the cut, no dead anchors at all: both attacks must be invisible. Chars a=1 root, b=2 child of a; mark [a..b] inner/inner mid 3; cut; beyond: ins x=10 child of b (newer than mark, grabbed) -- hand: covered {a,b,x}. Straggler companion: y=4 declared, child of b, mark mid 9 variant: y OLDER than the mark, NOT grabbed -- hand: covered {a,b}, y plain (Ex7's older-insert half across the cut).""" out = [] barrier = PState() barrier.ins(1, 'a', 0) barrier.ins(2, 'b', 1) barrier.add_mark(Mark(3, 'bold', 'add', 1, 2, 'before', 'after')) twin = barrier.copy() twin.ins(10, 'x', 2) ok = _bold_cps(twin) == 'abx' det = [f'twin={_bold_cps(twin)!r} (hand: abx)'] for attack in ('A', 'B'): settled, keep, cmap, mo, _cs = build_cut(barrier, attack) subj = apply_compaction(barrier, settled, keep, cmap, mo) subj.ins(10, 'x', 2) ok &= _bold_cps(subj) == 'abx' and rview(subj) == rview(twin) det.append(f'{attack}={_bold_cps(subj)!r}') out.append(('D4 growth across cut (newer insert grabbed)', ok, ' '.join(det))) barrier2 = PState() barrier2.ins(1, 'a', 0) barrier2.ins(2, 'b', 1) barrier2.add_mark(Mark(9, 'bold', 'add', 1, 2, 'before', 'after')) decl = [(4, 2)] twin2 = barrier2.copy() twin2.ins(4, 'y', 2) ok2 = _bold_cps(twin2) == 'ab' det2 = [f'twin={_bold_cps(twin2)!r} (hand: ab, y not grabbed)'] for attack in ('A', 'B'): settled, keep, cmap, mo, _cs = build_cut(barrier2, attack, declared_ins=decl) subj = apply_compaction(barrier2, settled, keep, cmap, mo) subj.ins(4, 'y', 2) ok2 &= _bold_cps(subj) == 'ab' and rview(subj) == rview(twin2) det2.append(f'{attack}={_bold_cps(subj)!r}') out.append(('D4 straggler older than mark not grabbed', ok2, ' '.join(det2))) return out def d6_no_retention_negative(): """D6 NEGATIVE (the refusal's justification): dropping dead mark-boundary anchors WITHOUT retention or rewrite must flip a read. D1 geometry, end-inner mark [s=L before, e=d after] mid 5, d dead, cut, NO continuation needed. Twin (hand): end=d dead scans left -> L; covered {L}: L BOLD. NONE-subject (hand): d has no record; the resolver cannot place the scan and degrades to the side's collapse branch (endSide=after -> collapse): covered {}: L PLAIN. FLIP. PASS companion: same compactor, mark with LIVE anchors [L..R], reads twin-identical -- the refusal is specifically about dead mark anchors.""" barrier = _d1_base('end-inner') twin = barrier.copy() ok = _bold_cps(twin) == 'L' settled, keep, cmap, mo, _cs = build_cut(barrier, 'NONE') subj = apply_compaction(barrier, settled, keep, cmap, mo) flip = _bold_cps(subj) == '' ok &= flip det = f'twin={_bold_cps(twin)!r} (hand: L) NONE={_bold_cps(subj)!r} (hand: empty, FLIP)' barrier2 = PState() barrier2.ins(1, 'R', 0) barrier2.ins(2, 'L', 0) barrier2.ins(3, 'd', 2) barrier2.add_mark(Mark(5, 'bold', 'add', 2, 1, 'before', 'after')) barrier2.delete(3) twin2 = barrier2.copy() settled, keep, cmap, mo, _cs = build_cut(barrier2, 'NONE') subj2 = apply_compaction(barrier2, settled, keep, cmap, mo) ok &= rview(subj2) == rview(twin2) and _bold_cps(subj2) == 'LR' det += f' | live-anchor companion: NONE={_bold_cps(subj2)!r} == twin ' \ f'{_bold_cps(twin2)!r} (hand: LR)' return ('D6 no-retention drop FLIPS a read (refusal justified)', ok, det) def d7_a3_family(): """D7: the A3 guarded pair-drop. (gamma) guarded positive: A=1 root, B=2 child of A; add bold [A..B] mid 3; remove bold [A..B] mid 4, same sides. Window (3,4) empty, no other bold marks. Drop pair + roots at the cut. Beyond: ins x=10 child of B. Twin (hand): add covers {A,B,x} (10>3 grabbed), remove covers {A,B,x} (10>4), remove wins everywhere: ALL PLAIN. Subject: no marks: ALL PLAIN. Identical. (beta) growth-window corner (a FINDING: the settled-removal guard alone is insufficient): add mid 3, char w=4 child of B (live, settled), remove mid 5. Unguarded drop flips w: twin has add covering {A,B,w} (4>3 grabbed) but remove covering only {A,B} (4>5 false), so w is BOLD; dropped subject: w PLAIN. FLIP. The (iv) window guard refuses this drop; guarded subject reads twin-identical. (alpha) settledness corner: add mid 3, remove mid 6, an in-flight straggler add mid 4 over the same span (concurrent, undelivered). Dropping the pair while ignoring the declaration: after delivery the straggler add(4) is unopposed: {A,B} BOLD; twin keeps remove(6) winning: PLAIN. FLIP. The (i)/(iii) guard sees the declared mid 4 and refuses; guarded subject reads twin-identical.""" out = [] # (gamma) -- with a DEAD anchor B so a retention root is genuinely freed. # Twin (hand): birth A x B, live A x; add(3) end=B dead scans left -> x, # covered {A,x}; remove(4) likewise {A,x}; remove wins: ALL PLAIN. # A3 subject: pair + root B dropped, no marks: ALL PLAIN. Identical. # Contrast: plain attack A retains B (root freed only by the pair-drop). barrier = PState() barrier.ins(1, 'A', 0) barrier.ins(2, 'B', 1) barrier.add_mark(Mark(3, 'bold', 'add', 1, 2, 'before', 'after')) barrier.add_mark(Mark(4, 'bold', 'remove', 1, 2, 'before', 'after')) barrier.delete(2) twin = barrier.copy() twin.ins(10, 'x', 1) _s, keepA, _c, _m, _cs = build_cut(barrier, 'A') settled, keep, cmap, mo, cs = build_cut(barrier, 'A3') subj = apply_compaction(barrier, settled, keep, cmap, mo) subj.ins(10, 'x', 1) ok = (cs['dropped_pairs'] == 1 and len(subj.marks) == 0 and 2 not in subj.shadow and 2 in keepA and _bold_cps(twin) == '' and rview(subj) == rview(twin)) out.append(('D7 A3 guarded pair-drop frees the root, reads identical', ok, f'pairs dropped={cs["dropped_pairs"]} root B: retained by A, ' f'dropped by A3; twin={_bold_cps(twin)!r} ' f'subj={_bold_cps(subj)!r} (hand: both empty)')) # (beta) barrier = PState() barrier.ins(1, 'A', 0) barrier.ins(2, 'B', 1) barrier.add_mark(Mark(3, 'bold', 'add', 1, 2, 'before', 'after')) barrier.ins(4, 'w', 2) barrier.add_mark(Mark(5, 'bold', 'remove', 1, 2, 'before', 'after')) twin = barrier.copy() ok_twin = _bold_cps(twin) == 'w' settled, keep, cmap, mo, cs_u = build_cut(barrier, 'A3', a3_guard=False) subj_u = apply_compaction(barrier, settled, keep, cmap, mo) flip = _bold_cps(subj_u) == '' and cs_u['dropped_pairs'] == 1 settled, keep, cmap, mo, cs_g = build_cut(barrier, 'A3', a3_guard=True) subj_g = apply_compaction(barrier, settled, keep, cmap, mo) guarded = cs_g['dropped_pairs'] == 0 and rview(subj_g) == rview(twin) out.append(('D7 A3 growth-window corner: unguarded FLIPS, guard refuses', ok_twin and flip and guarded, f'twin={_bold_cps(twin)!r} (hand: w) unguarded={_bold_cps(subj_u)!r} ' f'(hand: empty, FLIP) guarded drops={cs_g["dropped_pairs"]} (hand: 0)')) # (alpha) barrier = PState() barrier.ins(1, 'A', 0) barrier.ins(2, 'B', 1) barrier.add_mark(Mark(3, 'bold', 'add', 1, 2, 'before', 'after')) barrier.add_mark(Mark(6, 'bold', 'remove', 1, 2, 'before', 'after')) straggler = Mark(4, 'bold', 'add', 1, 2, 'before', 'after') twin = barrier.copy() twin.add_mark(straggler) ok_twin = _bold_cps(twin) == '' settled, keep, cmap, mo, cs_u = build_cut(barrier, 'A3', declared_marks=[], a3_guard=True) subj_u = apply_compaction(barrier, settled, keep, cmap, mo) subj_u.add_mark(straggler) flip = _bold_cps(subj_u) == 'AB' and cs_u['dropped_pairs'] == 1 settled, keep, cmap, mo, cs_g = build_cut(barrier, 'A3', declared_marks=[straggler], a3_guard=True) subj_g = apply_compaction(barrier, settled, keep, cmap, mo) subj_g.add_mark(straggler) guarded = cs_g['dropped_pairs'] == 0 and rview(subj_g) == rview(twin) out.append(('D7 A3 undeclared-straggler drop FLIPS, declared guard refuses', ok_twin and flip and guarded, f'twin={_bold_cps(twin)!r} (hand: empty) undeclared-drop=' f'{_bold_cps(subj_u)!r} (hand: AB, FLIP) declared drops=' f'{cs_g["dropped_pairs"]} (hand: 0)')) return out # ========================================================================= # Randomized DAG PBT. Multi-replica histories (text + mark ops, dead-anchor # marks included), random settled cuts (certified by a full exchange), # per-attack compaction on every replica, random continuations + merges + # straggler deliveries, multi-epoch (1..3 successive cuts). The subject is # compared to the never-compacted control twin at EVERY version, on two # levels: text (subject birth order must be the control birth order filtered # to the subject's records -- fatal for every attack) and marks (rview # equality -- fatal for A/A3, counted as the refutation/flip evidence for # B/NONE). # ========================================================================= class TrialFail(Exception): def __init__(self, info): super().__init__(str(info)) self.info = info def pbt_trial(seed, attack, res): from random import Random rng = Random(seed) nrep = rng.randint(2, 3) nepoch = rng.randint(1, 3) g = PState() g.ins(1, 'p', 0) g.ins(2, 'q', 1) nid = [2] def fresh(): nid[0] += 1 return nid[0] subj = [g.copy() for _ in range(nrep)] ctrl = [g.copy() for _ in range(nrep)] log = [] trial_div = [None] def compare(i, where): sb = subj[i].birth_order() cb = [x for x in ctrl[i].birth_order() if x in subj[i].shadow] if sb != cb: raise TrialFail({'kind': 'TEXT-ORDER', 'seed': seed, 'attack': attack, 'where': where, 'replica': i, 'subject': sb, 'expected': cb, 'log': log[-40:]}) rs, rc = rview(subj[i]), rview(ctrl[i]) if rs != rc: info = {'kind': 'READ-DIVERGENCE', 'seed': seed, 'attack': attack, 'where': where, 'replica': i, 'subject': rs, 'expected': rc, 'log': log[-40:]} if attack in ('A', 'A3'): raise TrialFail(info) if trial_div[0] is None: trial_div[0] = info def rand_op(i): r = rng.random() live = subj[i].live_order() if r < 0.45 or not live: x = fresh() a = rng.choice([0] + live) subj[i].ins(x, chr(97 + x % 26), a) ctrl[i].ins(x, chr(97 + x % 26), a) log.append((i, 'ins', x, a)) elif r < 0.65: x = rng.choice(live) subj[i].delete(x) ctrl[i].delete(x) log.append((i, 'del', x)) else: op = 'add' if rng.random() < 0.7 else 'remove' # a remove often reuses a visible add's exact boundaries+sides # (the pair shape A3 needs); the template must be identical in # subject and control (i.e., not rewritten away by H-B). templates = [t for t in ctrl[i].marks.values() if t.op == 'add' and subj[i].marks.get(t.mid) == t and isinstance(t.start_id, int) and t.start_id in subj[i].shadow and t.end_id in subj[i].shadow] if op == 'remove' and templates and rng.random() < 0.7: t = rng.choice(templates) m = Mark(fresh(), t.mtype, 'remove', t.start_id, t.end_id, t.startSide, t.endSide) else: birth = subj[i].birth_order() # dead anchors allowed i0 = rng.randrange(len(birth)) j0 = rng.randrange(i0, len(birth)) m = Mark(fresh(), rng.choice(['bold', 'ital']), op, birth[i0], birth[j0], rng.choice(['before', 'after']), rng.choice(['after', 'before'])) subj[i].add_mark(m) ctrl[i].add_mark(m) log.append((i, m.op, m.mid, m.mtype, m.start_id, m.end_id, m.startSide, m.endSide)) def maybe_merge(i): if nrep > 1 and rng.random() < 0.30: j = (i + 1 + rng.randrange(nrep - 1)) % nrep subj[j] = merge(subj[j], subj[i]) ctrl[j] = merge(ctrl[j], ctrl[i]) log.append((j, 'merge-from', i)) compare(j, 'post-merge') for epoch in range(nepoch): for _ in range(rng.randint(4, 10)): i = rng.randrange(nrep) rand_op(i) compare(i, f'epoch{epoch}-local') maybe_merge(i) # --- mint stragglers (declared in-flight, applied NOWHERE yet) pend_ins, pend_marks = [], [] if rng.random() < 0.55: r = rng.randrange(nrep) for _ in range(rng.randint(1, 2)): live = subj[r].live_order() a = rng.choice([0] + live) pend_ins.append((fresh(), '*', a)) if attack in ('A', 'A3') and rng.random() < 0.6: birth = subj[r].birth_order() i0 = rng.randrange(len(birth)) j0 = rng.randrange(i0, len(birth)) pend_marks.append(Mark(fresh(), rng.choice(['bold', 'ital']), 'add' if rng.random() < 0.7 else 'remove', birth[i0], birth[j0], rng.choice(['before', 'after']), rng.choice(['after', 'before']))) res['straggler_cuts'] += 1 log.append(('cut', epoch, [x[0] for x in pend_ins], [m.mid for m in pend_marks])) # --- barrier: full exchange = the settled-cut certificate Bs, Bc = subj[0], ctrl[0] for i in range(1, nrep): Bs = merge(Bs, subj[i]) Bc = merge(Bc, ctrl[i]) ctrl = [Bc.copy() for _ in range(nrep)] # --- compact every replica at the common cut decl_ins = [(x, a) for (x, _cp, a) in pend_ins] settled, keep, cmap, mo, cs = build_cut(Bs, attack, decl_ins, pend_marks) cst = apply_compaction(Bs, settled, keep, cmap, mo) subj = [cst.copy() for _ in range(nrep)] res['cuts'] += 1 res['retained_for_marks'].append(cs['retained_for_marks']) res['mark_records'].append(cs['mark_records']) res['dropped'] += cs['dropped'] res['dead'] += cs['dead'] res['pairs_dropped'] += cs['dropped_pairs'] if cs['mark_records'] > 0: assert cs['retained_for_marks'] <= 2 * cs['mark_records'], \ 'cost claim violated: retained > 2 per mark record' for i in range(nrep): compare(i, f'epoch{epoch}-post-compaction') # --- continuation: random ops + merges + straggler deliveries deliveries = [(('ins',) + op, i) for op in pend_ins for i in range(nrep)] deliveries += [(('mark', m), i) for m in pend_marks for i in range(nrep)] rng.shuffle(deliveries) for _ in range(rng.randint(4, 10) + len(deliveries)): if deliveries and rng.random() < 0.5: d, i = deliveries.pop() if d[0] == 'ins': _t, x, cp, a = d if x not in subj[i].shadow: # may have come by merge subj[i].ins(x, cp, a) if x not in ctrl[i].shadow: ctrl[i].ins(x, cp, a) log.append((i, 'deliver-ins', x, a)) else: subj[i].add_mark(d[1]) ctrl[i].add_mark(d[1]) log.append((i, 'deliver-mark', d[1].mid)) compare(i, f'epoch{epoch}-delivery') else: i = rng.randrange(nrep) rand_op(i) compare(i, f'epoch{epoch}-continuation') maybe_merge(i) while deliveries: # force the rest d, i = deliveries.pop() if d[0] == 'ins': _t, x, cp, a = d if x not in subj[i].shadow: subj[i].ins(x, cp, a) if x not in ctrl[i].shadow: ctrl[i].ins(x, cp, a) else: subj[i].add_mark(d[1]) ctrl[i].add_mark(d[1]) compare(i, f'epoch{epoch}-late-delivery') # --- end of trial: full exchange; subject replicas must agree among # themselves (the weaker H-B claim; trivial for A where twin-equality # already holds). Bs, Bc = subj[0], ctrl[0] for i in range(1, nrep): Bs = merge(Bs, subj[i]) Bc = merge(Bc, ctrl[i]) views = {tuple(rview(Bs))} for i in range(nrep): m2 = merge(subj[i], Bs) views.add(tuple(rview(m2))) if len(views) != 1: raise TrialFail({'kind': 'SUBJECT-CONVERGENCE', 'seed': seed, 'attack': attack, 'views': len(views), 'log': log[-40:]}) if trial_div[0] is not None: res['div_trials'] += 1 if res['first_div'] is None: res['first_div'] = trial_div[0] res['trials'] += 1 def run_pbt(attack, n, seed0=0): res = {'trials': 0, 'cuts': 0, 'straggler_cuts': 0, 'div_trials': 0, 'first_div': None, 'fatal': None, 'retained_for_marks': [], 'mark_records': [], 'dropped': 0, 'dead': 0, 'pairs_dropped': 0} for e in range(n): try: pbt_trial(seed0 * 1000003 + e, attack, res) except TrialFail as tf: res['fatal'] = tf.info break except AssertionError as ae: res['fatal'] = {'kind': 'ASSERT', 'seed': seed0 * 1000003 + e, 'attack': attack, 'msg': str(ae)} break return res def _fmt_div(info, limit=3): if info is None: return 'none' subj, exp = info.get('subject'), info.get('expected') diff = '' if isinstance(subj, list) and isinstance(exp, list): for a, b in zip(subj, exp): if a != b: diff = f' first-diff: subject {a} vs twin {b}' break return (f"seed={info['seed']} where={info['where']} replica={info['replica']}" f"{diff} log-tail={info['log'][-limit:]}") def main(): fast = '--fast' in sys.argv[1:] W = 74 print('=' * W) print('Peritext MARKS-layer GC: retention roots (H-A) vs early ' 're-anchoring (H-B)') print('=' * W) ok, msgs = selfcheck() print('-- selfcheck: model fidelity (embed_tree + peritext_read_model) --') for name, good, det in msgs: print(f' [{"PASS" if good else "FAIL"}] {name}') if not good: print(f' {det}') allok = ok print('\n-- directed cases (expected values hand-derived in comments) --') directed = [] directed += d1_family() directed.append(d1_double_death()) directed += d2_respan() directed.append(d3_same_dead_run()) directed += d4_growth_across_cut() directed.append(d6_no_retention_negative()) directed += d7_a3_family() for name, good, det in directed: allok &= good print(f' [{"PASS" if good else "FAIL"}] {name}') if not good: print(f' {det}') print('\n-- randomized DAG PBT (subject vs never-compacted twin, every ' 'version) --') nA, nB, n3, nN = (200, 200, 80, 80) if fast else (2000, 2000, 600, 600) verdicts = {} for attack, n in (('A', nA), ('B', nB), ('A3', n3), ('NONE', nN)): res = run_pbt(attack, n) verdicts[attack] = res t, d = res['trials'], res['div_trials'] print(f' attack {attack:4} : {t}/{n} trials, {res["cuts"]} cuts ' f'({res["straggler_cuts"]} with declared stragglers), ' f'{d} trials diverged from the twin') if res['fatal']: print(f' FATAL {res["fatal"]["kind"]}: {_fmt_div(res["fatal"]) if "where" in res["fatal"] else res["fatal"]}') if attack == 'A': rf, mr = res['retained_for_marks'], res['mark_records'] tot_r, tot_m = sum(rf), sum(mr) per = [r / m for r, m in zip(rf, mr) if m > 0] print(f' cost: retained-for-marks total {tot_r} over {tot_m} ' f'mark records across {len(rf)} cuts; ' f'mean {tot_r / max(1, len(rf)):.2f}/cut; ' f'per-mark mean {sum(per) / max(1, len(per)):.3f}, ' f'max {max(per) if per else 0:.3f} (claim bound: 2)') print(f' GC efficacy: {res["dropped"]}/{res["dead"]} settled-dead ' f'records dropped ' f'({100 * res["dropped"] / max(1, res["dead"]):.1f}%)') if attack == 'A3': print(f' guarded pair-drops fired: {res["pairs_dropped"]} ' f'(non-vacuity requires > 0)') if attack in ('B', 'NONE') and res['first_div']: print(f' first divergence: {_fmt_div(res["first_div"])}') a_ok = (verdicts['A']['fatal'] is None and verdicts['A']['div_trials'] == 0 and verdicts['A']['trials'] == nA) b_meta = (verdicts['B']['fatal'] is None and verdicts['B']['div_trials'] > 0) a3_ok = (verdicts['A3']['fatal'] is None and verdicts['A3']['div_trials'] == 0 and verdicts['A3']['trials'] == n3 and verdicts['A3']['pairs_dropped'] > 0) none_flips = (verdicts['NONE']['fatal'] is None and verdicts['NONE']['div_trials'] > 0) allok &= a_ok and b_meta and a3_ok and none_flips print('\n' + '=' * W) print('VERDICTS') print(f' H-A reads-identical : ' f'{"VALIDATED" if a_ok else "REFUTED/FAIL"} ' f'({verdicts["A"]["trials"]}/{nA} trials, 0 divergences required)') print(f' H-B reads-identical : ' f'{"REFUTED (as expected)" if b_meta else "UNEXPECTED"} ' f'({verdicts["B"]["div_trials"]}/{verdicts["B"]["trials"]} trials ' f'diverged; weaker claim [subject convergence] held in all trials)') print(f' A3 guarded pair-drop : ' f'{"VALIDATED" if a3_ok else "FAIL"} ' f'({verdicts["A3"]["pairs_dropped"]} drops fired, 0 divergences)') print(f' NONE no-retention (refusal) : ' f'{"FLIPS as required" if none_flips else "UNEXPECTED"} ' f'({verdicts["NONE"]["div_trials"]}/{verdicts["NONE"]["trials"]} ' f'trials flipped)') print('=' * W) print(f'OVERALL: {"ALL CHECKS PASS" if allok else "FAILURES PRESENT"}') return 0 if allok else 1 if __name__ == '__main__': raise SystemExit(main())