check_learned_width_cache.py

#!/usr/bin/env python3
"""Foreground165: independent width-dispatch, retained-mask and clock audits.

Small finite cases exercise the large-width algebra at reduced parameters;
they are not executions of the native theorem's separate small-width branch.
Python sets, array indices and object identity are semantic audit tools only.
"""
from collections import Counter
from itertools import product
from fractions import Fraction
import copy
import json
import random
import check_live_rank_transport as live
import check_cross_width_adoption as shared
import check_parity_cell_learning as learn


def replay_cache(widths,k):
    cache=[];seen=set()
    for n in widths:
        # Literal one-pop-per-actual-bit ruler test, not numeric selection.
        rulers={h:[None]*h for h in cache};under={h:False for h in cache}
        for _ in range(n):
            for h in cache:
                if rulers[h]:rulers[h].pop()
                else:under[h]=True
        matches=[h for h in cache if not under[h] and not rulers[h]]
        assert len(matches)<=1
        if matches:cache.remove(matches[0])
        else:
            if n in seen:return False
            if len(cache)==k:cache.pop(0)
            seen.add(n)
        cache.append(n)
    return True


def revisit(widths,k):
    for i,n in enumerate(widths):
        for j in range(i-1,-1,-1):
            if widths[j]==n:
                if len(set(widths[j+1:i]))>=k:return False
                break
    return True


def repair_extend(reader,n,newrows,r):
    # The sentinel's first and subsequent virtual columns have zero state.
    old=list(reader)
    while len(old)<n:old.append(((),(),(0,)*r))
    return shared.repair(old,n,newrows,r)


class Protocol:
    def __init__(self,r,k,counts,fixed_m=None):
        self.r=r;self.k=k;self.count=counts;self.fixed_m=fixed_m;self.cache=[];self.seen=set()
        self.active={};self.physical=[];self.reader=[];self.masks={};self.sets={}
        self.saved=[];self.group=live.invertibles(r);self.rejected=False
    def new_context(self):
        return {'rows':{},'state':((),live.identity(self.r)),'registered':{}}
    def step(self,n,key,x):
        if self.rejected:return
        # A candidate dictionary path is opened for every resident context
        # and the fresh empty context during the ordinary v spelling.
        proposals={};bits=[(x>>i)&1 for i in range(n)]
        for h in self.cache:
            context=self.active[h];row=context['rows'].get(key)
            if row and row['pending'] is not None:
                self.count['speculative_registration_jobs']+=1
                # Actual matching words supply exactly h symbols. Short
                # mismatches can stop a worker early; long ones finish/discard.
                self.count['speculative_service_units']+=min(n,h)
                if n>=h:
                    c=live.update(context['state'],context['registered'],key,
                                  row['pending'],h,self.r,self.group)
                    if c is not None:
                        state,registered,q,_=c
                        replacement=shared.physical(state,self.r)
                        reader=repair_extend(self.reader,h,replacement,self.r)
                        proposals[h]=(state,registered,q,reader,replacement)
                    else:proposals[h]=None
                if h!=n:self.count['discarded_registration_jobs']+=1
        matched=[h for h in self.cache if h==len(bits)]
        if not matched:
            if n in self.seen:
                self.count['reentry_rejections']+=1;self.rejected=True;return
            if len(self.cache)==self.k:
                victim=self.cache.pop(0)
                if not all(self.certified(row) for row in self.active[victim]['rows'].values()):
                    self.count['uncertified_eviction_rejections']+=1;self.rejected=True;return
                del self.active[victim];self.count['certified_evictions']+=1
            self.active[n]=self.new_context();self.seen.add(n)
            self.count['fresh_width_bootstraps']+=1
        else:self.cache.remove(n)
        self.cache.append(n)
        context=self.active[n]
        row=context['rows'].setdefault(key,{'points':set(),'candidate':None,
            'pending':None,'registered':None,'m':None})
        if n in proposals:
            proposal=proposals[n]
            if proposal is None or proposal[3] is None:
                self.count['rank_rejections']+=1;self.rejected=True;return
            state,registered,q,newreader,replacement=proposal
            oldregistered=context['registered']
            for oldkey,item in oldregistered.items():
                if oldkey!=key:assert registered[oldkey][1]==item[1]
            context['state']=state;context['registered']=registered
            row['registered']=row['pending'];row['pending']=None
            while len(self.physical)<n:self.physical.append((0,)*self.r)
            self.physical[n-1]=replacement
            for oldn in range(1,len(self.reader)+1):
                if oldn!=n:
                    for word in range(1<<oldn):
                        assert shared.query(newreader,word,oldn)==shared.query(self.reader,word,oldn)
                        self.count['other_width_preservation_queries']+=1
            self.reader=newreader
            self.masks[n]={v:item[1] for v,item in registered.items()}
            self.sets[n]={v:item[2] for v,item in registered.items()}
            self.count['committed_registrations']+=1
            self.count['nonidentity_transports']+=q!=live.identity(self.r)
        fresh=x not in row['points'];row['points'].add(x)
        B=1<<(n-self.r)
        if fresh:
            c=len(row['points'])
            tiers=[m for m in range(1,(1<<self.r)+1) if c==m*B-B//4+1 and (self.fixed_m is None or m==self.fixed_m)]
            if tiers:
                m=tiers[0];got,_,_=learn.recover(row['points'],n,self.r,m)
                row['candidate']=None if got is None else frozenset(got[0])
                row['pending']=row['candidate'];row['m']=m
                self.count['tier_events']+=1
        if row['candidate'] is not None and x not in row['candidate']:
            row['candidate']=None;row['pending']=None
        self.count['records']+=1
        if self.count['records']%97==0:
            self.saved.append((list(self.reader),copy.deepcopy(self.masks),copy.deepcopy(self.sets)))
        if self.count['records']%43==0:self.check(self.reader,self.masks,self.sets)
    def certified(self,row):
        return row['registered'] is not None and row['candidate']==row['registered'] \
            and row['points']==row['registered'] and row['pending'] is None
    def check(self,reader,masks,sets):
        for n,rows in masks.items():
            mask=0;literal=set()
            for key,g in rows.items():
                mask|=g;literal|=sets[n][key]
                for x in range(1<<n):
                    assert bool((g>>shared.query(reader,x,n))&1)==(x in sets[n][key])
                    self.count['row_queries']+=1
            for x in range(1<<n):
                assert bool((mask>>shared.query(reader,x,n))&1)==(x in literal)
                self.count['union_queries']+=1
    def finish(self):
        if self.rejected:return False
        if not all(self.certified(row) for c in self.active.values() for row in c['rows'].values()):
            self.count['final_certificate_rejections']+=1;return False
        self.check(self.reader,self.masks,self.sets)
        for reader,masks,sets in self.saved:
            self.check(reader,masks,sets);self.count['retained_versions']+=1
        self.count['accepted_histories']+=1;return True


def main():
    counts=Counter();rng=random.Random(165)
    for k in (1,2,3):
        for length in range(8):
            for widths in product((1,2,3,4),repeat=length):
                assert replay_cache(widths,k)==revisit(widths,k)
                counts['exact_revisit_histories']+=1
    # Unbounded-width pattern with shared future rank <=2. Reduced-width
    # spectral components are exact; native clock inequalities are separate.
    for known_density in (False,True):
        for k in (1,2,3):
            for trial in range(4):
                machine=Protocol(2,k,counts,3 if known_density else None)
                widths=list(range(6,10))
                if trial%2:widths.reverse()
                for j in range(0,len(widths),k):
                    queues=[]
                    for n in widths[j:j+k]:
                        bs=[1<<(n-2),1<<(n-1)]
                        for idx,mask in enumerate((7,11) if known_density else (7,10)):
                            points=sorted(learn.row(n,bs,mask));rng.shuffle(points)
                            q=[(n,'0'*6+str(idx),x) for x in points]
                            # Duplicate records have no distinct-count credit.
                            for _ in range(5):q.insert(rng.randrange(len(q)+1),rng.choice(q))
                            queues.append(q)
                    while any(queues):
                        q=rng.choice([q for q in queues if q]);machine.step(*q.pop(0))
                assert machine.finish()
                counts['known_density_histories' if known_density else 'hereditary_histories']+=1
                assert len(machine.seen)==4 and len(machine.cache)<=k
                counts['retired_width_queries']+=sum(1<<n for n in machine.seen if n not in machine.cache)
    # A fixed parallel tuple learns an unknown common density; all branches
    # read the same actual records and only validated roots may be selected.
    for actual_m in (1,2,3,4):
        records=[]
        for n in (6,7):
            bs=[1<<(n-2),1<<(n-1)]
            mask=(1<<actual_m)-1
            points=sorted(learn.row(n,bs,mask));rng.shuffle(points)
            records.extend((n,'0'*7,x) for x in points)
        accepted=[]
        for guess in (1,2,3,4):
            branch=Protocol(2,1,counts,guess)
            for record in records:branch.step(*record)
            if branch.finish():accepted.append(guess)
            counts['density_hypotheses_checked']+=1
        assert accepted==[actual_m]
        counts['unknown_common_density_cases']+=1
    # Ready masks do not excuse evicting an unfinished literal row.
    for k in (1,2,3):
        bad=Protocol(1,k,counts)
        for n in range(5,6+k):bad.step(n,'0'*6,0)
        assert bad.rejected
    # Input-derived intervals: arbitrary coefficient bits, many independent
    # full vectors, but at most one future prefix at every cut.
    for h in range(1,13):
        family=[];old=0;vectors=[]
        for i in range(h):
            n=2*(i+1);a=(rng.randrange(1,4)<<(2*i))
            while len(family)<n:family.append((0,))
            family[n-1]=(a,);vectors.append(a)
        bases,reader=shared.compile_vector(family)
        assert max(map(len,bases))==1 and len(live.basis(vectors))==h
        counts['independent_interval_families']+=1
    # The earliest raw prefix has unbounded rank, while every classifier
    # installed at the single known-density tier is already the final one.
    for n in range(4,10):
        r=2;B=1<<(n-r);bs=(1<<(n-2),1<<(n-1))
        for mask in (7,11,13,14):
            final=learn.row(n,bs,mask);order=sorted(final);rng.shuffle(order)
            assert len(live.minimal({order[0]},n))==n
            q=3*B-B//4+1
            candidate,_,_=learn.recover(set(order[:q]),n,r,3)
            assert candidate is not None and candidate[0]==final
            counts['known_density_raw_rank_witnesses']+=1
    # Exact rational inequalities: preparation/capture plus two full own
    # words (boundary alignment and registration) fit the distinct suffix.
    for r in range(1,9):
        for n in range(2*r+12,2*r+53):
            B=1<<(n-r);T=Fraction(n*B,256)
            delay=T+2*n+(T+2*n)/256+1
            assert delay < Fraction(n*B,64)
            assert delay+2*n < n*(Fraction(B,4)-1)
            counts['strict_horizon_inequalities']+=1
    print(json.dumps({'status':'passed',**counts},sort_keys=True))


if __name__=='__main__':main()