#!/usr/bin/env python3
"""Finite teaching interfaces for phi tail scores and ORIGINAL 2002 SWIM.
No sockets, authority changes, clock settings, or actual failure detection.
A fixed process-instance namespace is trusted; transport can lose/reorder/duplicate
messages. Probe acknowledgement identity is checked, but no cryptography is added.
Confirm is absorbing for this instance, regardless of incarnation. Full repair is
an explicit additional operation, not inferred from bounded piggyback attempts.
Exact Fraction window statistics; libm phi scores are approximate. stdout only.
"""
from fractions import Fraction as Q
from collections import deque
from dataclasses import dataclass,asdict
from copy import deepcopy
from math import erfc,sqrt,log10
from itertools import permutations,product
import json,random

def need(ok,why):
    if not ok:raise ValueError(why)
def rejected(call):
    try:call()
    except ValueError as e:return str(e)
    raise RuntimeError('required rejection missing')

class Phi:
    def __init__(self,window=4,scale_floor=1):
        need(type(window)is int and window>=2,'at least two intervals')
        need(type(scale_floor)is int and 1<=scale_floor<=10**9,'integer scale floor 1..1e9')
        self.W=window;self.floor=Q(scale_floor)
        self.gaps=deque();self.total=Q(0);self.squares=Q(0);self.seq=None;self.last=None;self.now=0
    def clock(self,t):
        need(type(t)is int and 0<=t<=10**9 and t>=self.now,'monotone bounded local tick');self.now=t
    def heartbeat(self,seq,t):
        need(type(seq)is int and seq>=0,'nonnegative sequence');self.clock(t)
        if self.seq is not None and seq<=self.seq:return dict(action='IGNORE_OLD',last=self.last,seq=self.seq)
        if self.last is not None:
            gap=t-self.last
            if len(self.gaps)==self.W:
                old=self.gaps.popleft();self.total-=old;self.squares-=old*old
            self.gaps.append(gap);self.total+=gap;self.squares+=gap*gap
        self.seq=seq;self.last=t
        return dict(action='ACCEPT',last=t,seq=seq,window=list(self.gaps))
    def query(self,t,thresholds=(1,2)):
        hs=tuple(thresholds);need(all(type(h)is int and 1<=h<=12 for h in hs),'threshold integers 1..12')
        self.clock(t)
        if len(self.gaps)<self.W:return dict(status='UNTRAINED',samples=len(self.gaps),decisions={h:'UNTRAINED'for h in hs})
        mu=self.total/self.W;v=self.squares/self.W-mu*mu;need(v>=0,'exact nonnegative variance')
        variance=max(v,self.floor*self.floor);lag=t-self.last;z=float(Q(lag)-mu)/sqrt(float(variance))
        out=dict(time=t,last=self.last,window=list(self.gaps),sum=self.total,sum_squares=self.squares,mean=mu,variance=v,effective_variance=variance,lag=lag,z=z)
        if z>=12:
            return dict(out,status='LOWER_BOUND',phi_lower_bound=30,strict=True,decisions={h:'SUSPECT'for h in hs})
        tail=erfc(z/sqrt(2))/2;need(tail>0,'no underflow in declared numeric branch')
        score=-log10(tail)
        decisions={h:('NUMERIC_BOUNDARY'if abs(score-h)<=1e-9 else'SUSPECT'if score>=h else'BELOW')for h in hs}
        return dict(out,status='APPROXIMATE',tail=tail,phi=score,decisions=decisions)

@dataclass(frozen=True)
class Update:
    member:str
    kind:str
    inc:object=None
    def __post_init__(self):
        need(self.kind in ('ALIVE','SUSPECT','CONFIRM'),'update kind')
        need(self.inc is None if self.kind=='CONFIRM'else type(self.inc)is int and self.inc>=0,'incarnation or canonical Confirm')
    def key(self):return (1,0,0)if self.kind=='CONFIRM'else(0,self.inc,int(self.kind=='SUSPECT'))

@dataclass
class State:
    info:Update
    deadline:object=None

class MemberView:
    def __init__(self,owner,members,suspicion=12,budget=2,batch=1):
        ids=tuple(members);need(len(ids)==len(set(ids)) and owner in ids,'fixed registered identities')
        need(all(type(x)is int and x>0 for x in (suspicion,budget,batch)),'positive timer/propagation limits')
        self.owner=owner;self.table={p:State(Update(p,'ALIVE',0))for p in ids};self.own_inc=0
        self.delay=suspicion;self.budget=budget;self.batch=batch;self.now=0
        self.pending={};self.order=0;self.timers=[];self.last_round=0;self.next_probe_at=0
    def clock(self,t):
        need(type(t)is int and t>=self.now,'monotone member clock');self.now=t
    def publish(self,u):
        self.order+=1;self.pending[u.member]=[u,0,self.order]
    def merge(self,u,t):
        need(isinstance(u,Update)and u.member in self.table,'registered update');self.clock(t)
        before=self.table[u.member].info
        if before.kind=='CONFIRM':return dict(action='IGNORE_TERMINAL',generated=None)
        # Only this member generates its incarnation. No crash-recovery reset.
        if u.member==self.owner and u.kind!='CONFIRM':
            need(u.inc<=self.own_inc,'unreachable future own incarnation')
            if u.kind=='SUSPECT'and u.inc==self.own_inc:
                self.own_inc+=1;fresh=Update(self.owner,'ALIVE',self.own_inc)
                self.table[self.owner]=State(fresh);self.publish(fresh)
                return dict(action='REFUTE',generated=fresh)
        if u.key()<=before.key():return dict(action='IGNORE_OLD_OR_DUPLICATE',generated=None)
        deadline=t+self.delay if u.kind=='SUSPECT'else None
        self.table[u.member]=State(u,deadline);self.publish(u)
        if deadline is not None:self.timers.append((u.member,u.inc,deadline))  # retained audit callbacks
        return dict(action='APPLY',generated=u)
    def fire(self,member,inc,deadline,t):
        need(member in self.table and type(deadline)is int and t>=deadline,'timer target/time');self.clock(t)
        state=self.table[member]
        if state.info.kind!='SUSPECT'or state.info.inc!=inc or state.deadline!=deadline:return dict(action='STALE_TIMER',generated=None)
        out=self.merge(Update(member,'CONFIRM'),t);out['confirmed_suspicion_inc']=inc;return out
    def expire(self,t):
        self.clock(t);due=[(p,s.info.inc,s.deadline)for p,s in self.table.items()if s.info.kind=='SUSPECT'and s.deadline<=t]
        return [self.fire(p,i,d,t)for p,i,d in due]
    def piggyback(self,t):
        self.clock(t);selected=sorted(self.pending,key=lambda p:(self.pending[p][1],self.pending[p][2]))[:self.batch];out=[]
        for p in selected:
            rec=self.pending[p];out.append(rec[0]);rec[1]+=1
            if rec[1]>=self.budget:del self.pending[p]
        return out
    def repair_from(self,other,t):
        need(set(self.table)==set(other.table),'same registered namespace');self.clock(t)
        incoming=[state.info for state in other.table.values()]
        return [self.merge(u,t)for u in incoming]
    def snapshot(self):
        return {p:dict(kind=s.info.kind,inc=s.info.inc,deadline=s.deadline)for p,s in self.table.items()}
    def candidates(self):return [p for p,s in self.table.items()if p!=self.owner and s.info.kind!='CONFIRM']

@dataclass(frozen=True)
class Packet:
    kind:str
    src:str
    dst:str
    origin:str
    round:int
    target:str

class Probe:
    def __init__(self,view,round,target,helpers,t,period=8,direct=2):
        need(type(round)is int and round>0 and type(period)is int and type(direct)is int and 0<direct<period,'probe timing/id')
        need(round>view.last_round and t>=view.next_probe_at,'fresh nonoverlapping probe round')
        view.clock(t);need(view.table[view.owner].info.kind!='CONFIRM','removed origin cannot initiate')
        need(target in view.candidates(),'eligible target')
        helpers=tuple(helpers);need(len(helpers)==len(set(helpers)) and all(p in view.table and p not in (view.owner,target) and view.table[p].info.kind!='CONFIRM'for p in helpers),'distinct helper identities')
        self.view=view;self.origin=view.owner;self.round=round;self.target=target;self.helpers=helpers
        self.start=t;self.direct=t+direct;self.end=t+period;self.indirect=False;self.done=False;self.acked=False;self.sent_ping=False
        view.last_round=round;view.next_probe_at=self.end
    def identity(self):return self.origin,self.round,self.target
    def ping(self):
        need(not self.sent_ping and self.view.last_round==self.round,'initial ping issued once in current round');self.sent_ping=True
        return Packet('PING',self.origin,self.target,*self.identity())
    def indirect_requests(self,t):
        self.view.clock(t);need(self.direct<=t<=self.end,'direct timeout within period')
        need(self.sent_ping,'initial ping before indirect phase')
        if self.done or self.acked or self.indirect or self.view.last_round!=self.round:return []
        self.indirect=True
        return [Packet('PING_REQ',self.origin,p,*self.identity())for p in self.helpers]
    def ack(self,msg,t):
        self.view.clock(t);need(isinstance(msg,Packet),'packet object')
        match=(msg.origin,msg.round,msg.target)==self.identity() and msg.dst==self.origin
        source_ok=(msg.kind=='ACK'and msg.src==self.target)or(msg.kind=='RELAY_ACK'and self.indirect and msg.src in self.helpers)
        if self.done or self.view.last_round!=self.round or t>self.end or not match or not source_ok:return 'IGNORE_ACK'
        self.acked=True;return 'ACCEPT_ACK'
    def finish(self,t):
        self.view.clock(t);need(t>=self.end,'whole probe period elapsed')
        if self.done:return dict(action='ALREADY_FINISHED')
        if self.view.last_round!=self.round:self.done=True;return dict(action='STALE_PROBE')
        need(self.sent_ping and(self.acked or self.indirect or not self.helpers),'probe phases executed before conclusion')
        self.done=True
        if self.acked:return dict(action='RESPONDED')
        current=self.view.table[self.target].info
        if current.kind=='CONFIRM':return dict(action='TARGET_ALREADY_REMOVED')
        out=self.view.merge(Update(self.target,'SUSPECT',current.inc),t)
        return dict(action='NO_MATCHING_ACK',membership=out)

def helper_ping(req):
    need(isinstance(req,Packet)and req.kind=='PING_REQ'and req.src==req.origin and req.dst not in (req.origin,req.target),'valid indirect request')
    return Packet('PING',req.dst,req.target,req.origin,req.round,req.target)
def target_ack(ping):
    need(isinstance(ping,Packet)and ping.kind=='PING'and ping.dst==ping.target,'target receives ping')
    return Packet('ACK',ping.target,ping.src,ping.origin,ping.round,ping.target)
def relay_ack(req,ack):
    need(isinstance(req,Packet)and req.kind=='PING_REQ'and isinstance(ack,Packet),'relay inputs')
    need(ack.kind=='ACK'and ack.src==req.target and ack.dst==req.dst and (ack.origin,ack.round,ack.target)==(req.origin,req.round,req.target),'matching target reply to helper')
    return Packet('RELAY_ACK',req.dst,req.origin,req.origin,req.round,req.target)

def main_trace():
    p=Phi();warm=[]
    for seq,t in enumerate([0,8,16,28,40]):
        p.heartbeat(seq,t);warm.append(p.query(t))
    scores=[p.query(t)for t in [50,54,55,58]]
    before=p.query(60);accepted=p.heartbeat(5,60);dup=[p.heartbeat(3,61),p.heartbeat(5,61)];after=p.query(75)
    need(after['mean']==13 and after['variance']==19 and p.last==60,'updated window and old heartbeat rejection')
    ids='ABCD';A=MemberView('A',ids);B=MemberView('B',ids);C=MemberView('C',ids)
    one=Probe(A,1,'B',['C'],0);direct=one.ping();request=one.indirect_requests(2)[0];ping=helper_ping(request);ack=target_ack(ping);relay=relay_ack(request,ack)
    need(one.ack(relay,5)=='ACCEPT_ACK'and one.finish(8)['action']=='RESPONDED','indirect route')
    two=Probe(A,2,'B',['C'],10);two.ping();two.indirect_requests(12);late=two.ack(relay,12);need(late=='IGNORE_ACK','old round ignored')
    two.finish(18);sus=Update('B','SUSPECT',0);C.merge(sus,19)
    need(A.table['B'].deadline==30 and C.table['B'].deadline==31,'observer local deadlines')
    branchA=deepcopy(A);branchC=deepcopy(C);branchB=deepcopy(B)
    alive=B.merge(sus,20)['generated'];A.merge(alive,22);C.merge(alive,24);old_sus=C.merge(sus,27)
    old_timers=[A.fire('B',0,30,30),C.fire('B',0,31,31)]
    need(all(x['action']=='STALE_TIMER'for x in old_timers),'old callback guard')
    alive2=branchB.merge(sus,20)['generated'];branchA.merge(alive2,22);confirmation=branchC.fire('B',0,31,31)
    late_alive=branchC.merge(alive2,32);branchA.merge(Update('B','CONFIRM'),33)
    need(branchA.table['B'].info.kind==branchC.table['B'].info.kind=='CONFIRM','original Confirm dominates higher Alive')
    same_time=[]
    for order in ['alive-first','timer-first']:
        v=MemberView('C',ids);v.merge(sus,19)
        if order=='alive-first':v.merge(alive,31);v.fire('B',0,31,31)
        else:v.fire('B',0,31,31);v.merge(alive,31)
        same_time.append(dict(order=order,B=v.snapshot()['B']))
    # Both piggyback attempts really consume budget even when transport drops them.
    sender=MemberView('A',ids);receiver=MemberView('C',ids);sender.merge(Update('D','CONFIRM'),0)
    lost=[sender.piggyback(t)for t in [1,2]];nothing=sender.piggyback(3);pre=receiver.snapshot();receiver.repair_from(sender,4)
    need(not nothing and pre['D']['kind']=='ALIVE'and receiver.table['D'].info.kind=='CONFIRM','finite rumor budget plus explicit repair')
    return dict(phi=dict(warmup=warm,scores=scores,before_same_time_heartbeat=before,accepted=accepted,duplicates=dup,updated_score=after,posterior_examples=[Q(100,1099),Q(100,101)]),swim=dict(first_round_packets=[direct,request,ping,ack,relay],first_round_send_times=[0,2,3,4,5],first_round_receive_times=[None,3,4,5,5],first_direct_ping='LOST',late_old_ack=late,timely_refutation=dict(alive=alive,A=A.snapshot(),C=C.snapshot(),old_suspect=old_sus,old_timers=old_timers),lost_refutation=dict(confirmation=confirmation,late_alive=late_alive,A=branchA.snapshot(),C=branchC.snapshot()),same_time=same_time,finite_budget=dict(lost_packets=lost,third_packet=nothing,before_repair=pre,after_repair=receiver.snapshot()),first_selection_probability=Q(175,256),expected_first_selection_rounds=Q(256,175),frozen_four_target_gap=7))

def tests():
    rng=random.Random(150015);counts=dict(phi_events=0,precedence_events=0,probe_cases=0,round_robin_pairs=0)
    for trial in range(160):
        W=rng.randrange(2,9);p=Phi(W);accepted=[];seq=-1;t=0
        for step in range(100):
            t+=rng.randrange(5);s=seq+rng.randrange(-2,4);s=max(s,0)
            if s>seq:accepted.append(t);seq=s
            p.heartbeat(s,t);out=p.query(t+rng.randrange(2));t=p.now
            gaps=[y-x for x,y in zip(accepted,accepted[1:])][-W:]
            need(list(p.gaps)==gaps and p.total==sum(gaps)and p.squares==sum(x*x for x in gaps),'literal window ledger')
            if len(gaps)==W:
                mu=Q(sum(gaps),W);variance=sum((Q(x)-mu)**2 for x in gaps)/W
                need(out['mean']==mu and out['variance']==variance,'centered exact variance')
            else:need(out['status']=='UNTRAINED','training boundary')
            counts['phi_events']+=1
    fixed=Phi();
    for s,t in enumerate([0,8,16,28,40]):fixed.heartbeat(s,t)
    prior=-1
    for t in range(40,100):
        out=fixed.query(t);score=out.get('phi',out.get('phi_lower_bound'));need(score>=prior or out['status']=='LOWER_BOUND','monotone mathematical score until displayed lower bound');prior=score
    need(fixed.query(100)['status']=='LOWER_BOUND','avoid tiny-tail underflow')
    choices=[Update('B',kind,i)for i in range(5)for kind in ['ALIVE','SUSPECT']]+[Update('B','CONFIRM')]
    for trial in range(200):
        v=MemberView('A','ABC');best=Update('B','ALIVE',0)
        for t in range(80):
            u=rng.choice(choices);before=v.table['B'];old_deadline=before.deadline;old_info=before.info
            v.merge(u,t)
            if u.key()>best.key():best=u
            need(v.table['B'].info==best,'information supremum')
            if u.key()<=old_info.key():need(v.table['B'].deadline==old_deadline,'duplicates do not extend suspicion')
            counts['precedence_events']+=1
    for sequence in product(['current','old','wrong-target','wrong-helper','late'],repeat=3):
        view=MemberView('A','ABCD');p=Probe(view,2,'B',['C'],0);p.ping();p.indirect_requests(2);expected=False
        for step,kind in enumerate(sequence):
            seq=1 if kind=='old'else 2;target='D'if kind=='wrong-target'else'B';src='D'if kind=='wrong-helper'else'C';t=9 if kind=='late'else 3+step
            if t<p.view.now:t=p.view.now
            packet=Packet('RELAY_ACK',src,'A','A',seq,target)
            valid=seq==2 and target=='B'and src=='C'and t<=8;expected|=valid
            need((p.ack(packet,t)=='ACCEPT_ACK')==valid,'ACK identity/time interface')
        result=p.finish(max(8,p.view.now));need((result['action']=='RESPONDED')==expected,'complete probe outcome');counts['probe_cases']+=1
    for first,second in product(permutations(range(4)),repeat=2):
        for item in range(4):
            gap=4+second.index(item)-first.index(item);need(gap<=7,'frozen permutation gap');counts['round_robin_pairs']+=1
    bad=[rejected(lambda:Phi(1)),rejected(lambda:Phi(scale_floor=0)),rejected(lambda:Phi().query(0,[13])),rejected(lambda:MemberView('A','AB').merge(Update('A','SUSPECT',9),0))]
    counts['explicit_rejections']=len(bad)
    return counts

def encode(x):
    if isinstance(x,Q):return str(x)
    if isinstance(x,(Update,Packet)):return asdict(x)
    raise TypeError(type(x).__name__)
if __name__=='__main__':print(json.dumps(dict(status='PASS',demonstration=main_trace(),checks=tests()),default=encode,ensure_ascii=False,indent=2))