verify_incremental_shared_tabulation.py

"""Finite semantic checks, NOT a local-machine or primitive-count verifier.

Host recursion/indices construct a shared immutable trie. An abstract service
clock tests the band deadline at its worst allowed job length. The written
proof supplies the correspondence from a band job to local primitives.
Small k,n outside the header promise are also used to test the identities.
"""
import itertools
import json
import random
from functools import lru_cache
from pathlib import Path


@lru_cache(None)
def universal(k):
    d = k*k + 2*k

    def build(bits):
        if len(bits) == d:
            matrix, a, b = bits[:k*k], bits[k*k:k*k+k], bits[k*k+k:]
            return sum(a[i]*matrix[i*k+j]*b[j]
                       for i in range(k) for j in range(k)) % 2
        return (build(bits+(0,)), build(bits+(1,)))

    return build(())


def walk(root, bits):
    for bit in bits:
        root = root[bit]
    return root


def compile_band(raw, n, k):
    assert len(raw) == n*k
    return tuple(walk(universal(k), tuple(
        raw[i*n+j] if j < n else 0
        for i in range(k) for j in range(start, start+k)))
        for start in range(0, n, k))


def snapshots(bits, n, k, quantum=5):
    # Each abstract job takes the worst allowed quantum*n*k service units.
    # Actual compilation is a host oracle; this is not a timing measurement.
    compiled = ()
    pending = ()
    current = ()
    remaining = 0
    yield (compiled, pending, current)
    for bit in bits:
        if pending:
            remaining = max(0, remaining-quantum)
            if remaining == 0:
                compiled += (compile_band(pending, n, k),)
                pending = ()
        current += (bit,)
        if len(current) == n*k:
            assert not pending, "a second completed band missed its deadline"
            pending, current = current, ()
            remaining = quantum*n*k
        assert len(pending)+len(current) <= 2*n*k
        yield (compiled, pending, current)


def answer(snapshot, u, v, n, k):
    compiled, pending, current = snapshot
    parity = 0
    row = 0
    for band in compiled:
        a = u[row:row+k]
        for col, root in zip(range(0, n, k), band):
            b = v[col:col+k]
            b += (0,)*(k-len(b))
            parity ^= walk(root, a+b)
        row += k
    for raw in (pending, current):
        for start in range(0, len(raw), n):
            present = raw[start:start+n]
            parity ^= (u[row]*sum(x*y for x, y in zip(present, v))) % 2
            row += 1
    return parity


def run():
    rng = random.Random(84003)
    checks = prefixes = 0
    saw_partial = saw_two_raw = saw_shared = False
    for n in range(1, 10):
        matrices = list(itertools.product((0, 1), repeat=n*n)) if n <= 2 else [
            tuple(rng.randrange(2) for _ in range(n*n)) for _ in range(3)]
        vectors = list(itertools.product((0, 1), repeat=n))
        pairs = list(itertools.product(vectors, repeat=2)) if n <= 3 else [
            (rng.choice(vectors), rng.choice(vectors)) for _ in range(12)]
        for k in range(1, min(3, n)+1):
            # A repeated data block selects the identical universal subtree.
            raw = (0,)*(n*k)
            roots = compile_band(raw, n, k)
            if len(roots) >= 2:
                assert roots[0] is roots[1]
                saw_shared = True
            for bits in matrices:
                saved = tuple(snapshots(bits, n, k))
                for t, snapshot in enumerate(saved):
                    prefixes += 1
                    compiled, pending, current = snapshot
                    covered = len(compiled)*n*k+len(pending)+len(current)
                    assert covered == t
                    saw_partial |= bool(t % n)
                    saw_two_raw |= bool(pending and current)
                    for u, v in pairs:
                        expected = sum(bits[p]*u[p//n]*v[p % n]
                                       for p in range(t)) % 2
                        assert answer(snapshot, u, v, n, k) == expected
                        checks += 1
                # Reuse the saved early snapshot after all later ones exist.
                assert answer(saved[0], pairs[0][0], pairs[0][1], n, k) == 0
    assert saw_partial and saw_two_raw and saw_shared
    result = {
        "status": "passed",
        "scope": "finite semantic trie/partition checks and abstract deadline clock; not local-machine verification",
        "n": "1 through 9; all matrices n<=2, three seeded matrices otherwise",
        "k": "1 through min(3,n); includes parameters outside header promise",
        "queries": "all vector pairs n<=3, twelve seeded pairs otherwise",
        "prefix_snapshots_checked": prefixes,
        "answers_checked": checks,
        "partial_rows_two_raw_bands_and_shared_subtrees": True,
        "resource_claims": "written proof only; no empirical asymptotic or primitive-count claim"
    }
    Path(__file__).with_name("incremental-shared-tabulation-output.json").write_text(
        json.dumps(result, indent=2)+"\n")
    print(json.dumps(result))


if __name__ == "__main__":
    run()