check_frozen_row_composition.py

"""Semantic replay of frozen-row composition through persistent capsule tries.

Tuple queues model the separately proved strict FIFO interface. Python timings
are NOT native cost evidence. The oracle uses ordinary sets and Boolean clauses.
"""
from dataclasses import dataclass
from itertools import product
import json
import random


@dataclass(frozen=True)
class Blank:
    pass


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


@dataclass(frozen=True)
class Capsule:
    frames: tuple
    tip: Node


B = Blank()


def view(t):
    if isinstance(t, Blank):
        return None, B, B
    if isinstance(t, Node):
        return t.value, t.left, t.right
    value, direction, other = t.frames[0]
    rest = t.frames[1:]
    child = Capsule(rest, t.tip) if rest else t.tip
    return (value, other, child) if direction else (value, child, other)


def descend(root, word):
    frames = ()
    current = root
    for bit in word:
        value, left, right = view(current)
        frames += ((value, int(bit), left if bit == '1' else right),)
        current = right if bit == '1' else left
    return frames, current


def close(builder, value):
    frames, current = builder
    _, left, right = view(current)
    tip = Node(value, left, right)
    return Capsule(frames, tip) if frames else tip


def lookup(root, word):
    # Read-only logical traversal can allocate suffix capsules, as in the proof.
    current = root
    for bit in word:
        _, left, right = view(current)
        current = right if bit == '1' else left
    return view(current)[0]


def insert(root, word, value):
    return close(descend(root, word), value)


@dataclass(frozen=True)
class Receipt:
    seen: object = B
    rows: tuple = ()
    degree: int = 0


def append_s(root, middle, target):
    outer = descend(root, middle)
    row = view(outer[1])[0]
    row = B if row is None else row
    row = insert(row, target, True)
    return close(outer, row)


def append_r(root, frozen_s, source, middle, d, base=True):
    outer = descend(root, source)
    payload = view(outer[1])[0]
    payload = Receipt() if payload is None else payload
    seen_path = descend(payload.seen, middle)
    present = view(seen_path[1])[0]
    row = lookup(frozen_s, middle)
    contribution = ((B if row is None else row),) if base else (() if row is None else row.rows)
    if present:
        return root
    if payload.degree == d:
        return None
    return close(outer, Receipt(close(seen_path, True), payload.rows + contribution,
                               payload.degree + 1))


def answer(root, source, target):
    payload = lookup(root, source)
    return payload is None or not any(lookup(row, target) for row in payload.rows)


def clauses_sat(r, s, a, b, names):
    # Independent brute-force Horn truth on three disjoint variable sorts.
    variables = [(sort, name) for sort in range(3) for name in names]
    for bits in product((False, True), repeat=len(variables)):
        val = dict(zip(variables, bits))
        if not val[0, a] or val[2, b]:
            continue
        if all(not val[0, i] or val[1, k] for i, k in r) and all(
                not val[1, k] or val[2, j] for k, j in s):
            return True
    return False


def run():
    names = ('0', '00')  # Prefix-related names must remain different.
    edges = list(product(names, repeat=2))
    relations = [set(e for n, e in enumerate(edges) if mask >> n & 1)
                 for mask in range(16)]
    exhaustive = 0
    for s in relations:
        sr = B
        for k, j in sorted(s):
            sr = append_s(sr, k, j)
        for r in relations:
            for d in (1, 2):
                root, overflow = B, False
                for i, k in sorted(r):
                    updated = append_r(root, sr, i, k, d)
                    if updated is None:
                        overflow = True
                        break
                    root = updated
                    # Repeated edges do not spend another fanout slot.
                    root = append_r(root, sr, i, k, d)
                allowed = all(sum(i == u for i, k in r) <= d for u in names)
                assert overflow != allowed
                if allowed:
                    for a, b in edges:
                        expected = clauses_sat(r, s, a, b, names)
                        assert answer(root, a, b) == expected
                        exhaustive += 1
    rng = random.Random(123)
    names = ('0', '00', '01', '1', '10', '101', '0'*257, '1'*263)
    versions, queries = 0, 0
    for trial in range(80):
        s, sr = set(), B
        for _ in range(30):
            k, j = rng.choice(names), rng.choice(names)
            s.add((k, j)); sr = append_s(sr, k, j)
        root, r = B, set()
        history = [(root, set(r))]
        for _ in range(30):
            i, k = rng.choice(names), rng.choice(names)
            candidate = append_r(root, sr, i, k, 3)
            if candidate is None:
                assert len({v for u, v in r if u == i} | {k}) > 3
                continue
            root = candidate; r.add((i, k)); history.append((root, set(r)))
        for old, old_r in history:
            versions += 1
            for a, b in product(names + ('111111',), repeat=2):
                expected = not any(i == a and j == b and k == v
                                   for i, k in old_r for v, j in s)
                assert answer(old, a, b) == expected
                queries += 1
    # A late update to S is a real stale-snapshot counterexample, not supported.
    sr = B
    root = append_r(B, sr, '0', '1', 1)
    sr = append_s(sr, '1', '0')
    assert answer(root, '0', '0')  # Frozen empty S says SAT.
    assert lookup(lookup(sr, '1'), '0')  # Updated S would say UNSAT.
    layered_queries = 0
    names = ('0', '00', '1')
    for trial in range(100):
        relation = {edge for edge in product(names, repeat=2) if rng.randrange(2)}
        previous = B
        for k, j in relation:
            previous = append_s(previous, k, j)
        for depth in range(1, 5):
            left = {(i, k) for i in names for k in rng.sample(names, rng.randrange(3))}
            new = B
            for i, k in sorted(left):
                new = append_r(new, previous, i, k, 2, base=(depth == 1))
                assert new is not None
                assert len(lookup(new, i).rows) <= 2**depth
            relation = {(i, j) for i, k in left for v, j in relation if k == v}
            previous = new
            for a, b in product(names, repeat=2):
                assert answer(previous, a, b) == ((a, b) not in relation)
                layered_queries += 1
    print(json.dumps(dict(status='pass', exhaustive_Horn_queries=exhaustive,
                          retained_versions=versions, retained_queries=queries,
                          stale_snapshot_counterexamples=1,
                          fixed_depth_queries=layered_queries), indent=2))


if __name__ == '__main__':
    run()