verify_horn_join.py

#!/usr/bin/env python3
"""Semantic reference for retained Join/Pin; not a pointer-cost implementation."""
from dataclasses import dataclass
from pathlib import Path
import hashlib
import json
import random
from verify_persistent_horn import Engine, State

B = frozenset((0,1))

@dataclass(frozen=True)
class Solver:
    z: frozenset
    failed: bool = False
    engine: object = None
    local: object = None
    left: object = None
    right: object = None

@dataclass(frozen=True)
class Term:
    kind: str
    data: object
    left: object = None
    right: object = None

@dataclass
class Events:
    z: int = 0
    k: int = 0
    D: int = 0
    A: int = 0
    E: int = 0
    calls: int = 0


def leaf(rules):
    e = Engine(3,rules,B)
    state,_ = e.extend(State(),(),True)
    z = frozenset() if state.conflict else e.model(state)&B
    return Term('leaf',tuple(rules)), Solver(z,state.conflict,e,state)


def extend(s, incoming, ev, initial=False):
    ev.calls += 1
    if s.failed:
        return s, ()
    if s.engine is not None:
        before = (s.engine.cost.activated,s.engine.cost.occurrences)
        local,delta = s.engine.extend(s.local,incoming)
        ev.A += s.engine.cost.activated-before[0]
        ev.E += s.engine.cost.occurrences-before[1]
        z = s.z | frozenset(delta)
        return Solver(z,local.conflict,s.engine,local),delta
    z = set() if initial else set(s.z)
    pending = []
    delta = []
    def seed(v):
        if v not in z:
            z.add(v);pending.append(v);delta.append(v)
    for v in incoming:
        seed(v)
    left,right = s.left,s.right
    while pending:
        v = pending.pop();ev.D += 1
        left,dl = extend(left,(v,),ev)
        if left.failed:
            return Solver(frozenset(z),True,left=left,right=right),tuple(delta)
        for w in dl:
            seed(w)
        right,dr = extend(right,(v,),ev)
        if right.failed:
            return Solver(frozenset(z),True,left=left,right=right),tuple(delta)
        for w in dr:
            seed(w)
    assert left.z == right.z == frozenset(z)
    return Solver(frozenset(z),left=left,right=right),tuple(delta)


def join(a,b):
    ta,sa=a;tb,sb=b
    term=Term('join',None,ta,tb)
    ev=Events()
    if sa.failed or sb.failed:
        return (term,Solver(frozenset(),True)),ev
    names=tuple(sa.z)+tuple(sb.z);ev.z=len(names)
    s,_=extend(Solver(frozenset(),left=sa,right=sb),names,ev,True)
    return (term,s),ev


def pin(h,names):
    t,s=h
    ev=Events(k=len(names))
    new,_=extend(s,names,ev)
    return (Term('pin',tuple(names),t),new),ev


def flatten(t,path=()):
    if t.kind=='leaf':
        def rename(v):
            return None if v is None else ('B',v) if v in B else ('P',path,v)
        return [(tuple(rename(v) for v in body),rename(head)) for body,head in t.data]
    if t.kind=='pin':
        return flatten(t.left,path)+[((),('B',v)) for v in t.data]
    return flatten(t.left,path+(0,))+flatten(t.right,path+(1,))


def reference(t):
    rules=flatten(t)
    variables=list({v for body,head in rules for v in (*body,head) if v is not None}|{('B',v) for v in B})
    models=[]
    for mask in range(1<<len(variables)):
        truth={v for j,v in enumerate(variables) if mask&(1<<j)}
        if all(not set(body)<=truth or head in truth for body,head in rules):
            models.append(truth)
    if not models:
        return None
    return frozenset(v for v in B if all(('B',v) in m for m in models))


def check(h):
    term,s=h
    expected=reference(term)
    assert s.failed == (expected is None)
    if expected is not None:
        assert s.z==expected
    return expected


def leaves(t):
    return 1 if t.kind=='leaf' else leaves(t.left) if t.kind=='pin' else leaves(t.left)+leaves(t.right)


def snapshot(s):
    if s.engine is not None:
        return (s.z,s.failed,repr(s.local))
    return (s.z,s.failed,snapshot(s.left) if s.left else None,snapshot(s.right) if s.right else None)


def main():
    rng=random.Random(4001)
    pool=[]
    for _ in range(30):
        rules=[]
        for j in range(rng.randrange(6)):
            h=rng.choice([None,0,1,2])
            choices=[v for v in range(3) if v!=h]
            body=tuple(rng.sample(choices,rng.randrange(min(len(choices),3 if h is None else 2)+1)))
            rules.append((body,h))
        pool.append(leaf(rules));check(pool[-1])
    joins=pins=0
    for _ in range(500):
        a=rng.choice(pool);saved=snapshot(a[1])
        if rng.randrange(2):
            b=rng.choice(pool)
            if leaves(a[0])+leaves(b[0])>6:
                continue
            saved_b=snapshot(b[1])
            result,ev=join(a,b);joins+=1
            assert saved_b==snapshot(b[1])
        else:
            result,ev=pin(a,[rng.randrange(2) for _ in range(rng.randrange(3))]);pins+=1
        check(result)
        assert snapshot(a[1])==saved
        pool.append(result)
    fact=leaf([((),0)])
    implication=leaf([((0,),1)])
    crossed,ev=join(fact,implication)
    assert check(crossed)==B and ev.A>=1
    # An empty boundary cache prevents descent, even into a deep composite.
    h=leaf([((0,),1)])
    for _ in range(20):
        h,ev=join(h,leaf([]))
        assert ev.z==ev.D==ev.A==ev.E==0
    seeded,ev=pin(h,[0])
    assert check(seeded)==B and ev.D>=20
    deep_events=vars(ev).copy()
    # Reusing a handle creates separate semantic private occurrences.
    private=leaf([((0,),2),((2,),1)])
    double,_=join(private,private)
    assert len({v for body,head in flatten(double[0]) for v in (*body,head) if v and v[0]=='P'})==2
    old=snapshot(private[1])
    double,_=pin(double,[0]);assert check(double)==B
    assert snapshot(private[1])==old
    # Sum counters falsely fires; maximum counters fails to fire.
    one_x={0};one_z={1};body={0,1}
    assert len(one_x&body)+len(one_x&body)==2 and not body<=one_x
    assert max(len(one_x&body),len(one_z&body))==1 and body<=one_x|one_z
    # Conditional state must not masquerade as the unconditioned formula.
    base=leaf([((0,),1)]);negative=leaf([((1,),None)])
    assert check(join(base,negative)[0])==frozenset()
    conditioned,_=pin(base,[0]);assert check(join(conditioned,negative)[0]) is None
    here=Path(__file__).resolve().parent
    names=['compositional-horn-join.md','verify_horn_join.py','verify_persistent_horn.py']
    out={'status':'passed','random_join_operations':joins,'random_pin_operations':pins,
         'retained_snapshots':'unchanged','cross_operand_old_fact':'passed',
         'private_occurrence_separation':'passed','empty_cache_no_descent':'passed',
         'deep_composite_pin_events':deep_events,'counter_sum_and_max_negative_controls':'passed',
         'conditional_assumption_scope':'passed',
         'scope':'Finite semantic Join/Pin histories against truth assignments; recursive sets and existing leaf checker, not pointer-cost or Lean verification.',
         'sha256':{n:hashlib.sha256((here/n).read_bytes()).hexdigest() for n in names}}
    (here/'horn-join-output.json').write_text(json.dumps(out,indent=2)+'\n');print(json.dumps(out,indent=2))

if __name__=='__main__':
    main()