verify_partial_restoration.py

#!/usr/bin/env python3
"""Finite certificates for partial restoration and reversible covers.

The script minimizes finite partial Moore systems with undefinedness included
in the behavior, tests whether each named operation is injective on its
defined domain, and records five boundary examples.  It proves no infinite-
state theorem and performs no scaffold compilation.
"""

from collections import deque
from dataclasses import dataclass
import json
from pathlib import Path


@dataclass(frozen=True)
class PartialMoore:
    states: tuple[str, ...]
    alphabet: tuple[str, ...]
    starts: tuple[str, ...]
    output: dict[str, object]
    delta: dict[tuple[str, str], str]


def reachable(machine: PartialMoore) -> set[str]:
    seen = set(machine.starts)
    queue = deque(machine.starts)
    while queue:
        state = queue.popleft()
        for letter in machine.alphabet:
            target = machine.delta.get((state, letter))
            if target is not None and target not in seen:
                seen.add(target)
                queue.append(target)
    return seen


def minimize(machine: PartialMoore):
    states = reachable(machine)
    by_output: dict[object, set[str]] = {}
    for state in sorted(states):
        by_output.setdefault(machine.output[state], set()).add(state)
    partition = list(by_output.values())

    while True:
        owner = {
            state: index
            for index, block in enumerate(partition)
            for state in block
        }
        refined: dict[tuple, set[str]] = {}
        for state in sorted(states):
            signature = (
                machine.output[state],
                tuple(
                    None
                    if machine.delta.get((state, letter)) is None
                    else owner[machine.delta[state, letter]]
                    for letter in machine.alphabet
                ),
            )
            refined.setdefault(signature, set()).add(state)
        next_partition = list(refined.values())
        if {frozenset(block) for block in next_partition} == {
            frozenset(block) for block in partition
        }:
            partition = next_partition
            break
        partition = next_partition

    partition = sorted(partition, key=lambda block: tuple(sorted(block)))
    owner = {
        state: index
        for index, block in enumerate(partition)
        for state in block
    }
    quotient_delta = {}
    for index, block in enumerate(partition):
        representative = min(block)
        for letter in machine.alphabet:
            target = machine.delta.get((representative, letter))
            if target is not None:
                quotient_delta[index, letter] = owner[target]

    quotient = PartialMoore(
        states=tuple(str(index) for index in range(len(partition))),
        alphabet=machine.alphabet,
        starts=tuple(str(owner[state]) for state in machine.starts),
        output={
            str(index): machine.output[min(block)]
            for index, block in enumerate(partition)
        },
        delta={
            (str(state), letter): str(target)
            for (state, letter), target in quotient_delta.items()
        },
    )
    return quotient, [sorted(block) for block in partition]


def operation_audit(machine: PartialMoore):
    operations = {}
    all_injective = True
    for letter in machine.alphabet:
        arrows = sorted(
            (state, machine.delta[state, letter])
            for state in machine.states
            if (state, letter) in machine.delta
        )
        targets: dict[str, list[str]] = {}
        for source, target in arrows:
            targets.setdefault(target, []).append(source)
        merges = [
            {"target": target, "sources": sources}
            for target, sources in sorted(targets.items())
            if len(sources) > 1
        ]
        injective = not merges
        all_injective = all_injective and injective
        operations[letter] = {
            "defined_arrows": arrows,
            "partial_injective": injective,
            "merge_witnesses": merges,
        }
    return all_injective, operations


def analyze(name: str, machine: PartialMoore):
    quotient, blocks = minimize(machine)
    injective, operations = operation_audit(quotient)
    source_injective, source_operations = operation_audit(machine)
    return {
        "name": name,
        "reachable_source_states": len(reachable(machine)),
        "residual_states": len(quotient.states),
        "residual_blocks": blocks,
        "source_operations_partial_injective": source_injective,
        "residual_operations_partial_injective": injective,
        "residual_operations": operations,
        "source_merge_witnesses": {
            letter: audit["merge_witnesses"]
            for letter, audit in source_operations.items()
            if audit["merge_witnesses"]
        },
    }


def fork_forward():
    return PartialMoore(
        states=("x0", "x1", "y0", "y1"),
        alphabet=("+h",),
        starts=("x0", "x1", "y0", "y1"),
        output={"x0": 0, "x1": 1, "y0": 2, "y1": 2},
        delta={("x0", "+h"): "y0", ("x1", "+h"): "y1"},
    )


def fork_two_way():
    machine = fork_forward()
    return PartialMoore(
        states=machine.states,
        alphabet=("+h", "-h"),
        starts=machine.starts,
        output=machine.output,
        delta={
            **machine.delta,
            ("y0", "-h"): "x0",
            ("y1", "-h"): "x1",
        },
    )


def alternating_parity():
    return PartialMoore(
        states=("q0", "q1", "b0", "b1"),
        alphabet=("+A", "-A", "+X", "-X"),
        starts=("q0", "q1", "b0", "b1"),
        output={"q0": 1, "q1": 0, "b0": 0, "b1": 0},
        delta={
            ("q0", "+A"): "b0",
            ("q1", "+A"): "b1",
            ("b0", "-A"): "q0",
            ("b1", "-A"): "q1",
            ("b0", "+X"): "q1",
            ("b1", "+X"): "q0",
            ("q1", "-X"): "b0",
            ("q0", "-X"): "b1",
        },
    )


def restricted_tag():
    states = ("e", "B0", "B1", "AB0", "AB1", "C0AB0", "C1AB1")
    forward = {
        ("e", "+B0"): "B0",
        ("e", "+B1"): "B1",
        ("B0", "+A"): "AB0",
        ("B1", "+A"): "AB1",
        ("AB0", "+C0"): "C0AB0",
        ("AB1", "+C1"): "C1AB1",
    }
    inverse = {
        (target, "-" + letter[1:]): source
        for (source, letter), target in forward.items()
    }
    return PartialMoore(
        states=states,
        alphabet=("+B0", "-B0", "+B1", "-B1", "+A", "-A", "+C0", "-C0", "+C1", "-C1"),
        starts=states,
        output={
            "e": "empty",
            "B0": 0,
            "AB0": 0,
            "C0AB0": 0,
            "B1": 1,
            "AB1": 1,
            "C1AB1": 1,
        },
        delta={**forward, **inverse},
    )


def bounded_chain():
    return PartialMoore(
        states=("X0", "X1", "X2"),
        alphabet=("+X", "-X"),
        starts=("X0", "X1", "X2"),
        output={"X0": "empty", "X1": "nonempty", "X2": "nonempty"},
        delta={
            ("X0", "+X"): "X1",
            ("X1", "+X"): "X2",
            ("X1", "-X"): "X0",
            ("X2", "-X"): "X1",
        },
    )


def presence_forward():
    return PartialMoore(
        states=("absent", "present"),
        alphabet=("+X",),
        starts=("absent", "present"),
        output={"absent": 0, "present": 1},
        delta={
            ("absent", "+X"): "present",
            ("present", "+X"): "present",
        },
    )


def main():
    cases = [
        analyze("fork_merge_forward", fork_forward()),
        analyze("fork_merge_two_way", fork_two_way()),
        analyze("alternating_parity_restricted", alternating_parity()),
        analyze("rightmost_tag_restricted", restricted_tag()),
        analyze("bounded_depth_two_chain", bounded_chain()),
        analyze("presence_forward", presence_forward()),
    ]
    payload = {
        "scope": (
            "exact minimization of six finite partial Moore systems with "
            "undefinedness observable; no infinite-state or scaffold proof"
        ),
        "claims_checked": {
            "fork_forward_residual_merges": cases[0]["residual_states"] == 3
            and not cases[0]["residual_operations_partial_injective"],
            "fork_two_way_residual_splits": cases[1]["residual_states"] == 4
            and cases[1]["residual_operations_partial_injective"],
            "alternating_parity_is_partial_reversible": cases[2]["residual_operations_partial_injective"],
            "restricted_tag_is_partial_reversible": cases[3]["residual_operations_partial_injective"],
            "bounded_chain_is_partial_reversible": cases[4]["residual_operations_partial_injective"],
            "presence_push_merges": not cases[5]["residual_operations_partial_injective"],
        },
        "cases": cases,
    }
    if not all(payload["claims_checked"].values()):
        raise AssertionError(payload["claims_checked"])
    target = Path(__file__).with_name("partial-restoration-output.json")
    target.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(payload, indent=2))


if __name__ == "__main__":
    main()