check_prefix_cylinder_composition.py

#!/usr/bin/env python3
"""Independent finite checks for reader30, not a native timing implementation.

The streaming arithmetic uses bit tuples and signed unit stacks. The oracle
uses literal sets and enumerates cylinders. Immutable snapshots are retained.
"""
from dataclasses import dataclass
from itertools import product, permutations
from collections import Counter
import json
import random

C = Counter()

def words(n):
    return tuple(''.join(x) for x in product('01', repeat=n))

def signed_step(s, direction):
    sign, units = s
    if not units:
        return direction, (None,)
    if sign == direction:
        return sign, (None,) + units
    return (sign, units[1:]) if len(units) > 1 else (0, ())

@dataclass(frozen=True)
class Row:
    prefix: str
    count_bits: tuple
    seen: frozenset
    good: bool
    quota: tuple

def insert(row, w, bounded=False):
    n = len(w)
    old_bits = row.count_bits if row else (0,) * (n + 1)
    old_p = row.prefix if row else None
    seen = row.seen if row else frozenset()
    quota = row.quota if row else (None,) * n
    duplicate = w in seen
    carry, ones, delta, prefix, candidate = 1, 0, (0, ()), '', []
    common = True
    for i, bit in enumerate(w):
        common = common and (old_p is None or (i < len(old_p) and old_p[i] == bit))
        if common:
            prefix += bit
        else:
            delta = signed_step(delta, 1)
        z = old_bits[i] ^ carry
        carry = old_bits[i] & carry
        candidate.append(z)
        if z:
            ones = min(2, ones + 1)
        elif ones == 0:
            delta = signed_step(delta, -1)
        C['streamed_bits'] += 1
    z = old_bits[n] ^ carry
    carry = old_bits[n] & carry
    candidate.append(z)
    assert not carry
    if z:
        ones = min(2, ones + 1)
    good = ones == 1 and not delta[1]
    if duplicate:
        assert prefix == old_p
        C['duplicate_updates'] += 1
        return row
    if bounded and not quota:
        C['degree_rejections'] += 1
        return None
    return Row(prefix, tuple(candidate), seen | {w}, good,
               quota[1:] if quota else ())

def oracle_p(s):
    first = next(iter(s))
    return next(first[:i] for i in range(len(first), -1, -1)
                if all(x.startswith(first[:i]) for x in s))

def audit(row, n):
    p = oracle_p(row.seen)
    cylinder = frozenset(p + t for t in words(n - len(p)))
    assert row.prefix == p
    assert row.good == (row.seen == cylinder)
    assert sum(b << i for i, b in enumerate(row.count_bits)) == len(row.seen)
    C['row_audits'] += 1

def trie_set(t, key):
    # Semantic persistent path copying, not a claim of delimiter-time execution.
    marked, zero, one = t or (False, None, None)
    if not key:
        return True, zero, one
    if key[0] == '0':
        zero = trie_set(zero, key[1:])
    else:
        one = trie_set(one, key[1:])
    return marked, zero, one

def trie_hits(t, word):
    if t and t[0]:
        return True
    for b in word:
        if not t:
            return False
        t = t[1 + int(b)]
        if t and t[0]:
            return True
    return False

@dataclass(frozen=True)
class Source:
    committed: tuple | None = None
    pending: str | None = None
    selected: tuple = ()

def select(s, row, width):
    t = s.committed
    if s.pending is not None:
        assert len(s.pending) <= width
        t = trie_set(t, s.pending)
        C['worker_prefix_bits'] += len(s.pending)
    return Source(t, row.prefix if row else None,
                  s.selected + ((row.seen if row else frozenset()),))

def query(s, b, width):
    return len(b) == width and (trie_hits(s.committed, b) or
           (s.pending is not None and b.startswith(s.pending)))

def source_audit(s, width):
    raw = frozenset().union(*s.selected)
    for b in words(width) + words(max(0, width - 1)) + ('0' * (width + 1),):
        assert query(s, b, width) == (b in raw)
        C['product_queries'] += 1

def main():
    rng = random.Random(147)
    snapshots = []
    for n in range(1, 4):
        universe = words(n)
        streams = product(universe, repeat=5) if n <= 2 else (
            tuple(rng.choice(universe) for _ in range(20)) for _ in range(1200))
        for stream in streams:
            row = None
            for w in stream:
                row = insert(row, w)
                audit(row, n)
                snapshots.append((row, n))
    # All orders of a complete width3 cylinder of size4, including duplicates.
    for order in permutations(('000', '001', '010', '011')):
        row = None
        for w in order + order[:2]:
            row = insert(row, w)
            audit(row, 3)
        assert row.good and row.prefix == '0'
        C['complete_orders'] += 1
    for n in range(1, 7):
        universe = words(n)
        rows = []
        for plen in range(n + 1):
            p = rng.choice(words(plen))
            members = list(p + tail for tail in words(n - plen))
            rng.shuffle(members)
            row = None
            for w in members:
                row = insert(row, w)
                audit(row, n)
            assert row.good
            rows.append(row)
        history = [Source()]
        for _ in range(120):
            old = rng.choice(history)
            new = select(old, rng.choice(rows + [None]), n)
            history.append(new)
            source_audit(old, n)
            source_audit(new, n)
            C['source_branches'] += 1
        for s in history:
            source_audit(s, n)
    for row, n in snapshots[::13]:
        audit(row, n)
        branch = insert(row, rng.choice(words(n)))
        audit(branch, n)
        audit(row, n)
        C['retained_row_branches'] += 1
    # Check global invalid-row count under interleaving, loss and recovery.
    table, raw, bad = {}, {}, 0
    for _ in range(3000):
        v, w = rng.choice(words(4)), rng.choice(words(4))
        old = table.get(v)
        new = insert(old, w)
        bad += int(not new.good) - int(old is not None and not old.good)
        table = dict(table, **{v: new})
        raw.setdefault(v, set()).add(w)
        expected = sum(set(p + t for t in words(4 - len(p))) != s
                       for s in raw.values() for p in [oracle_p(s)])
        assert bad == expected
        C['global_gate_updates'] += 1
    # Unbounded-degree controls inside d(v)<=|v|: width=2^r, degree=width.
    for r in range(1, 7):
        n = 1 << r
        p = '10' * ((n - r) // 2) + ('1' if (n - r) % 2 else '')
        row = None
        for tail in words(r):
            row = insert(row, p + tail, bounded=True)
            assert row is not None
        assert row.good and len(row.seen) == n and not row.quota
        assert insert(row, next(iter(row.seen)), bounded=True) is row
        outside = ('1' if p[0] == '0' else '0') + '0' * (n - 1)
        assert insert(row, outside, bounded=True) is None
        C['name_linear_controls'] += 1
    def cover_count(s, p=''):
        members = {x for x in s if x.startswith(p)}
        if not members:
            return 0
        if members == {p + t for t in words(4 - len(p))}:
            return 1
        return cover_count(s, p + '0') + cover_count(s, p + '1')
    x = {'0000', '0001', '1000', '1001'}
    y = {'0000', '0010', '1000', '1010'}
    assert oracle_p(x) == oracle_p(y) == '' and len(x) == len(y) == 4
    assert (cover_count(x), cover_count(y)) == (2, 4)
    C['equal_summary_frontier_counterexamples'] += 1
    print(json.dumps({'checks': dict(sorted(C.items())),
                      'native_recognizer_implemented': False,
                      'native_timing_evidence': False}, indent=2))

if __name__ == '__main__':
    main()