check_gap_hole_exports.py

#!/usr/bin/env python3
"""Independent persistent gap-list audit for arbitrary bounded-hole reuse.

Unary cells and list capsules are explicit immutable records. Tuple FIFOs and
run-head numerals stand for the separately proved strict queue/counter. No
loop in view/close follows a capsule tip; terminal commits are bounded. Sets
and integer decoding occur only in the independent assertions/test harness.
"""
from dataclasses import dataclass, replace
from itertools import product, combinations
from random import Random
import json
from check_single_hole_reuse import inc, decode
from check_colored_hole_reuse import Receipt, select, query


@dataclass(frozen=True, eq=False)
class Unit:
    tail: object = None


def length(p):
    result = 0
    while p is not None:
        result += 1
        p = p.tail
    return result


@dataclass(frozen=True, eq=False)
class Payload:
    gap: object
    ruler: object
    numeral: tuple


@dataclass(frozen=True, eq=False)
class Node:
    payload: Payload
    tail: object = None


@dataclass(frozen=True, eq=False)
class Cap:
    frames: tuple
    tip: Node | None


def close(frames, tip):
    assert tip is None or isinstance(tip, Node)
    return Cap(frames, tip) if frames else tip


def view(root):
    if root is None or isinstance(root, Node):
        return root
    assert isinstance(root, Cap) and root.frames
    # One FIFO front/dequeue and one capsule; no recursive normalization.
    return Node(root.frames[0], close(root.frames[1:], root.tip))


@dataclass(frozen=True, eq=False)
class Row:
    maximum: object = None
    gaps: object = None
    root: object = None


def record(old, word):
    """The algorithm depends on arriving symbols and local empty tests only."""
    assert word
    frames = ()
    head = view(old.root)
    successor = view(head.tail) if head is not None else None
    remaining = head.payload.gap if head is not None else None
    merge = successor.payload.gap if successor is not None else None
    elapsed = ruler = None
    numeral = ()
    old_maximum = old.maximum
    gaps = old.gaps
    passed = False
    extension = None
    extensions = deletions = prefix_frames = 0
    for symbol in word:
        assert symbol in '01'
        if passed:
            assert head is not None
            frames += (head.payload,)
            prefix_frames += 1
            head = successor
            successor = view(head.tail) if head is not None else None
            remaining = head.payload.gap if head is not None else None
            merge = successor.payload.gap if successor is not None else None
            elapsed = None
            passed = False
        if extension is not None:
            frames += (extension,)
            prefix_frames += 1
            gaps = Unit(gaps)
            extensions += 1
            elapsed = None
            extension = None
        ruler = Unit(ruler)
        numeral = inc(numeral)
        if old_maximum is not None:
            old_maximum = old_maximum.tail
            beyond = False
        else:
            beyond = True
        elapsed = Unit(elapsed)
        if head is not None:
            assert not beyond and remaining is not None
            remaining = remaining.tail
            if successor is not None:
                merge = Unit(merge)
            passed = remaining is None
        elif beyond:
            extension = Payload(elapsed, ruler, numeral)
    # Only these bounded actions occur after the final target bit.
    if beyond:
        new = Row(ruler, gaps, close(frames, None))
    elif passed:
        assert gaps is not None
        tip = None if successor is None else Node(
            replace(successor.payload, gap=merge), successor.tail)
        new = Row(old.maximum, gaps.tail, close(frames, tip))
        deletions = 1
    else:
        new = old
    assert prefix_frames <= len(word)
    return new, (extensions, deletions, prefix_frames)


def inspect(row, occupied):
    """Independent oracle; never called by record or the native algorithm."""
    h = max(occupied, default=0)
    assert length(row.maximum) == h
    expected = sorted(set(range(1, h+1))-set(occupied))
    found, cursor, previous = [], row.root, 0
    while cursor is not None:
        node = view(cursor)
        value = length(node.payload.ruler)
        assert decode(node.payload.numeral) == value
        assert length(node.payload.gap) == value-previous > 0
        found.append(value)
        cursor, previous = node.tail, value
    assert found == expected
    assert length(row.gaps) == len(expected)
    return tuple(found)


def main():
    rng = Random(18457)
    records = branches = merges = extensions = saved_queries = 0
    # Every short insertion ordering, duplicates included; earlier roots retained.
    for sequence in product(range(1, 7), repeat=4):
        row, occupied = Row(), set()
        history = [(row, frozenset())]
        for n in sequence:
            row, stats = record(row, '0'*n)
            occupied.add(n)
            inspect(row, occupied)
            records += 1; extensions += stats[0]; merges += stats[1]
            history.append((row, frozenset(occupied)))
        for old, old_set in history:
            inspect(old, old_set)
            n = rng.randrange(1, 10)
            fork, _ = record(old, '1'*n)
            inspect(fork, old_set | {n})
            inspect(old, old_set)
            branches += 1
    for _ in range(100):
        rows, sets = [Row() for _ in range(4)], [set() for _ in range(4)]
        retained = []
        for _ in range(100):
            u, n = rng.randrange(4), rng.randrange(1, 90)
            old = rows[u]
            rows[u], stats = record(old, ''.join(rng.choice('01') for _ in range(n)))
            sets[u].add(n)
            inspect(rows[u], sets[u])
            retained.append((rows[u], frozenset(sets[u])))
            records += 1; extensions += stats[0]; merges += stats[1]
        for old, old_set in rng.sample(retained, 10):
            inspect(old, old_set)
            n = rng.randrange(1, 110)
            fork, _ = record(old, '0'*n)
            inspect(fork, old_set | {n})
            inspect(old, old_set)
            branches += 1
    # Very late short deletion followed immediately by a one-bit query export.
    row, _ = record(Row(), '0'*80)
    occupied = {80}
    for n in range(3, 80):
        if n != 63:
            row, _ = record(row, '0'*n); occupied.add(n)
    assert inspect(row, occupied) == (1, 2, 63)
    old = row
    row, _ = record(row, '0'); occupied.add(1)
    assert inspect(row, occupied) == (2, 63)
    assert inspect(old, occupied-{1}) == (1, 2, 63)

    # End-to-end producer exports into the independently audited reuse kernel.
    literal_records = 0
    for bound in (1, 2, 3, 4):
        for _ in range(35):
            h = rng.randrange(3, 8)
            table, right_rows = {}, []
            for v in range(8):
                holes = set(rng.sample(range(1, h), rng.randrange(min(bound, h-1)+1)))
                name = bin(v+2)[2:]
                table[name] = holes
                right_rows += [(name, format(x, f'0{n}b'))
                    for n in range(1, h+1) if n not in holes for x in range(1 << n)]
            rng.shuffle(right_rows)
            stores, widths = {}, {}
            for name, word in right_rows:
                stores[name], _ = record(stores.get(name, Row()), word)
                widths.setdefault(name, set()).add(len(word))
                literal_records += 1
            exports = {name: inspect(stores[name], widths[name]) for name in table}
            assert all(set(exports[name]) == table[name] for name in table)
            states, selected = [Receipt() for _ in range(3)], [[] for _ in range(3)]
            snapshots = []
            for _ in range(30):
                u, name = rng.randrange(3), rng.choice(tuple(table))
                states[u] = select(states[u], name, exports[name], bound)
                selected[u].append(table[name])
                missing = frozenset(set.intersection(*selected[u]))
                snapshots.append((states[u], missing))
            for old, missing in snapshots:
                saved_queries += query(old, h, h+2, missing, bound)
    print(json.dumps(dict(status='passed', insertion_records=records,
        persistent_branch_checks=branches, hole_deletions=merges,
        confirmed_new_holes=extensions, literal_records=literal_records,
        composed_query_prefix_answers=saved_queries,
        late_short_deletion='passed', nonrecursive_capsule_view='asserted'), indent=2))


if __name__ == '__main__':
    main()