check_sparse_completion_certificates.py

#!/usr/bin/env python3
"""Finite evidence for sparse sibling completion, not a native runtime.

Immutable node tuples/frames model the serial materialized producer. Python
sets independently decide literal gates and memberships; strings, integers,
deque and random access in the harness are not claimed native primitives.
"""
from collections import deque
from itertools import permutations, product
import json
import random


EMPTY = (False, None, None)


def node(t):
    return EMPTY if t is None else t


def member(t, word):
    for bit in word:
        t = node(t)[1 + int(bit)]
    return node(t)[0]


class PairJob:
    def __init__(self, root, prefix):
        self.cursor = root
        self.route = deque(prefix)
        self.frames = []
        self.phase = 'down'
        self.steps = 0
        self.limit = 8 * (len(prefix) + 2)

    def step(self):
        self.steps += 1
        t = node(self.cursor)
        if self.phase == 'down' and self.route:
            b = int(self.route.popleft())
            self.frames.append((t[0], b, t[2 - b]))
            self.cursor = t[1 + b]
        elif self.phase == 'down':
            left, right = node(t[1]), node(t[2])
            self.cursor = (t[0], (True, left[1], left[2]),
                           (True, right[1], right[2]))
            self.phase = 'up'
        elif self.frames:
            terminal, b, sibling = self.frames.pop()
            self.cursor = ((terminal, sibling, self.cursor) if b else
                           (terminal, self.cursor, sibling))
        else:
            self.phase = 'done'
        assert self.steps <= self.limit


class Row:
    def __init__(self):
        self.literal = set()
        self.half = 0
        self.queue = deque()
        self.active = None
        self.root = None
        self.completed_pairs = set()
        self.active_prefix = None
        self.releases = 0
        self.actions = 0

    def bit(self):
        # Fixed rate, no stored idle credit and no delimiter service.
        for _ in range(32):
            if self.active is None:
                if not self.queue:
                    break
                self.active_prefix = self.queue.popleft()
                self.active = PairJob(self.root, self.active_prefix)
                self.actions += 1
            else:
                self.active.step()
                self.actions += 1
                if self.active.phase == 'done':
                    self.root = self.active.cursor
                    self.completed_pairs.add(self.active_prefix)
                    self.active = None

    def delimiter(self, word):
        assert word
        if word not in self.literal:
            mate = word[:-1] + str(1 - int(word[-1]))
            if mate not in self.literal:
                self.half += 1
                self.queue.append(word[:-1])
                self.releases += 1
            else:
                self.half -= 1
            self.literal.add(word)
        oracle_half = sum((p + '0' in self.literal) !=
                          (p + '1' in self.literal)
                          for p in {w[:-1] for w in self.literal})
        assert self.half == oracle_half
        if not self.half:
            assert self.active is None and not self.queue


def run_history(records, stats):
    rows = {}
    snapshots = []
    for rowid, word in records:
        r = rows.setdefault(rowid, Row())
        old_actions = r.actions
        for _ in word:
            r.bit()
            stats['target_bits'] += 1
        r.delimiter(word)
        stats['records'] += 1
        assert r.actions - old_actions <= 32 * len(word)
        # Worker root may overapproximate observed literals, but must exactly
        # represent committed pairs, even at an ineligible boundary.
        expected = {p + b for p in r.completed_pairs for b in '01'}
        probes = expected | r.literal | {word + '0', word[:-1], ''}
        for q in probes:
            assert member(r.root, q) == (q in expected)
            stats['intermediate_memberships'] += 1
        if all(not x.half for x in rows.values()):
            stats['passing_boundaries'] += 1
            for x in rows.values():
                assert x.active is None and not x.queue
                assert {p + b for p in x.completed_pairs for b in '01'} == x.literal
                snapshots.append((x.root, frozenset(x.literal)))
        else:
            stats['rejecting_boundaries'] += 1
    # Later work must not change earlier eligible roots.
    for root, oracle in snapshots:
        for q in oracle | {'', '0', '1', '000', '10101'}:
            assert member(root, q) == (q in oracle)
            stats['retained_memberships'] += 1
        stats['retained_snapshots'] += 1
    stats['histories'] += 1


def mixed_count(words, lo, hi):
    # Independent exhaustive universe, including internal terminals.
    universe = {''.join(b) for n in range(lo, hi + 1)
                for b in product('01', repeat=n)}
    prefixes = {w[:i] for w in universe for i in range(len(w) + 1)}
    return sum(bool(words & region) and bool(region - words)
               for p in prefixes
               for region in [{w for w in universe if w.startswith(p)}])


def main():
    stats = dict.fromkeys(['histories', 'records', 'target_bits',
                          'passing_boundaries', 'rejecting_boundaries',
                          'intermediate_memberships', 'retained_memberships',
                          'retained_snapshots', 'mixed_controls',
                          'subset_residuals', 'preparation_tradeoffs'], 0)
    # Nested terminal pairs: presence of descendants does not imply membership.
    for order in permutations(['0', '1', '00', '01', '000', '001']):
        run_history([(0, w) for w in order], stats)
    rng = random.Random(202)
    for _ in range(80):
        records = []
        for rowid in range(rng.randint(1, 5)):
            pairs = {''.join(rng.choice('01') for _ in range(rng.randint(0, 55)))
                     for _ in range(rng.randint(1, 25))}
            words = [p + b for p in pairs for b in '01']
            words += rng.choices(words, k=rng.randint(0, 30))
            records.extend((rowid, w) for w in words)
        rng.shuffle(records)
        run_history(records, stats)
    # First large word at the final boundary: rejected, no current-word credit.
    for n in [1, 2, 3, 16, 63, 128, 511]:
        r = Row()
        for _ in range(n):
            r.bit()
        r.delimiter('0' * n)
        assert r.half == 1 and len(r.queue) == 1 and r.actions == 0
        for _ in range(n):
            r.bit()
        r.delimiter('0' * (n - 1) + '1')
        assert not r.half and not r.queue and r.active is None
        assert member(r.root, '0' * n)
        assert member(r.root, '0' * (n - 1) + '1')
    # Distinct content with identical occupied-length profile.
    for n in range(2, 12):
        x = {'0' * n, '0' * (n - 1) + '1'}
        y = {'1' + '0' * (n - 1), '1' + '0' * (n - 2) + '1'}
        assert {len(w) for w in x} == {len(w) for w in y}
        assert '0' * n in x - y
    for lo in [2, 4]:
        hi = 2 * lo - 1
        comb = {'0' * (n - 1) + b for n in range(lo, hi + 1) for b in '01'}
        assert mixed_count(comb, lo, hi) == 3 * lo - 3
        full = {''.join(b) for b in product('01', repeat=lo)}
        full |= {'0' * (hi - 1) + b for b in '01'}
        if lo >= 4:
            assert mixed_count(full, lo, hi) >= 2 ** (lo + 1) - 1
        else:
            assert mixed_count(full, lo, hi) == 6
        stats['mixed_controls'] += 2
    # Residual languages distinguish all subset/depth combinations. Prepared
    # records are selected without knowing the later subset; average coverage
    # gives an actual subset with the stated missing-node bound.
    for m in [1, 2, 3]:
        q = 2 ** m
        codes = [format(i, f'0{m}b') for i in range(q)]
        for height in [1, 2, 5]:
            by_subset = []
            for mask in range(1, 2 ** q):
                roots = set()
                for depth in range(height + 1):
                    residual = frozenset('0' * (height - depth) + code + bit
                                         for i, code in enumerate(codes)
                                         if mask & (1 << i) for bit in '01')
                    roots.add(residual)
                assert len(roots) == height + 1
                by_subset.append(roots)
            all_roots = set().union(*by_subset)
            assert len(all_roots) == (height + 1) * (2 ** q - 1)
            stats['subset_residuals'] += len(all_roots)
            ordered = sorted(all_roots, key=lambda x: tuple(sorted(x)))
            for stride in [1, 2, 3, 7]:
                prepared = set(ordered[::stride])
                most_missing = max(len(roots - prepared) for roots in by_subset)
                assert ((height + 1 - most_missing) * len(by_subset)
                        <= len(prepared))
                stats['preparation_tradeoffs'] += 1
    print(json.dumps(stats, sort_keys=True))


if __name__ == '__main__':
    main()