check_cone_factorization.py

"""Finite checks for the cone-factorization proof draft.

This is a regression instrument, not a proof.  It independently exercises:

* the machine-to-cone dependence lemma on random tiny protocols;
* the spaced late-index capacity lemma;
* the exact small subcube-cover frontier; and
* the explicit upper-bound decoder for small late-index instances.
"""

from __future__ import annotations

from functools import lru_cache
from itertools import combinations, product
from random import Random


def cone(m: int, n: int, placement: tuple[int, ...]) -> tuple[int, ...]:
    cells: set[int] = set()
    for root in placement:
        if root >= 0:
            cells.update(range(max(0, root - n), root + 1))
    return tuple(sorted(cells))


def bits(word: int, length: int) -> tuple[int, ...]:
    return tuple((word >> (length - 1 - i)) & 1 for i in range(length))


def random_protocol_checks(seed: int = 20260728, trials: int = 200) -> int:
    """Generate protocols and verify output depends only on root config,
    cone contents, and continuation.
    """

    rng = Random(seed)
    checked = 0
    m, n, k = 5, 3, 2
    prefixes = tuple(product((0, 1), repeat=m))
    futures = tuple(product((0, 1), repeat=n))

    for _ in range(trials):
        state_count = rng.randint(2, 5)

        # The root configuration includes both finite control and placement.
        root_map: dict[tuple[int, ...], tuple[int, tuple[int, ...]]] = {}
        for prefix in prefixes:
            q = (
                sum((i + 1) * bit for i, bit in enumerate(prefix))
                + rng.randrange(state_count)
            ) % state_count
            placement = (
                4 - prefix[0],
                3 - prefix[1],
            )
            root_map[prefix] = (q, placement)

        move_table = {}
        state_table = {}
        for step in range(n):
            for q in range(state_count):
                for incoming in (0, 1):
                    for reads in product((0, 1), repeat=k):
                        key = (step, q, incoming, reads)
                        move_table[key] = tuple(
                            rng.choice((0, -1)) for _ in range(k)
                        )
                        state_table[key] = rng.randrange(state_count)

        output_table = {
            (q, reads): rng.randrange(2)
            for q in range(state_count)
            for reads in product((0, 1), repeat=k)
        }

        def read(prefix: tuple[int, ...], position: int) -> int:
            return prefix[position] if 0 <= position < m else 0

        def run(prefix: tuple[int, ...], future: tuple[int, ...]) -> int:
            q, initial = root_map[prefix]
            positions = list(initial)
            for step, incoming in enumerate(future):
                observed = tuple(read(prefix, p) for p in positions)
                key = (step, q, incoming, observed)
                moves = move_table[key]
                positions = [p + move for p, move in zip(positions, moves)]
                q = state_table[key]
            observed = tuple(read(prefix, p) for p in positions)
            return output_table[(q, observed)]

        # Equal root configuration plus equal full cone restriction must give
        # equal output on every common continuation.
        groups: dict[
            tuple[tuple[int, tuple[int, ...]], tuple[int, ...]],
            list[tuple[int, ...]],
        ] = {}
        for prefix in prefixes:
            config = root_map[prefix]
            accessible = cone(m, n, config[1])
            key = (config, tuple(prefix[i] for i in accessible))
            groups.setdefault(key, []).append(prefix)

        for group in groups.values():
            for future in futures:
                outputs = {run(prefix, future) for prefix in group}
                assert len(outputs) == 1
                checked += 1

    return checked


def spaced_capacity_checks(max_r: int = 9) -> int:
    checked = 0
    for r in range(max_r + 1):
        d = 2**r
        data_cells = tuple(j * (r + 1) for j in range(d))
        final = data_cells[-1]
        for root in range(-1, final + r + 2):
            reachable = set(range(max(0, root - r), root + 1))
            assert len(reachable.intersection(data_cells)) <= 1
            checked += 1
    return checked


def subcubes(d: int, free_limit: int) -> tuple[int, ...]:
    """All Boolean subcubes with at most free_limit varying coordinates.

    A bit in the returned mask denotes one data vector covered by the cube.
    """

    cubes: set[int] = set()
    coordinates = range(d)
    for free_size in range(min(d, free_limit) + 1):
        for free in combinations(coordinates, free_size):
            fixed = tuple(i for i in coordinates if i not in free)
            for fixed_values in product((0, 1), repeat=len(fixed)):
                mask = 0
                for free_values in product((0, 1), repeat=free_size):
                    vector = [0] * d
                    for i, value in zip(fixed, fixed_values):
                        vector[i] = value
                    for i, value in zip(free, free_values):
                        vector[i] = value
                    word = 0
                    for value in vector:
                        word = (word << 1) | value
                    mask |= 1 << word
                cubes.add(mask)
    return tuple(sorted(cubes))


def exact_subcube_cover(d: int, k: int) -> int:
    universe = (1 << (2**d)) - 1
    candidates = subcubes(d, k)
    by_point: list[list[int]] = [[] for _ in range(2**d)]
    for candidate in candidates:
        for point in range(2**d):
            if candidate & (1 << point):
                by_point[point].append(candidate)

    @lru_cache(maxsize=None)
    def solve(covered: int) -> int:
        if covered == universe:
            return 0
        missing = universe ^ covered
        point = (missing & -missing).bit_length() - 1
        return 1 + min(
            solve(covered | candidate) for candidate in by_point[point]
        )

    return solve(0)


def exact_frontier_checks() -> dict[int, tuple[int, ...]]:
    results = {}
    for d in (1, 2, 3, 4):
        frontier = tuple(exact_subcube_cover(d, k) for k in range(d + 1))
        expected = tuple(2 ** (d - k) for k in range(d + 1))
        assert frontier == expected
        results[d] = frontier
    return results


def explicit_index_decoder_checks(max_r: int = 4) -> int:
    checks = 0
    for r in range(max_r + 1):
        d = 2**r
        for k in range(d + 1):
            tracked = set(range(k))
            for word in range(2**d):
                data = bits(word, d)
                retained = tuple(data[i] for i in range(d) if i not in tracked)
                retained_index = {
                    i: j for j, i in enumerate(i for i in range(d) if i not in tracked)
                }
                for selected in range(d):
                    if selected in tracked:
                        output = data[selected]
                    else:
                        output = retained[retained_index[selected]]
                    assert output == data[selected]
                    checks += 1
    return checks


def run_checks() -> dict[str, object]:
    return {
        "random_protocol_factor_checks": random_protocol_checks(),
        "spaced_capacity_checks": spaced_capacity_checks(),
        "exact_small_frontiers": exact_frontier_checks(),
        "explicit_upper_decoder_checks": explicit_index_decoder_checks(),
        "violations": 0,
    }


if __name__ == "__main__":
    print(run_checks())