check_native_observation_shift.py

#!/usr/bin/env python3
"""Finite audit of native LMR views versus port-word endpoint observations."""

from collections import defaultdict
from dataclasses import dataclass
from itertools import product
import json
from pathlib import Path


MISSING = -1


@dataclass(frozen=True)
class Graph:
    labels: tuple
    edges: tuple

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


def base(degree):
    return Graph((None,), ((MISSING,) * degree,))


def graphs(degree, size):
    if size == 1:
        yield base(degree)
        return
    for graph in graphs(degree, size - 1):
        for label in (0, 1):
            for row in product(range(MISSING, size), repeat=degree):
                yield Graph(graph.labels + (label,), graph.edges + (row,))


def words(degree, horizon):
    return tuple(word for length in range(horizon + 1)
                 for word in product(range(degree), repeat=length))


def follow(graph, node, word):
    for port in word:
        if node == MISSING:
            break
        node = graph.edges[node][port]
    return node


def endpoint_observation(graph, node):
    if node == MISSING:
        return ("missing",)
    if graph.labels[node] is None:
        return ("base",)
    return ("label", graph.labels[node])


def port_table(graph, node, horizon):
    return tuple((word, endpoint_observation(graph, follow(graph, node, word)))
                 for word in words(len(graph.edges[0]), horizon))


def native_view(graph, node, radius):
    if node == MISSING:
        return None
    if radius == 0:
        return graph.labels[node]
    return (graph.labels[node], tuple(native_view(graph, child, radius - 1)
                                      for child in graph.edges[node]))


def truncate_port_table(table, degree, radius):
    observations = dict(table)

    def build(prefix, remaining):
        observation = observations[prefix]
        if observation[0] == "missing":
            return None
        label = None if observation[0] == "base" else observation[1]
        if remaining == 0:
            return label
        return (label, tuple(build(prefix + (port,), remaining - 1)
                             for port in range(degree)))

    return build((), radius)


def assert_functional_buckets(pairs, left_index, right_index):
    buckets = defaultdict(set)
    for pair in pairs:
        buckets[pair[left_index]].add(pair[right_index])
    assert all(len(values) == 1 for values in buckets.values())
    return len(buckets)


def stable_partition(all_graphs, native):
    states = [(index, node) for index, graph in enumerate(all_graphs)
              for node in range(len(graph.labels))]
    states.append((-1, MISSING))

    def label_of(state):
        index, node = state
        if node == MISSING:
            return None if native else ("missing",)
        label = all_graphs[index].labels[node]
        if native:
            return label
        return ("base",) if label is None else ("label", label)

    colors = {state: label_of(state) for state in states}
    rounds = 0
    while True:
        signatures = {}
        for state in states:
            index, node = state
            if node == MISSING:
                signature = ("missing-native",) if native else (
                    label_of(state), tuple(colors[state] for _ in range(len(all_graphs[0].edges[0])))
                )
            else:
                graph = all_graphs[index]
                children = tuple((-1, MISSING) if child == MISSING else (index, child)
                                 for child in graph.edges[node])
                signature = (label_of(state), tuple(colors[child] for child in children))
            signatures[state] = signature
        palette = {signature: number for number, signature in
                   enumerate(sorted(set(signatures.values()), key=repr))}
        updated = {state: palette[signature] for state, signature in signatures.items()}
        rounds += 1
        old_classes = defaultdict(set)
        new_classes = defaultdict(set)
        for state in states:
            old_classes[colors[state]].add(state)
            new_classes[updated[state]].add(state)
        if {frozenset(group) for group in old_classes.values()} == {
                frozenset(group) for group in new_classes.values()}:
            return updated, rounds, len(states)
        colors = updated


def audit_degree(degree, max_size):
    all_graphs = [graph for size in range(1, max_size + 1)
                  for graph in graphs(degree, size)]
    reconstruction_checks = 0
    native_to_port_buckets = 0
    deeper_native_to_port_buckets = 0
    frontier_collision_horizons = []
    deeper_strict_horizons = []

    for horizon in range(4):
        rows = []
        for graph in all_graphs:
            table = port_table(graph, graph.top, horizon)
            native = native_view(graph, graph.top, horizon)
            deeper = native_view(graph, graph.top, horizon + 1)
            assert truncate_port_table(table, degree, horizon) == native
            reconstruction_checks += 1
            rows.append((native, table, deeper))

        native_to_port = defaultdict(set)
        port_to_native = defaultdict(set)
        deeper_to_port = defaultdict(set)
        for native, table, deeper in rows:
            native_to_port[repr(native)].add(repr(table))
            port_to_native[repr(table)].add(repr(native))
            deeper_to_port[repr(deeper)].add(repr(table))
        assert all(len(values) == 1 for values in port_to_native.values())
        assert all(len(values) == 1 for values in deeper_to_port.values())
        native_to_port_buckets += len(port_to_native)
        deeper_native_to_port_buckets += len(deeper_to_port)
        if any(len(values) > 1 for values in native_to_port.values()):
            frontier_collision_horizons.append(horizon)
        if len({repr(row[2]) for row in rows}) > len({repr(row[1]) for row in rows}):
            deeper_strict_horizons.append(horizon)

    native_colors, native_rounds, state_count = stable_partition(all_graphs, native=True)
    causal_colors, causal_rounds, _ = stable_partition(all_graphs, native=False)
    native_to_causal = defaultdict(set)
    causal_to_native = defaultdict(set)
    for state in native_colors:
        native_to_causal[native_colors[state]].add(causal_colors[state])
        causal_to_native[causal_colors[state]].add(native_colors[state])
    assert all(len(values) == 1 for values in native_to_causal.values())
    assert all(len(values) == 1 for values in causal_to_native.values())

    return {
        "graphs": len(all_graphs),
        "states_in_joint_partition": state_count,
        "recursive_truncation_checks": reconstruction_checks,
        "port_to_native_equivalence_buckets": native_to_port_buckets,
        "deeper_native_to_port_equivalence_buckets": deeper_native_to_port_buckets,
        "same_radius_frontier_collision_horizons": frontier_collision_horizons,
        "deeper_native_strict_horizons": deeper_strict_horizons,
        "native_partition_rounds": native_rounds,
        "causal_partition_rounds": causal_rounds,
        "stable_equivalences_match": True,
    }


def boundary_witnesses():
    result = {}
    for horizon in range(1, 5):
        labels = (None,) + (0,) * horizon
        base_edges = [(MISSING,)]
        missing_edges = [(MISSING,)]
        for node in range(1, horizon + 1):
            target = node - 1
            base_edges.append((target,))
            missing_edges.append((MISSING if node == 1 else target,))
        left = Graph(labels, tuple(base_edges))
        right = Graph(labels, tuple(missing_edges))
        assert native_view(left, left.top, horizon) == native_view(right, right.top, horizon)
        assert port_table(left, left.top, horizon) != port_table(right, right.top, horizon)
        assert native_view(left, left.top, horizon + 1) != native_view(right, right.top, horizon + 1)
        result[str(horizon)] = True
    return result


def statement_markers():
    root = Path(__file__).resolve().parents[2]
    causal = (root / "research-artifacts/scaffold-proof-review/history/objects/"
              "2b0910f423c0a56136a7d1c293c57ff5f6361ee40bf5d777aa4d1da7aaaae716.txt").read_text()
    t153 = (root / "research-artifacts/finite-horizon-scaffold-capacity/finite-horizon-capacity.md").read_text()
    t159 = (root / "research-artifacts/distance-one-collapse/distance-one-collapse.md").read_text()
    t160 = (root / "research-artifacts/distance-two-normal-form/distance-two-normal-form.md").read_text()
    t161 = (root / "research-artifacts/structural-scaffold-normal-form/structural-scaffold-normal-form.md").read_text()
    checks = {
        "causal_same_radius_sentence_present": "exactly the restriction of the unfolding" in causal,
        "causal_all_depth_statement_present": "equality of every finite rooted neighborhood" in causal,
        "t153_native_base_missing_count_present": "a missing node and an unlabelled node have the same observation" in t153,
        "t159_uses_native_radius_one_summary": "s(p)=(q_p,N_1(p))" in t159,
        "t160_radius_one_reconstruction_sentence_present": "radius-one view of a shortcut-closed node" in t160,
        "t161_radius_two_reconstruction_present": "radius-two view reconstructs the complete source" in t161,
    }
    assert all(checks.values())
    return checks


result = {
    "status": "pass",
    "model": "LMR native recursive neighbourhoods with an unlabelled base and missing endpoint",
    "degree_one_through_five_nodes": audit_degree(1, 5),
    "degree_two_through_three_nodes": audit_degree(2, 3),
    "frontier_base_missing_witnesses": boundary_witnesses(),
    "statement_markers": statement_markers(),
    "proved_relation": {
        "native_m_is_recursive_truncation_of_port_words_through_m": True,
        "port_word_m_equality_implies_native_m_equality": True,
        "native_m_plus_one_equality_implies_port_word_m_equality": True,
        "all_native_depths_equal_iff_full_unfoldings_equal": True,
    },
    "scope": "Finite audit of the exact comparison and source markers; the unbounded equivalence is proved in Run 07.",
}

print(json.dumps(result, indent=2, sort_keys=True))