check_learned_row_keys.py

"""Finite semantic audit for learned row-key palettes; not native timing evidence."""
from dataclasses import dataclass, replace
from itertools import product, permutations
import json, random
import check_row_typed_composition as base
from check_rectangle_deficit import key, count, frame_count
from check_frozen_row_composition import B, lookup, insert
import check_pending_substitution as n

@dataclass(frozen=True)
class Learned:
    state: object
    palette: object = B

def initial(r): return Learned(base.initial(r))

def step(old, v, w, f, r, K, d=2, s=1):
    a,z=f(v),key(w)
    if not a or not z: return None
    branch=lookup(old.palette,a) or B
    code=lookup(branch,z)
    meta=lookup(old.state.keys,z)
    slots=meta.row_slots if meta else 0
    if code is None:
        if slots==r: return None
        c=slots; slots+=1
    else: c=code-1
    st=base.step(old.state,v,w,lambda _v,_z:c,r,K,d,s)
    if st is None: return None
    meta=replace(lookup(st.keys,z),row_slots=slots)
    st=replace(st,keys=insert(st.keys,z,meta))
    return Learned(st,insert(old.palette,a,insert(branch,z,c+1)))

stats=dict(sequences=0,prefixes=0,retained_versions=0,queries=0)

def oracle(seq,f,r,K,d=2,s=1):
    palettes={}; rows={}; cells={}; targets={}
    for v,w in seq:
        a,z=f(v),key(w)
        if not a or not z or abs(n.cls(v)-n.cls(z))>s:return None
        rows.setdefault(v,set()).add(z)
        if len(rows[v])>d:return None
        p=palettes.setdefault(z,[])
        if a not in p:p.append(a)
        if len(p)>r:return None
        c=p.index(a)
        cells.setdefault((z,c),set()).add((v,w))
        targets.setdefault(z,set()).add(w)
    rectangular=all(es==set(product({v for v,w in es},{w for v,w in es})) for es in cells.values())
    diversity=all(len({frozenset(v for v,t in seq if t==w) for w in ws})<=K for ws in targets.values())
    return palettes,cells,targets,rectangular and diversity

def inspect(st,seq,f,r,K,d=2,s=1):
    data=oracle(seq,f,r,K,d,s);assert data is not None
    palettes,cells,targets,ok=data
    incomplete=excess=0
    for z,p in palettes.items():
        meta=lookup(st.state.keys,z)
        assert meta.row_slots==len(p)
        for c,a in enumerate(p):assert lookup(lookup(st.palette,a),z)==c+1
        hist=[0]*(1<<r)
        for w in targets[z]:
            mask=sum(1<<c for c in range(r) if any(t==w for v,t in cells.get((z,c),set())))
            assert lookup(st.state.masks,w)==mask
            hist[mask]+=1
        assert [count(x) for x in meta.population]==hist
        assert meta.occupied==sum(bool(x) for x in hist)
        excess+=meta.occupied>K
        for c in range(r):
            es=cells.get((z,c),set());vc=len({v for v,w in es});tc=len({w for v,w in es})
            cell=meta.cells[c];debt=vc*tc-len(es)
            assert count(cell.predecessors)==vc and count(cell.targets)==tc
            assert frame_count(cell.deficit)==debt
            incomplete+=bool(debt)
    assert count(st.state.unfinished)==incomplete and count(st.state.excessive)==excess
    assert (not incomplete and not excess)==ok
    return ok

def check(seq,f,r=2,K=2,d=2,s=1,query=True):
    stats['sequences']+=1
    st=initial(r);versions=[st]
    for i,(v,w) in enumerate(seq):
        st=step(st,v,w,f,r,K,d,s)
        if st is None:
            assert oracle(seq,f,r,K,d,s) is None
            return None
        inspect(st,seq[:i+1],f,r,K,d,s);stats['prefixes']+=1
        versions.append(st)
    for i,old in enumerate(versions):
        inspect(old,seq[:i],f,r,K,d,s);stats['retained_versions']+=1
    ok=inspect(st,seq,f,r,K,d,s)
    if ok and query:
        roots=(B,)*r;left=set()
        for i,v in enumerate(sorted({v for v,w in seq})):
            u=str(i%2);left.add((u,v))
            roots=tuple(n.add_left(roots[c],st.state.lanes[c],u,v,d,s) for c in range(r))
            for u,w in product(('0','1'),sorted({w for v,w in seq})+['0001']):
                mask=lookup(st.state.masks,w) or 0;z=key(w)
                got=bool(mask) and z is not None and any(mask&(1<<c) and n.index_contains(lookup(roots[c],u) or {},z) for c in range(r))
                want=any((u,v) in left and t==w for v,t in seq)
                assert bool(got)==want;stats['queries']+=1
    return st

def main():
    rng=random.Random(137)
    edges=list(product(['10','010','11'],['10','010','0010']))
    for bits in range(1<<len(edges)):
        seq=[e for i,e in enumerate(edges) if bits>>i&1]
        mixed=seq[:];rng.shuffle(mixed)
        if mixed:mixed.append(mixed[0])
        for order in (seq,list(reversed(seq)),mixed):check(order,key)
    # All orders of a genuine two-row rectangle with equal streamed row keys.
    for seq in permutations(list(product(['10','010'],['10','010','0010']))):
        st=check(list(seq),key,K=1);assert st.state.unfinished is None
    # Slots are per target key, not global and not allocated per raw row.
    pool=list(product(['10','010','11','011'],['10','010','11','011']))
    for _ in range(300):check([rng.choice(pool) for _ in range(rng.randrange(1,25))],key)
    families=[]
    for m in (2,3,5,9):
        a='0'*m;b='1'*m;z='0'*(2*m+1)
        A=['0'*j+'1'+a for j in range(m+1)]
        Bv=['0'*j+'1'+b for j in range(m+1)]
        T=['0'*j+'1'+z for j in range(6)]
        seq=list(product(A,T[:3]))+list(product(Bv,T[3:]))
        rng.shuffle(seq)
        st=check(seq,key,d=1);assert st.state.unfinished is None
        assert lookup(st.state.keys,z).row_slots==2
        families.append({'rows':len(A)+len(Bv),'learned_keys':2,'targets':len(T)})
    # Same learned key is not permission to erase distinct neighborhoods.
    bad=check([('10','10'),('010','010')],key)
    assert bad.state.unfinished is not None
    assert check([('10','10'),('11','10'),('100','10')],key) is None
    assert check([('000','10')],key) is None
    print(json.dumps({**stats,'growing_families':families,'native_timing_evidence':False},indent=2))

if __name__=='__main__':main()