verify_regular_selection_horizon.py

#!/usr/bin/env python3
"""Exhaustive finite audit for FPRD-T173."""

from __future__ import annotations

from collections import deque
import itertools
import json
from pathlib import Path


BITS = (0, 1)


def source_dfas():
    """All complete labeled two-state DFAs over (visible bit, hidden tag)."""
    for entries in itertools.product(BITS, repeat=8):
        table = (tuple(entries[:4]), tuple(entries[4:]))
        for accepting_mask in range(4):
            accepting = frozenset(
                state for state in BITS if accepting_mask & (1 << state)
            )
            yield table, accepting


def selectors(states):
    """All labeled synchronous selectors with the requested state count."""
    choices = tuple(itertools.product(range(states), BITS))
    return tuple(
        table
        for table in itertools.product(choices, repeat=states * len(BITS))
    )


def source_step(table, state, visible, tag):
    return table[state][2 * visible + tag]


def selector_step(selector, state, visible):
    return selector[2 * state + visible]


def accepting_product_state(node, accepting):
    source_state, coordinates = node
    return source_state in accepting and all(
        selected_source_state not in accepting
        for _, selected_source_state in coordinates
    )


def shortest_missed_word(dfa, family):
    """BFS in the exact missed-word NFA from the proof."""
    table, accepting = dfa
    start = (0, tuple((0, 0) for _ in family))
    queue = deque([(start, ())])
    seen = {start}
    transitions = 0

    while queue:
        node, word = queue.popleft()
        if accepting_product_state(node, accepting):
            return word, len(seen), transitions

        source_state, coordinates = node
        for visible in BITS:
            selected_coordinates = []
            for selector, (selector_state, selected_source_state) in zip(
                family, coordinates
            ):
                next_selector_state, output_tag = selector_step(
                    selector, selector_state, visible
                )
                next_selected_source = source_step(
                    table, selected_source_state, visible, output_tag
                )
                selected_coordinates.append(
                    (next_selector_state, next_selected_source)
                )

            for hidden_tag in BITS:
                transitions += 1
                next_source = source_step(
                    table, source_state, visible, hidden_tag
                )
                target = (next_source, tuple(selected_coordinates))
                if target not in seen:
                    seen.add(target)
                    queue.append((target, word + (visible,)))

    return None, len(seen), transitions


def selected_word(selector, word):
    state = 0
    output = []
    for visible in word:
        state, tag = selector_step(selector, state, visible)
        output.append((visible, tag))
    return tuple(output)


def source_accepts(dfa, tagged_word):
    table, accepting = dfa
    state = 0
    for visible, tag in tagged_word:
        state = source_step(table, state, visible, tag)
    return state in accepting


def projected_accepts(dfa, word):
    return any(
        source_accepts(dfa, tuple(zip(word, tags)))
        for tags in itertools.product(BITS, repeat=len(word))
    )


def verify_witness(dfa, family, word):
    assert projected_accepts(dfa, word)
    assert all(
        not source_accepts(dfa, selected_word(selector, word))
        for selector in family
    )


def audit(families, selector_states):
    family_count = 0
    counterexamples = 0
    global_covers = 0
    visited_states = 0
    transition_checks = 0
    witness_checks = 0
    maximum_witness_length = -1
    tight_horizon_instances = 0

    for dfa in source_dfas():
        for family in families:
            family_count += 1
            width = len(family)
            bound = (2 ** (width + 1)) * (selector_states ** width)
            word, reached, transitions = shortest_missed_word(dfa, family)
            assert reached <= bound
            visited_states += reached
            transition_checks += transitions

            if word is None:
                global_covers += 1
                continue

            counterexamples += 1
            witness_checks += 1
            verify_witness(dfa, family, word)
            assert len(word) < bound
            maximum_witness_length = max(maximum_witness_length, len(word))
            if len(word) == bound - 1:
                tight_horizon_instances += 1

    return {
        "family_cases": family_count,
        "counterexample_cases": counterexamples,
        "global_cover_cases": global_covers,
        "visited_product_states": visited_states,
        "product_transition_checks": transition_checks,
        "direct_witness_checks": witness_checks,
        "maximum_shortest_witness_length": maximum_witness_length,
        "tight_horizon_instances": tight_horizon_instances,
    }


def main():
    one_state = selectors(1)
    one_state_families = tuple((selector,) for selector in one_state) + tuple(
        itertools.combinations(one_state, 2)
    )
    two_state = selectors(2)
    two_state_singletons = tuple((selector,) for selector in two_state)

    first = audit(one_state_families, selector_states=1)
    second = audit(two_state_singletons, selector_states=2)

    output = {
        "status": "pass",
        "scope": "complete two-state regular sources with bounded synchronous selectors",
        "two_state_source_dfas": 1024,
        "one_state_selectors": len(one_state),
        "one_state_family_widths": [1, 2],
        "one_state_audit": first,
        "two_state_selectors": len(two_state),
        "two_state_family_widths": [1],
        "two_state_audit": second,
        "verified_bound": "shortest missed word length < r^(m+1) s^m",
        "does_not_verify": [
            "tightness of the general product-state bound",
            "asynchronous selectors",
            "effective horizons for nonregular hidden sources",
            "the Kim-Park separation used by FPRD-T168",
        ],
    }
    Path(__file__).with_name("checker-output.json").write_text(
        json.dumps(output, indent=2) + "\n", encoding="utf-8"
    )
    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()