#!/usr/bin/env python3
"""Finite fixed-key snapshot/certification models; no persistence or DBMS claim."""
from dataclasses import dataclass,field
from collections import deque
from itertools import permutations,product
from copy import deepcopy
import json,random

def require(condition,message):
 if not condition:raise RuntimeError(message)
@dataclass
class Tx:
 name:str
 start:int
 end:int=0
 reads:dict=field(default_factory=dict)
 writes:dict=field(default_factory=dict)
 state:str='ACTIVE'

def dependency_graph(records,U):
 """Records already in increasing publication order; None creator means initial."""
 versions={x:[(0,None)] for x in range(U)};byname={t.name:t for t in records};edges={}
 require(len(byname)==len(records),'unique transaction identity')
 last=0
 for t in records:
  require(t.end>last and t.start<t.end,'commit-ordered records');last=t.end
  for x in t.writes:versions[x].append((t.end,t.name))
 def edge(a,b,kind,key):
  if a is not None and a!=b:edges.setdefault((a,b),[]).append((kind,key))
 for x,chain in versions.items():
  for (a,old),(b,new) in zip(chain,chain[1:]):edge(old,new,'ww',x)
 for t in records:
  for x,(stamp,creator,value) in t.reads.items():
   chain=versions[x];require((stamp,creator) in chain,'read source must exist')
   edge(creator,t.name,'wr',x)
   for c,writer in chain:
    if c>stamp:edge(t.name,writer,'rw',x)
 return edges

def topological(records,edges):
 names=[t.name for t in records];adj={n:[] for n in names};degree={n:0 for n in names}
 for a,b in edges:adj[a].append(b);degree[b]+=1
 todo=deque(n for n in names if degree[n]==0);order=[]
 while todo:
  u=todo.popleft();order.append(u)
  for v in adj[u]:
   degree[v]-=1
   if degree[v]==0:todo.append(v)
 return order if len(order)==len(names) else None

def concurrent(a,b):return a.start<b.end and b.start<a.end

def dangerous(records,edges,filters=True):
 """Adjacent concurrent rw edges; endpoints may be the same transaction."""
 by={t.name:t for t in records};incoming={t.name:[] for t in records};outgoing={t.name:[] for t in records}
 for (a,b),labels in edges.items():
  if any(k=='rw' for k,x in labels) and concurrent(by[a],by[b]):outgoing[a].append(b);incoming[b].append(a)
 for pivot in by:
  for left in incoming[pivot]:
   for right in outgoing[pivot]:
    I,P,O=by[left],by[pivot],by[right]
    if filters and not(O.end<=I.end and O.end<P.end):continue
    if filters and not I.writes and not O.end<I.start:continue
    return {'in':left,'pivot':pivot,'out':right,'left_keys':[x for k,x in edges[left,pivot] if k=='rw'],'right_keys':[x for k,x in edges[pivot,right] if k=='rw'],'in_read_only':not bool(I.writes),'out_commit':O.end,'in_start':I.start}
 return None

class Database:
 def __init__(self,U=16,initial=None,policy='ssi'):
  require(policy in ('si','occ','ssi'),'known policy');self.U=U;self.clock=0;self.policy=policy;initial={} if initial is None else initial
  self.versions={x:[(0,None,initial.get(x))] for x in range(U)};self.transactions={};self.committed=[];self.events=[]
 def tick(self):self.clock+=1;return self.clock
 def begin(self,name):
  require(name not in self.transactions,'new attempt needs new identity');t=Tx(name,self.tick());self.transactions[name]=t;self.events.append({'event':'begin','T':name,'start':t.start});return t
 def active(self,name):
  t=self.transactions[name];require(t.state=='ACTIVE','ordinary operation requires active transaction');return t
 def read(self,name,x):
  t=self.active(name);require(x in self.versions,'key in finite universe')
  if x in t.writes:return t.writes[x]
  row=next(v for v in reversed(self.versions[x]) if v[0]<t.start)
  if x in t.reads:require(t.reads[x]==row,'fixed snapshot reread')
  t.reads[x]=row;return row[2]
 def scan(self,name,a,b):
  require(0<=a<=b<=self.U,'half-open range');out={}
  for x in range(a,b):
   v=self.read(name,x)
   if v is not None:out[x]=v
  return out
 def write(self,name,x,value):
  t=self.active(name);require(x in self.versions,'key in finite universe');t.writes[x]=value
 def commit(self,name):
  # Validation and all publications are one indivisible manager event.
  t=self.active(name);t.end=self.tick();certificate=None;reason=None
  checks=t.reads if self.policy=='occ' else t.writes
  for x in checks:
   if self.versions[x][-1][0]>t.start:
    reason='STALE_READ' if self.policy=='occ' else 'WRITE_WRITE'
    certificate={'key':x,'snapshot':t.start,'read_source':t.reads.get(x),'latest':self.versions[x][-1]};break
  if reason is None and self.policy=='ssi':
   candidate=self.committed+[t];certificate=dangerous(candidate,dependency_graph(candidate,self.U))
   if certificate:reason='DANGEROUS_STRUCTURE'
  if reason:
   t.state='ABORTED';event={'event':'abort','T':name,'at':t.end,'reason':reason,'certificate':certificate}
  else:
   t.state='COMMITTED'
   for x,value in t.writes.items():self.versions[x].append((t.end,t.name,value))
   self.committed.append(t);event={'event':'commit','T':name,'at':t.end}
  self.events.append(event);return event
 def state(self):return {x:rows[-1][2] for x,rows in self.versions.items() if rows[-1][2] is not None}
 def graph(self):return dependency_graph(self.committed,self.U)

def serial_witness(records,U,initial,order):
 """Independent semantics of a proposed serial order, including all per-key writer orders."""
 by={t.name:t for t in records};current={x:(0,None,initial.get(x)) for x in range(U)};writers={x:[] for x in range(U)}
 for name in order:
  t=by[name]
  for x,row in t.reads.items():
   if current[x]!=row:return False
  for x,value in t.writes.items():current[x]=(t.end,t.name,value);writers[x].append(name)
 return all(writers[x]==[t.name for t in records if x in t.writes] for x in range(U))

def skew(policy,range_case=False):
 db=Database(initial={} if range_case else {0:1,1:1},policy=policy)
 for name,key in [('A',11 if range_case else 0),('B',12 if range_case else 1)]:
  db.begin(name)
  if range_case:
   require(db.scan(name,10,13)=={},'initial empty range');db.write(name,key,1)
  else:
   require(db.read(name,0)==db.read(name,1)==1,'initial paired state');db.write(name,key,0)
 for name in ['A','B']:db.commit(name)
 if policy=='si':require(len(db.committed)==2 and topological(db.committed,db.graph()) is None,'SI skew cycle')
 else:
  require(len(db.committed)==1,'validator excludes double decision');db.begin('B_retry')
  if range_case:require(db.scan('B_retry',10,13)=={11:1},'fresh retry sees occupied')
  else:require(db.read('B_retry',0)==0,'fresh retry sees changed basis')
  require(db.commit('B_retry')['event']=='commit','clean retry can finish')
 return {'policy':policy,'range':range_case,'events':db.events,'final':db.state()}

def readonly_three(policy,read_y=True):
 db=Database(U=2,initial={0:0,1:0},policy=policy)
 db.begin('P');db.read('P',1);db.begin('Q');db.write('Q',1,1);db.commit('Q')
 db.begin('R');db.read('R',0)
 if read_y:db.read('R',1)
 db.commit('R');db.write('P',0,1);last=db.commit('P')
 if policy=='si':
  require((topological(db.committed,db.graph()) is None)==read_y,'read-only edge closes exactly the cycle')
 else:require(last['event']=='abort','last updater rejected')
 return db

def exhaustive_histories():
 # Three transactions, each fixed read/write key sets; all six boundary events.
 orders=[o for o in permutations(range(6)) if all(o.index(2*i)<o.index(2*i+1) for i in range(3))]
 masks=[set(x for x in range(2) if i>>x&1) for i in range(4)]
 histories=0;cycles=0;false_positives=0;serial_checks=0;ssi_accepts=0
 for spec in product(range(4),repeat=6):
  reads=[masks[spec[2*i]] for i in range(3)];writes=[masks[spec[2*i+1]] for i in range(3)]
  # All 4096 R/W shapes crossed with all 90 begin/commit boundary orders.
  selected=orders
  for order in selected:
   db=Database(2,{0:0,1:0},'si')
   for event in order:
    i=event//2;name=str(i)
    if event%2==0:
     db.begin(name)
     for x in reads[i]:db.read(name,x)
     for x in writes[i]:db.write(name,x,i+1)
    else:db.commit(name)
   edges=db.graph();top=topological(db.committed,edges);cert=dangerous(db.committed,edges)
   witness=any(serial_witness(db.committed,2,{0:0,1:0},o) for o in permutations([t.name for t in db.committed]))
   require((top is not None)==witness,'graph/independent serial semantics');serial_checks+=1
   if top is None:require(cert is not None,'SI cycle has ordered/readonly-filtered danger');cycles+=1
   elif cert:false_positives+=1
   histories+=1
   safe=Database(2,{0:0,1:0},'ssi')
   for event in order:
    i=event//2;name=str(i)
    if event%2==0:
     safe.begin(name)
     for x in reads[i]:safe.read(name,x)
     for x in writes[i]:safe.write(name,x,i+1)
    else:
     safe.commit(name);g=safe.graph();require(topological(safe.committed,g) is not None and dangerous(safe.committed,g) is None,'SSI accepted prefix invariant')
   ssi_accepts+=len(safe.committed)
 return {'SI_histories':histories,'independent_serial_witness_checks':serial_checks,'SI_cycles':cycles,'acyclic_dangerous_structures':false_positives,'SSI_successful_commits':ssi_accepts}

def random_occ():
 rng=random.Random(81009);histories=0;events=0;rejects=0
 for trial in range(1000):
  U=5;initial={x:rng.randrange(3) for x in range(U)};db=Database(U,initial,'occ');next_name=0
  for step in range(35):
   active=[t.name for t in db.transactions.values() if t.state=='ACTIVE']
   if not active or (next_name<12 and rng.randrange(4)==0):db.begin(str(next_name));next_name+=1
   else:
    name=rng.choice(active);op=rng.randrange(3);x=rng.randrange(U)
    if op==0:db.read(name,x)
    elif op==1:db.write(name,x,rng.choice([None,0,1,2]))
    else:rejects+=db.commit(name)['event']=='abort'
   events+=1
  for t in db.transactions.values():
   if t.state=='ACTIVE':rejects+=db.commit(t.name)['event']=='abort';events+=1
  require(serial_witness(db.committed,U,initial,[t.name for t in db.committed]),'OCC commit-order serial witness');histories+=1
 return {'histories':histories,'events':events,'aborts':rejects}

def boundary_tests():
 blind={}
 for policy in ('si','occ','ssi'):
  db=Database(1,{0:0},policy);db.begin('C');db.begin('D');db.write('C',0,7);db.write('D',0,9);db.commit('C');db.commit('D')
  require(len(db.committed)==(2 if policy=='occ' else 1),'blind write rules differ');blind[policy]=db.state()[0]
 # Split validation is explicitly a faulty model: both inspect unchanged bases, then publish.
 db=Database(2,{0:1,1:1},'occ')
 for name,key in [('A',0),('B',1)]:db.begin(name);db.read(name,0);db.read(name,1);db.write(name,key,0)
 passes=[all(db.versions[x][-1][0]<t.start for x in t.reads) for t in db.transactions.values()]
 require(passes==[True,True],'both premature checks pass');wrong={0:0,1:0};require(sum(wrong.values())==0,'split-publish violates business constraint')
 # A no-active-updater snapshot and updates beginning later: readonly filter removes false structure.
 db=Database(2,{0:0,1:0},'ssi');db.begin('R');db.read('R',0);db.begin('P');db.read('P',1);db.begin('Q');db.write('Q',1,1);db.commit('Q');db.write('P',0,1);db.commit('P');db.commit('R')
 require(len(db.committed)==3 and dangerous(db.committed,db.graph(),False) is not None and dangerous(db.committed,db.graph()) is None,'safe snapshot temporal filter')
 A=Tx('A',2,3,{}, {0:1,1:1},'COMMITTED');B=Tx('B',1,4,{0:(0,None,0)}, {1:2},'COMMITTED');C=Tx('C',5,6,{}, {0:3,1:3},'COMMITTED')
 require(topological([A,B,C],dependency_graph([A,B,C],2)) is None,'fixed-version graph rejects')
 # Serial B,A,C preserves external reads and final writers but reorders y's intermediate versions.
 state={0:(None,0),1:(None,0)}
 for t in (B,A,C):
  for x,(_,source,value) in t.reads.items():require(state[x]==(source,value),'alternative view read source')
  for x,v in t.writes.items():state[x]=(t.name,v)
 require(state=={0:('C',3),1:('C',3)},'alternative view final writers')
 return {'fixed_order_cycle_has_other_view':['B','A','C'],'blind_write_final':blind,'faulty_separate_validation_passes':passes,'faulty_final':wrong,'safe_snapshot_events':db.events}

def main():
 result={'result':'PASS','write_skew':[skew(p) for p in ('si','occ','ssi')],'empty_range':[skew(p,True) for p in ('si','occ','ssi')],'readonly_true_cycle':readonly_three('si').events,'readonly_false_positive':{'si':readonly_three('si',False).events,'ssi':readonly_three('ssi',False).events},'boundaries':boundary_tests(),'finite_histories':exhaustive_histories(),'random_OCC':random_occ()}
 print(json.dumps(result,ensure_ascii=False,indent=2))
if __name__=='__main__':main()