check_stable_row_palettes.py

#!/usr/bin/env python3
"""Independent metric/palette semantics; no predecessor sets in maintained state.
Python dictionaries model name routing, not native timing. Unit stacks are actual
immutable counter representations. Literal sets occur only in the audit oracle.
"""
from dataclasses import dataclass
from collections import Counter
from itertools import product, permutations
import json, random
from check_bounded_type_dynamics import push, pop, one, count, columns

@dataclass(frozen=True)
class Header:
    sizes: tuple
    populations: tuple
    distances: tuple
    values: tuple
    cohorts: tuple

@dataclass(frozen=True)
class State:
    headers: dict
    targets: dict
    rows: dict


def initial(k):
    return Header((None,)*k,(None,)*k,tuple((None,)*k for _ in range(k)),
                  (0,)*((1<<k)-1),(None,)*((1<<k)-1))


def transform(f,b):
    return sum(((b>>src)&1)<<j for j,src in enumerate(f) if src is not None)


def step(st,v,w,z,k,d,events):
    old=st.headers.get(z,initial(k)); records=st.rows.get(v,{})
    label=records.get(z); b=0 if label is None else old.values[label]
    if label is None and len(records)==d:
        events['degree_rejections']+=1;return None
    src=st.targets.get(w)
    active=[i for i,c in enumerate(old.populations) if c is not None]
    if src is not None and b>>src&1:
        events['duplicates']+=1;return st
    match=[i for i in active if (one(old.sizes[i]) if src is None
           else i!=src and one(old.distances[src][i])) and b>>i&1]
    assert len(match)<=1
    changed=False
    if match:dst=match[0];events['joins']+=1
    elif src is not None and one(old.populations[src]):
        dst=src;changed=True;events['singleton_changes']+=1
    else:
        free=[i for i in range(k) if old.populations[i] is None]
        if not free:events['width_rejections']+=1;return None
        dst=free[0];changed=True;events['births' if src is None else 'splits']+=1
    sizes,pops=list(old.sizes),list(old.populations)
    dist=[list(r) for r in old.distances];f=list(range(k))
    if changed:
        sizes[dst]=push(None if src is None else old.sizes[src]);f[dst]=src
        for i in active:
            if i==dst:continue
            base=old.sizes[i] if src is None else old.distances[src][i]
            dist[dst][i]=dist[i][dst]=pop(base) if b>>i&1 else push(base)
        dist[dst][dst]=None
    if src!=dst:
        if src is not None:
            pops[src]=pop(pops[src])
            if pops[src] is None:
                events['retirements']+=1;sizes[src]=None
                for i in range(k):dist[src][i]=dist[i][src]=None
        pops[dst]=push(pops[dst])
    for i in range(k):
        if pops[i] is None:f[i]=None
    vals,cohorts=list(old.values),list(old.cohorts)
    if label is not None:cohorts[label]=pop(cohorts[label])
    occupied=[a for a,c in enumerate(cohorts) if c is not None]
    for a in occupied:vals[a]=transform(f,vals[a])
    assert len({vals[a] for a in occupied})==len(occupied)
    assert all(vals[a] for a in occupied)
    new_b=transform(f,b)|(1<<dst)
    same=[a for a in occupied if vals[a]==new_b]
    if same:a=same[0];events['cohort_joins']+=1
    else:
        a=next(a for a,c in enumerate(cohorts) if c is None)
        vals[a]=new_b;events['cohort_births']+=1
    cohorts[a]=push(cohorts[a])
    headers=dict(st.headers);headers[z]=Header(tuple(sizes),tuple(pops),
        tuple(tuple(r) for r in dist),tuple(vals),tuple(cohorts))
    targets=dict(st.targets);targets[w]=dst
    rows=dict(st.rows);record=dict(records);record[z]=a;rows[v]=record
    events['palette_updates']+=1
    return State(headers,targets,rows)


def audit(st,edges,key,k,stats):
    cs=columns(edges)
    for z,h in st.headers.items():
        local={w:p for w,p in cs.items() if key(w)==z};reps={};counts=Counter()
        for w,p in local.items():
            i=st.targets[w];counts[i]+=1
            if i in reps:assert reps[i]==p
            reps[i]=p
        assert len({frozenset(p) for p in reps.values()})==len(reps)
        rows=set().union(*local.values()) if local else set();cc=Counter()
        for v in rows:
            a=st.rows[v][z];cc[a]+=1
            expect=sum((v in reps.get(i,set()))<<i for i in range(k))
            assert h.values[a]==expect;stats['row_mask_checks']+=1
        for i in range(k):
            assert count(h.populations[i])==counts[i]
            assert count(h.sizes[i])==len(reps.get(i,set()))
            for j in range(k):
                expect=len(reps[i]^reps[j]) if i in reps and j in reps else 0
                assert count(h.distances[i][j])==expect
        for a,c in enumerate(h.cohorts):assert count(c)==cc[a]
        assert len({h.values[a] for a in cc})==len(cc)
    for v,record in st.rows.items():
        assert set(record)=={key(w) for x,w in edges if x==v}
    stats['audits']+=1


def run(seq,key,k,d,stats,events):
    st=State({}, {}, {});edges=set();saved=[]
    for v,w in seq:
        z=key(w);es=edges|{(v,w)};cs=columns(es)
        bad=any(len({key(t) for x,t in es if x==r})>d for r,t in es)
        bad |= any(len({frozenset(p) for t,p in cs.items() if key(t)==q})>k
                   for q in {key(t) for t in cs})
        new=step(st,v,w,z,k,d,events);assert (new is None)==bad
        if new is None:break
        # Every unnamed row's finite selector remains stable in its key.
        for x,rec in st.rows.items():
            if x!=v:assert new.rows[x]==rec
        st=new;edges=es;audit(st,edges,key,k,stats)
        if len(saved)<3:saved.append((st,edges.copy()))
        stats['prefixes']+=1
    for old,es in saved:
        audit(old,es,key,k,stats);stats['retained_versions']+=1
        if seq:
            v,w=seq[-1];branch=step(old,v,w,key(w),k,d,events)
            if branch is not None:
                audit(branch,es|{(v,w)},key,k,stats);stats['persistent_branches']+=1
    stats['streams']+=1
    return st,edges


def composition(st,edges,key,d,stats,rng):
    # Exact logical overlay; installation cost is addressed in the written proof.
    sources={};left=set();saved=[];rows=list(st.rows)+['absent']
    for _ in range(70):
        u=rng.choice(['0','10','111']);v=rng.choice(rows)
        committed,pending=sources.get(u,({},{}));committed=dict(committed);pending=dict(pending)
        jobs=[(z,a) for z,a in st.rows.get(v,{}).items()]
        buckets={len(z).bit_length()-1 for z,a in jobs}
        for j in buckets:
            for z,a in pending.pop(j,[]):
                committed[z]=committed.get(z,0)|st.headers[z].values[a]
        for z,a in jobs:pending.setdefault(len(z).bit_length()-1,[])
        for j in buckets:
            pending[j]=[(z,a) for z,a in jobs if len(z).bit_length()-1==j]
            assert len(pending[j])<=d
        sources=dict(sources);sources[u]=(committed,pending);left.add((u,v))
        if len(saved)<4:saved.append((sources,left.copy()))
    for srcs,rel in saved+[(sources,left)]:
        for u in ['0','10','111','absent']:
            committed,pending=srcs.get(u,({},{}))
            for w in list(st.targets)+['0011absent']:
                z=key(w);mask=committed.get(z,0)
                for q,a in pending.get(len(z).bit_length()-1,[]):
                    if q==z:mask|=st.headers[q].values[a]
                got=w in st.targets and bool(mask>>st.targets[w]&1)
                assert got==any(x==u and (v,w) in edges for x,v in rel)
                stats['product_queries']+=1
        stats['source_versions']+=1


def main():
    stats,events=Counter(),Counter();rng=random.Random(145)
    key=lambda w:w.split('11',1)[1]
    alphabet=list(product(['0','10'],['00110','01110','00111']))
    for k,d in product((1,2,3),(1,2)):
        for length in range(5):
            for seq in product(alphabet,repeat=length):run(seq,key,k,d,stats,events)
        for seq in permutations(alphabet):run(seq,key,k,d,stats,events)
    names=['0','00','01','1','10','1'*17]
    targets=['00'*j+'11'+z for j in range(1,5) for z in ['0','10','1111']]
    for _ in range(300):
        st,edges=run([(rng.choice(names),rng.choice(targets)) for _ in range(30)],
                     key,3,3,stats,events)
        composition(st,edges,key,3,stats,rng)
    for n in range(2,21):
        seq=[]
        for r in range(n):
            for z in ['0','10','1111']:
                ws=['00'*j+'11'+z for j in range(1,n+1)];rng.shuffle(ws)
                seq.extend((bin(r)[2:],w) for w in ws)
        st,edges=run(seq,key,2,3,stats,events)
        assert len(edges)==3*n*n
        composition(st,edges,key,3,stats,rng);stats['growing_families']+=1
    print(json.dumps({'checks':dict(sorted(stats.items())),
        'events':dict(sorted(events.items())), 'suffix_history_stored':False,
        'predecessor_sets_in_state':False,'finite_timing_evidence':False},indent=2))

if __name__=='__main__':main()