verify_stack_erasure.py

#!/usr/bin/env python3
"""Regular-invariant certificates plus finite traces; not a proof assistant.

DFA language emptiness verifies invariant closure and realizable local
classes without a stack-depth cutoff. General pushdown post* is imported
in the note, not implemented here. Trace and persistence checks are finite.
"""
from collections import deque
from dataclasses import dataclass
import itertools
import json
from pathlib import Path


@dataclass
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


def simple_dfa(kind):
    alphabet = ("A", "X")
    if kind == "A*":
        return DFA(alphabet, 0, frozenset({0}),
                   {(0, "A"): 0, (0, "X"): 1, (1, "A"): 1, (1, "X"): 1})
    if kind == "X*":
        d = simple_dfa("A*")
        d.delta = {(s, "X" if a == "A" else "A"): t for (s, a), t in d.delta.items()}
        return d
    delta = {(0, "X"): 0, (0, "A"): 1, (1, "A"): 1,
             (1, "X"): 2, (2, "A"): 2, (2, "X"): 2}
    if kind == "X*A*":
        return DFA(alphabet, 0, frozenset({0, 1}), delta)
    if kind == "X*A+":
        return DFA(alphabet, 0, frozenset({1}), delta)
    assert kind == "all"
    return DFA(alphabet, 0, frozenset({0}), {(0, a): 0 for a in alphabet})


def singleton_dfa(word, alphabet):
    sink = len(word) + 1
    delta = {(s, a): s + 1 if s < len(word) and word[s] == a else sink
             for s in range(sink + 1) for a in alphabet}
    return DFA(alphabet, 0, frozenset({len(word)}), delta)


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 a in left.alphabet:
            pair = (left.delta[x, a], right.delta[y, a])
            if pair not in seen:
                seen.add(pair)
                queue.append((*pair, word + (a,)))
    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 q in machine["initial"]:
        assert invariants[q].accepts(())
    obligations = 0
    for (q, top, letter), rules in machine["rules"].items():
        src = invariants[q]
        for target, operation in rules:
            dst = invariants[target]
            if top is None:
                if src.accepts(()):
                    assert dst.accepts(apply((), operation)), (q, letter, target)
            else:
                prefix = apply((top,), operation)
                failure = inclusion_failure(src, src.delta[src.start, top],
                                            dst, dst.advance(dst.start, prefix))
                assert failure is None, (q, top, letter, target, failure)
            obligations += 1
    return obligations


def local_classes(dfa, projection):
    queue = deque([(dfa.start, None, None, ())])
    seen = {(dfa.start, None, None)}
    classes = {}
    while queue:
        state, raw_top, projected_top, word = queue.popleft()
        if state in dfa.final:
            classes.setdefault((raw_top, projected_top), word)
        for a in dfa.alphabet:
            rt = a if raw_top is None else raw_top
            pt = projection[a] if projected_top is None else projected_top
            key = (dfa.delta[state, a], rt, pt)
            if key not in seen:
                seen.add(key)
                queue.append((*key, word + (a,)))
    return classes


def signatures(machine, q, top, letter, projection, controls):
    out = set()
    for target, operation in machine["rules"].get((q, top, letter), ()):
        if operation.startswith("push:"):
            image = projection[operation[5:]]
            mapped = "keep" if image is None else "push:" + image
        elif operation == "pop":
            mapped = "pop" if top is not None and projection[top] is not None else "keep"
        else:
            mapped = "keep"
        out.add((controls[target], mapped))
    return frozenset(out)


def certificate(machine, invariants, projection, controls):
    obligations = invariant_check(machine, invariants)
    groups = {}
    for q, invariant in invariants.items():
        for (top, projected_top), witness in local_classes(invariant, projection).items():
            groups.setdefault((controls[q], projected_top), []).append((q, top, witness))
    failures, quotient_rules, quotient_obs = [], {}, {}
    for (control, projected_top), cells in groups.items():
        q0, top0, word0 = cells[0]
        obs0 = machine["observe"](q0, top0)
        quotient_obs[control, projected_top] = obs0
        for q, top, word in cells:
            if machine["observe"](q, top) != obs0:
                failures.append({"kind": "observation", "control": control,
                                 "top": projected_top, "witnesses": [word0, word]})
        for a in machine["inputs"]:
            s0 = signatures(machine, q0, top0, a, projection, controls)
            quotient_rules[control, projected_top, a] = s0
            for q, top, word in cells[1:]:
                s = signatures(machine, q, top, a, projection, controls)
                if s != s0:
                    failures.append({"kind": "signatures", "control": control,
                                     "top": projected_top, "input": a,
                                     "witnesses": [[q0, list(word0)], [q, list(word)]],
                                     "signatures": [sorted(s0), sorted(s)]})
    return {"failures": failures, "rules": quotient_rules, "observations": quotient_obs,
            "groups": len(groups), "cells": sum(map(len, groups.values())),
            "closure_obligations": obligations}


def logging_example():
    states = ("grow", "fall", "yes", "no", "dead")
    inputs = ("c", "r", "#", "p")
    rules = {}
    for q, g, a in itertools.product(states, (None, "A", "X"), inputs):
        out = {("dead", "keep")}
        if q == "dead":
            out = {("dead", "keep")}
        elif q in ("yes", "no") and a == "p":
            out = {(q, "push:X"), (q, "keep")}
        elif q in ("grow", "fall") and a == "#":
            out = {("yes" if g is None else "no", "keep")}
        elif q == "grow" and a == "c" and g != "X":
            out = {("grow", "push:A")}
        elif q in ("grow", "fall") and a == "r" and g == "A":
            out = {("fall", "pop")}
        rules[q, g, a] = out
    machine = {"initial": ("grow",), "inputs": inputs, "rules": rules,
               "observe": lambda q, top: q == "yes"}
    invariants = {"grow": simple_dfa("A*"), "fall": simple_dfa("A*"),
                  "yes": simple_dfa("X*"), "no": simple_dfa("X*A+"),
                  "dead": simple_dfa("X*A*")}
    return machine, invariants, {"A": "A", "X": None}, {q: q for q in states}


def projected(config, projection, controls):
    q, word = config
    return controls[q], tuple(projection[g] for g in word if projection[g] is not None)


def successors(configs, rules, letter):
    return {(target, apply(word, op)) for q, word in configs
            for target, op in rules.get((q, word[0] if word else None, letter), ())}


def logging_audit():
    machine, invariants, pi, controls = logging_example()
    cert = certificate(machine, invariants, pi, controls)
    assert not cert["failures"]
    invalid = dict(invariants)
    invalid["yes"] = singleton_dfa((), ("A", "X"))
    try:
        invariant_check(machine, invalid)
    except AssertionError:
        invalid_rejected = True
    else:
        invalid_rejected = False
    assert invalid_rejected  # A log push must leave the proposed singleton.
    frontier = [({("grow", ())}, {("grow", ())})]
    checked_words = 0
    for length in range(8):
        following = []
        for source, target in frontier:
            assert {projected(c, pi, controls) for c in source} == target
            assert any(q == "yes" for q, _ in source) == any(q == "yes" for q, _ in target)
            for q, word in source:
                assert invariants[q].accepts(word)
            checked_words += 1
            if length < 7:
                for a in machine["inputs"]:
                    following.append((successors(source, machine["rules"], a),
                                      successors(target, cert["rules"], a)))
        frontier = following
    source = successors({("grow", ())}, machine["rules"], "#")
    for j in range(13):
        assert {len(word) for _, word in source} == set(range(j + 1))
        assert {len(projected(c, pi, controls)[1]) for c in source} == {0}
        source = successors(source, machine["rules"], "p")
    return {"regular_invariant": {"grow": "A*", "fall": "A*", "yes": "X*",
                                  "no": "X*A+", "dead": "X*A*"},
            "closure_obligations": cert["closure_obligations"],
            "realizable_cells": cert["cells"], "abstract_groups": cert["groups"],
            "certificate": "passed using DFA emptiness, without a depth cutoff",
            "invalid_singleton_log_invariant": "rejected as required",
            "trace_words_max_length": 7, "trace_words_checked": checked_words,
            "raw_spread_example_j": [0, 12]}


def delayed_pop_audit():
    states, inputs = ("q", "yes", "no"), ("a", "x", "r", "t")
    rules = {}
    for q, g, a in itertools.product(states, (None, "A", "X"), inputs):
        if q != "q":
            out = {(q, "keep")}
        elif a in ("a", "x"):
            out = {(q, "push:" + ("A" if a == "a" else "X"))}
        elif a == "r":
            out = {(q, "pop")}
        else:
            out = {("yes" if g is None else "no", "keep")}
        rules[q, g, a] = out
    machine = {"initial": ("q",), "inputs": inputs, "rules": rules,
               "observe": lambda q, g: q == "yes"}
    pi, controls = {"A": "A", "X": None}, {q: q for q in states}
    cert = certificate(machine, {q: simple_dfa("all") for q in states}, pi, controls)
    witness = next(f for f in cert["failures"] if f.get("input") == "r" and f["top"] == "A")
    left, right = {("q", ("A",))}, {("q", ("X", "A"))}
    assert {projected(c, pi, controls) for c in left} == {projected(c, pi, controls) for c in right}
    for a in "rt":
        left, right = successors(left, rules, a), successors(right, rules, a)
    assert any(q == "yes" for q, _ in left)
    assert not any(q == "yes" for q, _ in right)
    return {"certificate": "rejected as required", "pop_witness": witness,
            "distinguishing_suffix": "rt"}


def deeper_control_audit():
    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, str(i)): {("s" + str(i), "push:B" + str(i))} for i in range(2)}
    for i in range(2):
        rules["s" + str(i), "B" + str(i), "a"] = {("q" + str(i), "push:A")}
        rules["q" + str(i), "A", "z"] = {("e" + str(i), "push:C" + str(i))}
    machine = {"initial": ("init",), "inputs": ("0", "1", "a", "z"),
               "rules": rules, "observe": lambda q, g: False}
    pi = {a: a for a in alphabet}
    controls = {q: "q" if q in ("q0", "q1") else q for q in words}
    cert = certificate(machine, {q: singleton_dfa(w, alphabet) for q, w in words.items()}, pi, controls)
    assert len({projected((q, w), pi, controls) for q, w in words.items()}) == len(words)
    witness = next(f for f in cert["failures"] if f.get("input") == "z")
    return {"reachable_configurations": len(words), "projection_injective": True,
            "one_top_certificate": "rejected as required", "witness": witness}


def shortcut_audit():
    # Each record is an immutable (label, full_tail, saved_filtered_root) tuple.
    heap, snapshots = [], []
    pi = {"A": "a", "B": "b", "X": None}

    def decode(root, field, filtered=False):
        out = []
        while root is not None:
            label, tail, saved = heap[root]
            out.append(pi[label] if filtered else label)
            assert not filtered or pi[label] is not None
            root = (tail, saved)[field]
        return tuple(out)

    def visit(R, E, word, remaining):
        assert decode(R, 0) == word
        assert decode(E, 1, True) == tuple(pi[g] for g in word if pi[g] is not None)
        snapshots.append((R, E, word))
        if not remaining:
            return
        for operation in ("push:A", "push:B", "push:X", "pop", "keep"):
            before = len(heap)
            if operation.startswith("push:"):
                g = operation[5:]
                new_R = len(heap)
                heap.append((g, R, E))
                new_E = E if pi[g] is None else new_R
            elif operation == "pop" and R is not None:
                _, new_R, new_E = heap[R]
            else:
                new_R, new_E = R, E
            assert len(heap) - before <= 1
            visit(new_R, new_E, apply(word, operation), remaining - 1)

    visit(None, None, (), 7)
    for R, E, word in snapshots:
        assert decode(R, 0) == word
        assert decode(E, 1, True) == tuple(pi[g] for g in word if pi[g] is not None)
    return {"operation_alphabet": ["push A", "push B", "push X", "pop", "keep"],
            "all_sequences_through_length": 7, "retained_snapshots": len(snapshots),
            "allocations": len(heap), "max_allocations_per_update": 1,
            "full_and_filtered_decoders": "passed before and after all branches"}


def main():
    result = {"status": "passed", "regular_certificate": logging_audit(),
              "delayed_pop_negative": delayed_pop_audit(),
              "semantic_but_nonlocal_negative": deeper_control_audit(),
              "persistent_filtered_access": shortcut_audit(),
              "assurance": "Executable finite-automaton certificates and finite trace audits; no proof assistant.",
              "not_implemented": ["general pushdown post* saturation", "VPA determinization",
                                  "bounded-spread checker", "scaffold compiler"]}
    Path(__file__).with_name("stack-erasure-output.json").write_text(json.dumps(result, indent=2) + "\n")
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()