"""Original bounded in-memory LSM visibility/compaction exercise; no database or files opened."""
from itertools import product
import json

def validate(rows,oldest,older_keys):
 if type(rows) is not list or len(rows)>32 or any(type(r) is not tuple or len(r)!=3 or type(r[0]) is not str or not r[0] or len(r[0])>8 or type(r[1]) is not int or not 0<=r[1]<=100 or (r[2] is not None and type(r[2]) is not str) for r in rows):raise ValueError('bounded key/sequence/value rows')
 if len({(k,s) for k,s,v in rows})!=len(rows) or rows!=sorted(rows,key=lambda r:(r[0],-r[1])):raise ValueError('unique key-sequence and key asc/sequence desc order')
 if type(oldest) is not int or not 0<=oldest<=100 or type(older_keys) is not set or not older_keys<=set(k for k,s,v in rows):raise ValueError('explicit valid oldest snapshot and known possible-older keys')

def read_at(rows,key,snapshot):
 chosen=[(s,v) for k,s,v in rows if k==key and s<=snapshot]
 return max(chosen,key=lambda r:r[0])[1] if chosen else None

def independent_read(rows,key,snapshot):
 result=None;latest=-1
 for k,s,v in rows:
  if k==key and s<=snapshot and s>latest:result=v;latest=s
 return result

def compact(rows,oldest,older_keys):
 validate(rows,oldest,older_keys);out=[];trace=[];previous_key=None;previous_seq=101
 for k,s,v in rows:
  if k!=previous_key:previous_seq=101
  if previous_seq<=oldest:drop=True;reason='hidden_by_previous_version_visible_to_all_allowed_snapshots'
  elif v is None and s<=oldest and k not in older_keys:drop=True;reason='obsolete_tombstone_no_older_level'
  else:drop=False;reason='retain_version_or_tombstone'
  trace.append({'key':k,'seq':s,'value':v,'previous_seq':previous_seq,'drop':drop,'reason':reason})
  if not drop:out.append((k,s,v))
  previous_key=k;previous_seq=s
 return out,trace

def demo():
 full=[('a',3,None),('a',1,'old'),('a',0,'ancient'),('b',4,'new'),('b',2,'beta'),('c',5,'gamma')]
 kept,trace=compact(full,2,set());assert ('a',0,'ancient') not in kept and ('a',1,'old') in kept and ('a',3,None) in kept
 checks=[]
 for snap,key in product(range(2,6),['a','b','c']):
  before=independent_read(full,key,snap);after=read_at(kept,key,snap);assert before==after;checks.append({'snapshot':snap,'key':key,'before':before,'after':after})
 wrong_no_tombs=[r for r in full if r[2] is not None];assert read_at(wrong_no_tombs,'a',5)=='old' and read_at(full,'a',5) is None
 newest=[next(r for r in full if r[0]==k) for k in ['a','b','c']];assert read_at(newest,'a',2) is None and read_at(full,'a',2)=='old'
 partial=[('a',3,None),('a',1,'old')];older=[('a',0,'ancient')];partial_cases=[];total=0
 for exists in [False,True]:
  outside=older if exists else []
  for oldest in range(6):
   output,events=compact(partial,oldest,{'a'} if exists else set())
   for snap in range(oldest,6):assert read_at(output+outside,'a',snap)==independent_read(partial+outside,'a',snap);total+=1
   partial_cases.append({'older_level_has_a':exists,'oldest':oldest,'kept':output,'trace':events})
 assert compact(partial,3,set())[0]==[] and compact(partial,3,{'a'})[0]==[('a',3,None)]
 rejected=[]
 for label,rows,old,olderkeys in [('duplicate_key_seq',[('a',1,'x'),('a',1,'y')],0,set()),('unsorted',list(reversed(full)),2,set()),('negative_seq',[('a',-1,'x')],0,set()),('bool_snapshot',full,True,set()),('unknown_older_flag',full,2,{'z'}),('over_budget',[('a',i,'x') for i in range(33)],0,set())]:
  try:compact(rows,old,olderkeys)
  except ValueError:rejected.append(label)
  else:raise AssertionError('expected rejection')
 assert compact([],0,set())[0]==[]
 return {'original_sorted_history':full,'oldest_live_snapshot':2,'compacted':kept,'trace':trace,'all_key_snapshot_read_checks':checks,'wrong_drop_all_tombstones_current_a':'old','correct_current_a':None,'wrong_newest_only_snapshot2_a':None,'correct_snapshot2_a':'old','partial_compaction_cases':partial_cases,'partial_snapshot_oracles':total,'invalid_cases_rejected':rejected,'scope':'own bounded in-memory key/seq visibility model; None means logical absent/deleted only, not physical erasure; no LevelDB/SSTable/durability/privacy/IO/performance or Opera architecture claim'}
if __name__=='__main__':print(json.dumps(demo(),ensure_ascii=True,indent=2))
