check_input_dominated_classes.py

#!/usr/bin/env python3
"""Finite diagnostics for input-dominated maximum classes and binary balls.

The new unary two-terminal kernel uses immutable records and the maintained
strict FIFO prototype. Tuple slicing in that prototype is structural notation,
not a claim about Python execution time. Instrumented work counts queue calls
and their conceptual allocations, not a whole native recognizer's clock.
Python maps, integers, sets and exhaustive scans below are semantic oracles for
the inherited ordinary-name capsules and complete-level counters.
"""
from collections import Counter
from dataclasses import dataclass
import importlib.util
from itertools import product
import json
from pathlib import Path
import random

ROOT = Path(__file__).resolve().parents[2]
QUEUE = ROOT / "site/public/research-artifacts/foundations/verify_counter_free_real_time_queue.py"
spec = importlib.util.spec_from_file_location("strict_fifo", QUEUE)
fifo = importlib.util.module_from_spec(spec)
spec.loader.exec_module(fifo)


@dataclass(frozen=True)
class Cell:
    tail: object = None


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


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


class Meter:
    def __init__(self):
        self.cost = 0

    def enqueue(self, queue, value):
        queue, cost = fifo.enqueue(queue, value)
        self.cost += cost
        return queue

    def capsule(self, queue, tip):
        assert isinstance(tip, Node)
        if not queue[0]:
            return tip
        self.cost += 1
        return Cap(queue, tip)

    def view(self, root):
        self.cost += 1
        if root is None:
            return None, None
        if isinstance(root, Node):
            return root.value, root.child
        assert isinstance(root.tip, Node) and root.frames[0]
        value = root.frames[0][0]
        rest, cost = fifo.dequeue(root.frames)
        self.cost += cost
        return value, self.capsule(rest, root.tip)


def size(stack):
    """Oracle only; the kernel uses empty and singleton tests."""
    answer = 0
    while stack is not None:
        answer += 1
        stack = stack.tail
    return answer


def ruler(n):
    result = None
    for _ in range(n):
        result = Cell(result)
    return result


def snapshot(root, depth):
    meter = Meter()
    values = []
    for _ in range(depth + 1):
        value, root = meter.view(root)
        values.append(size(value))
    return values


def move_max(root, old_ruler, word, modulus):
    """Drive both changed addresses using actual word bits; no integer keys.

    Return root, new ruler, residue, source/destination zero crossings and
    per-symbol structural work. The caller retains old values on no-change.
    """
    meter = Meter()
    cursor = root
    left = old_ruler
    prefix = middle = fifo.EMPTY_QUEUE
    crossed = False
    old_value = None
    new_ruler = None
    residue = 0
    work = []
    for bit in word:
        assert bit in "01"
        start = meter.cost
        value, cursor = meter.view(cursor)
        if not crossed and left is None:
            crossed = True
            old_value = value
        elif not crossed:
            prefix = meter.enqueue(prefix, value)
            left = left.tail
        else:
            middle = meter.enqueue(middle, value)
        new_ruler = Cell(new_ruler)
        residue = (residue + 1) % modulus
        work.append(meter.cost - start + 1)
    if not crossed:
        return root, old_ruler, None, False, False, work + [0]
    start = meter.cost
    destination, child = meter.view(cursor)
    destination_empty = destination is None
    tip = Node(Cell(destination), child)
    between = meter.capsule(middle, tip)
    if old_ruler is not None:
        assert old_value is not None
        source_singleton = old_value.tail is None
        source = old_value.tail
    else:
        assert old_value is None
        source_singleton = False
        source = None
    changed = meter.capsule(prefix, Node(source, between))
    work.append(meter.cost - start + 3)
    return changed, new_ruler, residue, source_singleton, destination_empty, work


def occupy(root, word):
    """Single-terminal capsule; predecessor/successor occupancy give hole delta."""
    meter = Meter()
    cursor = root
    frames = fifo.EMPTY_QUEUE
    previous = None
    for _ in word:
        previous, cursor = meter.view(cursor)
        frames = meter.enqueue(frames, previous)
    current, child = meter.view(cursor)
    if current is not None:
        return root, 0
    following, _ = meter.view(child)
    delta = -(following is not None) + (len(word) > 1 and previous is None)
    return meter.capsule(frames, Node(Cell(), child)), delta


class Model:
    def __init__(self, modulus=2, bound=2):
        self.m = modulus
        self.k = bound
        self.rows = {}
        self.histogram = None
        self.diversity = 0
        self.by_residue = [0] * modulus
        self.holes = 0
        self.incomplete = 0
        self.max_work = 0
        self.history = []

    def insert(self, name, word):
        raw, old_ruler, old_residue, widths = self.rows.get(name, (frozenset(), None, 0, None))
        old_histogram = self.histogram
        old_depth = max((len(w) for r, *_ in self.rows.values() for w in r), default=0)
        old_snapshot = snapshot(old_histogram, old_depth)
        new, new_ruler, residue, leave, enter, work = move_max(
            self.histogram, old_ruler, word, self.m)
        self.max_work = max(self.max_work, *work)
        self.histogram = new
        self.diversity += enter - leave
        if leave:
            self.by_residue[old_residue] -= 1
        if enter:
            self.by_residue[residue] += 1
        if residue is None:
            residue = old_residue
        widths, delta = occupy(widths, word)
        self.holes += delta
        assert self.holes >= 0
        if word not in raw:
            count = sum(len(w) == len(word) for w in raw)
            if count == 0:
                self.incomplete += 1
            if count + 1 == 2 ** len(word):
                self.incomplete -= 1
            raw = raw | {word}
        self.rows[name] = (raw, new_ruler, residue, widths)
        assert snapshot(old_histogram, old_depth) == old_snapshot
        self.history.append((old_histogram, old_snapshot))
        self.audit()

    def audit(self):
        expected = Counter(max(map(len, raw)) for raw, *_ in self.rows.values())
        depth = max(expected, default=0)
        assert snapshot(self.histogram, depth) == [expected[j] for j in range(depth + 1)]
        assert self.diversity == len(expected)
        assert self.by_residue == [sum(j % self.m == r for j in expected) for r in range(self.m)]
        holes = incomplete = 0
        for raw, length, residue, widths in self.rows.values():
            counts = Counter(map(len, raw))
            assert size(length) == max(counts)
            assert residue == max(counts) % self.m
            assert snapshot(widths, max(counts)) == [int(j in counts) for j in range(max(counts) + 1)]
            holes += sum(j > 1 and j - 1 not in counts for j in counts)
            incomplete += sum(c != 2 ** j for j, c in counts.items())
        assert self.holes == holes
        assert self.incomplete == incomplete
        radii = set(expected)
        oracle = (all(raw == ball(max(map(len, raw))) for raw, *_ in self.rows.values())
                  and len(radii) <= self.k and len({j % self.m for j in radii}) == len(radii))
        assert self.accepts_shape() == oracle

    def accepts_shape(self):
        return not self.holes and not self.incomplete and self.diversity <= self.k and max(self.by_residue) <= 1

    def check_queries(self, left, sources, targets):
        assert self.accepts_shape()
        receipts = {}
        for source, intermediate in left:
            if intermediate in self.rows:
                raw, _, residue, _ = self.rows[intermediate]
                receipts.setdefault(source, {}).setdefault(residue, raw)
        checks = 0
        for source in sources:
            for target in targets:
                observed = any(target in raw for raw in receipts.get(source, {}).values())
                direct = any(u == source and target in self.rows.get(v, (set(),))[0] for u, v in left)
                assert observed == direct
                checks += 1
        return checks


def words(n):
    return {"".join(bits) for bits in product("01", repeat=n)}


def ball(n):
    return set().union(*(words(j) for j in range(1, n + 1)))


def main():
    histories = prefixes = queries = 0
    peak = 0
    # All short histories, including duplicated edges and proper-prefix names.
    alphabet = [(v, w) for v in ("0", "00") for w in ("0", "1", "00")]
    for n in range(5):
        for history in product(alphabet, repeat=n):
            model = Model()
            for v, w in history:
                model.insert(v, w)
                prefixes += 1
            histories += 1
            peak = max(peak, model.max_work)
    rng = random.Random(181)
    for _ in range(80):
        model = Model(3, 3)
        for _ in range(80):
            v = rng.choice(("0", "1", "00", "010101"))
            w = "".join(rng.choices("01", k=rng.randint(1, 12)))
            model.insert(v, w)
            prefixes += 1
        for root, prior in rng.sample(model.history, 8):
            assert snapshot(root, len(prior) - 1) == prior
        peak = max(peak, model.max_work)
        histories += 1
    # Complete arbitrary-order balls, gaps repaired in both directions, and
    # distinct final radii that deliberately collide under a fixed labeler.
    for radii in ((1, 2), (1, 3), (2, 5), (3, 4), (2, 2)):
        for _ in range(6):
            model = Model()
            pairs = [(v, w) for v, radius in zip(("0", "00"), radii) for w in sorted(ball(radius))]
            pairs += rng.choices(pairs, k=12)
            rng.shuffle(pairs)
            for v, w in pairs:
                model.insert(v, w)
                prefixes += 1
            assert model.accepts_shape() == (radii[0] == radii[1] or radii[0] % 2 != radii[1] % 2)
            if model.accepts_shape():
                left = [(u, v) for u in ("0", "00", "101") for v in ("0", "00", "111")]
                rng.shuffle(left)
                queries += model.check_queries(left + left[:3], ("0", "00", "101", "111"), ball(max(radii) + 1))
            peak = max(peak, model.max_work)
            histories += 1
    # Large intermediate diversity, short aliases, two final opposite-parity balls.
    model = Model()
    names = ["0", "1", "00", "01", "10"]
    for j, v in enumerate(names, 1):
        model.insert(v, "0" * j)
    assert model.diversity == 5
    pairs = [(v, w) for j, v in enumerate(names) for w in sorted(ball(6 + j % 2))]
    rng.shuffle(pairs)
    for v, w in pairs:
        model.insert(v, w)
        prefixes += 1
    assert model.accepts_shape() and model.diversity == 2
    queries += model.check_queries([("0", v) for v in names], ("0", "1"), ball(8))
    peak = max(peak, model.max_work)
    # Constant bound from the fixed number of queue calls, deliberately loose.
    assert peak <= 100
    print(json.dumps({"histories": histories + 1, "audited_prefixes": prefixes + 5,
                      "late_queries": queries, "maximum_kernel_structural_cost": peak,
                      "kernel_cost_bound": 100, "status": "passed"}, indent=2))


if __name__ == "__main__":
    main()