Research source · python

verify_synchronized_stack_selection.py

site/public/research-artifacts/synchronized-stack-selection/verify_synchronized_stack_selection.py

186 lines. Source is displayed for inspection; it is not executed by this page.

File fingerprint

SHA-256: 11582c94154f3ba152692d1b23df50f66d3b53b27884d6388e7692ac2d5ec07b

#!/usr/bin/env python3
"""Bounded zipped-stack audit for FPRD-T174."""

from __future__ import annotations

import itertools
import json
from pathlib import Path


CALL = 0
RETURN = 1
LOCAL = 2
ACTIONS = (CALL, RETURN, LOCAL)
TAGS = (0, 1)
WORDS = tuple(
    word
    for length in range(4)
    for word in itertools.product(ACTIONS, repeat=length)
)
SELECTORS = tuple(itertools.product(TAGS, repeat=len(ACTIONS)))
FAMILIES = tuple((selector,) for selector in SELECTORS) + tuple(
    itertools.combinations(SELECTORS, 2)
)


def machines():
    """A complete 4,096-machine deterministic visibly pushdown test family."""
    # Calls choose the next control from (state, hidden tag) and push the tag.
    # Locals use state xor tag. Returns choose the next control from
    # (state, popped tag), with a third top category for the empty stack.
    for call_next in itertools.product(TAGS, repeat=4):
        for return_next in itertools.product(TAGS, repeat=6):
            for accepting_mask in range(4):
                accepting = frozenset(
                    state for state in TAGS if accepting_mask & (1 << state)
                )
                yield call_next, return_next, accepting


def step(machine, state, stack, action, tag):
    call_next, return_next, _ = machine
    stack = list(stack)
    if action == CALL:
        stack.append(tag)
        state = call_next[2 * state + tag]
    elif action == RETURN:
        top_code = stack.pop() + 1 if stack else 0
        state = return_next[3 * state + top_code]
    else:
        state ^= tag
    return state, tuple(stack)


def accepts(machine, word, tags):
    state = 0
    stack = ()
    for action, tag in zip(word, tags):
        state, stack = step(machine, state, stack, action, tag)
    return state in machine[2]


def projected_accepts(machine, word):
    return any(
        accepts(machine, word, tags)
        for tags in itertools.product(TAGS, repeat=len(word))
    )


def selected_accepts(machine, selector, word):
    return accepts(machine, word, tuple(selector[action] for action in word))


def zipped_product_accepts(machine, family, word):
    """Run the one-stack synchronized product from the theorem."""
    selected_tags = tuple(
        tuple(selector[action] for action in word) for selector in family
    )
    for source_tags in itertools.product(TAGS, repeat=len(word)):
        source_state = 0
        selected_states = [0 for _ in family]
        product_stack = []
        expected_height = 0

        for position, action in enumerate(word):
            tags = (source_tags[position],) + tuple(
                tags_for_selector[position]
                for tags_for_selector in selected_tags
            )

            if action == CALL:
                expected_height += 1
                product_stack.append(tags)
                source_state = machine[0][2 * source_state + tags[0]]
                for index in range(len(family)):
                    selected_states[index] = machine[0][
                        2 * selected_states[index] + tags[index + 1]
                    ]
            elif action == RETURN:
                expected_height = max(0, expected_height - 1)
                tops = product_stack.pop() if product_stack else (None,) * len(tags)
                source_top = 0 if tops[0] is None else tops[0] + 1
                source_state = machine[1][3 * source_state + source_top]
                for index in range(len(family)):
                    top = tops[index + 1]
                    top_code = 0 if top is None else top + 1
                    selected_states[index] = machine[1][
                        3 * selected_states[index] + top_code
                    ]
            else:
                source_state ^= tags[0]
                for index in range(len(family)):
                    selected_states[index] ^= tags[index + 1]

            # One product-stack operation serves every component, so all
            # component heights are equal after every observed position.
            assert len(product_stack) == expected_height

        if source_state in machine[2] and all(
            state not in machine[2] for state in selected_states
        ):
            return True
    return False


def main():
    machine_count = 0
    projection_checks = 0
    selector_checks = 0
    product_language_checks = 0
    missed_instances = 0
    synchronized_steps = 0

    for machine in machines():
        machine_count += 1
        projected = {
            word: projected_accepts(machine, word) for word in WORDS
        }
        selected = {
            selector: {
                word: selected_accepts(machine, selector, word)
                for word in WORDS
            }
            for selector in SELECTORS
        }
        projection_checks += len(WORDS)
        selector_checks += len(SELECTORS) * len(WORDS)

        for family in FAMILIES:
            for word in WORDS:
                direct_missed = projected[word] and all(
                    not selected[selector][word] for selector in family
                )
                product_missed = zipped_product_accepts(machine, family, word)
                assert product_missed == direct_missed
                product_language_checks += 1
                synchronized_steps += len(word)
                missed_instances += int(product_missed)

    output = {
        "status": "pass",
        "scope": "bounded action-preserving deterministic visibly pushdown products",
        "deterministic_vpa_family": machine_count,
        "one_state_action_preserving_selectors": len(SELECTORS),
        "selector_families_width_at_most_two": len(FAMILIES),
        "observed_words": len(WORDS),
        "projection_checks": projection_checks,
        "selector_run_checks": selector_checks,
        "zipped_product_language_checks": product_language_checks,
        "synchronized_input_steps": synchronized_steps,
        "missed_word_instances": missed_instances,
        "does_not_verify": [
            "a closed-form horizon bound for visibly pushdown sources",
            "action-erasing projections",
            "nondeterministic source automata",
            "the external undecidability theorem of Kopczynski",
        ],
    }
    Path(__file__).with_name("checker-output.json").write_text(
        json.dumps(output, indent=2) + "\n", encoding="utf-8"
    )
    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()