verify_selective_horn.py

#!/usr/bin/env python3
"""Finite semantics and routing counts. Host sets do not verify pointer costs."""
from dataclasses import dataclass,replace
import random,json
from verify_persistent_horn import Engine,State
from verify_horn_join import Term,reference,leaves
B=frozenset((0,1))
@dataclass(frozen=True)
class Router:
    support:frozenset
    left:object=None
    right:object=None
    engine:object=None
    local:object=None
@dataclass(frozen=True)
class Handle:
    term:object
    router:object
    z:frozenset
    failed:bool=False
@dataclass
class Events:
    V:int=0
    A:int=0
    E:int=0
    deliveries:int=0
    routed:int=0

def prepare(rules):
    e=Engine(3,rules,B);s,d=e.extend(State(),(),True)
    support=frozenset(v for body,h in e.rules for v in body if v in B)
    return Handle(Term('leaf',tuple(rules)),Router(support,engine=e,local=s),frozenset(d),s.conflict)

def route(r,v,ev):
    ev.V+=1
    if v not in r.support:return r,(),False
    if r.engine is not None:
        ev.deliveries+=1
        before=(r.engine.cost.activated,r.engine.cost.occurrences)
        state,delta=r.engine.extend(r.local,(v,))
        ev.A+=r.engine.cost.activated-before[0]
        ev.E+=r.engine.cost.occurrences-before[1]
        return replace(r,local=state),delta,state.conflict
    left,dl,bad=route(r.left,v,ev)
    if bad:return replace(r,left=left),dl,True
    right,dr,bad=route(r.right,v,ev)
    return replace(r,left=left,right=right),dl+dr,bad

def drain(term,r,z,incoming):
    ev=Events();seen=set(z);pending=[]
    def seed(v):
        if v not in seen:seen.add(v);pending.append(v)
    for v in incoming:seed(v)
    while pending:
        v=pending.pop();ev.routed+=1
        r,delta,bad=route(r,v,ev)
        if bad:return Handle(term,r,frozenset(seen),True),ev
        for w in delta:seed(w)
    return Handle(term,r,frozenset(seen)),ev

def join(a,b):
    t=Term('join',None,a.term,b.term)
    r=Router(a.router.support|b.router.support,a.router,b.router)
    if a.failed or b.failed:return Handle(t,r,frozenset(),True),Events()
    # Replay all operand facts, including facts irrelevant to their old routers.
    return drain(t,r,(),tuple(a.z)+tuple(b.z))

def pin(a,vs):
    t=Term('pin',tuple(vs),a.term)
    if a.failed:return Handle(t,a.router,a.z,True),Events()
    return drain(t,a.router,a.z,vs)

def snap(r):
    if r.engine is not None:return (r.support,repr(r.local))
    return (r.support,snap(r.left),snap(r.right))

def check(a):
    want=reference(a.term)
    assert a.failed==(want is None)
    if not a.failed:assert a.z==want,(a,want)

def internal_subscribers(r,v):
    if r.engine is not None:return 0
    return int(v in r.support)+internal_subscribers(r.left,v)+internal_subscribers(r.right,v)

def main():
    rng=random.Random(4201);pool=[]
    for _ in range(25):
        rules=[]
        for j in range(rng.randrange(5)):
            head=rng.choice([None,0,1,2]);vs=[v for v in range(3) if v!=head]
            body=tuple(rng.sample(vs,rng.randrange(min(len(vs),3 if head is None else 2)+1)))
            rules.append((body,head))
        h=prepare(rules);check(h);pool.append(h)
    joins=pins=0
    for _ in range(500):
        a=rng.choice(pool);old=(a.z,snap(a.router))
        if rng.randrange(2):
            b=rng.choice(pool)
            if leaves(a.term)+leaves(b.term)>5:continue
            h,e=join(a,b);joins+=1
        else:
            h,e=pin(a,rng.sample(list(B),rng.randrange(3)));pins+=1
            if not h.failed:
                assert e.V==sum(1+2*internal_subscribers(a.router,v) for v in h.z-a.z)
        check(h);assert (a.z,snap(a.router))==old;pool.append(h)
    empty=prepare([]);h=empty
    for _ in range(63):h,_=join(h,empty)
    h,e=pin(h,(0,));check(h)
    assert (e.V,e.deliveries,e.A,e.E)==(1,0,0,0)
    # Forgotten global facts would fail here.
    g=prepare([((0,),1)]);cross,ce=join(h,g);check(cross)
    assert cross.z==B and ce.deliveries==1
    # Negative-headed bodies must also subscribe.
    bad,be=join(h,prepare([((0,),None)]));check(bad);assert bad.failed
    # A new private fact leads to a globally relevant boundary consequence.
    p=prepare([((0,),2),((2,),1)]);q,qe=join(h,p);check(q);assert q.z==B
    cycle=prepare([((0,),1),((1,),0)]);check(cycle);assert not cycle.z
    dup,de=join(p,p);dup,de=pin(dup,(0,));check(dup)
    assert de.deliveries==2
    # Conditional child states must never replace their original public handles.
    check(p);assert not p.z
    print(json.dumps(dict(random_joins=joins,random_pins=pins,
      empty_64_leaf_pin=vars(e),later_join=vars(ce),negative_body_conflict=True,
      private_chain=True,duplicate_occurrences=True,unseeded_cycle=True,
      successful_pin_routing_identity=True,
      scope='Finite three-variable leaves, two boundary variables, at most five leaves in random oracle checks; 64 empty leaves; no pointer-cost or proof-assistant verification.'),indent=2))
if __name__=='__main__':main()