#!/usr/bin/env python3
"""CS08 finite service-interface checker, standard library only.
Run: python foundations-service-checker.py
Not a complete Raft implementation: no network scheduler, elections liveness,
Byzantine behavior, real filesystems, or exhaustive implementation verification.
"""
from dataclasses import dataclass, field, asdict
from copy import deepcopy
from itertools import product, combinations, permutations
import json

@dataclass
class SessionMachine:
    balance: int = 100
    applied: int = 0
    sessions: dict = field(default_factory=lambda: {'s': (0, None, None)})
    effects: list = field(default_factory=list)

    def apply(self, index, sid, seq, delta):
        if index != self.applied + 1:
            raise ValueError('application must advance exactly one log slot')
        self.applied = index
        if sid not in self.sessions:
            return ('SESSION_EXPIRED_OR_UNKNOWN', None)
        high, old_delta, result = self.sessions[sid]
        if seq < high:
            return ('RESULT_FORGOTTEN', None)
        if seq == high:
            return ('OK', result) if delta == old_delta else ('IDENTITY_CONFLICT', None)
        if seq > high + 1:
            return ('SEQUENCE_GAP', None)
        self.balance += delta
        self.sessions[sid] = (seq, delta, self.balance)
        self.effects.append((sid, seq, delta))
        return ('OK', self.balance)

    def expire(self, index, sid):
        if index != self.applied + 1:
            raise ValueError('expiry must be ordered')
        self.applied = index
        self.sessions.pop(sid, None)

    def snapshot(self):
        return deepcopy(self)

@dataclass
class Entry:
    term: int
    operation: str
    value: int

@dataclass
class Replica:
    term: int = 4
    log: list = field(default_factory=list)
    commit: int = 0
    applied: int = 0
    value: int = 0

    def stamp(self):
        return (self.log[-1].term, len(self.log)) if self.log else (0, 0)

class Cluster:
    """Three replicas; explicit events, complete-log replication only.
    Election event is atomic for the supplied voter set. No timeout simulation.
    Nodes can retain an old local leadership belief after a new election.
    """
    def __init__(self):
        self.nodes = {x: Replica() for x in 'ABC'}
        self.leaders = {'A': 4}
        self.trace = []
        self.probes = {}

    def record(self, event):
        self.trace.append({'event': event, 'nodes': {
            k: {'term': v.term, 'log': [(e.term,e.operation,e.value) for e in v.log],
                'commit':v.commit,'applied':v.applied,'value':v.value}
            for k,v in self.nodes.items()}})

    def elect(self, who, term, voters):
        voters = set(voters)
        if who not in voters or len(voters) < 2:
            raise ValueError('not an election majority')
        candidate = self.nodes[who]
        if any(term <= self.nodes[v].term or candidate.stamp() < self.nodes[v].stamp()
               for v in voters):
            raise ValueError('term or log election restriction violated')
        for v in voters:
            self.nodes[v].term = term
        self.leaders[who] = term
        self.record(('elect',who,term,sorted(voters)))

    def append(self, who, operation, value=0):
        n = self.nodes[who]
        if self.leaders.get(who) != n.term:
            raise ValueError('not locally a leader')
        n.log.append(Entry(n.term, operation, value))
        self.record(('append',who,operation,value))
        return len(n.log)

    def replicate(self, who, target):
        a,b = self.nodes[who],self.nodes[target]
        if self.leaders.get(who) != a.term:
            raise ValueError('replication requires local leadership')
        if a.term < b.term:
            a.term=b.term
            self.record(('replicate-reject',who,target))
            return False
        if a.log[:b.commit] != b.log[:b.commit]:
            raise ValueError('would overwrite committed prefix')
        b.term = a.term
        b.log = deepcopy(a.log)
        b.commit = max(b.commit,min(a.commit,len(b.log)))
        self.record(('replicate',who,target))
        return True

    def commit(self, who, index, acknowledgers):
        n=self.nodes[who]; voters=set(acknowledgers)
        if self.leaders.get(who) != n.term:
            raise ValueError('commit requires local leadership')
        if index <= n.commit or index > len(n.log):
            raise ValueError('invalid commit index')
        if n.log[index-1].term != n.term:
            raise ValueError('direct commit needs current term')
        valid=[v for v in voters if self.nodes[v].term==n.term
               and self.nodes[v].log[:index]==n.log[:index]]
        if len(valid)<2:
            raise ValueError('no durable current-term majority')
        n.commit=index
        self.record(('commit',who,index,valid))

    def apply(self, who):
        n=self.nodes[who]
        while n.applied<n.commit:
            e=n.log[n.applied]
            if e.operation=='set': n.value=e.value
            elif e.operation=='add': n.value+=e.value
            elif e.operation!='noop': raise ValueError('unknown operation')
            n.applied+=1
        self.record(('apply',who))

    def begin_read(self, who, context):
        n=self.nodes[who]
        if self.leaders.get(who) != n.term:
            raise ValueError('leader read must start on a local leader')
        if context in self.probes: raise ValueError('read context must be fresh')
        anchored=any(e.term==n.term for e in n.log[:n.commit])
        self.probes[context]={'who':who,'term':n.term,'index':n.commit,
                              'anchored':anchored,'acks':set()}
        self.record(('read-start',who,context))

    def confirm(self, context, voter):
        p=self.probes[context]
        if self.nodes[voter].term > p['term']:
            origin=self.nodes[p['who']]
            origin.term=max(origin.term,self.nodes[voter].term)
        if self.nodes[voter].term <= p['term']:
            self.nodes[voter].term=p['term']
            p['acks'].add(voter)
        self.record(('read-confirm',context,voter))

    def read(self, context):
        p=self.probes[context];n=self.nodes[p['who']]
        if not p['anchored']: return ('BLOCKED_NO_CURRENT_TERM_COMMIT',None)
        if n.term != p['term'] or self.leaders.get(p['who'])!=n.term or len(p['acks'])<2: return ('BLOCKED_NO_QUORUM',None)
        if n.applied<p['index']: return ('BLOCKED_APPLY',None)
        return ('OK',n.value)

def majority(config, responders):
    return len(set(config)&set(responders))*2 > len(set(config))

def joint(old,new,responders):
    return majority(old,responders) and majority(new,responders)

@dataclass
class FencedResource:
    highest: int = 40
    value: int = 0
    def action(self, epoch, value=None):
        if epoch < self.highest: return False
        self.highest=epoch
        if value is not None: self.value=value
        return True

@dataclass
class Outbox:
    source_committed: bool = False
    sent: list = field(default_factory=list)
    marked: bool = False
    ack: bool = False
    seen: set = field(default_factory=set)
    stock: int = 5
    deliveries: int = 0
    deduplicate: bool = True
    def event(self, event):
        if event=='commit': self.source_committed=True
        elif event=='send':
            if not self.source_committed or self.marked: raise ValueError('no pending intention')
            self.sent.append('m7')
        elif event in ('deliver','deliver_ack'):
            if not self.sent: raise ValueError('no sent message')
            mid=self.sent.pop(0);self.deliveries+=1
            if not self.deduplicate or mid not in self.seen:
                self.stock-=1;self.seen.add(mid)
            if event=='deliver_ack': self.ack=True
        elif event=='mark':
            if not self.ack: raise ValueError('mark needs receiver confirmation')
            self.marked=True
        elif event=='relay_crash': self.ack=False
        else: raise ValueError(event)

def linearization(operations, initial):
    """Exhaustive witness for completed add/read histories, small inputs only."""
    for order in permutations(operations):
        position={op['id']:i for i,op in enumerate(order)}
        if any(x['end'] < y['start'] and position[x['id']]>position[y['id']]
               for x in operations for y in operations): continue
        value=initial; valid=True
        for op in order:
            if op['op']=='add':value+=op['arg']
            elif op['op']!='read':raise ValueError('bad history op')
            if op['result'] != value: valid=False;break
        if valid:return [op['id'] for op in order]
    return None

def install_suffix(local, included_index, included_term, applied):
    """local maps global index -> term. Snapshot bytes already validated."""
    if included_index <= applied: return ('IGNORE_OLD',dict(local),applied)
    matches=local.get(included_index)==included_term
    suffix={i:t for i,t in local.items() if i>included_index} if matches else {}
    return ('INSTALL',suffix,included_index)

def publication_trace(events, crash_after):
    """Two snapshot generations; old root=10, new root=12, log entries11/12.
    Crash loses unstabilized temporary bytes. Root publish is an atomic primitive.
    """
    stable_new=False;written=False;root=10;log={11,12}
    for e in events[:crash_after]:
        if e=='write':written=True
        elif e=='flush':stable_new=written
        elif e=='publish':root=12
        elif e=='trim':log.clear()
        else:raise ValueError(e)
    recoverable=(root==12 and stable_new) or (root==10 and {11,12}<=log)
    return dict(root=root,stable_new=stable_new,log=sorted(log),recoverable=recoverable)

def self_test():
    checks=0
    def check(condition):
        nonlocal checks
        checks+=1
        if not condition:raise AssertionError('check '+str(checks))
    examples={}
    m=SessionMachine(); responses=[]
    for q,d in [(1,10),(1,10),(2,5),(1,10)]:
        responses.append(m.apply(m.applied+1,'s',q,d))
    check(m.balance==115 and m.applied==4 and len(m.effects)==2)
    check(responses==[('OK',110),('OK',110),('OK',115),('RESULT_FORGOTTEN',None)])
    examples['session']={'responses':responses,'state':asdict(m)}
    # All length-five request words on three sequence numbers. Every prefix is
    # snapshotted and replayed; results, values, state and effects must match.
    for seqs in product([1,2,3],repeat=5):
        original=SessionMachine();prefixes=[original.snapshot()]
        for seq in seqs:
            original.apply(original.applied+1,'s',seq,seq*5)
            check(len({(s,q) for s,q,d in original.effects})==len(original.effects))
            prefixes.append(original.snapshot())
        for cut,prefix in enumerate(prefixes):
            restored=prefix.snapshot()
            for seq in seqs[cut:]:restored.apply(restored.applied+1,'s',seq,seq*5)
            check(asdict(restored)==asdict(original))
    expiry=SessionMachine();expiry.apply(1,'s',1,10);expiry.expire(2,'s')
    check(expiry.apply(3,'s',1,10)==('SESSION_EXPIRED_OR_UNKNOWN',None) and expiry.balance==110)
    broken=SessionMachine();broken.apply(1,'s',1,10);broken.sessions={'s':(0,None,None)}
    broken.apply(2,'s',1,10);check(broken.balance==120)
    examples['wrong_active_session_reset']=asdict(broken)
    cluster=Cluster();cluster.append('A','noop');cluster.replicate('A','B')
    cluster.commit('A',1,'AB');cluster.apply('A');cluster.replicate('A','C')
    cluster.elect('B',5,'BC');cluster.append('B','set',1);cluster.replicate('B','C')
    cluster.commit('B',2,'BC')
    cluster.begin_read('A','old');cluster.confirm('old','A');cluster.confirm('old','B')
    check(cluster.read('old')[0]=='BLOCKED_NO_QUORUM')
    cluster.begin_read('B','new');cluster.confirm('new','B');cluster.confirm('new','C')
    check(cluster.read('new')[0]=='BLOCKED_APPLY')
    cluster.apply('B');check(cluster.read('new')==('OK',1))
    examples['read']={'old_local_unsafe':cluster.nodes['A'].value,'old_safe':cluster.read('old'),
                      'new_safe':cluster.read('new'),'trace':cluster.trace}
    # Regression: new leader may not yet know the receiver's committed prefix.
    monotonic=Cluster();monotonic.append('A','noop');monotonic.replicate('A','B')
    monotonic.commit('A',1,'AB');monotonic.replicate('A','C')
    monotonic.elect('B',5,'BC');monotonic.append('B','noop');monotonic.replicate('B','C')
    check(monotonic.nodes['C'].commit==1)
    # Follower with a committed current-term entry cannot initiate a leader probe.
    monotonic.commit('B',2,'BC');monotonic.replicate('B','A');monotonic.apply('A')
    monotonic.append('B','set',1);monotonic.replicate('B','C');monotonic.commit('B',3,'BC');monotonic.apply('B')
    rejected=False
    try:monotonic.begin_read('A','follower')
    except ValueError:rejected=True
    check(rejected)
    # An observed higher-term response aborts even an already-confirmed probe.
    lost=Cluster();lost.append('A','noop');lost.replicate('A','B');lost.commit('A',1,'AB');lost.apply('A')
    lost.begin_read('A','q');lost.confirm('q','A');lost.confirm('q','B')
    check(lost.read('q')==('OK',0))
    lost.elect('B',5,'BC');lost.confirm('q','B')
    check(lost.read('q')[0]=='BLOCKED_NO_QUORUM' and lost.nodes['A'].term==5)
    history=[dict(id='w1',op='add',arg=10,result=110,start=1,end=5),
             dict(id='w2',op='add',arg=5,result=115,start=6,end=9)]
    check(linearization(history,100)==['w1','w2'])
    stale=[dict(id='w',op='add',arg=1,result=1,start=1,end=5),
           dict(id='r',op='read',arg=0,result=0,start=6,end=9)]
    check(linearization(stale,0) is None)
    examples['history']={'success_witness':linearization(history,100),'stale_witness':linearization(stale,0)}
    safe_events=['write','flush','publish','trim']
    safe=[publication_trace(safe_events,k) for k in range(5)]
    for state in safe:check(state['recoverable'])
    unsafe=[publication_trace(['trim','write','flush','publish'],k) for k in range(5)]
    check(not unsafe[1]['recoverable'])
    examples['snapshot_publication']={'safe':safe,'unsafe_trim_first':unsafe}
    check(install_suffix({11:5,12:5,13:6},12,5,10)==('INSTALL',{13:6},12))
    check(install_suffix({11:4,12:4,13:6},12,5,10)==('INSTALL',{},12))
    check(install_suffix({13:6,14:6},12,5,14)[0]=='IGNORE_OLD')
    valid=[]
    for size in range(6):
        for q in combinations('ABCDE',size):
            if joint('ABC','CDE',q):
                valid.append(''.join(q))
                for old in combinations('ABC',2):check(bool(set(old)&set(q)))
                for new in combinations('CDE',2):check(bool(set(new)&set(q)))
    check(joint('ABC','CDE','ACD') and not joint('ABC','CDE','ABD'))
    examples['joint_quorums']=valid
    for epochs in product([40,41,42],repeat=4):
        resource=FencedResource(); max_accepted=40
        for e in epochs:
            accepted=resource.action(e,e)
            check(accepted==(e>=max_accepted))
            if accepted:max_accepted=e
            check(resource.highest==max_accepted)
    resource=FencedResource();resource.action(41);resource.action(42);resource.action(42,9)
    stale_accept=resource.action(41,8);check(not stale_accept and resource.value==9)
    examples['fencing']={'old_accepted':stale_accept,**asdict(resource)}
    events=['commit','send','deliver','relay_crash','send','deliver_ack','mark']
    for dedup in [False,True]:
        out=Outbox(deduplicate=dedup);trace=[]
        for event in events:
            out.event(event);trace.append({'event':event,'stock':out.stock,'deliveries':out.deliveries,'marked':out.marked})
        check(out.stock==(4 if dedup else 3) and out.deliveries==2 and out.marked)
        examples['outbox_'+str(dedup)] = trace
    for copies in range(1,9):
        out=Outbox();out.event('commit')
        for _ in range(copies):out.event('send');out.event('deliver')
        check(out.stock==4 and out.deliveries==copies)
    return dict(status='passed',checks=checks,session_words=3**5,examples=examples,
                scope='Finite service interface models; not a complete Raft or durable storage implementation proof.')

if __name__=='__main__':
    print(json.dumps(self_test(),ensure_ascii=False,indent=2))