verify_persistent_horn.py

#!/usr/bin/env python3
"""Finite model checks, not a microinstruction or asymptotic proof certificate."""
from dataclasses import dataclass
from itertools import combinations
from pathlib import Path
import hashlib
import json
import random

@dataclass(frozen=True)
class Node:
    left: object = None
    right: object = None
    value: object = None

@dataclass
class Cost:
    reads: int = 0
    alloc: int = 0
    gets: int = 0
    puts: int = 0
    activated: int = 0
    occurrences: int = 0


def key(i, width):
    return tuple((i >> j) & 1 for j in reversed(range(width)))


def get(root, bits, cost, default=None):
    cost.gets += 1
    for bit in bits:
        cost.reads += 1
        if root is None:
            return default
        root = root.right if bit else root.left
    cost.reads += 1
    return default if root is None else root.value


def put(root, bits, value, cost):
    cost.puts += 1
    def rec(node, depth):
        cost.reads += 1
        cost.alloc += 1
        if depth == len(bits):
            return Node(value=value)
        left, right = (None, None) if node is None else (node.left, node.right)
        if bits[depth]:
            return Node(left, rec(right, depth+1))
        return Node(rec(left, depth+1), right)
    return rec(root, 0)


@dataclass(frozen=True)
class State:
    truth: object = None
    counters: object = None
    conflict: bool = False


class Engine:
    def __init__(self, n, rules, boundary=(), width=None):
        self.n = n
        self.rules = tuple((tuple(sorted(set(a))), h) for a, h in rules if h not in a)
        self.width = width or max(1, max(n, len(self.rules), 1).bit_length())
        self.cost = Cost()
        self.incidence = None
        self.clauses = None
        boundary = set(boundary)
        for v in range(n):
            self.incidence = put(self.incidence, key(v, self.width), (v in boundary, ()), self.cost)
        self.facts = []
        for j, (body, head) in enumerate(self.rules):
            assert len(body) <= (3 if head is None else 2)
            assert all(0 <= v < n for v in body)
            assert head is None or 0 <= head < n
            self.clauses = put(self.clauses, key(j, self.width), (len(body), head), self.cost)
            if not body:
                self.facts.append(head)
            for v in body:
                flag, inc = get(self.incidence, key(v, self.width), self.cost)
                self.incidence = put(self.incidence, key(v, self.width), (flag, (j, inc)), self.cost)
        self.preparation = vars(self.cost).copy()

    def extend(self, old, seeds, initialize=False):
        if old.conflict:
            return old, ()
        truth, counters = old.truth, old.counters
        work, delta = (), []
        conflict = False
        def mark(v):
            nonlocal truth, work
            flag, _ = get(self.incidence, key(v, self.width), self.cost)
            if not get(truth, key(v, self.width), self.cost, False):
                truth = put(truth, key(v, self.width), True, self.cost)
                work = (v, work)
                self.cost.activated += 1
                if flag:
                    delta.append(v)
        if initialize:
            for head in self.facts:
                if head is None:
                    conflict = True
                    break
                mark(head)
        if not conflict:
            for v in seeds:
                mark(v)
        while work and not conflict:
            v, work = work
            _, inc = get(self.incidence, key(v, self.width), self.cost)
            while inc and not conflict:
                j, inc = inc
                self.cost.occurrences += 1
                count = get(counters, key(j, self.width), self.cost, 0) + 1
                counters = put(counters, key(j, self.width), count, self.cost)
                arity, head = get(self.clauses, key(j, self.width), self.cost)
                assert count <= arity
                if count == arity:
                    if head is None:
                        conflict = True
                    else:
                        mark(head)
        return State(truth, counters, conflict), tuple(delta)

    def model(self, state):
        if state.conflict:
            return None
        return frozenset(v for v in range(self.n)
                         if get(state.truth, key(v, self.width), Cost(), False))

    def check_accounting(self):
        c = self.cost
        assert c.alloc == c.puts*(self.width+1)
        assert c.reads <= (c.gets+c.puts)*(self.width+1)


def oracle(n, rules, seeds):
    models = []
    for mask in range(1 << n):
        model = frozenset(v for v in range(n) if mask & (1 << v))
        if set(seeds) <= model and all(not set(body) <= model or head in model for body, head in rules):
            models.append(model)
    return frozenset.intersection(*models) if models else None


def compose(engines):
    states, pending = [], []
    for i, engine in enumerate(engines):
        state, delta = engine.extend(State(), (), True)
        states.append(state)
        pending.extend((1-i, v) for v in delta)
    notifications = len(pending)
    while pending and not any(s.conflict for s in states):
        target, v = pending.pop()
        states[target], delta = engines[target].extend(states[target], (v,))
        notifications += len(delta)
        pending.extend((1-target, w) for w in delta)
    return states, notifications


def main():
    checks = 0
    universe = []
    for size in range(3):
        for body in combinations(range(2), size):
            universe.append((body, None))
            for head in range(2):
                if head not in body:
                    universe.append((body, head))
    # All 256 normalized programs on two variables, every seed set, then every singleton extension.
    assert len(universe) == 8
    for mask in range(1 << len(universe)):
        rules = [r for j, r in enumerate(universe) if mask & (1 << j)]
        e = Engine(2, rules)
        initial, _ = e.extend(State(), (), True)
        for seedmask in range(4):
            seeds = {v for v in range(2) if seedmask & (1 << v)}
            state, _ = e.extend(initial, seeds)
            assert e.model(state) == oracle(2, rules, seeds)
            checks += 1
            for v in range(2):
                saved = repr(state)
                child, _ = e.extend(state, (v, v))
                assert e.model(child) == oracle(2, rules, seeds | {v})
                assert repr(state) == saved
                checks += 1
        e.check_accounting()
    rng = random.Random(3801)
    def random_rules(n):
        out = []
        for _ in range(rng.randrange(11)):
            h = rng.choice([None] + list(range(n)))
            body = tuple(rng.sample(range(n), rng.randrange(min(n, 3 if h is None else 2)+1)))
            out.append((body, h))
        return out
    forks = 0
    for _ in range(200):
        rules = random_rules(4)
        e = Engine(4, rules, range(2))
        initial, _ = e.extend(State(), (), True)
        versions = [(initial, frozenset())]
        for _ in range(20):
            parent, seeds = rng.choice(versions)
            v = rng.randrange(4)
            snapshot = repr(parent)
            child, _ = e.extend(parent, (v,))
            seeds = seeds | {v}
            assert e.model(child) == oracle(4, rules, seeds)
            assert repr(parent) == snapshot
            versions.append((child, seeds))
            forks += 1
        e.check_accounting()
    pairs = 0
    max_notifications = 0
    for _ in range(1000):
        rules = [random_rules(4), random_rules(4)]
        es = [Engine(4, r, (0, 1), width=4) for r in rules]
        states, sent = compose(es)
        assert sent <= 4
        combined = list(rules[0])
        def ren(v):
            return v if v is None or v < 2 else v+2
        combined += [(tuple(ren(v) for v in body), ren(h)) for body, h in rules[1]]
        expected = oracle(6, combined, ())
        if any(s.conflict for s in states):
            assert expected is None
        else:
            actual = es[0].model(states[0]) | frozenset(ren(v) for v in es[1].model(states[1]))
            assert actual == expected
        for e in es:
            assert e.cost.activated <= e.n
            assert e.cost.occurrences <= sum(len(a) for a, _ in e.rules)
            e.check_accounting()
        max_notifications = max(max_notifications, sent)
        pairs += 1
    # Negative three-body rule, duplicates, tautology, and explicit delayed inter-fragment conflict.
    r = [((0, 0), 1), ((1,), 1), ((0, 1, 2), None)]
    e = Engine(3, r)
    s, _ = e.extend(State(), (0, 2), True)
    assert e.model(s) == oracle(3, r, (0, 2)) is None
    es = [Engine(3, [((0,), 2), ((2,), 1)], (0, 1)),
          Engine(2, [((), 0), ((1,), None)], (0, 1))]
    ss, sent = compose(es)
    assert any(s.conflict for s in ss)
    chains = []
    for n in (16, 64, 256):
        e = Engine(n, [((v,), v+1) for v in range(n-1)])
        base, _ = e.extend(State(), (), True)
        start = e.cost.activated
        for _ in range(3):
            child, _ = e.extend(base, (0,))
            assert len(e.model(child)) == n
            assert e.model(base) == frozenset()
        assert e.cost.activated-start == 3*n
        e.check_accounting()
        chains.append({'variables': n, 'branches': 3, 'activation_events': 3*n})
    here = Path(__file__).resolve().parent
    result = {'status': 'passed', 'exhaustive_two_variable_programs': 256,
              'closure_and_extension_comparisons': checks, 'random_retained_version_extensions': forks,
              'random_independent_fragment_pairs': pairs, 'maximum_notifications_b2': max_notifications,
              'branch_repeated_work': chains,
              'scope': 'Finite high-level immutable-trie/event checks against assignment enumeration; no microinstruction, loader, asymptotic or Lean proof.',
              'sha256': {name: hashlib.sha256((here/name).read_bytes()).hexdigest() for name in
                         ('persistent-horn-closure.md', 'verify_persistent_horn.py', 'sat-boundary-dynamics.md')}}
    (here/'persistent-horn-output.json').write_text(json.dumps(result, indent=2)+'\n')
    print(json.dumps(result, indent=2))

if __name__ == '__main__':
    main()