verify_finite_summary_erasure.py

#!/usr/bin/env python3
"""Exact regular-invariant certificates for supplied finite stack summaries.

The DFA product explores every descriptor realized in a supplied regular
invariant, without a stack-depth cutoff. It checks invariant closure and the
finite local-signature criterion from finite-summary-erasure.md. It does not
implement general pushdown post* saturation or prove the theorem itself.
"""
from collections import deque
from dataclasses import dataclass
import json
from pathlib import Path


@dataclass(frozen=True)
class DFA:
    alphabet: tuple
    start: int
    final: frozenset
    delta: dict

    def advance(self, state, word):
        for symbol in word:
            state = self.delta[state, symbol]
        return state

    def accepts(self, word):
        return self.advance(self.start, word) in self.final


@dataclass(frozen=True)
class Monoid:
    elements: tuple
    identity: object
    table: dict
    generators: dict

    def mul(self, left, right):
        return self.table[left, right]

    def validate(self, alphabet):
        assert self.identity in self.elements
        assert set(self.generators) == set(alphabet)
        for x in self.elements:
            assert self.mul(self.identity, x) == x
            assert self.mul(x, self.identity) == x
            for y in self.elements:
                assert self.mul(x, y) in self.elements
                for z in self.elements:
                    assert self.mul(self.mul(x, y), z) == self.mul(x, self.mul(y, z))


def complete_dfa(alphabet, states, start, final, transitions, sink):
    delta = {}
    for state in states:
        for symbol in alphabet:
            delta[state, symbol] = transitions.get((state, symbol), sink)
    return DFA(tuple(alphabet), start, frozenset(final), delta)


def singleton_dfa(word, alphabet):
    sink = len(word) + 1
    states = tuple(range(sink + 1))
    transitions = {}
    for state in range(len(word)):
        transitions[state, word[state]] = state + 1
    return complete_dfa(alphabet, states, 0, {len(word)}, transitions, sink)


def pair_parity_dfa(parity, prefixed=False):
    alphabet = ("X", "A")
    sink = 5 if prefixed else 4
    if prefixed:
        # 0 expects the leading A. States 1/2 are pair boundaries of parity 0/1;
        # states 3/4 have just read X at those parities.
        states = tuple(range(6))
        transitions = {(0, "A"): 1, (1, "X"): 3, (3, "A"): 2,
                       (2, "X"): 4, (4, "A"): 1}
        final = {1 + parity}
        return complete_dfa(alphabet, states, 0, final, transitions, sink)
    # States 0/1 are pair boundaries of parity 0/1; 2/3 have just read X.
    states = tuple(range(5))
    transitions = {(0, "X"): 2, (2, "A"): 1,
                   (1, "X"): 3, (3, "A"): 0}
    return complete_dfa(alphabet, states, 0, {parity}, transitions, sink)


def star_dfa(symbol):
    return DFA((symbol,), 0, frozenset({0}), {(0, symbol): 0})


def inclusion_failure(left, left_start, right, right_start):
    queue = deque([(left_start, right_start, ())])
    seen = {(left_start, right_start)}
    while queue:
        x, y, word = queue.popleft()
        if x in left.final and y not in right.final:
            return word
        for symbol in left.alphabet:
            pair = (left.delta[x, symbol], right.delta[y, symbol])
            if pair not in seen:
                seen.add(pair)
                queue.append((*pair, word + (symbol,)))
    return None


def apply(word, operation):
    if operation == "pop":
        return word[1:] if word else word
    if operation == "keep":
        return word
    assert operation.startswith("push:")
    return (operation[5:],) + word


def invariant_check(machine, invariants):
    for state in machine["initial"]:
        assert invariants[state].accepts(())
    obligations = 0
    for (state, top, letter), rules in machine["rules"].items():
        source = invariants[state]
        for target, operation in rules:
            destination = invariants[target]
            if top is None:
                if source.accepts(()):
                    assert destination.accepts(apply((), operation))
            else:
                prefix = apply((top,), operation)
                failure = inclusion_failure(
                    source,
                    source.delta[source.start, top],
                    destination,
                    destination.advance(destination.start, prefix),
                )
                assert failure is None, (state, top, letter, target, failure)
            obligations += 1
    return obligations


def descriptors(dfa, monoid, projection):
    # Words are traversed top first. The key retains both the whole-word
    # summary and the summary after deleting the raw top.
    start = (dfa.start, None, None, monoid.identity, monoid.identity)
    queue = deque([(start, ())])
    seen = {start}
    realized = {}
    while queue:
        (dfa_state, raw_top, projected_top, summary, tail_summary), word = queue.popleft()
        if dfa_state in dfa.final:
            realized.setdefault(
                (raw_top, projected_top, summary, tail_summary), word
            )
        for symbol in dfa.alphabet:
            generator = monoid.generators[symbol]
            if raw_top is None:
                next_raw_top = symbol
                next_projected_top = projection[symbol]
                next_summary = generator
                next_tail_summary = monoid.identity
            else:
                next_raw_top = raw_top
                next_projected_top = projected_top
                if next_projected_top is None and projection[symbol] is not None:
                    next_projected_top = projection[symbol]
                next_summary = monoid.mul(summary, generator)
                next_tail_summary = monoid.mul(tail_summary, generator)
            key = (
                dfa.delta[dfa_state, symbol],
                next_raw_top,
                next_projected_top,
                next_summary,
                next_tail_summary,
            )
            if key not in seen:
                seen.add(key)
                queue.append((key, word + (symbol,)))
    return realized


def signatures(machine, state, descriptor, letter, monoid, projection, controls):
    raw_top, _, summary, tail_summary = descriptor
    output = set()
    for target, operation in machine["rules"].get((state, raw_top, letter), ()):
        if operation.startswith("push:"):
            symbol = operation[5:]
            next_summary = monoid.mul(monoid.generators[symbol], summary)
            image = projection[symbol]
            projected_operation = "keep" if image is None else "push:" + image
        elif operation == "pop" and raw_top is not None:
            next_summary = tail_summary
            projected_operation = "pop" if projection[raw_top] is not None else "keep"
        else:
            next_summary = summary
            projected_operation = "keep"
        output.add(((controls[target], next_summary), projected_operation))
    return frozenset(output)


def certificate(machine, invariants, monoid, projection, controls):
    alphabet = next(iter(invariants.values())).alphabet
    monoid.validate(alphabet)
    closure_obligations = invariant_check(machine, invariants)
    groups = {}
    for state, invariant in invariants.items():
        for descriptor, witness in descriptors(invariant, monoid, projection).items():
            raw_top, projected_top, summary, _ = descriptor
            group = (controls[state], summary, projected_top)
            groups.setdefault(group, []).append((state, descriptor, witness))

    failures = []
    quotient_rules = {}
    quotient_observations = {}
    for group, cells in groups.items():
        state0, descriptor0, witness0 = cells[0]
        raw_top0 = descriptor0[0]
        observation0 = machine["observe"](state0, raw_top0)
        quotient_observations[group] = observation0
        for state, descriptor, witness in cells[1:]:
            if machine["observe"](state, descriptor[0]) != observation0:
                failures.append({
                    "kind": "observation",
                    "group": group,
                    "witnesses": [[state0, list(witness0)], [state, list(witness)]],
                })
        for letter in machine["inputs"]:
            signature0 = signatures(
                machine, state0, descriptor0, letter, monoid, projection, controls
            )
            quotient_rules[group, letter] = signature0
            for state, descriptor, witness in cells[1:]:
                signature = signatures(
                    machine, state, descriptor, letter, monoid, projection, controls
                )
                if signature != signature0:
                    failures.append({
                        "kind": "signatures",
                        "group": group,
                        "input": letter,
                        "witnesses": [[state0, list(witness0)], [state, list(witness)]],
                        "signatures": [sorted(signature0), sorted(signature)],
                    })
    return {
        "failures": failures,
        "groups": len(groups),
        "descriptors": sum(map(len, groups.values())),
        "closure_obligations": closure_obligations,
        "rules": quotient_rules,
        "observations": quotient_observations,
    }


def trivial_monoid(alphabet):
    return Monoid((0,), 0, {(0, 0): 0}, {symbol: 0 for symbol in alphabet})


def parity_monoid():
    elements = (0, 1)
    return Monoid(
        elements,
        0,
        {(x, y): x ^ y for x in elements for y in elements},
        {"X": 1, "A": 0},
    )


def union_monoid():
    elements = (0, 1)
    return Monoid(
        elements,
        0,
        {(x, y): x | y for x in elements for y in elements},
        {"X": 1},
    )


def rightmost_tag_monoid(alphabet):
    elements = ("e", "0", "1")
    table = {(x, y): y if y != "e" else x for x in elements for y in elements}
    generators = {symbol: "e" for symbol in alphabet}
    generators.update({"B0": "0", "B1": "1"})
    return Monoid(elements, "e", table, generators)


def parity_restoration_example():
    states = ("q0", "q1", "b0", "b1", "d0", "d1")
    inputs = ("u", "v", "x", "a")
    rules = {}
    for parity in (0, 1):
        q, build, drop = f"q{parity}", f"b{parity}", f"d{parity}"
        if parity == 0:
            rules[q, None, "u"] = {(build, "push:A")}
        rules[q, "X", "u"] = {(build, "push:A")}
        rules[build, "A", "v"] = {(f"q{1 - parity}", "push:X")}
        rules[q, "X", "x"] = {(f"d{1 - parity}", "pop")}
        rules[drop, "A", "a"] = {(q, "pop")}
    machine = {
        "initial": ("q0",),
        "inputs": inputs,
        "rules": rules,
        "observe": lambda state, top: state == "q0",
    }
    invariants = {
        "q0": pair_parity_dfa(0),
        "q1": pair_parity_dfa(1),
        "b0": pair_parity_dfa(0, prefixed=True),
        "b1": pair_parity_dfa(1, prefixed=True),
        "d0": pair_parity_dfa(0, prefixed=True),
        "d1": pair_parity_dfa(1, prefixed=True),
    }
    projection = {"X": None, "A": "A"}
    controls = {state: state[0] for state in states}
    summary = certificate(machine, invariants, parity_monoid(), projection, controls)
    no_summary = certificate(
        machine, invariants, trivial_monoid(("X", "A")), projection, controls
    )
    assert not summary["failures"]
    assert any(failure["kind"] == "observation" for failure in no_summary["failures"])
    return {
        "regular_invariants": {
            "q0": "(X A)^(even)",
            "q1": "(X A)^(odd)",
            "b_i_and_d_i": "A followed by the corresponding parity language",
        },
        "summary": "parity of X frames",
        "summary_certificate": "passed",
        "without_summary": "rejected as required",
        "closure_obligations": summary["closure_obligations"],
        "realized_descriptors": summary["descriptors"],
        "abstract_local_groups": summary["groups"],
    }


def frame_set_negative():
    machine = {
        "initial": ("q",),
        "inputs": ("p", "r"),
        "rules": {
            ("q", None, "p"): {("q", "push:X")},
            ("q", "X", "p"): {("q", "push:X")},
            ("q", "X", "r"): {("q", "pop")},
        },
        "observe": lambda state, top: False,
    }
    invariants = {"q": star_dfa("X")}
    result = certificate(machine, invariants, union_monoid(), {"X": None}, {"q": "q"})
    failure = next(f for f in result["failures"] if f.get("input") == "r")
    assert {tuple(witness[1]) for witness in failure["witnesses"]} == {("X",), ("X", "X")}
    return {
        "summary": "set of labels present",
        "certificate": "rejected as required",
        "witness": failure,
        "reason": "X and XX have the same current summary but different summaries after pop",
    }


def missing_bit_example():
    alphabet = ("A", "B0", "B1", "C0", "C1")
    words = {
        "init": (), "s0": ("B0",), "s1": ("B1",),
        "q0": ("A", "B0"), "q1": ("A", "B1"),
        "e0": ("C0", "A", "B0"), "e1": ("C1", "A", "B1"),
    }
    rules = {
        ("init", None, "0"): {("s0", "push:B0")},
        ("init", None, "1"): {("s1", "push:B1")},
    }
    for bit in (0, 1):
        rules[f"s{bit}", f"B{bit}", "a"] = {(f"q{bit}", "push:A")}
        rules[f"q{bit}", "A", "z"] = {(f"e{bit}", f"push:C{bit}")}
    machine = {
        "initial": ("init",),
        "inputs": ("0", "1", "a", "z"),
        "rules": rules,
        "observe": lambda state, top: False,
    }
    invariants = {state: singleton_dfa(word, alphabet) for state, word in words.items()}
    projection = {symbol: symbol for symbol in alphabet}
    controls = {state: "q" if state in ("q0", "q1") else state for state in states_from(words)}
    repaired = certificate(
        machine, invariants, rightmost_tag_monoid(alphabet), projection, controls
    )
    unrepaired = certificate(
        machine, invariants, trivial_monoid(alphabet), projection, controls
    )
    assert not repaired["failures"]
    assert any(failure.get("input") == "z" for failure in unrepaired["failures"])
    return {
        "summary": "rightmost B0/B1 tag",
        "with_summary": "passed",
        "without_summary": "rejected as required",
        "reachable_configurations": len(words),
    }


def states_from(mapping):
    return tuple(mapping.keys())


def main():
    result = {
        "status": "passed",
        "parity_pop_restoration": parity_restoration_example(),
        "frame_set_pop_obstruction": frame_set_negative(),
        "missing_bit_repair": missing_bit_example(),
        "assurance": (
            "Exact DFA invariant closure and finite descriptor checks, without a stack-depth cutoff; "
            "the general theorem remains a written proof."
        ),
        "not_implemented": [
            "general pushdown post* saturation",
            "automatic summary synthesis",
            "minimal summary computation",
            "scaffold generation",
            "proof assistant",
        ],
    }
    output = Path(__file__).with_name("finite-summary-erasure-output.json")
    output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")
    print(json.dumps(result, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()