check_partition_adoption.py

#!/usr/bin/env python3
"""Exact certificate model, supplied finite partitions for testing only.
No native timing or input-decomposition theorem is asserted.
"""
from collections import Counter
from copy import deepcopy
from itertools import product, combinations
import json
import random


class Certificate:
    def __init__(self, partitions):
        self.parts = partitions
        self.stage = 0
        self.edges = set()
        self.tags = {}
        self.counts = Counter()
        self.inherited = set()
        self.remaining = 0
        # Oracle state is never read by the update decisions.
        self.credited = set()
        self.obligations = set()

    def cells(self, stage=None):
        p = self.parts[self.stage if stage is None else stage]
        out = {}
        for v, c in enumerate(p):
            out.setdefault(c, set()).add(v)
        return out

    def adopt(self, stage):
        assert stage > self.stage and self.remaining == 0
        oldcells, newcells = self.cells(), self.cells(stage)
        assert all(any(vs <= os for os in oldcells.values()) for vs in newcells.values())
        full = {c for c, vs in oldcells.items() if self.counts[c] == len(vs)}
        partial = {c for c, n in self.counts.items() if n and c not in full}
        self.remaining = sum(self.counts[c] for c in partial)
        self.inherited = {c for c, vs in newcells.items()
                          if any(vs <= oldcells[o] for o in full)}
        self.counts = Counter({c: len(newcells[c]) for c in self.inherited})
        self.stage = stage
        # Separate direct-set oracle for forgotten and inherited incidences.
        self.credited = set().union(*(newcells[c] for c in self.inherited)) if self.inherited else set()
        self.obligations = self.edges - self.credited

    def add(self, v):
        c = self.parts[self.stage][v]
        old = v in self.edges
        if c in self.inherited or self.tags.get(v) == self.stage:
            return
        if old:
            assert self.remaining > 0
            self.remaining -= 1
            self.obligations.remove(v)
        self.edges.add(v)
        self.tags[v] = self.stage
        self.counts[c] += 1
        self.credited.add(v)

    def valid(self):
        return self.remaining == 0 and all(n == len(self.cells()[c])
                                          for c, n in self.counts.items() if n)

    def audit(self):
        cells = self.cells()
        assert self.credited <= self.edges
        assert self.obligations == self.edges - self.credited
        assert self.remaining == len(self.obligations)
        assert +self.counts == Counter(self.parts[self.stage][v] for v in self.credited)
        assert all(n <= len(cells[c]) for c, n in self.counts.items())
        if self.remaining == 0:
            complete = all(not (vs & self.edges) or vs <= self.edges for vs in cells.values())
            assert self.valid() == complete
        if self.valid():
            selected = set().union(*(cells[c] for c, n in self.counts.items() if n)) if self.counts else set()
            assert selected == self.edges


def audit_queries(m, stats):
    m.audit()
    stats['prefix_checks'] += 1
    if m.valid():
        n = len(m.parts[0])
        for bits in product((0, 1), repeat=n):
            src = {v for v, bit in enumerate(bits) if bit}
            got = any(m.counts[m.parts[m.stage][v]] for v in src)
            assert bool(got) == bool(src & m.edges)
            stats['product_queries'] += 1


def main():
    stats = Counter()
    # Exhaustive old subsets and all possible once-per-edge replay subsets.
    for n in range(1, 7):
        parts = [[0] * n, [v % 2 for v in range(n)], list(range(n))]
        for mask in range(1 << n):
            old = {v for v in range(n) if mask >> v & 1}
            m = Certificate(parts)
            for v in sorted(old):
                m.add(v)
            m.adopt(1)
            audit_queries(m, stats)
            expected_debt = 0 if len(old) == n else len(old)
            assert m.remaining == expected_debt
            for replay_mask in range(1 << n):
                q = deepcopy(m)
                replay = {v for v in old if replay_mask >> v & 1}
                for v in sorted(replay):
                    q.add(v)
                    q.add(v)  # duplicate cannot pay twice
                    audit_queries(q, stats)
                assert q.remaining == (0 if len(old) == n else len(old - replay))
                if q.remaining == 0:
                    q.adopt(2)
                    audit_queries(q, stats)
                stats['exhaustive_cases'] += 1
    # Multi-stage streams, fresh edges during outstanding debt and retained versions.
    rng = random.Random(141)
    for _ in range(1800):
        n = rng.randrange(2, 7)
        parts = [[0] * n, [v % 2 for v in range(n)], list(range(n))]
        m, retained = Certificate(parts), []
        for step in range(24):
            if len(retained) < 5:
                retained.append(deepcopy(m))
            if m.remaining == 0 and m.stage < 2 and rng.randrange(4) == 0:
                m.adopt(m.stage + 1)
                stats['adoptions'] += 1
            else:
                m.add(rng.randrange(n))
            audit_queries(m, stats)
        for old in retained:
            audit_queries(old, stats)
            stats['retained_versions'] += 1
        stats['random_streams'] += 1
    # Complete coarse cells transfer, while only the incomplete cell owes work.
    m = Certificate([[0,0,1,1,1],[0,1,2,3,3]])
    for v in [0,1,2]: m.add(v)
    m.adopt(1)
    assert m.inherited == {0,1} and m.remaining == 1
    m.add(0); assert m.remaining == 1
    m.add(3); assert m.remaining == 1  # fresh edge cannot erase forgotten old one
    m.add(2); assert m.remaining == 0 and not m.valid()
    m.add(4); assert m.valid()
    audit_queries(m, stats)
    # Run139: v and x initially agree, y already differs. Target is {v,y}.
    m = Certificate([[0,0,1],[0,1,2],[0,1,2]])
    m.add(0);m.add(2)
    assert not m.valid()
    m.adopt(1)
    assert m.remaining == 1 and not m.valid()
    m.add(2); assert m.remaining == 1  # inherited complete class is not replay credit
    try:
        m.adopt(2)
    except AssertionError:
        stats['outstanding_debt_adoption_refused'] += 1
    else:
        raise AssertionError('adoption forgot outstanding debt')
    m.add(0); assert m.valid()
    audit_queries(m, stats)
    # Naive reset has zero new coverage deficit but silently forgets b.
    m = Certificate([[0,0,0],[0,1,2]])
    m.add(0);m.add(1);m.adopt(1);m.add(0)
    assert all(c == len(m.cells()[x]) for x,c in m.counts.items() if c)
    assert m.remaining == 1 and not m.valid()
    stats['false_zero_control'] = 1
    # Opaque-cardinality adversary, even with promised size and equal transcript.
    for n in range(2, 9):
        for r in range(1, n):
            for unseen in combinations(range(n), r+1):
                a, b = set(unseen[:-1]), set(unseen[1:])
                queried = set(range(n)) - set(unseen)
                assert len(a) == len(b) == r and a != b
                assert all((v in a) == (v in b) for v in queried)
                stats['indistinguishable_pairs'] += 1
    # Exact coverage at a balanced binary split: every short transcript has a swap.
    for m in range(2, 7):
        left, right = set(range(m)), set(range(m, 2*m))
        for q in range(m):
            for query in combinations(range(2*m), q):
                Q = set(query)
                x, y = min(left-Q), min(right-Q)
                good, bad = left, (left-{x}) | {y}
                assert len(good) == len(bad) == m
                assert all((v in good) == (v in bad) for v in Q)
                assert good in (left,right) and bad not in (left,right)
                stats['coverage_swap_transcripts'] += 1
    # Cross-control: actual reader26 anchor triggers, not a supplied arbitrary split.
    from check_ancestral_selectors import Model
    for size in range(2, 13):
        C0, C1 = list(range(size)), list(range(size, 2*size))
        marker, extra = 2*size, 2*size+1
        good = set(C0 + [marker, extra])
        bad = (good - {C0[0]}) | {C1[0]}
        for chosen, expected_types in [(good, 2), (bad, 3)]:
            seq = [(v,'a') for v in C0+C1+[marker]] + [(extra,'b')]
            seq += [(v,'t') for v in sorted(chosen)]
            seq += [(marker,'b')] + [(v,'b') for v in C0] + [(extra,'a')]
            old = Model(k=2, K=2)
            for edge in seq:
                old.add(*edge)
                old.audit()
            assert len(old.cuts) == 3 and not old.accepted()
            raw_types = {frozenset(v for v,w in old.edges if w == t) for t in 'abt'}
            assert len(raw_types) == expected_types
            parts = [[tuple(c[v][0] for c in old.cuts[:j+1])
                      for v in range(2*size+2)] for j in range(3)]
            repaired = Certificate(parts)
            for v in chosen: repaired.add(v)
            repaired.adopt(2)
            assert repaired.remaining == size+1
            for v in chosen: repaired.add(v)
            repaired.audit()
            assert repaired.valid() == (expected_types == 2)
            stats['anchor_realization_controls'] += 1
    stats['native_timing_evidence'] = False
    print(json.dumps(stats, indent=2, sort_keys=True))


if __name__ == '__main__':
    main()