check_prefix_frontier_composition.py

#!/usr/bin/env python3
"""Reader31 semantic checker: cutoff capsules, cover counts and retained unions.
Tuple FIFOs model the established strict queue interface, not native timings.
The independent oracle uses literal sets and bottom-up full-tree enumeration.
"""
from dataclasses import dataclass
from itertools import product, permutations
from collections import Counter
import random
import json
from check_frozen_row_composition import B, lookup, insert as mark_insert

C = Counter()

@dataclass(frozen=True)
class Empty:
    pass

@dataclass(frozen=True)
class Leaf:
    pass

@dataclass(frozen=True)
class Cut:
    frames: tuple
    cutoff: tuple

E, L = Empty(), Leaf()

def view(t):
    if isinstance(t, Empty):
        return False, E, E
    if isinstance(t, Leaf):
        return True, E, E
    assert t.frames
    direction, other = t.frames[0]
    rest = t.frames[1:]
    cut = t.cutoff[1:] if t.cutoff else ()
    assert len(cut) <= len(rest)
    child = Cut(rest, cut) if rest else L
    return (not t.cutoff, other, child) if direction else (not t.cutoff, child, other)

@dataclass(frozen=True)
class Row:
    root: object = E
    count: tuple = ()
    quota: tuple | None = None

def add(row, w, quota=False):
    frames, depth, cut = (), (), ()
    cursor, candidate, underflow = row.root, row.count, False
    for bit in w:
        full, zero, one = view(cursor)
        sibling = zero if bit == '1' else one
        sibling_full = view(sibling)[0]
        depth = (None,) + depth
        frames += ((int(bit), sibling),)
        if sibling_full:
            if candidate:
                candidate = candidate[1:]
            else:
                underflow = True
        else:
            cut = depth
            candidate, underflow = row.count, False
        cursor = one if bit == '1' else zero
        C['path_steps'] += 1
    duplicate = view(cursor)[0]
    if duplicate:
        C['duplicates'] += 1
        return row
    assert not underflow
    q = row.quota if row.quota is not None else (None,) * len(w)
    if quota and not q:
        C['quota_rejections'] += 1
        return None
    return Row(Cut(frames, cut), (None,) + candidate, q[1:] if quota else None)

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

def literal_cover(s, width, p=''):
    target = {p + x for x in words(width - len(p))}
    if not target & s:
        return ()
    if target <= s:
        return (p,)
    return literal_cover(s, width, p+'0') + literal_cover(s, width, p+'1')

def audit(row, s, width):
    stack = [(row.root, '')]
    while stack:
        root, p = stack.pop()
        full, zero, one = view(root)
        expect = {p+x for x in words(width-len(p))} <= s
        assert full == expect, (p, s)
        if len(p) < width:
            stack.extend(((zero,p+'0'), (one,p+'1')))
        C['fullness_views'] += 1
    assert len(row.count) == len(literal_cover(s, width))
    C['count_audits'] += 1

def cover(root):
    # Persistent path snapshots are tuple FIFOs in this semantic model.
    stack, out, visits = [(root, '')], [], 0
    while stack:
        t, p = stack.pop(); visits += 1
        if isinstance(t, Empty):
            continue
        full, z, o = view(t)
        if full:
            out.append(p)
        else:
            stack.extend(((o,p+'1'), (z,p+'0')))
    return tuple(out), visits

def member(root, w):
    for bit in w:
        _, z, o = view(root)
        root = o if bit == '1' else z
    return view(root)[0]

@dataclass(frozen=True)
class Source:
    committed: object = B
    pending: object = E
    literal: frozenset = frozenset()

def select(src, row, literal, width, k):
    prefixes, visits = cover(src.pending)
    assert len(prefixes) <= k
    assert visits <= 1 + 2*k*width
    t = src.committed
    for p in prefixes:
        t = mark_insert(t,p,True)
    C['frontier_installs'] += len(prefixes)
    return Source(t,row.root,src.literal|literal)

def marked(root,b):
    if lookup(root,''):
        return True
    return any(lookup(root,b[:i]) for i in range(1,len(b)+1))

def query(src,b,width):
    return len(b)==width and (marked(src.committed,b) or member(src.pending,b))

def check_source(src,width):
    for b in words(width) + words(width-1) + ('0'*(width+1),):
        assert query(src,b,width)==(b in src.literal)
        C['product_queries'] += 1

def main():
    rng=random.Random(148)
    histories=[]
    for width in range(1,4):
        universe=words(width)
        for mask in range(1<<len(universe)):
            s={w for i,w in enumerate(universe) if mask>>i&1}
            orders=[tuple(sorted(s)),tuple(sorted(s,reverse=True))]
            shuffled=list(s);rng.shuffle(shuffled);orders.append(tuple(shuffled))
            for seq in orders:
                row,raw=Row(),frozenset()
                for w in seq+seq[:2]:
                    row=add(row,w);raw=raw|{w};audit(row,raw,width)
                    histories.append((row,raw,width))
                p,visits=cover(row.root)
                assert p==literal_cover(raw,width)
                C['exhaustive_set_orders'] += 1
    for seq in permutations(words(3)):
        row=Row()
        for w in seq:row=add(row,w)
        assert len(row.count)==1 and view(row.root)[0]
        C['full_cube_permutations'] += 1
    for old,raw,width in histories[::7]:
        w=rng.choice(words(width));new=add(old,w)
        audit(new,raw|{w},width);audit(old,raw,width)
        C['retained_branches'] += 1
    for width in range(2,7):
        universe=words(width)
        k=3
        rows=[]
        for _ in range(50):
            ps=[rng.choice(universe)[:rng.randrange(width+1)] for _ in range(k)]
            s=frozenset(w for w in universe if any(w.startswith(p) for p in ps))
            row=Row();order=list(s);rng.shuffle(order)
            for w in order:row=add(row,w)
            audit(row,s,width);assert len(row.count)<=k
            rows.append((row,s))
        snapshots=[Source()]
        for _ in range(150):
            old=rng.choice(snapshots);row,s=rng.choice(rows)
            new=select(old,row,s,width,k);snapshots.append(new)
            check_source(old,width);check_source(new,width)
            C['source_branches'] += 1
    # Unbounded intermediate frontier, final one cylinder, last insertion merges width levels.
    for width in range(1,9):
        row=Row();universe=words(width)
        first=[w for w in universe if w[-1]=='0']
        second=[w for w in universe if w[-1]=='1']
        for w in first:row=add(row,w)
        assert len(row.count)==1<<(width-1)
        for w in second[:-1]:row=add(row,w)
        assert len(row.count)==width
        row=add(row,second[-1]);assert len(row.count)==1
        assert view(row.root)[0]
        C['late_cascade_families'] += 1
    # Interleaved row gate, final-only acceptance, and quota duplication.
    width,k=4,2;rows={};raw={};bad=0
    for _ in range(3000):
        v,w=rng.choice(words(width)),rng.choice(words(width))
        old=rows.get(v,Row());new=add(old,w)
        bad+=int(len(new.count)>k)-int(len(old.count)>k)
        rows=dict(rows,**{v:new});raw.setdefault(v,set()).add(w)
        assert bad==sum(len(literal_cover(s,width))>k for s in raw.values())
        C['global_gate_updates']+=1
    for width in range(2,9):
        row=Row();universe=words(width)
        for w in universe[:width]:row=add(row,w,quota=True)
        assert row is not None and row.quota==()
        assert add(row,universe[0],quota=True) is row
        assert add(row,universe[width],quota=True) is None
    print(json.dumps({'checks':dict(sorted(C.items())),
                      'native_recognizer_implemented':False,'timing_evidence':False},indent=2))

if __name__=='__main__':main()