check_antimirov_independent.py

#!/usr/bin/env python3
"""Independent finite audit for the regular-expression left quotient.

The construction side uses Antimirov partial derivatives over a raw binary
syntax tree parsed from infix strings.  The comparison side uses Thompson
epsilon-NFAs and subset reachability.  It does not call the Brzozowski
derivative, normalization, or nullability code in check_left_quotient.py.
"""

from __future__ import annotations

import hashlib
import itertools
import json
from collections import defaultdict, deque
from functools import lru_cache
from pathlib import Path

ALPHABET = ("a", "b")
ZERO = ("zero",)
ONE = ("one",)

A_SOURCES = (
    "0", "1", "a", "b", "(a|b)", "(a.b)", "(b.a)", "(a)*", "(b)*",
    "((a|b))*", "((a)*.b)", "(a.(b)*)", "((a.b))*", "((b.a))*",
    "((a|b).a)", "(a.(a|b))", "((a.b)|(b.a))", "((a)*.(b)*)",
    "(((a)*.b))*", "(((a.b))*.a)", "((a.b)|1)", "((a|1).b)",
    "(((a.b)|1))*", "(((a|b))*.a.b)", "((a)*|(b)*)",
    "((a)*.b.(a)*)", "(((a|b).(a|b)))*",
)

B_SOURCES = (
    "0", "1", "a", "b", "(a|b)", "(a.b)", "(b.a)", "(a)*", "(b)*",
    "((a|b))*", "((a)*.b)", "(a.(b)*)", "((a.b))*", "((a.b)|1)",
    "((a)*.b.(a)*)",
)


class Parser:
    def __init__(self, source: str):
        self.source = source
        self.index = 0

    def parse(self):
        expression = self.parse_union()
        if self.index != len(self.source):
            raise ValueError(f"trailing input at {self.index}: {self.source!r}")
        return expression

    def peek(self):
        return self.source[self.index] if self.index < len(self.source) else None

    def take(self, expected=None):
        token = self.peek()
        if token is None or (expected is not None and token != expected):
            raise ValueError(f"expected {expected!r} at {self.index}: {self.source!r}")
        self.index += 1
        return token

    def parse_union(self):
        expression = self.parse_concat()
        while self.peek() == "|":
            self.take("|")
            expression = ("alt", expression, self.parse_concat())
        return expression

    def parse_concat(self):
        expression = self.parse_postfix()
        while self.peek() == ".":
            self.take(".")
            expression = ("cat", expression, self.parse_postfix())
        return expression

    def parse_postfix(self):
        expression = self.parse_atom()
        while self.peek() == "*":
            self.take("*")
            expression = ("star", expression)
        return expression

    def parse_atom(self):
        token = self.peek()
        if token == "(":
            self.take("(")
            expression = self.parse_union()
            self.take(")")
            return expression
        self.take()
        if token == "0":
            return ZERO
        if token == "1":
            return ONE
        if token in ALPHABET:
            return ("lit", token)
        raise ValueError(f"unexpected {token!r}")


def parse(source: str):
    return Parser(source).parse()


def residual_cat(left, right):
    if left == ZERO or right == ZERO:
        return ZERO
    if left == ONE:
        return right
    if right == ONE:
        return left
    return ("cat", left, right)


@lru_cache(maxsize=None)
def antimirov_nullable(expression) -> bool:
    tag = expression[0]
    if tag in ("zero", "lit"):
        return False
    if tag in ("one", "star"):
        return True
    if tag == "alt":
        return antimirov_nullable(expression[1]) or antimirov_nullable(expression[2])
    if tag == "cat":
        return antimirov_nullable(expression[1]) and antimirov_nullable(expression[2])
    raise ValueError(tag)


@lru_cache(maxsize=None)
def partial(expression, letter: str):
    tag = expression[0]
    if tag in ("zero", "one"):
        return frozenset()
    if tag == "lit":
        return frozenset((ONE,)) if expression[1] == letter else frozenset()
    if tag == "alt":
        return partial(expression[1], letter) | partial(expression[2], letter)
    if tag == "cat":
        left, right = expression[1], expression[2]
        result = {residual_cat(item, right) for item in partial(left, letter)}
        result.discard(ZERO)
        if antimirov_nullable(left):
            result.update(partial(right, letter))
        return frozenset(result)
    if tag == "star":
        loop = expression
        return frozenset(
            item
            for item in (residual_cat(step, loop) for step in partial(expression[1], letter))
            if item != ZERO
        )
    raise ValueError(tag)


def partial_carrier(start):
    states = {start}
    edges = {}
    queue = deque((start,))
    while queue:
        state = queue.popleft()
        for letter in ALPHABET:
            targets = partial(state, letter)
            edges[(state, letter)] = targets
            for target in targets:
                if target not in states:
                    states.add(target)
                    queue.append(target)
    return states, edges


def compose(left, right):
    by_middle = defaultdict(set)
    for middle, target in right:
        by_middle[middle].add(target)
    return {
        (source, target)
        for source, middle in left
        for target in by_middle.get(middle, ())
    }


def closure(relation, states):
    result = set(relation) | {(state, state) for state in states}
    changed = True
    while changed:
        expanded = result | compose(result, result)
        changed = expanded != result
        result = expanded
    return result


def relation_semantics(expression, states, edges):
    identity = {(state, state) for state in states}

    @lru_cache(maxsize=None)
    def interpret(node):
        tag = node[0]
        if tag == "zero":
            return frozenset()
        if tag == "one":
            return frozenset(identity)
        if tag == "lit":
            return frozenset(
                (state, target)
                for state in states
                for target in edges[(state, node[1])]
            )
        if tag == "alt":
            return interpret(node[1]) | interpret(node[2])
        if tag == "cat":
            return frozenset(compose(interpret(node[1]), interpret(node[2])))
        if tag == "star":
            return frozenset(closure(interpret(node[1]), states))
        raise ValueError(tag)

    return interpret(expression)


def partial_run(states, word: str):
    current = frozenset(states)
    for letter in word:
        current = frozenset(
            target for state in current for target in partial(state, letter)
        )
    return current


def prepare_antimirov(a_expression, b_expression):
    states, edges = partial_carrier(b_expression)
    relation = relation_semantics(a_expression, states, edges)
    residuals = frozenset(
        target for source, target in relation if source == b_expression
    )
    return residuals, len(states)


def antimirov_accepts(residuals, suffix: str):
    return any(antimirov_nullable(state) for state in partial_run(residuals, suffix))


class Thompson:
    def __init__(self):
        self.next_state = 0
        self.epsilon = defaultdict(set)
        self.transitions = defaultdict(set)

    def fresh(self):
        state = self.next_state
        self.next_state += 1
        return state

    def compile(self, expression):
        tag = expression[0]
        start, final = self.fresh(), self.fresh()
        if tag == "zero":
            return start, final
        if tag == "one":
            self.epsilon[start].add(final)
            return start, final
        if tag == "lit":
            self.transitions[(start, expression[1])].add(final)
            return start, final
        if tag == "alt":
            left = self.compile(expression[1])
            right = self.compile(expression[2])
            self.epsilon[start].update((left[0], right[0]))
            self.epsilon[left[1]].add(final)
            self.epsilon[right[1]].add(final)
            return start, final
        if tag == "cat":
            left = self.compile(expression[1])
            right = self.compile(expression[2])
            self.epsilon[start].add(left[0])
            self.epsilon[left[1]].add(right[0])
            self.epsilon[right[1]].add(final)
            return start, final
        if tag == "star":
            body = self.compile(expression[1])
            self.epsilon[start].update((final, body[0]))
            self.epsilon[body[1]].update((final, body[0]))
            return start, final
        raise ValueError(tag)

    def epsilon_closure(self, states):
        result = set(states)
        queue = deque(result)
        while queue:
            state = queue.popleft()
            for target in self.epsilon[state]:
                if target not in result:
                    result.add(target)
                    queue.append(target)
        return frozenset(result)

    def step(self, states, letter: str):
        targets = {
            target
            for state in states
            for target in self.transitions[(state, letter)]
        }
        return self.epsilon_closure(targets)


def prepare_thompson(a_expression, b_expression):
    a_machine, b_machine = Thompson(), Thompson()
    a_start, a_final = a_machine.compile(a_expression)
    b_start, b_final = b_machine.compile(b_expression)
    start = (
        a_machine.epsilon_closure((a_start,)),
        b_machine.epsilon_closure((b_start,)),
    )
    reachable = {start}
    queue = deque((start,))
    while queue:
        a_states, b_states = queue.popleft()
        for letter in ALPHABET:
            target = (
                a_machine.step(a_states, letter),
                b_machine.step(b_states, letter),
            )
            if target not in reachable:
                reachable.add(target)
                queue.append(target)
    accepted_b_states = frozenset(
        b_states for a_states, b_states in reachable if a_final in a_states
    )
    return b_machine, b_final, accepted_b_states, len(reachable)


def thompson_accepts(prepared, suffix: str):
    machine, final, starts, _ = prepared
    for start in starts:
        states = start
        for letter in suffix:
            states = machine.step(states, letter)
        if final in states:
            return True
    return False


def words_up_to(maximum_length: int):
    return [
        "".join(letters)
        for length in range(maximum_length + 1)
        for letters in itertools.product(ALPHABET, repeat=length)
    ]


def main() -> int:
    a_expressions = tuple(parse(source) for source in A_SOURCES)
    b_expressions = tuple(parse(source) for source in B_SOURCES)
    suffixes = words_up_to(4)
    transcript = hashlib.sha256()
    mismatches = []
    max_carrier = 0
    max_product = 0

    for a_index, a_expression in enumerate(a_expressions):
        for b_index, b_expression in enumerate(b_expressions):
            residuals, carrier_size = prepare_antimirov(a_expression, b_expression)
            thompson = prepare_thompson(a_expression, b_expression)
            max_carrier = max(max_carrier, carrier_size)
            max_product = max(max_product, thompson[3])
            for suffix in suffixes:
                construction = antimirov_accepts(residuals, suffix)
                oracle = thompson_accepts(thompson, suffix)
                transcript.update(
                    f"{a_index}:{b_index}:{suffix}:{int(construction)}:{int(oracle)}\n".encode()
                )
                if construction != oracle:
                    mismatches.append({
                        "a_index": a_index,
                        "b_index": b_index,
                        "suffix": suffix,
                        "antimirov": construction,
                        "thompson": oracle,
                    })

    prior_path = Path(__file__).with_name("certificate.json")
    prior = json.loads(prior_path.read_text()) if prior_path.exists() else {}
    digest = transcript.hexdigest()
    certificate = {
        "certificate_type": "fprd-left-quotient-independent-audit.v1",
        "alphabet": list(ALPHABET),
        "quotienting_expressions": len(a_expressions),
        "target_expressions": len(b_expressions),
        "expression_pairs": len(a_expressions) * len(b_expressions),
        "suffixes": len(suffixes),
        "maximum_suffix_length": 4,
        "membership_comparisons": len(a_expressions) * len(b_expressions) * len(suffixes),
        "mismatch_count": len(mismatches),
        "maximum_antimirov_carrier_states": max_carrier,
        "maximum_thompson_product_states": max_product,
        "transcript_sha256": digest,
        "matches_v2_semantic_transcript": digest == prior.get("transcript_sha256"),
        "checker_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        "construction": "Antimirov partial derivatives on an independently parsed raw binary AST",
        "oracle": "Thompson epsilon-NFAs with subset-product reachability",
        "shared_mechanics": ["fixed infix expression corpus", "alphabet and case ordering"],
        "not_shared": ["regular-expression normalization", "nullability", "derivative code", "acceptance algorithm"],
        "mismatches": mismatches,
    }
    print(json.dumps(certificate, indent=2, sort_keys=True))
    return 1 if mismatches or not certificate["matches_v2_semantic_transcript"] else 0


if __name__ == "__main__":
    raise SystemExit(main())