#!/usr/bin/env python3
"""Finite index-growth teaching models. No disk, network, or concurrency runtime.
Normal Python required. Tests include expensive whole-state checks and snapshots.
"""
from collections import Counter
from dataclasses import dataclass, field
from copy import deepcopy
from itertools import product
import json
import random

@dataclass
class Bucket:
    depth: int
    signature: int
    records: dict = field(default_factory=dict)

class Extendible:
    def __init__(self, capacity=2, width=4):
        if capacity<1 or width<1:raise ValueError('positive capacity/width required')
        self.capacity,self.width=capacity,width
        self.g=1;self.directory=[0,1]
        self.buckets={0:Bucket(1,0),1:Bucket(1,1)};self.next_id=2
        self.splits=0;self.directory_copies=0
    def hash(self,key):return key % (1<<self.width)
    def location(self,key):return self.directory[self.hash(key) % (1<<self.g)]
    def get(self,key):return self.buckets[self.location(key)].records.get(key)
    def put(self,key,value):
        bucket=self.buckets[self.location(key)]
        if key in bucket.records:
            bucket.records[key]=value;return True
        # Detect the exact finite-width impossibility before mutating structure.
        if sum(self.hash(k)==self.hash(key) for k in bucket.records)>=self.capacity:
            return False
        while True:
            bid=self.location(key);bucket=self.buckets[bid]
            if len(bucket.records)<self.capacity:
                bucket.records[key]=value;return True
            d=bucket.depth
            if d>=self.width:raise AssertionError('preflight should have caught collision')
            if d==self.g:
                self.directory_copies+=len(self.directory)
                self.directory=self.directory+self.directory
                self.g+=1
            other=self.next_id;self.next_id+=1
            new=Bucket(d+1,bucket.signature+(1<<d));self.buckets[other]=new
            bucket.depth=d+1
            for j in range(len(self.directory)):
                if self.directory[j]==bid and (j>>d)&1:self.directory[j]=other
            for k,v in list(bucket.records.items()):
                if (self.hash(k)>>d)&1:
                    new.records[k]=v;del bucket.records[k]
            self.splits+=1
    def delete(self,key):
        bid=self.location(key);bucket=self.buckets[bid]
        if key not in bucket.records:return False
        del bucket.records[key]
        while self.buckets[bid].depth>0:
            bucket=self.buckets[bid];d=bucket.depth
            buddy_sig=bucket.signature^(1<<(d-1))
            other=self.directory[buddy_sig];buddy=self.buckets[other]
            if buddy.depth!=d or len(bucket.records)+len(buddy.records)>self.capacity:break
            assert other!=bid
            keep,remove=(bid,other) if bucket.signature<buddy.signature else (other,bid)
            self.buckets[keep].records.update(self.buckets[remove].records)
            self.buckets[keep].depth=d-1
            self.buckets[keep].signature&=(1<<(d-1))-1
            for j in range(len(self.directory)):
                if self.directory[j]==remove:self.directory[j]=keep
            del self.buckets[remove];bid=keep
        while self.g>0 and all(b.depth<self.g for b in self.buckets.values()):
            half=len(self.directory)//2
            assert self.directory[:half]==self.directory[half:]
            self.directory=self.directory[:half];self.g-=1
        return True
    def check(self,oracle):
        assert len(self.directory)==1<<self.g and 0<=self.g<=self.width
        assert set(self.directory)==set(self.buckets)
        observed={}
        aliases=Counter(self.directory)
        for bid,b in self.buckets.items():
            assert 0<=b.depth<=self.g and 0<=b.signature<(1<<b.depth)
            assert aliases[bid]==1<<(self.g-b.depth)
            assert len(b.records)<=self.capacity
            expected={j for j in range(1<<self.g) if j%(1<<b.depth)==b.signature}
            assert expected=={j for j,p in enumerate(self.directory) if p==bid}
            for key,value in b.records.items():
                assert key not in observed and self.location(key)==bid
                assert self.hash(key)%(1<<b.depth)==b.signature
                observed[key]=value
        assert observed==oracle
    def snapshot(self):
        return {'global_depth':self.g,'directory':self.directory.copy(),
                'buckets':[{'id':bid,'depth':b.depth,'signature':b.signature,
                            'keys':list(b.records)} for bid,b in sorted(self.buckets.items())],
                'physical_buckets':len(self.buckets),'records':sum(len(b.records) for b in self.buckets.values())}

class Linear:
    def __init__(self,capacity=2,initial=2,hash_function=lambda k:k):
        if capacity<1 or initial<1:raise ValueError('positive capacity and initial count required')
        self.capacity,self.initial=capacity,initial
        self.level,self.s=0,0;self.buckets=[{} for _ in range(initial)]
        self.hash=hash_function
    @property
    def base(self):return self.initial*(1<<self.level)
    def location(self,key):
        h=self.hash(key)
        if h<0:raise ValueError('nonnegative hash required')
        a=h%self.base
        return h%(2*self.base) if a<self.s else a
    def get(self,key):return self.buckets[self.location(key)].get(key)
    def put(self,key,value):
        at=self.location(key);bucket=self.buckets[at]
        if key in bucket:
            bucket[key]=value;return None
        bucket[key]=value
        return self.split() if len(bucket)>self.capacity else None
    def split(self):
        n=self.base;at=self.s;new=n+at
        assert new==len(self.buckets)
        left,right={},{}
        for key,value in self.buckets[at].items():
            destination=self.hash(key)%(2*n)
            assert destination in [at,new]
            (left if destination==at else right)[key]=value
        self.buckets[at]=left;self.buckets.append(right)
        self.s+=1
        if self.s==n:self.level+=1;self.s=0
        return at
    def delete(self,key):
        bucket=self.buckets[self.location(key)]
        if key not in bucket:return False
        del bucket[key];return True
    def shrink(self):
        if len(self.buckets)==self.initial:return False
        if self.s:
            self.s-=1
        else:
            self.level-=1;self.s=self.base-1
        last=self.buckets.pop()
        assert not (last.keys() & self.buckets[self.s].keys())
        self.buckets[self.s].update(last)
        return True
    def check(self,oracle):
        assert 0<=self.s<self.base and len(self.buckets)==self.base+self.s
        observed={}
        for at,bucket in enumerate(self.buckets):
            for key,value in bucket.items():
                assert key not in observed
                assert self.location(key)==at
                observed[key]=value
        assert observed==oracle
    def snapshot(self):
        return {'base':self.base,'split_pointer':self.s,'main_buckets':len(self.buckets),
                'buckets':[list(b) for b in self.buckets],
                'data_pages':sum(max(1,(len(b)+self.capacity-1)//self.capacity) for b in self.buckets)}

@dataclass(frozen=True)
class Node:
    high: float
    right: str | None
    keys: tuple = ()
    children: tuple = ()  # (inclusive lower fence, child page ID)

def read_step(pages,current,key):
    node=pages[current] # Atomic complete immutable page image in this model.
    if key>=node.high:
        if node.right is None:raise AssertionError('finite high fence needs right page')
        return node.right,None
    if node.children:
        candidates=[child for low,child in node.children if low<=key]
        if not candidates:raise AssertionError('reader approached target from the right')
        return candidates[-1],None
    return None,key in node.keys

def lookup(pages,key,start='R'):
    current=start;path=[]
    while current is not None:
        path.append(current)
        assert len(path)<=len(pages)+1
        current,answer=read_step(pages,current,key)
    return answer,path

def initial_blink():
    return {'R':Node(float('inf'),None,children=((float('-inf'),'A'),(50,'C'))),
            'A':Node(50,'C',(10,20,30,40)),
            'C':Node(float('inf'),None,(50,60))}

def write_blink(pages,index):
    if index==0:pages['B']=Node(50,'C',(30,35,40))
    elif index==1:pages['A']=Node(30,'B',(10,20))
    elif index==2:pages['R']=Node(float('inf'),None,children=((float('-inf'),'A'),(30,'B'),(50,'C')))
    else:raise ValueError('writer step out of range')

def blink_tests():
    old_keys={10,20,30,40,50,60};histories=0
    def explore(key,pages,writer,current,start_published,answer,path):
        nonlocal histories
        if answer is not None:
            if key in old_keys:assert answer
            elif key!=35:assert not answer
            elif start_published:assert answer
            if key==35 and answer:assert writer>=2
            histories+=1
            return
        if writer<3:
            new=dict(pages);write_blink(new,writer)
            explore(key,new,writer+1,current,start_published,answer,path)
        if start_published is None:start_published=writer>=2
        nxt,result=read_step(pages,current,key)
        explore(key,pages,writer,nxt,start_published,result,path+[current])
    for key in range(71):explore(key,initial_blink(),0,'R',None,None,[])
    # Canonical stale-parent path.
    pages=initial_blink();write_blink(pages,0);write_blink(pages,1)
    answer,path=lookup(pages,40)
    assert answer and path==['R','A','B']
    assert lookup(pages,30)[0] and not lookup(pages,29)[0]
    write_blink(pages,2);assert lookup(pages,40)==(True,['R','B'])
    # Incorrect intermediate states lose a key that predates the insertion.
    bad=initial_blink();bad['A']=Node(50,'C',(10,20))
    assert lookup(bad,40)[0] is False
    bad=initial_blink();bad['A']=Node(30,'B',(10,20))
    try:lookup(bad,40)
    except KeyError:pass
    else:raise AssertionError('unpublished right page was silently usable')
    # Right correction at an internal level, not only at leaves.
    upper={'R':Node(float('inf'),None,children=((float('-inf'),'I'),(100,'K'))),
           'I':Node(50,'J',children=((float('-inf'),'A'),(30,'B'))),
           'J':Node(100,'K',children=((50,'C'),(70,'D'))),
           'K':Node(float('inf'),None,children=((100,'E'),(150,'F'))),
           'A':Node(30,'B',(10,20)),'B':Node(50,'C',(30,40)),
           'C':Node(70,'D',(50,60)),'D':Node(100,'E',(70,80)),
           'E':Node(150,'F',(100,110)),'F':Node(float('inf'),None,(150,160))}
    assert lookup(upper,60)==(True,['R','I','J','C'])
    rng=random.Random(6101);queries=0;max_right_moves=0
    for _ in range(120):
        keys=rng.sample([k for k in range(120) if k not in {0,2,4,6}],40)
        pages={'R':Node(float('inf'),None,children=((float('-inf'),'Z'),(0,'0'))),
               'Z':Node(0,'0',(-2,-1)),
               '0':Node(float('inf'),None,(0,2,4,6))};oracle={-2,-1,0,2,4,6};next_id=1
        for key in keys:
            _,path=lookup(pages,key);leaf=path[-1];old=pages[leaf]
            ordered=sorted(old.keys+(key,));oracle.add(key)
            if len(ordered)<=4:
                pages[leaf]=Node(old.high,old.right,tuple(ordered))
            else:
                split=len(ordered)//2;right_id=str(next_id);next_id+=1
                pages[right_id]=Node(old.high,old.right,tuple(ordered[split:]))
                pages[leaf]=Node(ordered[split],right_id,tuple(ordered[:split]))
            # Root repair is deliberately delayed throughout this finite run.
            for query in range(120):
                answer,path=lookup(pages,query)
                assert answer==(query in oracle)
                max_right_moves=max(max_right_moves,len(path)-2);queries+=1
    return {'writer_reader_histories':histories,'repeated_split_runs':120,
            'delayed_parent_queries':queries,'max_extra_right_pages':max_right_moves,
            'stale_parent_path':['R','A','B'],'repaired_path':['R','B'],
            'internal_right_path':['R','I','J','C']}

def extendible_tests():
    e=Extendible();oracle={};trace=[]
    for key in [5,9,1,3,7,13]:
        assert e.put(key,key*10);oracle[key]=key*10;e.check(oracle)
        trace.append({'insert':key,**e.snapshot()})
    assert e.g==3 and len(e.buckets)==4
    final=e.snapshot()
    deletion=[]
    for key in [13,9,7,3]:
        assert e.delete(key);del oracle[key];e.check(oracle)
        deletion.append({'delete':key,**e.snapshot()})
    assert deletion[0]['global_depth']==3 and deletion[1]['global_depth']==2
    assert e.g==0 and len(e.buckets)==1 and set(oracle)=={1,5}
    collide=Extendible();assert collide.put(1,10);assert collide.put(17,170)
    before=collide.snapshot();assert not collide.put(33,330)
    assert collide.snapshot()==before and collide.get(1)==10 and collide.get(17)==170
    rng=random.Random(6102);events=0;rejections=0
    for _ in range(350):
        e=Extendible(rng.randint(1,4),rng.randint(2,6));oracle={}
        for step in range(100):
            key=rng.randrange(128)
            if rng.randrange(3):
                value=step+1000
                same_hash=sum(k%(1<<e.width)==key%(1<<e.width) for k in oracle if k!=key)
                expected=key in oracle or same_hash<e.capacity
                before=e.snapshot()
                accepted=e.put(key,value)
                assert accepted==expected
                if accepted:oracle[key]=value
                else:
                    assert e.snapshot()==before;rejections+=1
            else:
                expected=key in oracle;assert e.delete(key)==expected;oracle.pop(key,None)
            e.check(oracle)
            for q in [key,rng.randrange(128)]:assert e.get(q)==oracle.get(q)
            events+=1
    return {'insert_trace':trace,'final_before_delete':final,'delete_trace':deletion,
            'random_events':events,'finite_hash_rejections':rejections}

def linear_tests():
    h=Linear();oracle={};trace=[]
    for key in [5,9,1,3,7,13]:
        split=h.put(key,key*10);oracle[key]=key*10;h.check(oracle)
        trace.append({'insert':key,'split_bucket':split,**h.snapshot()})
    assert h.base==4 and h.s==1 and len(h.buckets)==5
    assert h.location(13)==1 and h.location(4)==4
    assert h.snapshot()['data_pages']==6
    final=h.snapshot();split=h.put(17,170);oracle[17]=170;h.check(oracle)
    assert split==1 and h.s==2 and list(h.buckets[1])==[9,1,17]
    assert list(h.buckets[5])==[5,13]
    after17=h.snapshot()
    # Restore the six-key run and perform two structural inverse steps.
    h=Linear();oracle={}
    for k in [5,9,1,3,7,13]:h.put(k,k*10);oracle[k]=k*10
    assert h.shrink();h.check(oracle);assert (h.base,h.s)==(4,0)
    assert h.shrink();h.check(oracle);assert (h.base,h.s)==(2,1)
    assert len(h.buckets[1])==6
    address_checks=0
    for initial,level in product([1,2,3,5],range(5)):
        n=initial*(1<<level)
        for s in range(n):
            count=n+s
            for key in range(400):
                # Independent fold-back description using the actual bucket count.
                candidate=key%(2*n)
                oracle_address=candidate if candidate<count else candidate-n
                first=key%n;actual=key%(2*n) if first<s else first
                assert actual==oracle_address and 0<=actual<count
                address_checks+=1
    rng=random.Random(6103);events=0
    for _ in range(300):
        h=Linear(rng.randint(1,4),rng.choice([1,2,3]),lambda k:k%31);oracle={}
        for step in range(100):
            event=rng.randrange(5);key=rng.randrange(128)
            if event<=1:h.put(key,step+1);oracle[key]=step+1
            elif event==2:
                assert h.delete(key)==(key in oracle);oracle.pop(key,None)
            elif event==3:h.split()
            else:h.shrink()
            h.check(oracle)
            for q in [key,rng.randrange(128)]:assert h.get(q)==oracle.get(q)
            events+=1
    return {'insert_trace':trace,'six_key_final':final,'after_insert_17':after17,
            'address_fold_checks':address_checks,'random_mixed_events':events}

def main():
    if not __debug__:
        raise RuntimeError('Run without -O: this checker requires active assertions.')
    result={'status':'PASS','scope':'Finite structural models; no real DBMS concurrency or durability claim',
            'blink':blink_tests(),'extendible':extendible_tests(),'linear':linear_tests()}
    print(json.dumps(result,ensure_ascii=False,indent=2))
if __name__=='__main__':main()