check_ancestral_selectors.py

#!/usr/bin/env python3
"""Independent set oracle for bounded snapshot selectors; no native timing claim."""
from collections import Counter
from copy import deepcopy
from itertools import product, permutations, combinations
import random
import json


class Model:
    def __init__(self, k=2, K=2):
        self.k, self.K = k, K
        self.anchors, self.edges, self.codes = [], set(), {}
        self.H, self.G = Counter(), Counter()
        self.pops, self.cuts, self.late = [], [], set()
        self.targets, self.cells = {}, {}
        self.invalid = False

    def extend(self, code):
        h, p = code
        return h + (p,) * (len(self.pops) - len(h)), p

    def cut(self):
        # Finite table reindex, not a scan of literal row receipts.
        self.H = Counter({(h + (p,), p): n for (h, p), n in self.H.items() if n})
        pop = Counter()
        for (h, p), n in self.H.items():
            pop[h] += n
        self.pops.append(pop)
        # Independent oracle snapshots, reconstructed only from raw edges.
        rows = {v for v, w in self.edges}
        self.cuts.append({v: tuple(sum(1 << j for j, a in enumerate(self.anchors)
                                       if (v, a) in self.edges) for _ in [0])
                          for v in rows})

    def add(self, v, w):
        if self.invalid or (v, w) in self.edges:
            return
        known = w in self.anchors or w in self.targets
        if not known and len(self.anchors) < self.k:
            self.anchors.append(w)
        anchor = w in self.anchors
        if not anchor and not self.pops:
            self.cut()
        if self.pops and v not in self.codes:
            self.invalid = True
            return
        old = self.codes.get(v)
        h, p = self.extend(old) if old else ((), 0)
        if anchor:
            if old:
                self.H[h, p] -= 1
            p |= 1 << self.anchors.index(w)
            self.H[h, p] += 1
            self.edges.add((v, w))
            if self.pops and w not in self.late:
                self.late.add(w)
                self.cut()
                h += (p,)
            self.codes[v] = (h, p)
        else:
            self.codes[v] = (h, p)
            if w not in self.targets:
                self.targets[w] = (len(self.pops) - 1, frozenset())
            t, mask = self.targets[w]
            c = h[:t+1]
            if mask:
                self.G[t, mask] -= 1
            if c not in mask:
                self.cells.setdefault((t, c), [self.pops[t][c], 0, 0])[1] += 1
                mask = mask | {c}
            self.cells[t, c][2] += 1
            self.G[t, mask] += 1
            self.targets[w] = (t, mask)
            self.edges.add((v, w))

    def selector(self, w, code):
        h, p = code
        if w in self.anchors:
            return bool(p & (1 << self.anchors.index(w)))
        if w not in self.targets:
            return False
        t, mask = self.targets[w]
        c = (h + (p,) * max(0, t+1-len(h)))[:t+1]
        return c in mask

    def types(self):
        atoms = [a for a, n in self.H.items() if n]
        masks = {tuple(self.selector(w, a) for a in atoms) for w in self.anchors}
        for (t, mask), n in self.G.items():
            if n:
                masks.add(tuple(h[:t+1] in mask for h, p in atoms))
        return len(masks)

    def accepted(self):
        return not self.invalid and all(V*T == E for V, T, E in self.cells.values()) and self.types() <= self.K

    def audit(self):
        if self.invalid:
            return
        profiles = {v: sum(1 << j for j, a in enumerate(self.anchors)
                           if (v, a) in self.edges) for v in self.codes}
        histories = {v: tuple(c[v][0] for c in self.cuts) for v in self.codes}
        actualH = Counter((histories[v], profiles[v]) for v in self.codes)
        assert +self.H == actualH
        for v, code in self.codes.items():
            assert self.extend(code) == (histories[v], profiles[v])
        for t, pop in enumerate(self.pops):
            assert pop == Counter(histories[v][:t+1] for v in self.codes)
        for (t, c), (V, T, E) in self.cells.items():
            rows = {v for v in self.codes if histories[v][:t+1] == c}
            targets = {w for w, (birth, mask) in self.targets.items()
                       if birth == t and any((v, w) in self.edges for v in rows)}
            assert (V, T, E) == (len(rows), len(targets),
                                 sum((v, w) in self.edges for v in rows for w in targets))
        valid = all(all((v, w) in self.edges for v in self.codes
                        if histories[v][:t+1] == histories[x][:t+1])
                    for w, (t, _) in self.targets.items()
                    for x in self.codes if (x, w) in self.edges)
        assert valid == all(V*T == E for V, T, E in self.cells.values())
        if valid:
            columns = {frozenset(v for v in self.codes if (v, w) in self.edges)
                       for w in self.anchors + list(self.targets)}
            assert self.types() == len(columns)
            for v, code in self.codes.items():
                for w in self.anchors + list(self.targets):
                    assert self.selector(w, code) == ((v, w) in self.edges)


def check_sequence(seq, stats, k=2, K=2):
    m = Model(k, K)
    retained = []
    for edge in seq:
        if len(retained) < 5:
            retained.append(deepcopy(m))
        m.add(*edge)
        m.audit()
        stats['prefix_checks'] += 1
    for old in retained:
        old.audit()
        stats['retained_versions'] += 1
    stats['sequences'] += 1
    if m.accepted():
        rows = list(m.codes)
        for bits in product((0, 1), repeat=len(rows)):
            chosen = {v for v, b in zip(rows, bits) if b}
            codes = {m.codes[v] for v in chosen}
            for w in m.anchors + list(m.targets) + ['missing']:
                got = any(m.selector(w, c) for c in codes)
                expect = any((v, w) in m.edges for v in chosen)
                assert got == expect
                stats['product_queries'] += 1
    return m


def main():
    stats = Counter()
    # Exhaustive small orders exercise discovery, duplicates and invalid new rows.
    base = [(0,'a'),(1,'a'),(1,'b'),(0,'t'),(1,'t'),(0,'b')]
    for seq in permutations(base):
        check_sequence(seq, stats)
    rng = random.Random(140)
    for _ in range(1500):
        seq = [(rng.randrange(4), rng.choice('abtu')) for _ in range(rng.randrange(1, 18))]
        check_sequence(seq, stats)
    for k in (1, 3):
        for _ in range(200):
            seq = [(rng.randrange(4), rng.choice('abcuv'))
                   for _ in range(rng.randrange(1, 20))]
            check_sequence(seq, stats, k=k, K=4)
    m3 = check_sequence([(0,'a'),(1,'b'),(2,'c'),(0,'t'),(1,'t'),(2,'t'),
                         (1,'a'),(2,'b'),(0,'c'),(2,'a'),(0,'b'),(1,'c')],
                        stats, k=3, K=4)
    assert m3.accepted() and len(m3.pops) == 4
    # General family with two final types and arbitrarily many late anchor edges.
    for n in range(2, 7):
        P, Q = list(range(n)), list(range(n, n+2))
        for size in range(1, n):
            for U in combinations(P, size):
                seq = [(v,'a') for v in P] + [(v,'b') for v in Q]
                seq += [(v,'t') for v in P+Q]
                seq += [(v,'b') for v in U] + [(Q[0],'a')]
                seq += [(v,'u') for v in list(U)+Q] + [(Q[1],'a')]
                m = check_sequence(seq, stats)
                assert m.accepted() and len(m.pops) == 3
                stats['two_type_families'] += 1
    # Run139 early split target must still fail despite a later matching profile.
    seq = [(0,'a'),(1,'a'),(2,'a'),(2,'b'),(0,'t'),(2,'t'),(0,'b')]
    assert not check_sequence(seq, stats).accepted()
    # A cut does not mutate untouched row codes; left readout interprets them.
    m = check_sequence([(0,'a'),(1,'a'),(2,'b'),(0,'t'),(1,'t'),(2,'t'),
                        (0,'b'),(2,'a')], stats, K=4)
    assert len(m.codes[1][0]) < len(m.pops)
    assert m.selector('t', m.codes[1])
    stats['native_timing_evidence'] = False
    print(json.dumps(stats, indent=2, sort_keys=True))


if __name__ == '__main__':
    main()