check_density_independent_readers.py

#!/usr/bin/env python3
"""Assessment166: independent literal oracles for reader conversion and replay.

Python sets and elimination model completed finite algebra, not native costs.
The queue test counts reversal/consumption operations with fixed-rate service.
Native row learning/registration deadlines are proved in reader44 and audited
separately by exact inequalities. No free EOF drain is used by the queue test.
"""
from collections import Counter
from fractions import Fraction
import copy
import json
import random
import check_cross_width_adoption as shared
import check_cross_width_parity as cut
import check_parity_cell_learning as learn


def spectrum(points,n):
    return cut.basis(a for a,v in learn.direct_transform(points,n).items() if a and v)


def future_rank(family):
    return max((len(cut.basis(a&((1<<t)-1) for rows in family[t-1:] for a in rows))
                for t in range(1,len(family)+1)),default=0)


def mask_for(points,bs):
    mask=0
    for x in points:mask |= 1<<cut.linear(bs,x)
    return mask


def convert(global_reader,local_reader,local_masks,count):
    """Only finite instruction matrices are used here, not coefficients/sets."""
    E=();gd=0;result=[]
    for t,((T,b,O),(U,d,P)) in enumerate(zip(global_reader,local_reader)):
        global_next=tuple(a|(bit<<gd) for a,bit in zip(T,b))
        local_next=tuple(cut.row_times(a,E)|(bit<<gd) for a,bit in zip(U,d))
        E=tuple(cut.coordinates(a,global_next) for a in local_next)
        D=tuple(cut.coordinates(cut.row_times(a,E),O) for a in P)
        H=0
        for z in range(1<<len(O)):
            if (local_masks[t]>>cut.linear(D,z))&1:H |= 1<<z
        result.append(H);gd=len(T);count['conversion_columns']+=1
    return result


def extend(reader,n,r):
    result=list(reader)
    while len(result)<n:result.append(((),(),(0,)*r))
    return result


class Protocol:
    def __init__(self,r,k,count):
        self.r=r;self.k=k;self.R=(k+1)*r;self.count=count
        self.active={};self.lru=[];self.seen=set();self.retired={}
        self.family=[];self.reader=[];self.rows={};self.saved=[]
    def enter(self,n):
        if n in self.active:
            self.lru.remove(n);self.lru.append(n);return
        assert n not in self.seen
        if len(self.active)==self.k:
            old=self.lru.pop(0);assert self.active[old]['ready']
            self.retired[old]=self.active.pop(old);self.count['retirements']+=1
        self.active[n]={'rows':{},'ready':True};self.seen.add(n);self.lru.append(n)
        while len(self.family)<n:self.family.append((0,)*self.r)
        self.reader=extend(self.reader,n,self.r)
    def register(self,n,key,points):
        self.enter(n);points=frozenset(points)
        local=self.rows.setdefault(key,{'family':[],'reader':[],'masks':[],'sets':{}})
        while len(local['family'])<n:
            local['family'].append((0,)*self.r);local['masks'].append(0)
        bs=spectrum(points,n);assert len(bs)<=self.r
        bs=tuple(bs)+(0,)*(self.r-len(bs))
        local['reader']=extend(local['reader'],n,self.r)
        got=shared.repair(local['reader'],n,bs,self.R);assert got is not None
        local['reader']=got;local['family'][n-1]=bs
        local['masks'][n-1]=mask_for(points,bs);local['sets'][n]=points
        assert future_rank(local['family'])<=self.R
        self.active[n]['rows'][key]=points
        support={a for ps in self.active[n]['rows'].values() for a in spectrum(ps,n)}
        basis=cut.basis(support);self.count['maximum_provisional_rank']=max(self.count['maximum_provisional_rank'],len(basis))
        if len(basis)>self.r:
            self.active[n]['ready']=False;self.count['local_index_failures']+=1
            return
        new=tuple(basis)+(0,)*(self.r-len(basis))
        got=shared.repair(self.reader,n,new,self.R);assert got is not None
        for m in range(1,len(self.reader)+1):
            if m!=n:
                for x in range(1<<m):
                    assert shared.query(got,x,m)==shared.query(self.reader,x,m)
                    self.count['other_width_queries']+=1
        self.reader=got;self.family[n-1]=new;self.active[n]['ready']=True
        assert max((len(rec[0]) for rec in got),default=0)==future_rank(self.family)
        self.count['rank_annotation_checks']+=1
        if future_rank(self.family)>self.r:self.count['working_rank_above_final_bound']+=1
    def check(self):
        assert all(c['ready'] for c in self.active.values())
        assert future_rank(self.family)<=self.r
        profiles={};D=len(self.reader)
        for key,l in self.rows.items():
            lr=extend(l['reader'],D,self.r);lm=l['masks']+[0]*(D-len(l['masks']))
            profiles[key]=convert(self.reader,lr,lm,self.count)
            for n in range(1,D+1):
                points=l['sets'].get(n,())
                for x in range(1<<n):
                    z=shared.query(self.reader,x,n)
                    assert bool((profiles[key][n-1]>>z)&1)==(x in points)
                    local_z=shared.query(lr,x,n)
                    assert bool((lm[n-1]>>local_z)&1)==(x in points)
                    self.count['row_and_conversion_queries']+=1
        keys=list(self.rows)
        # Class-local committed masks plus one independently readable pending row.
        M={};P={};literal={}
        for key in keys+keys[::2]:
            for n in self.rows[key]['sets']:
                j=n.bit_length()-1
                if j in P:
                    old=P[j]
                    for width in range(1<<j,min(1<<(j+1),D+1)):
                        M[width]=M.get(width,0)|profiles[old][width-1]
                P[j]=key
            # Repeated class entries in the harness are harmless idempotent unions.
            for n,points in self.rows[key]['sets'].items():literal.setdefault(n,set()).update(points)
        for n in range(1,D+1):
            j=n.bit_length()-1;pending=self.rows.get(P.get(j),None)
            for x in range(1<<n):
                committed=bool((M.get(n,0)>>shared.query(self.reader,x,n))&1)
                raw=False
                if pending is not None and n<=len(pending['reader']):
                    raw=bool((pending['masks'][n-1]>>shared.query(pending['reader'],x,n))&1)
                assert (committed or raw)==(x in literal.get(n,()))
                self.count['pending_union_queries']+=1
        self.saved.append(copy.deepcopy((self.reader,profiles,{k:v['sets'] for k,v in self.rows.items()})))
    def check_saved(self):
        for reader,profiles,sets in self.saved:
            for key,ps in sets.items():
                for n,points in ps.items():
                    for x in range(1<<n):
                        assert bool((profiles[key][n-1]>>shared.query(reader,x,n))&1)==(x in points)
                        self.count['retained_queries']+=1
            self.count['retained_versions']+=1


class Ahead:
    def __init__(self,prefix,j):
        # prefix is the existing chronological audit list; native starts from its
        # persistent reverse-stack root, not a copied array.
        self.rear=list(prefix);self.frozen=[];self.front=[];self.build=[]
        self.reversing=False;self.processed=[];self.state=0;self.j=j
    def work(self,bit,count):
        self.rear.append(bit)
        for _ in range(8):
            if self.reversing:
                self.build.append(self.frozen.pop());count['queue_reversals']+=1
                if not self.frozen:
                    self.front=self.build;self.build=[];self.reversing=False
            elif self.front:
                x=self.front.pop();self.processed.append(x)
                self.state=(3*self.state+x+self.j)%5;count['queue_consumptions']+=1
            elif self.rear:
                self.frozen=self.rear;self.rear=[];self.reversing=True
                # Administrative changes are bounded separately; no item consumed.
                self.build.append(self.frozen.pop());count['queue_reversals']+=1
                if not self.frozen:
                    self.front=self.build;self.build=[];self.reversing=False
            else:break


def replay_audit(count,rng):
    for length in (2,3,4,7,8,15,16,31,32,127,256,1024):
        for trial in range(8):
            bits=[rng.randrange(2) for _ in range(length)];prefix=[bits[0]]
            j=0;current_state=bits[0]%5;ahead=Ahead(prefix,1)
            for n,bit in enumerate(bits[1:],2):
                prefix.append(bit);current_state=(3*current_state+bit+j)%5
                ahead.work(bit,count)
                if n&(n-1)==0:
                    assert ahead.processed==prefix
                    j+=1;current_state=ahead.state;ahead=Ahead(prefix,j+1)
                    count['caught_up_boundaries']+=1
                oracle=0
                for b in prefix:oracle=(3*oracle+b+j)%5
                assert current_state==oracle
                count['exact_prefix_answers']+=1


def main():
    count=Counter();rng=random.Random(166)
    for k in (1,2,3):
        for trial in range(3):
            p=Protocol(2,k,count);widths=[4,6,8]
            if trial%2:widths.reverse()
            for n in widths:
                for j in range(n):
                    key='0'*40+format(j,'04b')
                    p.register(n,key,learn.row(n,[1<<j],1))
                for j in range(n):
                    key='0'*40+format(j,'04b');p.register(n,key,set(range(1<<n)))
                for j,mask in enumerate((1,3,7,15)):
                    key='1'*41+format(j,'03b')
                    p.register(n,key,learn.row(n,[1<<(n-2),1<<(n-1)],mask))
                p.check()
            p.check_saved();count['completed_histories']+=1
    # Interleaved widths give row-local temporary future rank above final r.
    p=Protocol(1,2,count)
    p.register(3,'0'*40,learn.row(3,[1],1));p.check()
    p.register(5,'1'*40,learn.row(5,[2],1))
    p.register(7,'1'*40,learn.row(7,[4],1))
    assert future_rank(p.family)==3
    assert future_rank(p.rows['1'*40]['family'])==2
    assert future_rank(p.family)>p.r;count['final_rank_rejections']+=1
    p.register(5,'1'*40,set(range(32)));p.register(7,'1'*40,set(range(128)))
    p.check();p.check_saved();count['interleaved_recovery_histories']+=1
    # Direct independent random matrix families and masks, including width gaps.
    for trial in range(40):
        family=[];locals_=[[] for _ in range(3)];masks=[[] for _ in range(3)];literal=[{} for _ in range(3)]
        for n in range(1,9):
            bs=cut.basis(rng.sample(range(1,1<<n),min(2,(1<<n)-1)))
            family.append(bs)
            for i in range(3):
                mask=rng.randrange(1<<(1<<len(bs)));points=learn.row(n,bs,mask)
                lb=spectrum(points,n);locals_[i].append(lb)
                masks[i].append(mask_for(points,lb));literal[i][n]=points
        _,gr=shared.compile_vector(family)
        for i in range(3):
            _,lr=shared.compile_vector(locals_[i]);profile=convert(gr,lr,masks[i],count)
            for n in range(1,9):
                for x in range(1<<n):
                    assert bool((profile[n-1]>>shared.query(gr,x,n))&1)==(x in literal[i][n])
                    count['random_conversion_queries']+=1
    for n in (6,7,8):
        B=1<<(n-2)
        for mask in range(1,16):
            points=sorted(learn.row(n,[1<<(n-2),1<<(n-1)],mask));rng.shuffle(points)
            m=mask.bit_count();q=m*B-B//4+1
            got,_,_=learn.recover(set(points[:q]),n,2,m)
            assert got is not None and got[0]==set(points)
            count['mixed_density_samples']+=1
    for r in range(1,9):
        for n in range(2*r+12,2*r+53):
            L=1<<(n-r)
            assert Fraction(n*L,8)-2*n>Fraction(n*L,16)>0
            count['strict_horizon_checks']+=1
    replay_audit(count,rng)
    assert count['local_index_failures'] and count['working_rank_above_final_bound']
    print(json.dumps({'status':'passed',**count},sort_keys=True))


if __name__=='__main__':main()