verify_future_observers.py

#!/usr/bin/env python3
"""Finite audit for FPRD-T158 finite future observers."""

from __future__ import annotations

import itertools
import json
from dataclasses import dataclass
from pathlib import Path


MISSING = -1
SELF = -2


@dataclass(frozen=True)
class Scaffold:
    labels: tuple[int, ...]
    edges: tuple[tuple[int, ...], ...]

    @property
    def top(self) -> int:
        return len(self.labels) - 1


def view(scaffold: Scaffold, node: int, radius: int):
    if node == MISSING:
        return ("missing",)
    if radius == 0:
        return ("label", scaffold.labels[node])
    return (
        "node",
        scaffold.labels[node],
        tuple(view(scaffold, target, radius - 1) for target in scaffold.edges[node]),
    )


def radius(distance: int, horizon: int) -> int:
    if horizon == 0 or distance == 0:
        return 0
    return distance + (horizon - 1) * (distance - 1)


def descriptor_words(degree: int, distance: int) -> list[tuple[int, ...]]:
    return [
        word
        for length in range(distance + 1)
        for word in itertools.product(range(degree), repeat=length)
    ]


def follow(scaffold: Scaffold, start: int, descriptor: tuple[int, ...]) -> int:
    node = start
    for port in descriptor:
        if node == MISSING:
            return MISSING
        node = scaffold.edges[node][port]
    return node


def extend(scaffold: Scaffold, label: int, descriptors: tuple[object, ...]) -> Scaffold:
    new = len(scaffold.labels)
    targets: list[int] = []
    for descriptor in descriptors:
        if descriptor == SELF:
            targets.append(new)
        elif descriptor == MISSING:
            targets.append(MISSING)
        else:
            targets.append(follow(scaffold, scaffold.top, descriptor))
    return Scaffold(scaffold.labels + (label,), scaffold.edges + (tuple(targets),))


def score(item) -> int:
    if isinstance(item, int):
        return item + 17
    if isinstance(item, str):
        return sum(ord(char) for char in item)
    return 19 + sum((index + 3) * score(value) for index, value in enumerate(item))


@dataclass(frozen=True)
class TinyMachine:
    seed: int
    degree: int
    distance: int

    def transition(self, state: int, symbol: int, scaffold: Scaffold) -> tuple[int, int, tuple[object, ...]]:
        root_view = ("missing",) if not scaffold.labels else view(scaffold, scaffold.top, self.distance)
        code = score((self.seed, state, symbol, root_view))
        options: list[object] = [MISSING, SELF, *descriptor_words(self.degree, self.distance)]
        next_state = (state + symbol + code + self.seed) % 3
        label = (code // 3 + state + symbol) & 1
        descriptors = tuple(options[(code + 7 * port) % len(options)] for port in range(self.degree))
        return next_state, label, descriptors

    def accepting(self, state: int) -> int:
        return int(state == (self.seed % 3))


def step(machine: TinyMachine, state: int, scaffold: Scaffold, symbol: int) -> tuple[int, Scaffold]:
    next_state, label, descriptors = machine.transition(state, symbol, scaffold)
    return next_state, extend(scaffold, label, descriptors)


def run_from(machine: TinyMachine, state: int, scaffold: Scaffold, word: tuple[int, ...]) -> tuple[int, Scaffold]:
    for symbol in word:
        state, scaffold = step(machine, state, scaffold, symbol)
    return state, scaffold


def answer_from(machine: TinyMachine, state: int, scaffold: Scaffold, word: tuple[int, ...]) -> int:
    final_state, _ = run_from(machine, state, scaffold, word)
    return machine.accepting(final_state)


def empty_scaffold(degree: int) -> Scaffold:
    return Scaffold((), ())


def scaffold_family(node_count: int, degree: int):
    label_choices = itertools.product(range(2), repeat=node_count)
    edge_options = [
        list(itertools.product([MISSING, *range(node)], repeat=degree))
        for node in range(node_count)
    ]
    for labels in label_choices:
        for edges in itertools.product(*edge_options):
            yield Scaffold(tuple(labels), tuple(edges))


def observer(machine: TinyMachine, state: int, scaffold: Scaffold, suffixes: tuple[tuple[int, ...], ...]) -> tuple[int, ...]:
    return tuple(answer_from(machine, state, scaffold, suffix) for suffix in suffixes)


def audit_compiled_control() -> dict[str, int]:
    machines = [TinyMachine(seed, 1 + seed % 2, 1 + (seed // 2) % 2) for seed in range(8)]
    suffix_pool = [word for length in range(3) for word in itertools.product((0, 1), repeat=length)]
    observer_families = tuple(
        tuple(suffix_pool[(seed + offset * offset + 2 * offset) % len(suffix_pool)] for offset in range(3))
        for seed in range(7)
    )
    prefixes = [word for length in range(7) for word in itertools.product((0, 1), repeat=length)]

    compiled_runs = 0
    observer_invariant_checks = 0
    graph_identity_checks = 0
    boolean_output_checks = 0

    for machine in machines:
        for suffixes in observer_families:
            for prefix in prefixes:
                base = empty_scaffold(machine.degree)
                direct_state, direct_graph = run_from(machine, 0, base, prefix)

                compiled_state = 0
                compiled_graph = base
                compiled_vector = observer(machine, compiled_state, compiled_graph, suffixes)
                for symbol in prefix:
                    old_state, old_graph = compiled_state, compiled_graph
                    compiled_state, compiled_graph = step(machine, old_state, old_graph, symbol)
                    compiled_vector = tuple(
                        answer_from(machine, old_state, old_graph, (symbol,) + suffix)
                        for suffix in suffixes
                    )

                assert compiled_state == direct_state
                assert compiled_graph == direct_graph
                assert compiled_vector == observer(machine, direct_state, direct_graph, suffixes)
                observer_invariant_checks += len(suffixes)
                graph_identity_checks += 1

                vector_index = sum(bit << index for index, bit in enumerate(compiled_vector))
                direct_vector = observer(machine, direct_state, direct_graph, suffixes)
                direct_index = sum(bit << index for index, bit in enumerate(direct_vector))
                for truth_table in range(1 << (1 << len(suffixes))):
                    compiled_answer = (truth_table >> vector_index) & 1
                    assert compiled_answer == ((truth_table >> direct_index) & 1)
                    boolean_output_checks += 1
                compiled_runs += 1

    return {
        "tiny_machines": len(machines),
        "observer_families": len(observer_families),
        "compiled_runs": compiled_runs,
        "observer_invariant_checks": observer_invariant_checks,
        "graph_identity_checks": graph_identity_checks,
        "boolean_output_checks": boolean_output_checks,
    }


def audit_locality() -> dict[str, int]:
    families = [
        (TinyMachine(0, 1, 2), 5),
        (TinyMachine(4, 2, 1), 4),
    ]
    suffixes = ((), (0,), (1, 0))
    old_scaffolds = 0
    future_observer_evaluations = 0
    same_view_comparisons = 0
    readable_view_classes = 0

    for machine, node_count in families:
        material = list(scaffold_family(node_count, machine.degree))
        old_scaffolds += len(material)
        distance = max(radius(machine.distance, len(suffix) + 1) for suffix in suffixes)
        for state in range(3):
            for symbol in (0, 1):
                representatives: dict[object, tuple[int, ...]] = {}
                for scaffold in material:
                    key = view(scaffold, scaffold.top, distance)
                    signature = tuple(
                        answer_from(machine, state, scaffold, (symbol,) + suffix)
                        for suffix in suffixes
                    )
                    future_observer_evaluations += len(suffixes)
                    if key in representatives:
                        assert representatives[key] == signature
                        same_view_comparisons += 1
                    else:
                        representatives[key] = signature
                readable_view_classes += len(representatives)

    return {
        "old_scaffold_machine_instances": old_scaffolds,
        "future_observer_evaluations": future_observer_evaluations,
        "same_view_signature_comparisons": same_view_comparisons,
        "readable_view_classes": readable_view_classes,
    }


def chain(depth: int, target: int) -> Scaffold:
    labels = tuple([target] + [0] * depth)
    edges = tuple((MISSING,) if node == 0 else (node - 1,) for node in range(depth + 1))
    return Scaffold(labels, edges)


def reveal_after_horizon(scaffold: Scaffold, distance: int, horizon: int) -> int:
    path = (0,) * distance
    current = scaffold
    for _ in range(horizon - 1):
        current = extend(current, 0, (path,))
    target = follow(current, current.top, path)
    return current.labels[target]


def audit_sharpness() -> dict[str, int]:
    sharp_radius_witnesses = 0
    equal_smaller_views = 0
    separated_observer_updates = 0
    for distance in range(1, 5):
        for suffix_length in range(6):
            horizon = suffix_length + 1
            required = radius(distance, horizon)
            left = chain(required, 0)
            right = chain(required, 1)
            assert view(left, left.top, required - 1) == view(right, right.top, required - 1)
            assert reveal_after_horizon(left, distance, horizon) == 0
            assert reveal_after_horizon(right, distance, horizon) == 1
            sharp_radius_witnesses += 1
            equal_smaller_views += 1
            separated_observer_updates += 1
    return {
        "sharp_radius_witnesses": sharp_radius_witnesses,
        "equal_radius_minus_one_views": equal_smaller_views,
        "separated_observer_updates": separated_observer_updates,
    }


def audit_radius_formula() -> dict[str, int]:
    checks = 0
    for distance in range(5):
        for suffix_length in range(9):
            expected = 0 if distance == 0 else distance + suffix_length * (distance - 1)
            assert radius(distance, suffix_length + 1) == expected
            assert radius(distance, suffix_length + 1) >= distance
            checks += 2
    return {"radius_formula_checks": checks}


def main() -> None:
    result: dict[str, object] = {
        "scope": {
            "compiled_control": "8 deterministic scaffold machines, 7 three-suffix observers, all binary prefixes through length 6, all 256 Boolean observers",
            "locality": "all binary-labelled backward scaffolds: degree 1 on 5 nodes and degree 2 on 4 nodes",
            "sharpness": "distance 1..4 and suffix length 0..5",
        },
        **audit_compiled_control(),
        **audit_locality(),
        **audit_sharpness(),
        **audit_radius_formula(),
        "status": "pass",
    }
    output = Path(__file__).with_name("checker-output.json")
    output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()