verify_nested_restoration_interfaces.py

#!/usr/bin/env python3
"""Finite checks for nested restoration interfaces.

The script checks supplied finite covers and subsequential boundary maps,
then exhaustively explores bounded nested traces. It does not prove the
unbounded theorems or synthesize an unknown stable cover.
"""
from collections import defaultdict, deque
from itertools import permutations, product
import json
from pathlib import Path

from verify_partial_restoration import minimize
from verify_regular_history_restoration import fork, moore, solve


class Cover:
    def __init__(self, states, alphabet, start, update, inverse, output):
        self.states = tuple(states)
        self.alphabet = tuple(alphabet)
        self.start = start
        self.update = update
        self.inverse = inverse
        self.output = output

    def word(self, start, letters):
        state = start
        for letter in letters:
            state = self.update[state, letter]
        return state


class Subseq:
    def __init__(self, states, start, step, final):
        self.states = tuple(states)
        self.start = start
        self.step = step
        self.final = final


def compatibility(cover, transducer):
    """Exact finite-product test for sigma(T(w)) as a function of sigma(w)."""
    initial = (cover.start, transducer.start, cover.start)
    seen, queue = {initial}, deque([initial])
    while queue:
        in_state, trans_state, out_state = queue.popleft()
        for letter in cover.alphabet:
            next_in = cover.update.get((in_state, letter))
            trans_step = transducer.step.get((trans_state, letter))
            if next_in is None or trans_step is None:
                continue
            next_trans, emitted = trans_step
            try:
                next_out = cover.word(out_state, emitted)
            except KeyError:
                return None, {"reason": "transducer output is not legal"}
            item = (next_in, next_trans, next_out)
            if item not in seen:
                seen.add(item)
                queue.append(item)
    values = defaultdict(set)
    for in_state, trans_state, out_state in seen:
        try:
            final_out = cover.word(out_state, transducer.final[trans_state])
        except KeyError:
            return None, {"reason": "terminal output is not legal"}
        values[in_state].add(final_out)
    conflicts = {str(state): sorted(map(str, targets))
                 for state, targets in values.items() if len(targets) != 1}
    if conflicts:
        return None, {"product_states": len(seen), "conflicts": conflicts}
    induced = {state: next(iter(targets)) for state, targets in values.items()}
    return induced, {"product_states": len(seen), "induced": {str(k): str(v)
                                                               for k, v in induced.items()}}


def parity_cover():
    return Cover(
        states=(0, 1), alphabet=("a", "b"), start=0,
        update={(0, "a"): 1, (1, "a"): 0, (0, "b"): 0, (1, "b"): 1},
        inverse={(0, "a"): 1, (1, "a"): 0, (0, "b"): 0, (1, "b"): 1},
        output={0: 0, 1: 1},
    )


def copier(final_suffix):
    return Subseq(
        states=("q",), start="q",
        step={("q", letter): ("q", letter) for letter in ("a", "b")},
        final={"q": final_suffix},
    )


def last_b_transformer():
    step = {}
    for state in ("empty", "a", "b"):
        step[state, "a"] = ("a", "")
        step[state, "b"] = ("b", "")
    return Subseq(("empty", "a", "b"), "empty", step,
                  {"empty": "", "a": "", "b": "a"})


def parity(word):
    return word.count("a") % 2


def transformed(word, mode, outcome=0):
    if mode == "resume":
        return word
    if mode == "append":
        return word + ("a" if outcome else "b")
    if mode == "last_b":
        return "a" if word.endswith("b") else ""
    raise AssertionError(mode)


def bounded_nested_check(mode, depth):
    """Compare raw histories with saved finite summaries on every trace."""
    cover = parity_cover()
    maps = {}
    if mode == "resume":
        maps[0] = maps[1] = compatibility(cover, copier(""))[0]
    elif mode == "append":
        maps[0] = compatibility(cover, copier("b"))[0]
        maps[1] = compatibility(cover, copier("a"))[0]
    else:
        raise AssertionError(mode)
    # Source: (active_word, tuple(suspended_words)).
    # Target: (active_summary, tuple(suspended_summaries)).
    initial = (("", ()), (cover.start, ()))
    seen, queue = {initial}, deque([(initial, 0)])
    checked = 0
    max_nesting = 0
    while queue:
        ((word, source_stack), (state, target_stack)), steps = queue.popleft()
        assert state == parity(word)
        assert tuple(parity(w) for w in source_stack) == target_stack
        checked += 1
        max_nesting = max(max_nesting, len(source_stack))
        if steps == depth:
            continue
        successors = []
        for letter in cover.alphabet:
            successors.append((((word + letter, source_stack),
                                (cover.update[state, letter], target_stack)), "+"))
        if word:
            letter = word[-1]
            successors.append((((word[:-1], source_stack),
                                (cover.inverse[state, letter], target_stack)), "-"))
        successors.append(((("", source_stack + (word,)),
                            (cover.start, target_stack + (state,))), "call"))
        if source_stack:
            outcome = parity(word)
            caller = source_stack[-1]
            caller_state = target_stack[-1]
            next_word = transformed(caller, mode, outcome)
            next_state = maps[outcome][caller_state]
            successors.append((((next_word, source_stack[:-1]),
                                (next_state, target_stack[:-1])), "return"))
        for nxt, _ in successors:
            item = (nxt, steps + 1)
            if (nxt, steps + 1) not in seen:
                seen.add((nxt, steps + 1))
                queue.append(item)
    return {"mode": mode, "trace_depth": depth, "configurations_checked": checked,
            "maximum_nesting_reached": max_nesting}


def permutation_search(max_states):
    """Finite corroboration: no small permutation DFA recognizes ends-in-b."""
    totals = {}
    witness_words = [""] + ["".join(w) for n in range(1, max_states + 3)
                            for w in product("ab", repeat=n)]
    for size in range(1, max_states + 1):
        states = tuple(range(size))
        checked = 0
        successes = 0
        for pa in permutations(states):
            for pb in permutations(states):
                for mask in range(1 << size):
                    checked += 1
                    ok = True
                    for word in witness_words:
                        q = 0
                        for letter in word:
                            q = pa[q] if letter == "a" else pb[q]
                        if bool(mask & (1 << q)) != word.endswith("b"):
                            ok = False
                            break
                    successes += int(ok)
        assert successes == 0
        totals[str(size)] = {"candidates": checked, "successes": successes,
                             "words_used": len(witness_words)}
    return totals


def main():
    cover = parity_cover()
    resume_map, resume_evidence = compatibility(cover, copier(""))
    append_a_map, append_a_evidence = compatibility(cover, copier("a"))
    append_b_map, append_b_evidence = compatibility(cover, copier("b"))
    last_b_map, last_b_evidence = compatibility(cover, last_b_transformer())
    assert resume_map == {0: 0, 1: 1}
    assert append_a_map == {0: 1, 1: 0}
    assert append_b_map == {0: 0, 1: 1}
    assert last_b_map is None and last_b_evidence["conflicts"]

    bad_machine, _ = minimize(fork("a"))
    assert solve(bad_machine)["possible"] is False
    good_machine, _ = minimize(moore(
        "st", "a", {"s": 0, "t": 1}, {("s", "a"): "t", ("t", "a"): "s"}))
    assert solve(good_machine)["states"] == 2

    result = {
        "scope": ("Exact finite-product compatibility for four supplied transformers; "
                  "all nested traces through recorded depths; local cover construction "
                  "on two finite machines; bounded permutation enumeration. "
                  "Not an unbounded proof or unknown-cover synthesis."),
        "compatibility": {
            "pure_resume": resume_evidence,
            "append_a": append_a_evidence,
            "append_b": append_b_evidence,
            "last_b_rejected": last_b_evidence,
        },
        "nested_traces": [
            bounded_nested_check("resume", 8),
            bounded_nested_check("append", 7),
        ],
        "local_forbidden_merge_rejected": True,
        "local_parity_cover_states": 2,
        "permutation_automata_ends_in_b_search": permutation_search(4),
        "proof_status": ("Proposition 3 for all finite sizes is proved in writing by "
                         "surjectivity of the b permutation; enumeration only corroborates n<=4."),
    }
    target = Path(__file__).with_name("nested-restoration-interfaces-output.json")
    target.write_text(json.dumps(result, indent=2) + "\n")
    print(json.dumps({"compatibility_cases": 4,
                      "nested_configurations": sum(x["configurations_checked"]
                                                   for x in result["nested_traces"]),
                      "permutation_sizes": 4,
                      "last_b_rejected": True}))


if __name__ == "__main__":
    main()