"""Two-token session admission and exact-version two-round causal reads.
Fixed honest origins; atomic local APIs; no real RPC, clocks, fsync or recovery.
Standard library; stdout only. All checks also run under Python -O.
"""
from dataclasses import dataclass
from itertools import combinations,product,permutations
import json,random

def need(ok,msg):
    if not ok:raise ValueError(msg)
def rejected(call):
    try:call()
    except ValueError as e:return str(e)
    raise RuntimeError('expected rejection')
def vmax(a,b):
    need(len(a)==len(b),'same vector dimension');return tuple(max(x,y)for x,y in zip(a,b))
def covered(a,b):return len(a)==len(b) and all(x<=y for x,y in zip(a,b))

@dataclass(frozen=True)
class Write:
    origin:int
    sequence:int
    key:str
    value:str
    clock:int
    dep:tuple
    @property
    def dot(self):return(self.origin,self.sequence)
    @property
    def stamp(self):return(self.clock,self.origin)

class Replica:
    def __init__(self,origin,m):
        need(type(m)is int and m>0 and type(origin)is int and 0<=origin<m,'fixed origins')
        self.origin=origin;self.m=m;self.f=(0,)*m;self.clock=0;self.issued=0;self.applied={};self.latest={}
    def receive(self,w):
        need(isinstance(w,Write)and type(w.origin)is int and 0<=w.origin<self.m,'write origin')
        need(len(w.dep)==self.m and all(type(x)is int and x>=0 for x in w.dep),'dependency vector')
        need(type(w.sequence)is int and w.sequence>0 and w.dep[w.origin]==w.sequence-1,'source prefix dependency')
        need(type(w.key)is str and type(w.value)is str and type(w.clock)is int and w.clock>0,'record fields')
        if w.dot in self.applied:
            need(self.applied[w.dot]==w,'identity cannot change');return 'DUPLICATE'
        if w.sequence!=self.f[w.origin]+1 or not covered(w.dep,self.f):return 'WAIT'
        # Only legitimate immutable messages are inputs. Record histories below
        # are retained for auditing, not a global sequencer used by the sender.
        self.applied[w.dot]=w;f=list(self.f);f[w.origin]=w.sequence;self.f=tuple(f);self.clock=max(self.clock,w.clock)
        old=self.latest.get(w.key)
        if old is None or old.stamp<w.stamp:self.latest[w.key]=w
        return 'APPLIED'
    def create(self,key,value):
        need(type(key)is str and type(value)is str,'string key/value')
        need(self.f[self.origin]==self.issued,'source identity has not been reused')
        w=Write(self.origin,self.issued+1,key,value,self.clock+1,self.f)
        need(self.receive(w)=='APPLIED','new write applies locally');self.issued+=1;return w
    def show(self):return dict(frontier=self.f,values={k:w.value for k,w in self.latest.items()})

class Session:
    def __init__(self,m,flags=('RYW','MR','MW','WFR')):
        need(type(m)is int and m>0,'session dimension');self.m=m;self.flags=frozenset(flags)
        need(self.flags<=frozenset(('RYW','MR','MW','WFR')),'known session guarantees');self.R=(0,)*m;self.W=(0,)*m
    def lower(self,operation):
        need(operation in ('read','write'),'operation');readflag,writeflag=('MR','RYW')if operation=='read'else('WFR','MW')
        return vmax(self.R if readflag in self.flags else(0,)*self.m,self.W if writeflag in self.flags else(0,)*self.m)
    def request(self,node,operation,key,value=None):
        need(node.m==self.m and type(key)is str,'matching request');lower=self.lower(operation)
        if not covered(lower,node.f):return dict(status='INCOMPLETE',lower=lower,frontier=node.f)
        if operation=='read':
            w=node.latest.get(key);f=node.f;self.R=vmax(self.R,f)
            return dict(status='OK',lower=lower,frontier=f,value=None if w is None else w.value,dot=None if w is None else w.dot)
        w=node.create(key,value);v=list(self.W);v[w.origin]=max(v[w.origin],w.sequence);self.W=tuple(v)
        return dict(status='OK',lower=lower,frontier=node.f,write=w)
    def show(self):return dict(read_token=self.R,write_token=self.W)

@dataclass(frozen=True)
class Version:
    key:str
    number:int
    value:object
    deps:tuple
    def dependencies(self):return dict(self.deps)

class VersionStore:
    """Single local cluster abstraction; each key has a serialized writer.
    Dict histories and pins. Methods are atomic. Explicit cancel, no TTL.
    """
    def __init__(self,keys):
        keys=list(keys);need(all(type(k)is str for k in keys)and len(set(keys))==len(keys),'distinct string keys')
        self.current={k:0 for k in keys};self.history={k:{0:Version(k,0,None,())}for k in keys};self.pins={};self.active=set();self.serial=0
    def put(self,key,value,parents=()):
        need(key in self.current and type(value)is str,'known key and immutable string value');deps={};allparents=[(key,self.current[key])]+list(parents)
        for k,v in allparents:
            need(k in self.history and type(v)is int and v in self.history[k],'published retained predecessor')
            r=self.history[k][v]
            if v:deps[k]=max(deps.get(k,0),v)
            for j,u in r.deps:deps[j]=max(deps.get(j,0),u)
        need(all(v<=self.current[k] for k,v in deps.items()),'dependencies already published')
        v=self.current[key]+1;r=Version(key,v,value,tuple(deps.items()));self.history[key][v]=r;self.current[key]=v;return r
    def begin(self):
        self.serial+=1;tx=self.serial;self.active.add(tx);return tx
    def latest_and_pin(self,tx,key):
        need(tx in self.active and key in self.current,'active transaction and key')
        need((tx,key)not in self.pins,'one first read per key')
        v=self.current[key];self.pins[tx,key]=v;return self.history[key][v]
    def exact(self,tx,key,version):
        need(tx in self.active and(tx,key)in self.pins,'active pinned handle')
        need(type(version)is int and self.pins[tx,key]<=version<=self.current[key],'forward target')
        return self.history[key].get(version)  # None means violated/missing retention.
    def close(self,tx):
        need(tx in self.active,'live transaction');self.active.remove(tx)
        for p in list(self.pins):
            if p[0]==tx:del self.pins[p]
    def gc(self,key):
        need(key in self.current,'known key');floor=self.current[key]
        for(t,k),v in self.pins.items():
            if k==key:floor=min(floor,v)
        removed=[]
        for v in list(self.history[key]):
            if v<floor:removed.append(v);del self.history[key][v]
        return dict(floor=floor,removed=removed)

class GetTransaction:
    def __init__(self,store,keys,context=None):
        self.store=store;self.keys=tuple(keys);need(len(set(self.keys))==len(self.keys)and all(k in store.current for k in self.keys),'requested keys')
        self.keyset=set(self.keys);self.first={};self.results={};self.targets={};self.remaining=set();self.phase='FIRST';self.context={}if context is None else context
        need(all(k in store.current and type(v)is int and 0<=v<=store.current[k]for k,v in self.context.items()),'cluster satisfies prior context')
        self.tx=store.begin()if self.keys else None
        if not self.keys:self.phase='DONE'
    def first_read(self,key):
        need(self.phase=='FIRST'and key in self.keyset and key not in self.first,'first collection slot')
        r=self.store.latest_and_pin(self.tx,key);self.first[key]=r;return r
    def plan(self):
        need(self.phase=='FIRST'and len(self.first)==len(self.keys),'complete first round')
        wanted=set(self.keys);self.targets={k:r.number for k,r in self.first.items()};self.results=dict(self.first)
        for r in self.first.values():
            for k,v in r.deps:
                if k in wanted:self.targets[k]=max(self.targets[k],v)
        self.remaining={k for k in self.keys if self.targets[k]>self.first[k].number};self.phase='SECOND';return dict(self.targets)
    def second_read(self,key):
        need(self.phase=='SECOND'and key in self.remaining,'pending exact target')
        r=self.store.exact(self.tx,key,self.targets[key])
        if r is None:self.cancel('RETRY');return None
        need(r.key==key and r.number==self.targets[key],'exact identity');self.results[key]=r;self.remaining.remove(key);return r
    def finish(self):
        if self.phase=='DONE'and not self.keys:return dict(status='OK',values={},versions={})
        need(self.phase=='SECOND'and not self.remaining,'all required results')
        for r in self.results.values():
            for k,v in r.deps:
                if k in self.results:need(v<=self.results[k].number,'result dependency closure')
        updated=dict(self.context)
        for r in self.results.values():
            updated[r.key]=max(updated.get(r.key,0),r.number)
            for k,v in r.deps:updated[k]=max(updated.get(k,0),v)
        self.store.close(self.tx);self.context.clear();self.context.update(updated);self.phase='DONE'
        return dict(status='OK',values={k:r.value for k,r in self.results.items()},versions={k:r.number for k,r in self.results.items()})
    def cancel(self,status='CANCELLED'):
        need(status in ('RETRY','CANCELLED')and self.phase in ('FIRST','SECOND'),'cancellable transaction')
        self.store.close(self.tx);self.phase=status;self.results={};return dict(status=status)

def session_demo():
    nodes=[Replica(i,3)for i in range(3)];A,B,C=nodes;s=Session(3);events=[]
    def call(node,op,key,value=None):
        r=s.request(node,op,key,value);visible={k:v for k,v in r.items()if k!='write'};events.append(dict(at=node.origin,operation=op,key=key,result=visible,session=s.show()));return r
    a1=call(A,'write','x','draft')['write'];call(B,'read','x');B.receive(a1);call(B,'read','x');c1=C.create('z','side');b1=call(B,'write','y','reply')['write'];call(C,'write','z','done');events.append(dict(deliver='b1 to C before a1',status=C.receive(b1)));C.receive(a1);C.receive(b1);c2=call(C,'write','z','done')['write'];call(A,'read','z')
    for w in (b1,c1,c2):need(A.receive(w)=='APPLIED','catch-up order')
    call(A,'read','z');need(A.f==(1,1,2)and s.R==(1,1,2)and s.W==(1,1,2),'terminal tokens')
    # Four distinct failures when each specific gate is absent.
    failures={}
    for omitted in ('RYW','MR','MW','WFR'):
        X,Y,Z=[Replica(i,3)for i in range(3)];u=Session(3,set(('RYW','MR','MW','WFR'))-{omitted})
        if omitted=='RYW':u.request(X,'write','x','1');failures[omitted]=u.request(Y,'read','x')['value']is None
        elif omitted=='MR':X.create('x','1');u.request(X,'read','x');failures[omitted]=u.request(Y,'read','x')['value']is None
        elif omitted=='MW':u.request(X,'write','library','new');w=u.request(Y,'write','program','needs-new')['write'];Z.receive(w);failures[omitted]='program'in Z.latest and'library'not in Z.latest
        else:X.create('post','text');u.request(X,'read','post');w=u.request(Y,'write','reply','answer')['write'];Z.receive(w);failures[omitted]='reply'in Z.latest and'post'not in Z.latest
    need(all(failures.values()),'separate failed guarantees')
    return dict(events=events,final=[n.show()for n in nodes],missing_gate_failures=failures)

def two_round_demo():
    store=VersionStore(('x','y'));store.put('x','public');store.put('y','old');context={};t=GetTransaction(store,('x','y'),context);t.first_read('x');store.put('x','private');store.put('y','personal',[('x',2)]);t.first_read('y');targets=t.plan();store.put('y','sanitized');store.put('x','public',[('y',3)])
    latest_bad=dict(x=store.current['x'],y=t.first['y'].number);need(store.history['x'][3].dependencies()['y']>latest_bad['y'],'latest introduces dependency')
    gc_before={k:store.gc(k)for k in store.current};t.second_read('x');answer=t.finish();need(answer['versions']=={'x':2,'y':2},'exact second round');gc_after={k:store.gc(k)for k in store.current}
    chain=VersionStore(('x','z','y'))
    for k in chain.current:chain.put(k,'old')
    u=GetTransaction(chain,('x','z','y'));u.first_read('x');u.first_read('z');chain.put('x','new');chain.put('z','middle',[('x',2)]);chain.put('y','result',[('z',2)]);u.first_read('y');chain_targets=u.plan()
    for k in list(u.remaining):u.second_read(k)
    chain_answer=u.finish();need(chain_answer['versions']=={'x':2,'z':2,'y':2},'transitive closure')
    # Legal explicit cancellation preserves context and invalidates old handles.
    v=GetTransaction(chain,('x',),context);v.first_read('x');old_context=dict(context);v.cancel();need(context==old_context,'cancel cannot publish context');late=rejected(lambda:chain.exact(v.tx,'x',2))
    return dict(first_versions={'x':1,'y':2},targets=targets,latest_bad=latest_bad,gc_before=gc_before,answer=answer,context=context,gc_after=gc_after,chain_targets=chain_targets,chain_answer=chain_answer,cancelled_handle_error=late)

def tests():
    rng=random.Random(171717);counts=dict(session_events=0,accepted_reads=0,accepted_writes=0,dependency_waits=0,get_transactions=0,interleaved_puts=0,gc_steps=0)
    for _ in range(100):
        nodes=[Replica(i,3)for i in range(3)];sessions=[Session(3)for i in range(3)];messages=[]
        for step in range(160):
            if messages and rng.randrange(2):
                node=rng.choice(nodes);w=rng.choice(messages);status=node.receive(w);counts['dependency_waits']+=status=='WAIT'
            else:
                node=rng.choice(nodes);s=rng.choice(sessions);op=rng.choice(('read','write'));key=rng.choice(('x','y','z'));before=(s.R,s.W,node.f,len(node.applied));res=s.request(node,op,key,str(step))
                if res['status']=='INCOMPLETE':need(before==(s.R,s.W,node.f,len(node.applied)),'rejected request changes nothing')
                elif op=='write':messages.append(res['write']);counts['accepted_writes']+=1
                else:counts['accepted_reads']+=1
            for node in nodes:
                need(len(node.applied)==sum(node.f),'no prefix holes')
                for w in node.applied.values():need(covered(w.dep,node.f),'all disclosed deps applied')
            counts['session_events']+=1
    for _ in range(1200):
        keys=('x','y','z');store=VersionStore(keys)
        for k in keys:store.put(k,'initial')
        t=GetTransaction(store,keys);order=list(keys);rng.shuffle(order)
        for key in order:
            for __ in range(rng.randrange(5)):
                k=rng.choice(keys);parents=[(j,store.current[j])for j in keys if rng.randrange(2)];store.put(k,str(counts['interleaved_puts']),parents);counts['interleaved_puts']+=1
            t.first_read(key)
            for k in keys:store.gc(k);counts['gc_steps']+=1
        t.plan();targets=dict(t.targets)
        for k in list(t.remaining):
            for __ in range(rng.randrange(4)):
                j=rng.choice(keys);store.put(j,'later',[(z,store.current[z])for z in keys]);counts['interleaved_puts']+=1
            for j in keys:store.gc(j);counts['gc_steps']+=1
            need(t.second_read(k)is not None,'pin preserves exact version')
        result=t.finish();need(result['versions']==targets,'fixed targets');need(not store.pins and not store.active,'pins released');counts['get_transactions']+=1
    # Explicitly violated storage contract returns RETRY, never latest or context.
    store=VersionStore(('x','y'));store.put('x','old');store.put('y','old');context={};t=GetTransaction(store,('x','y'),context);t.first_read('x');store.put('x','new');store.put('y','dependent',[('x',2)]);t.first_read('y');t.plan();del store.history['x'][2]
    need(t.second_read('x')is None and t.phase=='RETRY'and not context and not store.pins,'missing history fails closed')
    need(GetTransaction(VersionStore(()),()).finish()==dict(status='OK',values={},versions={}),'empty result')
    return counts

def main():
    print(json.dumps(dict(status='PASS',session=session_demo(),two_round=two_round_demo(),checks=tests()),ensure_ascii=False,indent=2))
if __name__=='__main__':main()