verify_replay_slice.py

#!/usr/bin/env python3
"""Exhaustive small-model falsifier for the adaptive replay-slice theorem.

The proof is the well-founded replay induction in the public note. This checker
enumerates every two-source, two-derived-cell adaptive history in a small model,
every observer, every source assignment, and every perturbation that agrees on
the computed replay slice. It also checks the exact old-history pullback law for
the split between the first and second derived cells.
"""

from __future__ import annotations

from itertools import product
import json


SOURCES = ("s0", "s1")
DERIVED = ("d0", "d1")


def specs(refs: tuple[str, ...]):
    return product(refs, repeat=3)


def run_adaptive(spec: tuple[str, str, str], values: dict[str, int]):
    selector, zero_ref, one_ref = spec
    selector_value = values[selector]
    chosen = one_ref if selector_value else zero_ref
    trace = (selector, chosen)
    return values[chosen] ^ selector_value, trace


def execute(
    source_values: tuple[int, int],
    d0_spec: tuple[str, str, str],
    d1_spec: tuple[str, str, str],
    observer_spec: tuple[str, str, str],
):
    values = dict(zip(SOURCES, source_values, strict=True))
    traces: dict[str, tuple[str, ...]] = {}
    values["d0"], traces["d0"] = run_adaptive(d0_spec, values)
    values["d1"], traces["d1"] = run_adaptive(d1_spec, values)
    output, observer_trace = run_adaptive(observer_spec, values)

    support = set(observer_trace)
    pending = [name for name in DERIVED if name in support]
    seen = set()
    while pending:
        name = pending.pop()
        if name in seen:
            continue
        seen.add(name)
        support.update(traces[name])
        pending.extend(
            ref for ref in traces[name] if ref in DERIVED and ref not in seen
        )

    return {
        "values": values,
        "traces": traces,
        "observer_trace": observer_trace,
        "output": output,
        "support": frozenset(support),
    }


def close_in_base(seeds: set[str], base_trace: tuple[str, ...]):
    closure = set(seeds)
    if "d0" in closure:
        closure.update(base_trace)
    return closure


def main():
    spec_count = 0
    executions = 0
    perturbation_pairs = 0
    supported_cell_checks = 0
    composition_checks = 0
    shallow_slice_failures = 0
    first_shallow_witness = None

    for d0_spec in specs(SOURCES):
        for d1_spec in specs(SOURCES + ("d0",)):
            for observer_spec in specs(SOURCES + DERIVED):
                spec_count += 1
                for source_values in product((0, 1), repeat=2):
                    executions += 1
                    left = execute(source_values, d0_spec, d1_spec, observer_spec)
                    supported_sources = left["support"].intersection(SOURCES)

                    for changed_values in product((0, 1), repeat=2):
                        if any(
                            changed_values[SOURCES.index(name)]
                            != source_values[SOURCES.index(name)]
                            for name in supported_sources
                        ):
                            continue
                        perturbation_pairs += 1
                        right = execute(
                            changed_values, d0_spec, d1_spec, observer_spec
                        )
                        assert right["output"] == left["output"]
                        assert right["observer_trace"] == left["observer_trace"]
                        assert right["support"] == left["support"]
                        for name in DERIVED:
                            if name in left["support"]:
                                supported_cell_checks += 1
                                assert right["values"][name] == left["values"][name]
                                assert right["traces"][name] == left["traces"][name]

                    # Composition split: H contains d0; K contains d1 and O.
                    continuation_supported = {
                        name for name in ("d1",) if name in left["support"]
                    }
                    old_cells = set(SOURCES + ("d0",))
                    old_demands = set(left["observer_trace"]).intersection(old_cells)
                    for name in continuation_supported:
                        old_demands.update(
                            set(left["traces"][name]).intersection(old_cells)
                        )
                    pulled_back = close_in_base(
                        old_demands, left["traces"]["d0"]
                    )
                    assert pulled_back == set(left["support"]).intersection(old_cells)
                    composition_checks += 1

                    # Negative control: direct observer queries alone are not a replay slice.
                    shallow_sources = set(left["observer_trace"]).intersection(SOURCES)
                    if shallow_sources != supported_sources:
                        for changed_values in product((0, 1), repeat=2):
                            if any(
                                changed_values[SOURCES.index(name)]
                                != source_values[SOURCES.index(name)]
                                for name in shallow_sources
                            ):
                                continue
                            right = execute(
                                changed_values, d0_spec, d1_spec, observer_spec
                            )
                            if (
                                right["output"] != left["output"]
                                or right["observer_trace"] != left["observer_trace"]
                                or right["support"] != left["support"]
                            ):
                                shallow_slice_failures += 1
                                if first_shallow_witness is None:
                                    first_shallow_witness = {
                                        "source": source_values,
                                        "changed": changed_values,
                                        "d0_spec": d0_spec,
                                        "d1_spec": d1_spec,
                                        "observer_spec": observer_spec,
                                        "observer_queries": left["observer_trace"],
                                        "full_support": sorted(left["support"]),
                                        "left_output": left["output"],
                                        "right_output": right["output"],
                                    }
                                break

    assert first_shallow_witness is not None
    report = {
        "ok": True,
        "model": {
            "source_cells": 2,
            "derived_cells": 2,
            "adaptive_program": "query selector, then one of two references",
        },
        "counts": {
            "program_observer_combinations": spec_count,
            "executions": executions,
            "slice_preserving_perturbation_pairs": perturbation_pairs,
            "supported_cell_checks": supported_cell_checks,
            "composition_pullback_checks": composition_checks,
            "shallow_slice_failures": shallow_slice_failures,
        },
        "negative_control": {
            "description": "Observer queries without transitive dependency closure are insufficient.",
            "first_witness": first_shallow_witness,
        },
    }
    print(json.dumps(report, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()